我怎样才能从队列中创建一个 IObservable，这样序列就不会在队列为空时结束？

Question

我正在研究使用 .NET (Rx) 的响应式扩展的东西，我想要一个从队列（或类似队列）获取输入的序列。

我试过这样做：

    static readonly Queue<DeviceTransaction> TransactionQueue = new Queue<DeviceTransaction>();
    //...
    var observableTransactionSource = TransactionQueue.ToObservable();
    //...
    observableTransactionSource.Subscribe(transactionObserver);

它在一定程度上起作用，但当队列为空时序列完成。我不想要一个空队列来结束序列。空并不意味着结束，它只是意味着'no more at the moment'.

有没有办法在队列为空时停止序列完成，或者我应该以不同的方式考虑整个问题？

Answer 1

调用 ToObservable() 充满了问题，正如我所解释的 here 它只是要使用 IEnumerable<T> 并使用队列的快照。

在这种情况下，您最好使用 Subject<T> 来支持您的事件。由于 Rx 语法指定您必须序列化事件的传递，因此它已经具有排队语义。只需就 post 事件的主题致电 OnNext<T>。

如果您需要确保在 post 编辑事件后发生的订阅不会错过事件，请使用 ReplaySubject<T>。

如果使用的主题与您有关，那么您可能需要查看 this blog post。总而言之，您对队列的使用表明可以在此处使用主题，但您可能需要考虑是否可以使用像 Observable.FromEvent.

这样的转换方法

我怎样才能从队列中创建一个 IObservable，这样序列就不会在队列为空时结束？

How can I make an IObservable from a queue, so that the sequence doesn't end when the queue is empty?

c#

reactive-programming

system.reactive