Rxjs：如何实现 Observable.first()？

Question

我正在学习 RxJs 并试图了解 Observable.first() 方法是如何工作的。 Source code 很乱。

有没有一种简单的方法来实现这个方法来理解它是如何工作的？

我想确切地了解它在内部是如何工作的，而不是如何实现它。

Answer 1

从我们看到的文档中

只发出第一个值。或者只发出通过某些测试的第一个值。

因此，基本行为（仅发出第一个值）是通过使用订阅者级字段跟踪第一个值是否已发出来完成的。

那么predicate、resultSelector 或defaultValues 只是对该基本代码的一点补充（基本上是filter、map 或default）。

如果你想看一下这个操作符的具体代码，相关部分是FirstSubscriber._next => FirstSubscriber._emit => FirstSubscriber._emitFinal，它使用this._emitted标志来跟踪它是否已经发出。

编辑：如果您想定义一个自定义 Rx 扩展，它可以在不使用其他基本运算符（例如 take(1)）的情况下完成 first() 所做的事情，我会这样做：

Observable.prototype.first = function() {
    const source = this;
    return new Observable(observer => {
        let hasSentValue = false;

        return source.subscribe(
            v => {
                if(!hasSentValue) {
                    observer.next(v);
                    observer.complete();
                    hasSentValue = true;
                }
            },
            err => observer.error(err),
            () => {
                if(!hasSentValue) {
                    observer.error("Empty stream");
                }
            });
    });
};

注意：未测试