问题标签 [system.reactive]

目前，我正在使用 RX 框架来实现类似工作流的消息处理管道。本质上，我有一个消息生产者（反序列化网络消息并在主题上调用 OnNext()），并且我有几个消费者。

注意： if 和 transform 是我编写的扩展方法，它们只返回一个 IObservable。

消费者执行以下操作：

commerceRequest然后被另一个类似的管道消耗，并一直持续到顶部，最后有人调用Subscribe()最终管道。我遇到的问题是来自基地的消息不会传播，除非直接在某处对消息调用订阅。

如何将消息推送到堆栈顶部？我知道这是一种非正统的方法，但我觉得它使代码非常容易理解消息发生了什么。如果您觉得这是一个非常糟糕的主意，有人可以提出另一种方法吗？

我正在为我的第一个简单的“hello world”RX 应用程序而苦苦挣扎。我正在使用 VS2010 RC，以及最新的 RX 下载。

以下是简单的控制台应用程序；

我在 Where 子句中得到一个 ArgumentNullException。这是堆栈；

我想知道是否可以使用队列（特别是 ConcurrentQueue）作为 IObservable 的来源？就像是;

我想这真的没有意义，因为没有任何东西被出队。我正在尝试实现一个非阻塞进程，它可以订阅被推送给观察者的“消息”，因此使用队列。我确信我应该能够用 RX 做到这一点，但似乎无法理解它！

我会对有关如何实施的任何建议感兴趣。谢谢！

在我目前正在处理的系统中，我有许多组件被定义为接口和基类。系统的每个部分都有一些特定的点，它们与系统的其他部分交互。

例如，数据准备组件准备了一些数据，这些数据最终需要进入数据处理部分，通信组件需要查询不同组件的状态以中继到外部等。

目前，我使用“上帝对象”或对系统不同部分有深入了解的对象将系统的这些部分粘合在一起。它在这里注册事件并将结果传送给那里的方法，在这里创建一个回调方法并在那里返回该方法的结果，并通过多线程队列传递许多请求以进行处理，因为它“知道”某些操作已经在 STA 线程等上运行

虽然它很方便，但让我担心的是，这种类型非常了解系统中其他人的设计方式。我更喜欢一个更通用的集线器，它可以提供可以公开事件或方法或回调或可以使用这些的实例。

我已经看到更多关于反应式框架的 IObservable/IObserver 功能的信息，并且这些功能正在被纳入 .NET 4.0（我相信）。

我可以利用这种模式来帮助替换我的“上帝对象”吗？我该怎么做呢？是否有任何资源可以将此模式用于此特定目的？

我有一个要并行处理的可观察集合，然后在过滤时观察处理后的值，最后订阅一个接收过滤值的处理程序。

我的示例在语法上是正确的并且编译得很好，当我运行代码时，Where会评估执行过滤的语句。但是没有数据通过订阅。如果我删除AsParallel以便通过常规完成处理，则IEnumerable数据会通过并且一切都按预期工作。

这是我的示例，对字符串进行了一些处理：

下一个奇怪的事情是，如果我使用TakeWhile运算符，在我看来，它在概念上类似于 Where，观察 ParallelQuery 按预期工作：

向订阅添加一些日志记录代码表明在ToObservable转换之前接收到数据，但不是在转换之后：

第 4 行的 lambda 中的断点被命中，而第 6 行的 lambda 中的断点从未被命中。

为什么将TakeWhile数据传递给订阅者而Where不会？

如果它很重要，我会在 Visual Studio 2010 RC 中使用针对 .Net 4.0 Framework Client Profile 的项目开发我的代码。

更新：根据@Sergeys 的回答，我重新设计了Where过滤器的位置。以下代码按预期工作：

processedStrings必须首先将初始的 observable转换为 enumerable 以使其并行化，然后将其转换回 observable 以订阅最终结果，这仍然感觉有点尴尬。

I am playing with Rx in .Net3.5SP1 and trying the 101 Rx Samples. I am trying the first sample (Start - Run Code Asynchronously) but it doesn't seem to actually run asynchronously. For example,

Outputs

Is there an explanation for this? Am I doing something wrong? Is this expected behaviour?

UPDATE

I would have thought it would have said

But the main thread is blocked while the supposedly Asynch call happens.

那么，使用 Reactive Framework (Rx) 实现 Silverlight 本地化以实现即时 UI 更改的最佳方法是什么？我将非常有义务举例。

这是一个即时更改 UI 的示例。但我无法弄清楚它是如何工作的。

我真正想做的就是完成两个异步操作，一个接一个。例如

下载网站 X。完成后，下载网站 Y。

根据 ILMerge 文档，它允许合并具有冲突类型的程序集。我正在尝试合并 FSharp.Core.dll 和 System.Reactive.dll 但我无法弄清楚命令行参数。

ILMerge 抱怨System.IObservable<T>. 我怎么能告诉它解决这个冲突。

谢谢和问候，叉子

有没有这个的简写版本。我想调用一个序列，然后在它完成后调用另一个序列。