c# - 使用 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();
});

当我模拟关闭套接字时，存在缓冲区中存在一些无法处理的数据的情况 - 有什么办法可以避免这种行为？

当我以 500 毫秒的延迟从另一个进程发送一些消息时，它将像下面的示例一样工作:

1. 20条消息将收入
2. 将调用一些逻辑 - 订阅者逻辑
3. 4条消息将收入
4. 模拟关闭方法将被调用

当调用“关闭方法”时，我需要立即处理缓冲区中的所有数据并关闭应用程序，甚至无需等待缓冲区超时。缓冲延迟时间由用户定义，所以我不想等待调用订阅者逻辑，因为它可能会很长。

Answer 1

欢迎使用 StackOverflow!

缓冲区方法的现有重载不支持时间、计数和门。但是有一个重载会在序列产生值时触发缓冲区关闭。因此，我们将通过合并缓冲区关闭的所有条件的可观察量来简单地创建一个序列。

看看这个演示。

onTime 将在指定的时间段后产生一个值。

onCount 将在 x 项通过后产生第一个值。

onClose 将立即在订阅时产生一个值 - 但在我们决定连接之前我们不会连接到它。

            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();

这将根据计数或时间产生一个输出，直到按下返回键，此时它会立即关闭并显示缓冲区中的可用内容。