问题标签 [rxjs7]

我知道有很多关于标题主题的问题，但我无法找到有助于我跟进的清晰简洁的问题/回复。

假设我们有 x 数量的 web api 调用返回 observables Observable<cs[]>... 见

函数restructureABCs()依赖于 as/bs & cs。因为，bs & cs 不相互依赖。当前的实现很丑陋，可以改进，但我不确定应该使用哪个运算符。使用 concatMap 对我来说没有意义，因为（根据我可能有缺陷的理解）多个流将被合并为一个，这是我不想要的。我只需要确保在调用restructureABCs()函数之前加载 as/bs & cs。

Angular13，rxjs7.4

我有在 Angular 13 / RxJs 7 中获得下载进度的工作解决方案。

首先我定义了一些枚举：

接下来，我实现了一个共享服务来处理跟踪进度：

然后，我只需订阅它并获得正确的下载进度：

这一切都很好，我的组件库中有一个进度条，显示了真实的进度。

问题是......我如何获得结果文件？

由于 Angular 12.2.x 分支最近添加了对 rxjs 7.0.0 的支持，这是否意味着严格支持 7.0.x 或者我们甚至可以使用 7.5.x？

我有一个 observable，它将从 SignalR 集线器接收多个实时交易值（可能每秒很多）。我想要实现的是一个连续（每 10 秒）输出最后 10 秒内发生的 5 笔交易的样本的可观察值。

我编写了一个可观察的管道来尝试通过将所有接收到的交易添加到缓冲区中 10 秒来实现这一点，然后使用“concatMap”和“from”为缓冲区数组中的每个交易创建一个可观察的对象。然后，创建另一个缓冲区，收集 5 个值并发出它们。

但是，管道不断发出它在 10 秒窗口中接收到的所有值，但以 5 个为一组。我尝试take(5)在 concat 映射之后在管道中添加一个，它对于第一批 5 个值可以正常工作，但是然后可观察的“完成”并停止侦听新值。我还尝试在 concatMap 之后添加一个带有索引的过滤器，如下所示：

这适用于第一批 5 个值，但随后会不断过滤掉每个值，因此永远不会创建第二个 5 个缓冲区。此外，过滤器的这个用例似乎已被弃用。

我不确定我是否在这里忽略了一些明显的东西，但是我查看了许多 rxjs 运算符并且找不到实现此目的的方法

所以，这是我的简单 GET 请求：

这是我使用结果的地方：

但是，在我当前的 RxJS 7.5 版本中，不推荐使用上述订阅方法......所以我使用将错误处理程序移动到error: () => {}订阅调用内的通知函数的首选语法。但是，我的 formCriteria 对象不再从具有类型 InspectionSummaryFormField[] 的可观察对象中解包，而是使用“任何”类型。

这只是一个测试版问题还是我在这里遗漏了一些重要的东西？

我一直在看视频和阅读文章，但到目前为止，它们要么在我的脑海中，要么过于简单化。（在这里寻找一个好的金发姑娘型实用解释。）

我有一个具有可观察属性的角度服务，就像这样，我要测试的属性是message$：

message$根本不应该发射，直到主题以某种方式获得下一步。

我发现的所有测试（至少对我有意义）都定义了 3 个可观察对象： a source$、expected$和result$，然后比较最后两个，因此：

现在，如果我想测试我的服务的message$observable，我将如何更改上述测试？我会用 替换source$吗const source$ = myService.message$？那是对的吗？

如果我这样做，我可以测试它不会发射（WOOHOO！）。但是我怎样才能测试它，以便在“a”处源不发射，但在“b”之前或在“b”处，我调用myService.someFunction("hello")以便在“b”处，值是“hello”？

谢谢！