问题标签 [rx.net]

我想使用一个可以随时填充的 IObservable。

我有这个扩展方法：

我使用如下：

推送新项目后：

IObservable 被消耗。

我的问题是，当我添加很多项目时，我不想使用已添加的第一个项目，而是最后添加的项目（如堆栈，而不是队列）。

我怎样才能做到这一点？BehaviorSubject 是否适合堆栈消耗行为？

我想使用 Rx Buffer 功能：

这意味着自上次发出后缓冲区达到 5 或 30 秒的大小时会发生发出（发布给订阅者）。

但我需要能够按需发出 - 例如当我收到高优先级序列项目时。然后我想将它添加到 observable ( source.OnNext()) 并以某种方式强制它发出（这意味着返回缓冲区中的所有元素并清除它）。

我知道我可以添加以下代码：

并调用 flusher.OnNext(highPriorityItem) 我会发出它。

但在这种情况下，我有两个具有两个不同发射的独立序列。当缓冲区已满或特定项目按顺序出现时，我需要一个发射。

Force flush count-type Observable.Buffer c#和Force flush to Observable.Buffer c#似乎不适合我

我一直很高兴通过以下方式使用 RX 在 WPF 应用程序中进行一些 API 调用：

但是由于各种原因，我认为我可能需要使用 SelectMany 运算符进行调用并对流进行一些处理。

我希望在 api 调用中可能会有一些错误。例如，API 端点可能不可用。API 调用之前的某些解析失败。等等。我希望 Hot Observable 继续。我还需要显示一个标准的 IsLoading 微调器。

现在我也明白了，一旦收到 OnError 序列就不应该继续。我明白这一点......我只是不喜欢它。

有了这个，问题是：使用 Retry() 是实现热可观察的正确方法吗？不管错误如何，它都可以继续运行？

下面重写的代码有效，但感觉很糟糕：

我看过一些代码示例，人们使用嵌套流重新订阅故障流。从我读过的内容来看，这似乎是一种常见的方法，但对我来说，这似乎将原本应该是简单的场景变成了难以理解的情况。

System.Web.Http 端点可以是 IObservable 吗？可以在仍然使用 Route 和 AcceptVerbs 等属性的同时完成吗？

基本上，我想将以下代码示例转换为 IObservable 流，因此我通过响应式扩展订阅和处理它：

编辑：我正在寻找更优雅的解决方案，例如https://blogs.msdn.microsoft.com/jeffva/2010/09/15/rx-on-the-server-part-5-of-n-observable- asp-net（可从 IHttpHandler 观察到）但希望在七年后更成熟？

所有消息都应该发布到消息总线：

我想进行单元测试以验证发布方法是否被正确调用：

但无论我尝试什么，测试都不会停止。

更新：

这是完整的生产代码：

目前，我只是使用普通的旧委托/事件观察者模式来处理按钮上的点击事件。所以，我想我会去 Rx。

我的第一次尝试是球。我想我会选择你在 WinForms 示例中看到的最简单的重载：

但我有一个例外，Android 按钮没有点击事件。这是史诗般的反思失败。我猜也许我错过了一个 dll（太好了，另一个 dll 让我的 apk 更大）。

无论如何，我想我会以不同的重载继续前进。现在看起来像这样：

还是失败了。没有例外。但是当我点击按钮时，只有蟋蟀和风滚草。我是要订阅那个吗？那会是什么样子，因为订阅似乎已经在超载中发生了？

Xamarin 足够棘手，但 Rx 只是让它变得非常困难。尤其是在没有很多人写博客的情况下（至少我的谷歌搜索没有什么太有用的东西。我的代码基于我找到的一些零碎的东西。

谢谢

给定 2 个热可观察对象 t1 和 t2，我将如何 GoupJoin 以便我从 t2 获取所有事件，这些事件发生在 t1 中每个事件之前 x 秒和 y 秒之后？

鉴于：

t1 -----A-----B-----C

t2 --1--2--3--4--5--6

如果 t1 相隔 2 秒，t2 相隔 1 秒，我们正在寻找每个 t1 事件两侧相隔 1 秒的 t2 事件，结果如下。

结果：

{ A, [1,2,3] }

{ B, [3,4,5] }

{ C, [5,6] }

下面是一个真实的例子，我们需要解决上述问题。我们有一个电子邮件流和另一个文本消息流。我们需要发出另一个流的结果，其中包含电子邮件，并且文本消息发生在电子邮件发送时间之前或之后的 1 分钟内。

我正在实现一个进程间消息传递系统，例如客户端可以从服务器请求某些数据。服务器需要能够以部分回复的形式发回其回复，并且还需要能够通知客户端以防发生任何异常。

目前，我通过 3 种消息类型执行此操作：

例如，客户端请求数据，服务器发回 0-N PartialResponse 消息，后跟 ResponseError 或 ResponseComplete。

我正在使用的库（以 NetMQ 作为其传输层的 Obvs）会将所有可能的消息流公开为

虽然这个可观察的流永远不会完成，我相信也永远不会出错（除非可能是某些 Obvs/NetMQ 内部异常等）。

我想将其转换为 IObservable<PartialResponse>，它在原始流推送 ResponseComplete 消息时完成，并在遇到 ResponseError 消息或输入流中的实际错误时出错。例如：

这段代码实际上应该可以工作，但可能很糟糕——尤其是因为它直接订阅了输入的可观察序列。

有没有更好/更清洁的方法来转换可观察序列？

在Intro to Rx中说：

BehaviorSubjects 通常与类属性相关联。由于它们始终具有值并且可以提供更改通知，因此它们可以成为属性支持字段的候选者。

但是我真的找不到一个明确的例子来说明如何做到这一点。

如果我理解正确，建议这样做：

我对此有几个问题：

• 这确实是提议的内容吗？
• 这种模式是否有一个通用名称，我可以进一步谷歌吗？因为我不知道该怎么称呼它，所以我的搜索结果很短。
• 这被认为是好的做法吗？还是有更好的方法来做同样的事情（即创建一个可观察的领域）？
• 你会建议什么命名约定？

所以我有一个包含一个可观察对象作为属性的对象：

我还有一个瓷砖列表：

现在我想为瓦片列表制作一个可观察的，所以当任何瓦片的类型发生变化时都会发生一个事件（并且可观察到的应该包括它是哪个瓦片），例如：

这可能吗？如果是这样，怎么办？