非同期関数で作成されたobservableを取り消す

Question

私はrxを使い慣れていて、ドットネットでリアクティブエクステンションを使っていくつかのネットワークコードに取り組んでいます。私の問題は、非同期関数で作成したtcpClientのオブザーバブルが、私が供給しているトークンを介して取り消しをトリガするときに期待通りに完了していないことです。私はこれを実行すると、ローカルホストに接続する場合

public static class ListenerExtensions 
{ 
    public static IObservable<TcpClient> ToListenerObservable(
     this IPEndPoint endpoint, 
     int backlog) 
    { 
     return new TcpListener(endpoint).ToListenerObservable(backlog); 
    } 

    public static IObservable<TcpClient> ToListenerObservable(
     this TcpListener listener, 
     int backlog) 
    { 
     return Observable.Create<TcpClient>(async (observer, token) => 
     { 
      listener.Start(backlog); 

      try 
      { 
       while (!token.IsCancellationRequested) 
        observer.OnNext(await Task.Run(() => listener.AcceptTcpClientAsync(), token)); 
       //This never prints and onCompleted is never called. 
       Console.WriteLine("Completing.."); 
       observer.OnCompleted(); 
      } 
      catch (System.Exception error) 
      { 
       observer.OnError(error); 
      } 
      finally 
      { 
       //This is never executed and my progam exits without closing the listener. 
       Console.WriteLine("Stopping listener..."); 
       listener.Stop(); 
      } 
     }); 
    } 
} 
class Program 
{ 
    static void Main(string[] args) 
    { 
     var home = new IPEndPoint(IPAddress.Parse("127.0.0.1"), 2323); 
     var cancellation = new CancellationTokenSource(); 

     home.ToListenerObservable(10) 
      .Subscribe(
       onNext: c => Console.WriteLine($"{c.Client.RemoteEndPoint} connected"), 
       onError: e => Console.WriteLine($"Error: {e.Message}"), 
       onCompleted:() => Console.WriteLine("Complete"), // Never happens 
       token: cancellation.Token); 

     Console.WriteLine("Any key to cancel"); 
     Console.ReadKey(); 
     cancellation.Cancel(); 
     Thread.Sleep(1000); 
    } 
} 

：ここで私はとの問題を抱えているコードの簡易版である2323年は、私が接続されているtcpClientsのシーケンスを取得することがわかります。しかし、私がcancelationトークンのキャンセルをトリガーすると、プログラムはリスナーを閉じずに、期待どおりにonCompletedイベントを出さずに終了します。私は間違って何をしていますか？

Answer 1

だから私はここでいくつかのことを混乱させた。 This articleが私の進路を助けました。実際、そこからコードをコピーしてしまいました。

public static class TaskCancellations 
{ 
    public static async Task<T> WithCancellation<T>(
     this Task<T> task, 
     CancellationToken token) 
    { 
     var cancellation = new TaskCompletionSource<bool>(); 
     using (token.Register(s => 
      (s as TaskCompletionSource<bool>).TrySetResult(true), cancellation)) 
     { 
      if (task != await Task.WhenAny(task, cancellation.Task)) 
       throw new OperationCanceledException(token); 
      return await task; 
     } 
    } 
} 

そして、このようなtcplistenerでそれを使用します。今、予想通り

public static IObservable<TcpClient> ToListenerObservable(
    this TcpListener listener, 
    int backlog) 
{ 
    return Observable.Create<TcpClient>(async (observer, token) => 
    { 
     listener.Start(backlog) 
     try 
     { 
      while (!token.IsCancellationRequested) 
      { 
       observer.OnNext(await listener.AcceptTcpClientAsync() 
        .WithCancellation(token)); 
      } 
     } 
     catch (OperationCanceledException) 
     { 
      Console.WriteLine("Completing..."); 
      observer.OnCompleted(); 
     } 
     catch (System.Exception error) 
     { 
      observer.OnError(error); 
     } 
     finally 
     { 
      Console.WriteLine("Stopping listener..."); 
      listener.Stop(); 
     } 
    }); 
} 

すべてが動作します。

Answer 2

あまりにも多くのコードを書くことは避けてください。キャンセルトークンを使用してください。ここでは、標準のRxオペレータから離れることなく、あなたがやっていることをやり遂げる方法があります。私はこのコードを完全にテストすることができませんでしたので、少し微調整が必​​要な場合があります。ここで重要なことは、キャンセルトークンのように尋ねcompleterある

var query = Observable.Create<TcpClient>(o => 
{ 
    var home = new IPEndPoint(IPAddress.Parse("127.0.0.1"), 2323); 
    var listener = new TcpListener(home); 
    listener.Start(); 
    return 
     Observable 
      .Defer(() => Observable.FromAsync(() => listener.AcceptTcpClientAsync())) 
      .Repeat() 
      .Subscribe(o); 
}); 

var completer = new Subject<Unit>(); 
var subscription = 
    query 
     .TakeUntil(completer) 
     .Subscribe(
      onNext: c => Console.WriteLine($"{c.Client.RemoteEndPoint} connected"), 
      onError: e => Console.WriteLine($"Error: {e.Message}"), 
      onCompleted:() => Console.WriteLine("Complete")); 

Console.WriteLine("Enter to cancel"); 
Console.ReadLine(); 
completer.OnNext(Unit.Default); 
Thread.Sleep(1000); 

：

はこれを試してみてください。それはOnCompletedコールなしで終了するsubscription.Dispose()とは異なり、自然にサブスクリプションを終了します。