反应式扩展：如何创建可观察的占位符？

Question

我有一个方法，getObs()，它返回一个 observable，应该由所有调用者共享。但是，当有人调用 getObs() 时，该 observable 可能不存在，并且创建它是一个异步操作，所以我的想法是返回一个占位符 observable，一旦创建，它就会被真正的 observable 替换。

我的基本尝试是这样的：

var createSubject = new Rx.Subject();
var placeholder = createSubject.switchLatest();

如果调用 'getObs()' 时真正的 observable 不存在，我可以在哪里返回 placeholder。创建真正的 observable 后，我使用 createSubject.onNext(realObservable)，然后将其传递给 switchLatest()，为任何订阅者解包。

但是，为此目的使用 Subject 和 switchLatest 似乎有点矫枉过正，所以我想知道是否有更直接的解决方案？

Answer 1

如果获取 observable 本身的行为是异步的，那么您也应该将其建模为 observable。

例如...

var getObsAsync = function () {
    return Rx.Observable.create(function (observer) {
        var token = startSomeAsyncAction(function (result) {
                // the async action has completed!
                var obs = Rx.Observable.fromArray(result.dataArray);
                token = undefined;
                observer.OnNext(obs);
                observer.OnCompleted();
            }),
            unsubscribeAction = function () {
                if (asyncAction) {
                    stopSomeAsyncAction(token);
                }
            };            

        return unsubscribeAction;
    });
};

var getObs = function () { return getObsAsync().switchLatest(); };

如果您想共享该 observable 的单个实例，但不希望获得 observable直到有人实际订阅，那么您可以这样做：

// source must be a Connectable Observable (ie the result of Publish or Replay)
// will connect the observable the first time an observer subscribes
// If an action is supplied, then it will call the action with a disposable
// that can be used to disconnect the observable.
// idea taken from Rxx project
Rx.Observable.prototype.prime = function (action) {
    var source = this;
    if (!(source instanceof Rx.Observable) || !source.connect) {
        throw new Error("source must be a connectable observable");
    }

    var connection = undefined;
    return Rx.Observable.createWithDisposable(function (observer) {
        var subscription = source.subscribe(observer);

        if (!connection) {
            // this is the first observer.  Connect the underlying observable.
            connection = source.connect();
            if (action) {
                // Call action with a disposable that will disconnect and reset our state
                var disconnect = function() {
                    connection.dispose();
                    connection = undefined;
                };
                action(Rx.Disposable.create(disconnect));
            }
        }

        return subscription;
    });
};

var globalObs = Rx.Observable.defer(getObs).publish().prime();

现在在任何地方都可以使用 globalObs 而不必担心它：

// location 1
globalObs.subscribe(...);

// location 2
globalObs.select(...)...subscribe(...);

请注意，实际上没有人需要调用 getObs，因为您只需设置一个全局可观察对象，当有人订阅时，它将（通过 defer）为您调用 getObs。

Answer 2

您可以在事后使用主题来连接来源：

var placeholder = new Subject<YourType>();
// other code can now subscribe to placeholder, best expose it as IObservable

创建源时：

var asyncCreatedObs = new ...;
placeholder.Subscribe(asyncCreatedObs);
// subscribers of placeholder start to see asyncCreatedObs