问题标签 [rxjs]

我正在 TDDing RxJS 解决方案并使用bufferWithTime为什么示例代码中的res得到值[]（空数组）？这是我的代码或 RxJS 库中的问题吗？在带有 rx 版本 2.2.27 的 node.js v0.10.30 上运行它。

以下可以在 nodejs coffeescript REPL 中运行

所以我知道 Promises 和函数响应式编程。

我想要做的是调用一个异步函数并 .then(callback) 做一些事情，但我不想立即定义回调。它让我想起了 RxJS Observable，只是没有所有花哨的功能..

也许有人知道我在找什么。

下面的代码片段是功能性的（从某种意义上说它正在工作;-)），但充其量似乎很蹩脚......

任何人都可以提出一种方法来使它更可组合或至少不那么难看吗？

代码基于此页面上的示例： Wrap an Existing API with RxJS

最近我开始研究RxJS和 RxJava（来自 Netflix）库，它们致力于响应式编程的概念。

Node.js 在事件循环的基础上工作，它为您提供了异步编程的所有工具，而后续的节点库（如“集群”）可帮助您充分利用多核机器。Node.js 还为您提供了 EventEmitter 功能，您可以在其中订阅事件并对其进行异步操作。

另一方面，如果我理解正确，RxJS（以及一般的反应式编程）的工作原理是事件流，订阅事件流，异步转换事件流数据。

所以，问题是在 Node.js 中使用 Rx 包意味着什么。Node 的事件循环、事件发射器和对 Rx 的流和订阅的订阅有多么不同。

以下是我用作示例的 HTML 片段：

JS 控制台中的输出为：

现在，“stuff”进行了一些处理（基本上将函数应用于当前值），这是我从 ReactJs + RxJS 的 TodoMVC 示例（https://github.com/fdecampredon/react-rxjs-todomvc）中获得的想法.

我试图实现的输出是让第二个订阅者在订阅的那一刻也看到“4”。我将 RxJS 与 ReactJS 结合使用，因此组件在卸载时取消订阅（由于路由更改），并在再次安装时再次订阅。

我已经删除了样板以达到重点

// a.js

// b.js（需要上面导出的模块）

我期望看到的：

{item: 'a', a: 'a'}打印在控制台中

发生了什么：

Hit!之前打印了 24 次{item: 'a', a: 'a'}。这意味着zip从 中获取所有值aStream，缓冲它们，然后做它应该做的事情。

我如何获得相同的功能zip提供但懒惰？我的目标是使用无限流/可观察的并用有限（异步）流压缩它。

编辑

通过 runnable 查看/编辑它：RX Zip test Edit 2 Code updated based on answer -> no output now。

在我的系统中，我有一个源，两个“步骤”将源映射到一个新值，然后是一个将这两个步骤组合起来以创建最终值的总和。这个系统的初始运行按我希望的那样工作，产生一个 3 的总和。

输出：

但是，如果我为 source 输入一个新值，我会得到如下两个结果：

第一个是中间结果……当新的源值通过系统的一部分时，当所有值都完成传播时，我得到最终结果。

所以我的问题是，推荐的配置方法是什么，以便推入源的新值将原子地通过系统并且只生成一个总和结果？

谢谢！

我有一些代码片段如下

videoEntries.json 是一个 JSON 序列化的 videoEntry 对象数组。我希望订阅 A 和订阅 B 都能接收由 videosNeedFix observable 发出的数据。

但是，根据控制台日志，只有订阅 A 会收到数据，而订阅 B 不会。如果我交换两个订阅的顺序，只有订阅 B 会看到数据。为什么 observable 只向第一个订阅发送数据？

我想使用 angular-rx 作为结果的简单刷新按钮。如果用户单击刷新按钮，则重新加载结果。如果用户在 1 秒内单击刷新按钮 100 次，则仅加载最新结果。如果结果由于某种原因失败，这并不意味着刷新按钮应该停止工作。

为了实现最后一点，即使失败，我也想保留订阅（或重新订阅），但我不知道该怎么做？

这不起作用，但这是一个简单的示例，我尝试在错误时重新订阅：

我认为这很常见，应该有一些标准做法来实现这一点？

更新

使用建议的解决方案，这是我所做的测试：

这些是我的测试结果：

1. 单击了刷新按钮
2. obsSubject.onNext() 调用
3. map 函数返回 1。
4. subscribe onNext 被触发
5. 单击了刷新按钮
6. obsSubject.onNext() 调用
7. 映射函数抛出错误
8. 进入catch函数
9. subscribe onNext 被触发
10. 单击了刷新按钮
11. obsSubject.onNext() 调用
12. 没有什么。我需要继续订阅

我的理解是 catch 应该保留订阅，但我的测试表明它没有。为什么？

我正在查看 RxJS 文档以获取类似 bacon's 的flatMapWithConcurrencyLimit内容，但什么也没看到。知道它是否在 Rx 中可用吗？