将数据从一个observable传递到另一个observable的更好方法(A better way to pass data from one observable to another)

我已经创建了两个用于测试Observables函数，每个函数都返回一个Observable ：

foo() { return new Observable(observer => { let i = 0; setInterval( () => { if(i === 10) { observer.complete(); } else { observer.next(i); i++; } }, 1000); // If I call observer.complete() here then it never completes }); } bar(fooResult) { return new Observable(observer => { let j = 0; setInterval( () => { if(fooResult) { observer.next('j -> '+j+' plus fooResult '+JSON.stringify(fooResult)); observer.complete(); } else { observer.next(j); j++; } }, 2000); }); }

并像这样使用它们：

提出的问题是：

有没有更好的方法将数据从完成Observable传递到另一个也返回Observable的函数？构建一个数组似乎很麻烦，我不能做以下事情，因为Observable的complete callback部分没有传递像progressUpdate和onError这样的参数：

(complete) => { this.exampleProductService.bar(complete).// rest of code }

我尝试将第一个函数的结果赋给变量，然后传递该变量，但是，正如预期的那样，我得到了一个Observable而不是我想要的结果。

关于我如何做以上事情有什么不正确的吗？

谢谢

PS这是一个Angular 2应用程序！

I've created two functions for testing Observables that each return an Observable:

foo() { return new Observable(observer => { let i = 0; setInterval( () => { if(i === 10) { observer.complete(); } else { observer.next(i); i++; } }, 1000); // If I call observer.complete() here then it never completes }); } bar(fooResult) { return new Observable(observer => { let j = 0; setInterval( () => { if(fooResult) { observer.next('j -> '+j+' plus fooResult '+JSON.stringify(fooResult)); observer.complete(); } else { observer.next(j); j++; } }, 2000); }); }

And make use of them like this:

Make question(s) are:

Is there a better way to pass data from the completion of an Observable to another function that also returns an Observable? Building up an array seems cumbersome and I can't do the following as the complete callback part of the Observable doesn't pass a parameter like progressUpdate and onError:

(complete) => { this.exampleProductService.bar(complete).// rest of code }

I tried assigning the result of the first function to a variable and then passing that variable but, as expected, I got an Observable and not the result I wanted.

Is there anything incorrect about how I'm doing the above?

Thanks

P.S. This is an Angular 2 application!

最满意答案

我认为你的功能有点过于复杂。首先，当工厂函数已经可用时，不要使用构造函数，在这种情况下@Meir指出的interval或timer ，尽管在这种情况下它会更详细。

其次， bar函数实际上没有多大意义，因为你似乎正在等待你已经知道已经完成的事情的完成（因为你没有订阅它直到Observable的完成块由foo ）。

我根据你声明的目标重构，在开始第二个之前等待一个Observable完成，同时使用第二个中第一个的结果。

// Factory function to emit 10 items, 1 every second function foo() { return Observable.interval(1000) .take(10); } // Lifts the passed in value into an Observable and stringfys it function bar(fooResult) { return Rx.Observable.of(fooResult) .map(res => JSON.stringify(fooResult)) }

现在使用它们时你会这样做：

foo() // Log the side effects of foo .do( x => console.log(`foo next() -> ${x}`), err => console.error(`foo error: ${JSON.stringify(err)}`), () => console.log('foo finished') ) // Collect the results from foo and only emit when foo completes .reduce((total, diff) => [...total, diff], []) // Pass the result from the reduce on to bar .concatMap(fooResult => bar(fooResult)) //Subscribe to the results of bar .subscribe( res => console.log(`bar next() -> ${res}`), err => console.error(`bar error: ${JSON.stringify(err)}`), () => console.log('bar finished') );

注意上面我也摆脱了全局状态，这是函数式编程的诅咒。在任何可能的情况下，您的州应该本地化到流。

请参阅此处的工作示例： http ： //jsbin.com/pexayohoho/1/edit？js，console

I think your functions are a little overly complex. For one, don't use the constructor when there are factory functions already available, in this case interval or timer as @Meir pointed out, though in this case it would be more verbose.

Second, the bar function doesn't really make much sense as it stands, since you seem to be waiting for something to complete which you already know has completed (since you don't subscribe to it until the completion block of the Observable generated by foo).

I refactored according to your stated goal of waiting for one Observable to complete before starting a second one, while using the results of the first in the second.

// Factory function to emit 10 items, 1 every second function foo() { return Observable.interval(1000) .take(10); } // Lifts the passed in value into an Observable and stringfys it function bar(fooResult) { return Rx.Observable.of(fooResult) .map(res => JSON.stringify(fooResult)) }

Now when using them you would do instead:

foo() // Log the side effects of foo .do( x => console.log(`foo next() -> ${x}`), err => console.error(`foo error: ${JSON.stringify(err)}`), () => console.log('foo finished') ) // Collect the results from foo and only emit when foo completes .reduce((total, diff) => [...total, diff], []) // Pass the result from the reduce on to bar .concatMap(fooResult => bar(fooResult)) //Subscribe to the results of bar .subscribe( res => console.log(`bar next() -> ${res}`), err => console.error(`bar error: ${JSON.stringify(err)}`), () => console.log('bar finished') );

Notice above I am also getting rid of global state, which is the anathema of functional programming. Where ever possible your state should be localized to the stream.

See the working example here: http://jsbin.com/pexayohoho/1/edit?js,console

更多推荐