问题标签 [rxjs-pipeable-operators]

我正在尝试测试通过共享/热可观察对象发送的消息的延迟。我注意到，当我在单个共享 observable 上有多个观察者时，单个观察者会从单个消息中调用n次（其中n是共享 observable 上的观察者数量）。

我用10 个 Observers和1 Message Per Observer运行下面的代码，每个观察者每条消息被调用 10 次（意味着总共 100 个 observer.next() 调用）。根据我对观察者/可观察者的理解，每个观察者每条消息应该只被调用一次。我只是在share()这里错误地使用了运算符吗？还是我对它的理解总体上存在缺陷？

运行完成后，如果“发送的消息总数”为x，则“接收的消息总数”将为x^2

我对 RxJS 很陌生，我想知道我是否做得对……在下面的 ngOnInit() 函数中，我得到一个客户端对象，然后通过管道传输它……

有没有更好的方法来执行下面的重复 switchMap/map 操作？

我的代码有效...但我想知道是否应该采用更优雅的方法...

我正在使用 Redux Observable 并且需要在从史诗中触发动作时解决时间问题。

我有一组要循环的项目，以便对每个项目进行 AJAX 调用。收到 AJAX 响应后，我想立即执行一些操作。在原始数组中的每个项目的所有 AJAX 响应都返回后，我想触发更多操作。

timer即使原始数组尚未完成循环，如何在过期后立即触发这些操作？

在 rxjs 可观察链中，我如何在经过一定时间后访问可观察的当前值？本质上，我正在寻找类似tap 操作符的东西，但只有在经过一定时间而没有从 observable 中看到值的情况下才会执行。所以实际上它就像是点击和超时的组合。

我在想像下面这样的东西

这是一个虚构的例子，“tapAfterTime”函数不是真实的。但基本思想是，如果订阅后 2000 毫秒过去了，并且 observable 没有看到大于 5 的值，那么无论 observable 的当前值是什么，都执行 tapAfterTime 回调函数。如果我们在 2000 毫秒之前看到大于 5 的值，那么 tapAfterTime 回调将永远不会运行，但 map 函数将始终按预期运行。

是否有运营商来实现这一点或运营商的任何组合？

我有一个带有分页的帖子列表，我订阅了查询参数?page=以调用服务 API

现在，我有一个新要求，用户还可以通过使用Reactive Form选择 Category 下拉列表来过滤列表。

使用 Reactive Form，我可以在此处使用此代码订阅观察者

我的问题是，我怎样才能使用 Angular 和 RXJS 来简化它？

上面示例中的方法并不是真正的 DRY，因为它们调用的是相同的 API，只是请求参数不同

谢谢。

我想用一个 observable 执行一些操作，比如将 20 添加到其属性之一

然后我需要检查其他 observable 的条件，如果它是真的，然后执行进一步的操作，比如将 40 添加到第一个 observable 的属性之一

组合代码：

在这里，如果 otherObservable 的条件为真，那么我需要将 40 添加到 count 中，最后在订阅中我应该得到第一个 observable 的实际对象。

有人可以帮我解释一下吗？应引入注释代码以实现要求，以便

示例 observable 应该从第一个 observable 发出对象

StackBlitz

我调用了一个返回 observable 的 HTTP 服务（它是第三方库的一部分，因此我无法更改其内部代码），并且它在订阅我想处理的用例时抛出错误小路。

我有这样的事情：

我的服务等级：

我的消费者：

因此，对于当前的代码示例，我要做的是，对于状态码为 409 的情况，使订阅成功。

我使用 Angular 6，我想过滤异步管道的结果，然后在 UI 中呈现它们。

这是我现在的代码

和模板

从管道中，我得到了一些 ID 和名称。我已经有一个 id 数组。我想过滤管道的 id，而不是渲染已经在数组中的那些。

所以，这就是我尝试做的。

新版本的管道过滤器

假设那checkIfResultIdInArray是我创建的函数。过滤并返回所有不在array. 所以最终出现在模板中的 ids/names 不是{id:1,name:'one'},{id:2,name:'two'}.

或者也许我可以以某种方式过滤模板？

这与我遇到的错误无关，而是语法问题。

工作流程很简单：

• 发出返回布尔值的 HTTP 请求
• 如果布尔值为真，则继续
• 如果布尔值为 false，则记录警告并停止流。

为了管理它，我当前的代码是这样的：

样板

当前代码

我不知道为什么，但它对我来说并不自然和优化。

我认为会有一些东西可以简化语法，就像这样

有没有类似的东西，或者我有正确的语法？

我有一个 RxJS Observable，它对底层数据结构发出一系列更改——特别是来自 AngularFirestoreCollection 的 snapshotChanges()。

• 我目前正在将其映射到一组纯 JavaScript 对象以供我的应用程序使用。
• 此数组不受任何保护，使用代码可能会意外修改此结构。
• 每当底层数据源发出时，都会重新构建整个数组，即使数组中只有一个（或有时没有）项目实际发生了变化。
• 正因为如此，所有引用每次都会更改，使得更改检测比需要的更难 - 并且真的减慢了我的应用程序。

相反，我想做的是使用Immer来维护一个不可变的结构，以便在结构上与“新”数组共享未更改的数据。

我无法解决的是如何pipe()关闭可观察对象，以便管道除了最新输出之外还snapshotChanges()可以访问先前发出的不可变数据（或首次默认值） 。snapshotChanges()

在代码中，我基本上已经拥有的是：

我本质上是在寻找这样的东西，我不知道XXX(...)应该是什么：

我觉得运算符接近expand我需要的，但它似乎只在后续运行中传递先前发出的值，而我也需要新发出的snapshotChanges.

给定一个 RxJS Observable 管道，我怎样才能操作这个 Observable 的排放物，同时还能访问管道之前的排放物？