javascript - 在 RxJS Observable 的 onNext 中等待异步操作

Question

我有一个以正常方式使用的 RxJS 序列...

但是，在可观察的“onNext”处理程序中，某些操作将同步完成，但其他操作需要异步回调，需要在处理输入序列中的下一项之前等待。

...有点困惑如何做到这一点。有任何想法吗？谢谢!

someObservable.subscribe(
    function onNext(item)
    {
        if (item == 'do-something-async-and-wait-for-completion')
        {
            setTimeout(
                function()
                {
                    console.log('okay, we can continue');
                }
                , 5000
            );
        }
        else
        {
            // do something synchronously and keep on going immediately
            console.log('ready to go!!!');
        }
    },
    function onError(error)
    {
        console.log('error');
    },
    function onComplete()
    {
        console.log('complete');
    }
);

Answer 1

您想要执行的每个操作都可以建模为可观察的。甚至同步操作也可以用这种方式建模。然后，您可以使用 map 将序列转换为序列序列，然后使用 concatAll 压平序列。

someObservable
    .map(function (item) {
        if (item === "do-something-async") {
            // create an Observable that will do the async action when it is subscribed
            // return Rx.Observable.timer(5000);

            // or maybe an ajax call?  Use `defer` so that the call does not
            // start until concatAll() actually subscribes.
            return Rx.Observable.defer(function () { return Rx.Observable.ajaxAsObservable(...); });
        }
        else {
            // do something synchronous but model it as an async operation (using Observable.return)
            // Use defer so that the sync operation is not carried out until
            // concatAll() reaches this item.
            return Rx.Observable.defer(function () {
                return Rx.Observable.return(someSyncAction(item));
            });
        }
    })
    .concatAll() // consume each inner observable in sequence
    .subscribe(function (result) {
    }, function (error) {
        console.log("error", error);
    }, function () {
        console.log("complete");
    });

为了回复您的一些评论...在某些时候您需要对函数流施加一些期望。在大多数语言中，当处理可能异步的函数时，函数签名是异步的，并且函数的实际异步与同步性质被隐藏为函数的实现细节。无论您使用的是 javaScript Promise、Rx observables、c# Tasks、c++ Futures 等，都是如此。函数最终返回 Promise/observable/task/future/etc，如果函数实际上是同步的，那么它返回的对象是刚刚完成。

话虽如此，由于这是 JavaScript，您可以作弊:

var makeObservable = function (func) {
    return Rx.Observable.defer(function () {
        // execute the function and then examine the returned value.
        // if the returned value is *not* an Rx.Observable, then
        // wrap it using Observable.return
        var result = func();
        return result instanceof Rx.Observable ? result: Rx.Observable.return(result);
    });
}

someObservable
    .map(makeObservable)
    .concatAll()
    .subscribe(function (result) {
    }, function (error) {
        console.log("error", error);
    }, function () {
        console.log("complete");
    });