将 RxJS Observable 收集到数组

问题描述

我想使用 RxJS 来桥接"异步事件世界与同步世界.具体来说，我想创建一个函数，该函数返回在某个时间间隔内收集的事件数组.

I'd like to use RxJS to "bridge" async world of events with sync world.Specifically I want to create an function which returns an array of events collected during some time interval.

我可以创建 Observable 来做我想要的

I can create Observable which does what I want

    var source = Rx.Observable
    .interval(100 /* ms */)
    .bufferWithTime(1000).take(1)

我可以很好地打印正确的值

I can print correct values just fine

    var subscription = source.subscribe(
        function (x) {
            console.log('Next: ' + JSON.stringify(x));
        },
        function () {
            console.log('Completed');
        });

这个打印

    [0,1,2,3,4,5,6,7,8]
    Completed

但我想要的是将此数组分配给变量.从概念上讲，我想要类似的东西

But want I want is to assign this array to variable. Conceptually I want something like

varcollectedDuringSecond = source.toPromise.getValue()

这个想法是 getValue 会阻塞，所以在上面的行完成后 collectDuringSecond 将包含 [0,1,2,3,4,5,6,7,8]

The idea is that getValue would block so after the line above is done collectedDuringSecond will contain [0,1,2,3,4,5,6,7,8]

推荐答案

JavaScript 中的同步事件编程具有高度限制性.事实上，在很多情况下这可能是不可能的.我尝试使用 Rx 进行黑客攻击，看看是否可以在不修改 Rx 源的情况下提供同步接口，并且(有充分的理由)直接使用 JavaScript 是不可能的.

Synchronous event programming in JavaScript is highly restrictive. In fact, it may be impossible in a lot of cases. I tried hacking around with Rx to see if I could provide a synchronous interface without modifying the Rx source, and (for good reason) it's not possible with straight JavaScript.

我建议将 Observable 作为 API 的一部分公开，并允许消费者从那里处理它(当然要推动使用 Rx；).

I would suggest exposing the Observable as part of your API, and allowing consumers to handle it from there (with a nudge to use Rx, of course ;).

function MyClass () {

    this.getArrayOfStuffAsObservable = function () {
        return Rx.Observable.interval(100)
            .bufferWithTime(1000).take(1);
    };

    // this is optional and I don't recommend it, since you already have Rx available.
    // additionally, consumers will probably miss the fact that you can dispose
    // of the subscription.
    this.getArrayOfStuff = function (callback) {
        var value;
        return this.getArrayOfStuffAsObservable()
            .subscribe(
                function (x) {
                    value = x;
                },
                function (err) {
                    callback(err);
                },
                function () {
                    if (hasValue) {
                        callback(undefined, value);
                    } else {
                        callback('did not receive value');
                    }
                });

    };
};

作为附加说明，您可能希望将 toArray 与 take 结合使用，而不是 bufferWithTime 用于此特定em> 示例(这实际上是做同一件事的两种方式，但一种基于时间，另一种基于项目数).toArray 创建一个 Observable，它将收集底层 observable 的所有值，并在底层 Observable 完成时将这些值作为数组产生.

As an additional note, you may want to use toArray in conjunction with take instead of bufferWithTime for this specific example (it's really two ways of doing the same thing, but one is based on time and the other based on item count). toArray creates an Observable which will collect all of the values of the underlying observable, and yield those values as an array when the underlying Observable completes.

this.getArrayOfStuffAsObservable = function () {
    return Rx.Observable.interval(100)
        .take(10)
        .toArray();
};

这篇关于将 RxJS Observable 收集到数组的文章就介绍到这了，希望我们推荐的答案对大家有所帮助，也希望大家多多支持！