RxExtensionsでUDPソケットをリッスンするのを止める適切な方法

Question

UDPソケットを聴く必要があります.10秒後またはバッファ内の100アイテム後に、いくつかのロジックを呼び出す必要があります。一般的には正常に動作しますが、リスニングソケットを正しく停止する方法はわかりません。

var ip = new IPEndPoint(IPAddress.Parse("127.0.0.1"), 1234); 
var socket = new UdpClient(ip); 

var cancellationTokenSource = new CancellationTokenSource(); 
var observable = 
    Observable 
    .FromAsync(socket.ReceiveAsync) 
    .DoWhile(() => !cancellationTokenSource.IsCancellationRequested) 
    .Buffer(TimeSpan.FromSeconds(10), 100); 

var subscribtion = observable.Subscribe(o => 
{ 
    //logic 
}); 

//simulate close method from another thread 
Task.Factory.StartNew(() => 
{ 
    Task.Delay(TimeSpan.FromSeconds(12)).Wait(); 
    cancellationTokenSource.Cancel(); 
    socket.Close(); 
    subscribtion.Dispose(); 
}); 

私は閉じたソケットをシミュレートすると、処理できないバッファ内の一部のデータが存在する状況が存在するが - 、この動作を回避するために、どのような方法ですか？

私は500msの遅延で別のプロセスからいくつかのメッセージを送信し、それは以下の例のような作品になります。

1. 20のメッセージが収入
2. いくつかのロジックを呼び出すことになるだろう - 加入者のロジックを
3. 4のメッセージ収入
4. は「closeメソッドは、」起動された場合、closeメソッドが

を起動しますシミュレートします私はすぐにバッファタイムアウトを待つことなく、バッファとクローズアプリケーションのすべてのデータを処理する必要があります。バッファ遅延時間はユーザーによって定義されるため、かなり長い時間がかかる可能性があるため、invokeサブスクライバロジックを待つ必要はありません。

Answer 1

ようこそStackOverflow！

バッファメソッドの既存のオーバーロードは、時間、カウント、ゲートをサポートしていません。しかし、シーケンスが値を生成するときにバッファークローズをトリガーするオーバーロードがあります。したがって、バッファクローズのためのすべての条件のオブザーバブルをマージしてシーケンスを作成するだけです。

このデモをご覧ください。

onTimeは、指定した期間の後に値を生成します。

onCountは、xアイテムが経過した後に最初の値を生成します。

onCloseはすぐにsubscribeに値を生成しますが、私たちが決定するまでは接続しません。

  var producer = Observable.Interval(TimeSpan.FromSeconds(0.2)); 
      var source = producer.Publish().RefCount(); 

      var onClose = Observable.Return(0L).Publish(); 
      var onTime = Observable.Timer(TimeSpan.FromSeconds(2)); 
      var onCount = source.Skip(10); 

      var bufferClose = Observable.Merge(onClose, onTime, onCount); 

     var subscription = 
      source 
      .Buffer(() => bufferClose) 
      .Subscribe(list => Console.WriteLine(string.Join(",", list))); 

      Console.WriteLine("Waiting for close"); 
      Console.ReadLine(); 

      onClose.Connect(); //signal 
      subscription.Dispose(); 

      Console.WriteLine("Closed"); 

      Console.ReadLine(); 

リターンが押されるまでは、それがすぐにバッファで利用できるものと閉じたときに、カウントまたは時間に基づいて出力を生成します。