RX.netのDropQueueメカニズム

Question

私はRX.netで背圧問題を遭遇しましたが、解決策が見つかりません。私は観測可能なリアルタイムのログメッセージを持っています。私は、彼らがワイヤを介して送信される前に、logObservableからリアルタイムのログメッセージをシリアライズTCPインタフェースを介して公開する

var logObservable = /* Observable stream of log messages */ 

。だから私は、次の手順を実行します

foreach (var message in logObservable.ToEnumerable()) 
{ 
    // 1. Serialize message 
    // 2. Send it over the wire. 
} 

背圧のシナリオは、例えば発生した場合、問題が発生して.ToEnumerable()相手側のクライアントがストリームを一時停止する場合問題は、.ToEnumerable()がメモリを大量に使用するアイテムをキャッシュすることです。私はDropQueueのようなものを探しています。最後の10個のメッセージをバッファリングするだけです。

var observableStream = logObservable.DropQueue(10).ToEnumerable(); 

これはこの問題を解決する正しい方法ですか？背圧の問題を避けるために、このような仕組みを実装することを知っていますか？

Answer 1

マイDropQueue実装：

public static IEnumerable<TSource> ToDropQueue<TSource>(
     this IObservable<TSource> source, 
     int queueSize, 
     Action backPressureNotification = null, 
     CancellationToken token = default(CancellationToken)) 
    { 
     var queue = new BlockingCollection<TSource>(new ConcurrentQueue<TSource>(), queueSize); 
     var isBackPressureNotified = false; 

     var subscription = source.Subscribe(
      item => 
      { 
       var isBackPressure = queue.Count == queue.BoundedCapacity; 

       if (isBackPressure) 
       { 
        queue.Take(); // Dequeue an item to make space for the next one 

        // Fire back-pressure notification if defined 
        if (!isBackPressureNotified && backPressureNotification != null) 
        { 
         backPressureNotification(); 
         isBackPressureNotified = true; 
        } 
       } 
       else 
       { 
        isBackPressureNotified = false; 
       } 

       queue.Add(item); 
      }, 
      exception => queue.CompleteAdding(), 
      () => queue.CompleteAdding()); 

     token.Register(() => { subscription.Dispose(); }); 

     using (new CompositeDisposable(subscription, queue)) 
     { 
      foreach (var item in queue.GetConsumingEnumerable()) 
      { 
       yield return item; 
      } 
     } 
    }