如何从方法内外中断同步订阅？

Question

问题：我订阅了一个永无止境的消息服务，我的代码需要检查任何消息是否满足条件，如果满足，然后在处理所有消息并返回true之前关闭订阅。如果我已经处理了所有的消息，但条件不满足，那么我需要关闭订阅并返回false。

例如，条件是foo=5:

message dataset early success :
msg1: foo=1
msg2: foo=2
msg3: foo=5 <= condition satisfied, return true and stop processing
msg4: foo=6

message dataset failure :
msg1: foo=1
msg2: foo=2
msg3: foo=3 
msg4: foo=4 <= no more messages, return false and stop processing

我使用的订阅有一个同步方法，我必须传递异步EventHandler。下面是适用于这两种情况的功能代码，LastMessageReceivedDateTime跟踪最后一次接收消息的时间（以标识消息的结尾），_conditionstatistified告诉我是否获得了数据：

private DateTime lastMessageReceivedDateTime;
private bool _conditionSatisfied;

public Task<bool> CheckSubscription(IThirdParyCode connection)
{
     var subscription = connection.Subscribe(async (obj, args) =>
     {
         lastMessageReceivedDateTime = DateTime.Now;
         if(args.Message.foo == 5)
         {
              _conditionSatisfied = true;
         }
     });

     while (lastMessageReceivedDateTime.AddSeconds(1) > DateTime.Now  && !_conditionSatisfied)
     {
         Thread.Sleep(500);
     }

     subscription?.Unsubscribe();
     return _activityCheckSatisfied;
}

更多信息:连接的类型是IstanConnection（来自NATS)，Subscribe的签名是：

IStanSubscription Subscribe(string subject, StanSubscriptionOptions options,
    EventHandler<StanMsgHandlerArgs> handler);

我简化了签名，把重点放在我有问题的代码上。

Answer 1

根据您的代码示例，我可以假设，如果在最后一条消息的一秒钟内没有新消息，则消息流结束。可以修改解决方案以消除活动等待循环，并将其替换为单个await调用。它将基于两项任务：

1. 第一个任务将跟踪成功完成（在您的示例中_conditionsetfreed)并将由taskcompletionsource.setresult
设置
2. 第二个任务将尝试通过使用CancellationToken任务包装器（这种包装器的示例实现）和CancellationTokenSource.cancelAfter的组合来发出流结束的信号，后者将尝试在每次迭代后取消任务并延迟。这将替换LastMessageReceivedDateTime.addSeconds(1)>DateTime.nowCondition.

修改后的代码应该如下所示：

private CancellationTokenSource streamEndCancellation = new CancellationTokenSource();
private TaskCompletionSource<bool> satisfiedCompletionSource = new TaskCompletionSource<bool>();

public async Task<bool> CheckSubscription(IThirdParyCode connection)
{
     // CancellationTokenTaskSource is in third-party library and not part of .NET
     var streamEndSource = new CancellationTokenTaskSource<bool>(streamEndCancellation.Token);

     var subscription = connection.Subscribe(async (obj, args) =>
     {
         lastMessageReceivedDateTime = DateTime.Now;
         if(args.Message.foo == 5)
         {
             satisfiedCompletionSource.SetResult(true);
         }
         streamEndCancellation.CancelAfter(1000);
     });

     Task<bool> actualTask = await Task.WhenAny<bool>(satisfiedCompletionSource.Task, streamEndSource.Task);
          
     subscription?.Unsubscribe();
     return !actualTask.IsCanceled;
}