问题标签 [reactive-programming]

我是反应式扩展初学者。Gideon Engelberth 在我的问题中给了我关于 Reactive Extension 的出色回答。

如何使用 LINQ 或 Rx 在一个方法链上将 img url 转换为 HTML 中的 BASE64 字符串

现在我有第二个问题，即如何将 IObservable 返回值分配给 XDocument 实例。

Gideon 给了我下面的样品。

我想这样做。布丁好像很难...

我想在反应香蕉中实现某种类型的事件限制。它应该这样工作，如果从最后一个通过的事件到达的时间少于 delta 秒，则不会让事件通过。如果它没有通过，那么它会被存储并在最后一次触发事件的 delta 秒后触发。

下面是一个为时间戳数字列表实现此功能的程序。有没有可能把它翻译成 reactive-banana ？

另外，在反应香蕉中，我如何在其他事件出现后 x 秒触发一个事件？

我希望能够添加这样的扩展方法：

在哪里：

产生输出：

关于如何实施上述方法的任何建议？有没有我可以使用的内置方法？（假设：同步/立即订阅）

首先是一些背景知识，我编写了一个名为 Duplitiy（在 github.com 上）的开源 .NET 库，它使用FileSystemWatcher复制两个目录之间的所有文件更改。

我编写了一个FileSystemObservable实现的类IObservable<FileSystemChange>（它使用FSWatcher来包装实际的FileSystemWatcher）。当创建、修改或删除文件或目录时，这些更改将通过Subject<FileSystemChange>使用响应式扩展发布。

然后我使用以下订阅订阅这个 observable。

更改将被缓冲，直到至少有 2 秒的时间段没有任何更改，最长为 1 分钟。这是因为在删除目录时，FileSystemWatcher会通知所有包含的文件和目录。我们可以通过吞下目录中包含的更改来优化行为，并简单地删除订阅者中的父级。这由PrioritizeFileSystemChanges过滤器处理。它还允许我们忽略在缓冲区窗口中创建和随后删除的文件，再次减少目标上的 IO 操作。

这可行，尽管目前以一种幼稚的方式，不支持失败/重试。

但是我的问题是，这个 observable 的订阅者可能需要花费合理的时间来处理每个更改。例如，将大文件复制到慢速文件系统。当当前正在复制的同一文件发生新的文件系统更改时，我如何处理中止正在进行的操作。或者，如果文件包含在缓冲列表中但未完成，如何将其删除或排除？

我假设需要对原始 observable 进行另一个订阅，但不确定如何最好地共享状态或修改待处理的任务？必须按照接收到的顺序处理更改，这表示队列。但是，新的文件系统更改可能会应用于需要取消或删除的排队操作。队列不是为乱序删除而设计的。

例如，如果我们当前正在复制文件Foo\Bar.txt并且Foo目录已被删除。然后必须取消目录和所有子目录的任何正在进行或未决的更改。这可能是任务并行库的用例，还是我可以采取一些反应式方法？

也欢迎任何 github 拉取请求！

我有一个类似于以下内容的 RX 代码：

几天一次，程序停止进入 onEvent 方法，并且永远不会恢复。（即使这样 OnNext 调用仍在发生）。

没有对订阅调用 dispose。

我使用 rx-experimental 1.11111。

什么可能导致这种情况？

我目前正在学习 Haskell 和 Yampa，并且对反应函数有疑问。

正如您在类型签名中看到的，感知函数的部分输出是函数的当前调用和上一次调用之间的时间差。在我看到的例子中，这个时间差是“手动”计算的，使用 IORef 来保持前一次调用的值。

您必须使用外部状态来跟踪时间差，这似乎很奇怪，为什么不在 reactimate 函数中进行此计算？IORef 是处理它的好方法吗？

迈出 Rx 的第一步，我被困在这里：

令我惊讶的是，上面的代码产生了

请注意，在两个订阅之后手动调用 Connect() 不是我想要的，尽管输出符合预期。

我怎样才能做到这一点？

我有一个代码：

问题是这个第一次调用是一个冷的可观察的，所以在另一个这样的调用中：

myObserver 根本不会触发 onNext。显然是因为第一个电话发布了1,2,3, END. 我不想让 observable 调用“END”，因为我想稍后继续订阅。是否有一个功能可以轻松为我做到这一点？

我目前正在开发 Windows Phone 应用程序，我想使用 Reactive Extensions 来创建异步以获得更好的 UI 体验。

我使用 MVVM 模式：我的 View 在我的 ViewModel 中有一个 ListBox 绑定到 Items 的 ObservableCollection。项目具有传统属性，如名称或 IsSelected。

我在 App.xaml.cs 中实例化我的 MainViewModel：

我在 MainPage_Loaded 中加载我的数据。

我从 ViewModel 加载数据（稍后我将把代码移到模型中）：

它工作正常，项目在视图中更新。

现在我想创建异步。在我从 View 而不是 LoadData 调用的新方法中使用传统的 BeginInvoke 可以正常工作。

我使用了一个名为 CurrentDispatcher 的属性，它在 App.xaml.cs 中填充了 App.Current.RootVisual.Dispatcher。

但我想使用反应式扩展。因此，我尝试了 Rx 的不同元素，例如 ToAsync 或 ToObservable，但是当我将项目添加到清单时，我遇到了一些“UnauthorizedAccessException 未处理”和“无效的跨线程访问”。

我尝试观察其他线程，因为错误可能来自 UI 和后台线程之间的混合，但它不起作用。也许我不像在那种特殊情况下那样使用 Rx？

任何帮助将不胜感激。

回答后编辑：

这是一个很好用的代码！

如您所见，我以前没有正确使用。现在我可以使用 ObservableCollection 并在订阅中使用它。这是完美的！非常感谢！

下面是我尝试将 System.Diagnostics.Process 转换为 IConnectableObservable。这个解决方案有一个问题：我想连续收听标准输出和错误，并使用事件 Process.Exited 作为 OnCompleted 的触发器。不幸的是，我发现 Process.Exited在输出缓冲区为空之前引发。这意味着如果没有我使用线程睡眠的丑陋解决方法，我可以重现不通过 OnNext 语句提供输出的情况。

Q1：您看到此问题的任何解决方法吗？

Q2：关于 System.Reactive：在我的解决方案中我可以做得更好吗？

问候，

马库斯