在 RxJS 4.0 中，我可以执行以下操作：

现在 RxJS 5.0 放弃了对timestamp.

所以这是我试图弄清楚如何使用 rxjs 实现的场景：

1. 从文件/数据库/等加载一些元数据。元数据中的每个元素都有一个 id 和附加信息——比如实际数据的位置。目前，我在应用程序启动时异步加载所有这些元数据。加载此数据后，Observable 调用完成。最终我可能会添加刷新功能

2. 在应用程序的稍后时间点，我将需要根据元数据中可用的内容加载特定的数据集。我目前正在尝试使用fetchData(ids:string[]):Observable 之类的函数来执行此操作。这是我不清楚如何在 rxjs 范式下进行的地方。我同样不确定如何使用fetchDatum(id:string):Observable 之类的函数请求单个项目

我当然可以使用过滤器仅对从 IMetadata Observable 发出的与列表中的名称之一匹配的 IMetadata 项目进行操作 - 但我还需要确认在 IMetadata Observable 排放中找到所有请求的项目，如果不是我需要出错。

因此，如果有人请求 id = "Bob" 的 IMetadata - 但源 Observable 没有发出这样的 IMetadata，那么它需要出错。或者如果他们请求 { "Shirley", "Rex", "Samantha" } 并且没有 "Rex" 的数据，那么它应该出错。

我考虑过在这里使用 Rx.Subject ，但从我读到的内容来看，这在 rxjs 范式下通常是不可取的。请告知在 rxjs 范式下哪些方法适用于这种情况。谢谢！

尝试使用 RxJS v5 构建计划，其中某些事件可以触发计划重新加载。当前使用 3 个源 - schedule$、event$ 和 userNotification$（示例如下）。

我已经尝试了许多不同的策略，并且当 reloadSchedule 事件时间命中时，我一直很奇怪，比如递归重新加载。有没有办法让下游数据 (event$) 干净地触发上游 (schedule$) 重新加载，而不会有任何动作/通知从以前的计划项目中挥之不去？

我有一个函数，它以 ID 作为参数并根据 ID 进行 HTTP 调用。它返回一个 Observable。我希望具有不同参数的调用是完全独立的，但它们会相互影响。

我将
RxJS 5.0.0-
beta.2 Angular 2.0.0-beta.7 用于 http 调用
Typescript 1.8.2

代码：

预期行为：

我打电话给
getCharacterDetails(1).subscribe((details) => { this.details = details }));
第一个 console.log 打印我的 EMPTY_MODEL。
第二个 console.log 打印 ID 1 的字符模型。

然后我调用
getCharacterDetails(2).subscribe((details) => { this.details = details });
第一个 console.log 打印我的 EMPTY_MODEL。
第二个 console.log 打印 ID 2 的字符模型。

实际行为：

我打电话给
getCharacterDetails(1).subscribe((details) => { this.details = details }));
第一个 console.log 打印我的 EMPTY_MODEL。
第二个 console.log 打印 ID 1 的字符模型。

然后我调用 getCharacterDetails(2).subscribe((details) => { this.details = details });
第一个 console.log 打印 ID 1 的字符模型。 <- 问题
第二个 console.log 打印 ID 2 的字符模型。

为什么这两个电话不完全独立？第二个电话怎么知道第一个电话的数据？

我想采用 Observable<T[]> 并将其转换为 Observable<T> 以便 Observable<T[]> 中的每个数组都被分解，然后数组的各个元素分别通过可观察的<T>。

是否有这样做的标准操作员？我四处寻找，但没有找到任何东西。谢谢。

在被指向 concatMap/flatMap 的方向后，我想出了以下通用解决方案：

在 Rxjs5 中找不到 Rx.Observable.pairs，我只需将对象转换为 Observable 并检查每个属性的更改。有任何想法吗？

I have a scenario where I want to fetch some piece of data from a server, and where the user can request a refresh of this data. The rest of the page needs to update to reflect the currently loaded iteration of the data.

I am picturing this where I have a hot Observable that publishes the data that it loads. I don't want to retain all the old iterations of the data as 1. I only care about the latest iteration of the data and 2. it could lead to an out-of-memory exception if the user refreshes enough in a given session.

However, I DO want to retain the last published value so that if I dynamically bring up a new component that needs access to that same data, it isn't sending out a new request unnecessarily. For this I need an Observable to sit ontop of the hot observable which will only retain and emit the last emission from the hot observable. Here's a diagram illustrating this idea:

echoStream is subscribed to the dataStream. The subscription1, subscription2, and subscription3 are all subscribed to echoStream, but they subscribe at different points. At the time of subscription they get the last value that was emitted from the dataStream, and receive subsequent updates from dataStream.

echoStream is a bit of a mix of a Hot and Cold Observable, having a limited history retention.

Does rxjs provide a standard operator for setting up something like echoStream in the above example?

I want to run this example with rxjs5. But it doesn't work. I've stucked on #41 line. It says that map returns Subject and it doesn't have .takeUntil method. What is the best way to implement it? Thanks

I have been experimenting with RxJS for two weeks now, and although I love it in principle I just cannot seem to find and implement the correct pattern for managing state. All articles and questions appear to agree:

• Subject should be avoided where possible in favor of just pushing state through via transformations;
• .getValue() should be deprecated entirely; and
• .do should perhaps be avoided except for DOM manipulation?

The problem with all such suggestions is that none of the literature appears to directly say what you should be using instead, besides "you'll learn the Rx way and stop using Subject".

But I cannot find a direct example anywhere that specifically indicates the correct way to perform both additions and removals to a single stream/object, as the consequence of multiple other stream inputs, in a stateless and functional manner.

Before I get pointed in the same directions again, problems with uncovered literature are:

• The Introduction to Reactive Programming You've been missing: great starting text, but does not specifically address these questions.
• The TODO example for RxJS comes with React and involves explicit manipulation of Subjects as proxies for React Stores.
• http://blog.edanschwartz.com/2015/09/18/dead-simple-rxjs-todo-list/ : explicitly uses a state object for addition and removal of items.

My perhaps 10th rewrite of the standard TODO follows - My prior iterations covered include:

• starting with a mutable 'items' array - bad as state is explicit and imperatively managed
• using scan to concatenate new items to an addedItems$ stream, then branching another stream where the removed items were deleted - bad as the addedItems$ stream would grow indefinitely.
• discovering BehaviorSubjectand using that - seemed bad since for each new updatedList$.next() emission, it requires the previous value to iterate, meaning that Subject.getValue() is essential.
• trying to stream the result of the inputEnter$ addition events into filtered removal events - but then every new stream creates a new list, and then feeding that into the toggleItem$ and toggleAll$ streams means that each new stream is dependent on the previous, and so causing one of the 4 actions (add, remove, toggle item or toggle all) requires the whole chain to be unnecessarily run through again.

Now I have come full circle, where I am back to using both Subject (and just how is it supposed to be successively iterated upon in any way without using getValue()?) and do, as show below. Myself and my colleague agree this is the clearest way, yet it of course seems the least reactive and most imperative. Any clear suggestions on the correct way for this would be much appreciated!

Edit

In relation to @user3743222 very helpful answer, I can see how representing state as an additional input can make a function pure and thus scan is the best way to represent a collection evolving over time, with a snapshot of its previous state up to that point as an additional function parameter.

However, this was already how I approached my second attempt, with addedItems$ being a scanned stream of inputs:

The obvious solution would be to just have items = [], and manipulate it directly, or const items = new BehaviorSubject([]) - but then the only way to iterate on it appears to be using getValue to expose the previous state, which Andre Stalz (CycleJS) has commented on in the RxJS issues as something that shouldn't really be exposed (but again, if not, then how is it usable?).

I guess I just had an idea that with streams, you weren't supposed to use Subjects or represent anything via a state 'meatball', and in the first answer I'm not sure how this doesn't introduce mass chained streams which are orphaned/grow infinitely/have to build on each other in exact sequence.

我正在寻找一些关于如何识别哪个流进入一个merge或combineLatest函数的最佳实践建议，以便只对新流进行操作。

在 TODO 应用程序的上下文中，我有传入的添加和删除流，我想将它们组合起来，以便我的列表编辑可以在单个流中以无状态的方式进行。输出是一个集成了 add(concat) 和 remove(filter) 事件的列表，否则您似乎最终会得到包含所有添加或删除事件的其他流，因此会无限增长。

遇到的问题包括：

• usingmerge不指示传入的是哪个流；
• usingcombineLatest不指示哪个流触发了订阅的流，因此您不能只执行与该流相关的操作。
• 使用withLatestFrom导致不更新withLatestFrom输入源的新列表，因此除非下一个对此列表感兴趣的流订阅该列表，否则该列表将不同步（或者您的每个新列表都基于它之前的列表，这会导致不必要地重新执行之前发生的所有转换……）。

目前发现的似乎次优的方法包括：

• Cycle JS TODO 应用程序将type属性直接分配给早期流的创建对象以进行识别，我认为应该避免使用直接识别流来自何处的 RxJS 方法？
• 接受来自http://www.jisaacks.com/manipulating-rxjs-streams/的建议，并将添加流和删除流的输出拆分为[null, {addItem}]and [removeItem, {null}]，这样当merge同时用于添加和删除事件时，我仍然可以识别传入的正在更新的流，以便我可以在单个流中执行添加和删除（但随后我想添加切换事件等，但这似乎也不正确，因为我需要创建流输出了解所有其他潜在的流（以 [null, null, myOutput, null, null 等结束）。

非常欢迎任何最佳实践建议。