Observable wrapper for AsyncProducerConsumerQueue

Question

所以我使用以下代码为 Stephen Cleary 的AsyncProducerConsumerQueue<T>创建了一个可观察的包装器。

我想知道这里是否有人知道我如何以更简单的方式做到这一点？

• 它可以在没有包装类的情况下编写吗？
• 是否有可能防止将多个包装器应用于一个队列的错误？
• 我可以在第一次订阅时连接它，而不是通过直接调用Connect吗？如果是这样，这意味着什么？
• 最后，你会怎么做？

using Nito.AsyncEx;
using System.Reactive;
static async Task ExampleUsage() {
var queue = new AsyncProducerConsumerQueue<int>();
var observable = queue.AsConnectableObservable();
await queue.EnqueueAsync(1);
observable.Subscribe(Console.WriteLine);
observable.Connect();
await queue.EnqueueAsync(2);
}
public static class AsyncExExtensions {
public static IConnectableObservable<T> AsConnectableObservable<T>(this AsyncProducerConsumerQueue<T> queue) {
return new ConnectableObservableForAsyncProducerConsumerQueue<T>(queue);
}
}
class ConnectableObservableForAsyncProducerConsumerQueue<T> : IConnectableObservable<T> {
readonly AsyncProducerConsumerQueue<T> Queue;
long _isConnected = 0;
ImmutableList<IObserver<T>> Observers = ImmutableList<IObserver<T>>.Empty;
public ConnectableObservableForAsyncProducerConsumerQueue(AsyncProducerConsumerQueue<T> queue) {
Queue = queue;
}
public IDisposable Connect() {
if (Interlocked.Exchange(ref _isConnected, 1) == 1) throw new InvalidOperationException("Observable cannot be connected more than once.");
var cts = new CancellationTokenSource();
var token = cts.Token;
Task.Run(async () => {
try {
while (true) {
token.ThrowIfCancellationRequested();
var @event = await Queue.DequeueAsync(token).ConfigureAwait(false);
foreach (var observer in Observers)
observer.OnNext(@event);
}
} catch (Exception x) when (x is OperationCanceledException || x is InvalidOperationException) {
foreach (var observer in Observers)
observer.OnCompleted();
}
});
return Disposable.Create(() => {
cts.Cancel();
cts.Dispose();
});
}
readonly object subscriberListMutex = new object();
public IDisposable Subscribe(IObserver<T> observer) {
lock (subscriberListMutex) {
Observers = Observers.Add(observer);
}
return Disposable.Create(() => {
lock (subscriberListMutex) {
Observers = Observers.Remove(observer);
}
});
}
}

Answer 1

免责声明：我不是专家，所以这个答案可能有一些方面我忽略了 - 谨慎使用！

请考虑以下两个演示。对于有多个观察者的情况，这些行为会有所不同。在第一个演示中，观察者将竞争队列中的项目，在第二个演示中，他们将各自收到一份副本。

演示 #1 - 冷可观察

var queue = new AsyncProducerConsumerQueue<int>();
// This is a cold observable, so each observer is fed by its own individual dequeue loop
// and therefore will be 'competing' with other observers for queued items.
var coldObservable = Observable
// Create an observable that asynchronously waits for items to become available on the
// queue and then emits them to the observer. This will be cancelled when the observer
// is unsubscribed. 
.Create<int>(async (observer, cancellationToken) =>
{
while (true)
{
var item = await queue.DequeueAsync(cancellationToken).ConfigureAwait(false);
Console.WriteLine($"Dequeued {item}");
observer.OnNext(item);
}
})
// If an InvalidOperationException is thrown by the above, continue with
// an empty observable instead of the error. This effectively catches an
// `OnError(InvalidOperationException)` and turns it into an `OnCompleted()`.
.Catch<int, InvalidOperationException>(exn =>
{
Console.WriteLine("Caught InvalidOperation");
return Observable.Empty<int>();
});
Console.WriteLine("TEST COLD");
await queue.EnqueueAsync(1);
Console.WriteLine("Enqueued 1");
Console.WriteLine("Subscribing A");
coldObservable.Subscribe(
item => Console.WriteLine($"A received {item}"),
() => Console.WriteLine("A completed"));
Console.WriteLine("Subscribing B");
coldObservable.Subscribe(
item => Console.WriteLine($"B received {item}"),
() => Console.WriteLine("B completed"));
await queue.EnqueueAsync(2);
Console.WriteLine("Enqueued 2");
await queue.EnqueueAsync(3);
Console.WriteLine("Enqueued 3");
queue.CompleteAdding();
Console.WriteLine("Completed adding");
Console.WriteLine("Waiting...");
await Task.Delay(2000);
Console.WriteLine("DONE");
// TEST COLD
// Enqueued 1
// Subscribing A
// Dequeued 1
// A received 1
// Subscribing B
// Enqueued 2
// Enqueued 3
// Completed adding
// Waiting...
// Dequeued 2
// Dequeued 3
// A received 2
// B received 3
// Caught InvalidOperation
// Caught InvalidOperation
// A completed
// B completed
// DONE

演示 #2 - 热门可观察

var queue = new AsyncProducerConsumerQueue<int>();
var coldObservable = // defined same as above
// This is a hot observable, so each observer receives the same items from the queue.
var hotObservable = coldObservable
// Publish the cold observable to create an `IConnectableObservable` that will subscribe
// to the dequeue loop when connected and emit the same items to all observers.
.Publish()
// Automatically connect to the published observable when the first observer subscribes
// and automatically disconnect when the last observer unsubscribes. This means that the
// first observer will receive any items queued before it subscribes, but additional
// observers will only receive items queued after they subscribed.
.RefCount();
Console.WriteLine("TEST HOT");
await queue.EnqueueAsync(1);
Console.WriteLine("Enqueued 1");
Console.WriteLine("Subscribing A");
hotObservable.Subscribe(
item => Console.WriteLine($"A received {item}"),
() => Console.WriteLine("A completed"));
Console.WriteLine("Subscribing B");
hotObservable.Subscribe(
item => Console.WriteLine($"B received {item}"),
() => Console.WriteLine("B completed"));
await queue.EnqueueAsync(2);
Console.WriteLine("Enqueued 2");
await queue.EnqueueAsync(3);
Console.WriteLine("Enqueued 3");
queue.CompleteAdding();
Console.WriteLine("Completed adding");
Console.WriteLine("Waiting...");
await Task.Delay(2000);
Console.WriteLine("DONE");
// TEST HOT
// Enqueued 1
// Subscribing A
// Dequeued 1
// A received 1
// Subscribing B
// Enqueued 2
// Enqueued 3
// Dequeued 2
// Completed adding
// Waiting...
// A received 2
// B received 2
// Dequeued 3
// A received 3
// B received 3
// Caught InvalidOperation
// A completed
// B completed
// DONE

要回答您的原始问题：

它可以在没有包装类的情况下编写吗？

是的，请参阅上面的演示。

是否有可能防止将多个包装器应用于一个队列的错误？

上面演示的方法不会阻止其他方将项目取消排队(或对队列执行任何其他操作)。如果要确保仅公开给定队列的单个IObservable<T>，请考虑通过创建在内部创建和管理自己的AsyncProducerConsumerQueue的ObservableProducerConsumerQueue<T>来封装队列本身。您可以公开仅委托给内部队列的EnqueueAsync方法，并使用上面的演示可观察量之一将可观察量公开为属性或实现IObservable<T>接口。

我可以在第一个订阅时使其连接，而不是通过直接调用 Connect 吗？如果是这样，这意味着什么？

演示 #2 显示了此行为并描述了其含义。如果您希望能够在连接之前订阅观察器，请跳过RefCount调用并像以前一样使用Publish返回的IConnectableObservable。

最后，你会怎么做？

如上所述，我将封装队列并使用上面演示的方法之一公开IObservable或IConnectableObservable。