可觀察到流

Question

目的：

我使用StackExchange Redis的客戶端。我的目標是從客戶端公開的Pub Sub Subscriber創建一個Observable流，然後可以依次通過Observables支持訂閱，其中每個訂閱者都有自己的通過LINQ過濾的訂閱。 （出版工作按計劃進行，這個問題純粹是圍繞訂閱事件流上的特定通道。）

背景：

我使用Redis的酒吧子作爲事件源CQRS應用程序的一部分。具體用例是將事件發佈給多個訂閱者，然後更新各種閱讀模型，發送電子郵件等。

這些訂戶都需要過濾它們處理的事件類型，爲此我期待使用Rx帶有LINQ的.Net（Reactive Extensions）到 在事件流上提供了一個過濾標準，以便有效地處理僅對感興趣的事件作出反應。使用這種方法可以省去註冊處理程序的事件總線實現，並且允許我通過部署1-n向系統添加新投影。每個微服務都有1-n訂閱了具有自己的事件流的可觀察對象具體過濾器。

我曾嘗試：

1）I已經創建了一個類從ObservableBase繼承，重寫SubscribeCore方法，該方法從觀測量接收的訂閱請求，將它們存儲在ConcurrentDictionary，並且作爲各Redis的通知到達從通道中，遍歷已註冊的Observable訂閱者，並調用傳遞RedisValue的OnNext方法。

2）我創建了一個Subject，它也接受來自Observables的訂閱，並調用它們的OnNext方法。再次，受試者的使用似乎被許多人所詬病。

問題：

我試圖做的功能（至少在表面上），與各種性能層面的方法，但feel like a hack，那我不是它的目的的方式使用的Rx。

我看到很多意見，應儘可能使用內置的Observable方法，例如Observable.FromEvent，但似乎不可能使用StackExchange Redis客戶端訂閱API，至少在我眼中。

我也明白，用於接收流，並轉發到多個觀察員的優選的方法是使用一個ConnectableObservable，這似乎被設計爲非常場景我面對（每個微服務將在內部具有1 -n訂閱觀察對象）。目前，我無法理解如何將ConnectableObservable連接到來自StackExchange Redis的通知，或者如果它提供的實際好處超過了Observable。

UPDATE：

雖然完成是不是在我的情況的問題（處理是罰款），錯誤處理爲重要;例如隔離在一個用戶中檢測到的錯誤，以防止所有訂閱終止。

Answer 1

這裏是你可以用它來創建一個ISubscriber的IObservable<RedisValue>擴展方法和RedisChannel：

public static IObservable<RedisValue> WhenMessageReceived(this ISubscriber subscriber, RedisChannel channel) 
{ 
    return Observable.Create<RedisValue>(async (obs, ct) => 
    { 
     await subscriber.SubscribeAsync(channel, (_, message) => 
     { 
      obs.OnNext(message); 
     }).ConfigureAwait(false); 

     return Disposable.Create(() => subscriber.Unsubscribe(channel)); 
    }); 
} 

由於沒有Redis的渠道完成後所產生的IObservable將永遠不會完成，但您可能降IDisposable訂閱取消訂閱Redis頻道（這將由許多Rx運營商自動完成）。

用途可能是像這樣：

var subscriber = connectionMultiplexer.GetSubscriber(); 

var gotMessage = await subscriber.WhenMessageReceived("my_channel") 
    .AnyAsync(msg => msg == "expected_message") 
    .ToTask() 
    .ConfigureAwait(false); 

或按你的例子​​：

var subscriber = connectionMultiplexer.GetSubscriber(); 

var sendEmailEvents = subscriber.WhenMessageReceived("my_channel") 
    .Select(msg => ParseEventFromMessage(msg)) 
    .Where(evt => evt.Type == EventType.SendEmails); 

await sendEmailEvents.ForEachAsync(evt => 
{ 
    SendEmails(evt); 
}).ConfigureAwait(false); 

其他微服務可以過濾不同。