Làm thế nào để chờ một sự kiện trong C#, với thời gian chờ và hủy

Question

Vì vậy, yêu cầu của tôi là để chức năng của tôi chờ phiên bản event Action<T> đến từ một lớp khác và một chủ đề khác, và xử lý nó trên chủ đề của tôi bị gián đoạn bởi thời gian chờ hoặc CancellationToken.

Tôi muốn tạo một hàm chung mà tôi có thể sử dụng lại. Tôi đã xoay xở để tạo ra một vài lựa chọn làm (tôi nghĩ) những gì tôi cần, nhưng cả hai dường như phức tạp hơn tôi tưởng tượng nó phải có.

Cách sử dụng

Chỉ cần được rõ ràng, việc sử dụng mẫu của chức năng này sẽ trông như thế này, nơi serialDevice là phun ra các sự kiện trên một sợi riêng biệt:

var eventOccurred = Helper.WaitForSingleEvent<StatusPacket>(
    cancellationToken, 
    statusPacket => OnStatusPacketReceived(statusPacket), 
    a => serialDevice.StatusPacketReceived += a, 
    a => serialDevice.StatusPacketReceived -= a, 
    5000, 
    () => serialDevice.RequestStatusPacket()); 

Lựa chọn 1-ManualResetEventSlim

Tùy chọn này không phải là xấu, nhưng việc xử lý Dispose của ManualResetEventSlim là lộn xộn hơn nó có vẻ như nó phải được. Nó cho ReSharper phù hợp với việc tôi đang truy cập vào những thứ đã được sửa đổi/xử lý trong phạm vi đóng cửa, và thật khó để làm theo vì vậy tôi thậm chí không chắc nó là chính xác. Có lẽ có điều gì đó mà tôi bỏ lỡ có thể làm sạch nó, đó sẽ là sở thích của tôi, nhưng tôi không thấy nó một cách dễ dàng. Đây là mã.

public static bool WaitForSingleEvent<TEvent>(this CancellationToken token, Action<TEvent> handler, Action<Action<TEvent>> subscribe, Action<Action<TEvent>> unsubscribe, int msTimeout, Action initializer = null) 
{ 
    var eventOccurred = false; 
    var eventResult = default(TEvent); 
    var o = new object(); 
    var slim = new ManualResetEventSlim(); 
    Action<TEvent> setResult = result => 
    { 
     lock (o) // ensures we get the first event only 
     { 
      if (!eventOccurred) 
      { 
       eventResult = result; 
       eventOccurred = true; 
       // ReSharper disable AccessToModifiedClosure 
       // ReSharper disable AccessToDisposedClosure 
       if (slim != null) 
       { 
        slim.Set(); 
       } 
       // ReSharper restore AccessToDisposedClosure 
       // ReSharper restore AccessToModifiedClosure 
      } 
     } 
    }; 
    subscribe(setResult); 
    try 
    { 
     if (initializer != null) 
     { 
      initializer(); 
     } 
     slim.Wait(msTimeout, token); 
    } 
    finally // ensures unsubscription in case of exception 
    { 
     unsubscribe(setResult); 
     lock(o) // ensure we don't access slim 
     { 
      slim.Dispose(); 
      slim = null; 
     } 
    } 
    lock (o) // ensures our variables don't get changed in middle of things 
    { 
     if (eventOccurred) 
     { 
      handler(eventResult); 
     } 
     return eventOccurred; 
    } 
} 

Lựa chọn 2-bỏ phiếu mà không có một WaitHandle

Các WaitForSingleEvent chức năng ở đây là sạch hơn nhiều. Tôi có thể sử dụng ConcurrentQueue và do đó thậm chí không cần khóa. Nhưng tôi không thích chức năng bỏ phiếu Sleep và tôi không thấy bất kỳ cách nào xung quanh nó với phương pháp này. Tôi muốn chuyển một số WaitHandle thay vì Func<bool> để làm sạch Sleep, nhưng lần thứ hai tôi làm điều đó tôi đã có toàn bộ mess Dispose để dọn dẹp lại.

public static bool WaitForSingleEvent<TEvent>(this CancellationToken token, Action<TEvent> handler, Action<Action<TEvent>> subscribe, Action<Action<TEvent>> unsubscribe, int msTimeout, Action initializer = null) 
{ 
    var q = new ConcurrentQueue<TEvent>(); 
    subscribe(q.Enqueue); 
    try 
    { 
     if (initializer != null) 
     { 
      initializer(); 
     } 
     token.Sleep(msTimeout,() => !q.IsEmpty); 
    } 
    finally // ensures unsubscription in case of exception 
    { 
     unsubscribe(q.Enqueue); 
    } 
    TEvent eventResult; 
    var eventOccurred = q.TryDequeue(out eventResult); 
    if (eventOccurred) 
    { 
     handler(eventResult); 
    } 
    return eventOccurred; 
} 

public static void Sleep(this CancellationToken token, int ms, Func<bool> exitCondition) 
{ 
    var start = DateTime.Now; 
    while ((DateTime.Now - start).TotalMilliseconds < ms && !exitCondition()) 
    { 
     token.ThrowIfCancellationRequested(); 
     Thread.Sleep(1); 
    } 
} 

Câu hỏi

Tôi không đặc biệt quan tâm đối với một trong các giải pháp này, tôi cũng không chắc chắn 100% một trong hai trong số đó là 100% chính xác. Hoặc là một trong những giải pháp này tốt hơn các giải pháp khác (tính tự nhiên, hiệu quả, v.v.), hoặc có cách nào dễ hơn hoặc chức năng tích hợp để đáp ứng những gì tôi cần làm ở đây không?

Cập nhật: Câu trả lời hay nhất

Sửa đổi giải pháp TaskCompletionSource bên dưới. Không có đóng cửa dài, ổ khóa, hoặc bất cứ điều gì cần thiết. Có vẻ khá đơn giản. Có lỗi nào ở đây không?

public static bool WaitForSingleEvent<TEvent>(this CancellationToken token, Action<TEvent> onEvent, Action<Action<TEvent>> subscribe, Action<Action<TEvent>> unsubscribe, int msTimeout, Action initializer = null) 
{ 
    var tcs = new TaskCompletionSource<TEvent>(); 
    Action<TEvent> handler = result => tcs.TrySetResult(result); 
    var task = tcs.Task; 
    subscribe(handler); 
    try 
    { 
     if (initializer != null) 
     { 
      initializer(); 
     } 
     task.Wait(msTimeout, token); 
    } 
    finally 
    { 
     unsubscribe(handler); 
     // Do not dispose task http://blogs.msdn.com/b/pfxteam/archive/2012/03/25/10287435.aspx 
    } 
    if (task.Status == TaskStatus.RanToCompletion) 
    { 
     onEvent(task.Result); 
     return true; 
    } 
    return false; 
} 

Cập nhật 2: Một giải pháp tuyệt vời

Hóa ra rằng BlockingCollection công trình giống như ConcurrentQueue mà còn có phương pháp chấp nhận một thời gian chờ và hủy thẻ. Một điều thú vị về giải pháp này là nó có thể được cập nhật để thực hiện một WaitForNEvents khá dễ dàng:

public static bool WaitForSingleEvent<TEvent>(this CancellationToken token, Action<TEvent> handler, Action<Action<TEvent>> subscribe, Action<Action<TEvent>> unsubscribe, int msTimeout, Action initializer = null) 
{ 
    var q = new BlockingCollection<TEvent>(); 
    Action<TEvent> add = item => q.TryAdd(item); 
    subscribe(add); 
    try 
    { 
     if (initializer != null) 
     { 
      initializer(); 
     } 
     TEvent eventResult; 
     if (q.TryTake(out eventResult, msTimeout, token)) 
     { 
      handler(eventResult); 
      return true; 
     } 
     return false; 
    } 
    finally 
    { 
     unsubscribe(add); 
     q.Dispose(); 
    } 
} 

Answer 1

Bạn có thể sử dụng Rx để chuyển đổi sự kiện thành có thể quan sát, sau đó đến một tác vụ và cuối cùng chờ đợi tác vụ đó với mã thông báo/thời gian chờ của bạn.

Một lợi thế này có trên bất kỳ giải pháp hiện có nào, là nó gọi unsubscribe trên chuỗi sự kiện, đảm bảo mà trình xử lý của bạn sẽ không được gọi hai lần. (Trong giải pháp đầu tiên của bạn, bạn làm việc xung quanh điều này bằng cách tcs.TrySetResult thay vì tcs.SetResult, nhưng nó luôn luôn tốt đẹp để thoát khỏi một "TryDoSomething" và chỉ cần đảm bảo DoSomething luôn hoạt động).

Một ưu điểm khác là tính đơn giản của mã. Nó cơ bản là một dòng. Vì vậy, bạn thậm chí không cần một chức năng độc lập. Bạn có thể nội tuyến nó để rõ ràng chính xác mã của bạn là gì, và bạn có thể tạo ra các biến thể trên chủ đề mà không cần một tấn thông số tùy chọn (như tùy chọn initializer hoặc cho phép chờ đợi trên N sự kiện, hoặc bỏ qua thời gian chờ/hủy trong trường hợp nơi họ không cần thiết). Và bạn có cả số tiền trả về bool val và thực tế result trong phạm vi khi kết thúc, nếu điều đó hữu ích.

using System.Reactive.Linq; 
using System.Reactive.Threading.Tasks; 
... 
public static bool WaitForSingleEvent<TEvent>(this CancellationToken token, Action<TEvent> onEvent, Action<Action<TEvent>> subscribe, Action<Action<TEvent>> unsubscribe, int msTimeout, Action initializer = null) { 
    var task = Observable.FromEvent(subscribe, unsubscribe).FirstAsync().ToTask(); 
    if (initializer != null) { 
     initializer(); 
    } 
    try { 
     var finished = task.Wait(msTimeout, token); 
     if (finished) onEvent(task.Result); 
     return finished; 
    } catch (OperationCanceledException) { return false; } 
} 

Answer 2

Bạn có thể sử dụng để tạo ra một TaskCompletetionSourceTask mà bạn có thể đánh dấu là hoàn thành hoặc bị hủy bỏ.Dưới đây là một thực thể cho một sự kiện cụ thể:

public Task WaitFirstMyEvent(Foo target, CancellationToken cancellationToken) 
{ 
    var tcs = new TaskCompletionSource<object>(); 
    Action handler = null; 
    var registration = cancellationToken.Register(() => 
    { 
     target.MyEvent -= handler; 
     tcs.TrySetCanceled(); 
    }); 
    handler =() => 
    { 
     target.MyEvent -= handler; 
     registration.Dispose(); 
     tcs.TrySetResult(null); 
    }; 
    target.MyEvent += handler; 
    return tcs.Task; 
} 

Trong C# 5 bạn có thể sử dụng nó như thế này:

private async Task MyMethod() 
{ 
    ... 
    await WaitFirstMyEvent(foo, cancellationToken); 
    ... 
} 

Nếu bạn muốn chờ đợi cho sự kiện này một cách đồng bộ, bạn cũng có thể sử dụng phương pháp Wait :

private void MyMethod() 
{ 
    ... 
    WaitFirstMyEvent(foo, cancellationToken).Wait(); 
    ... 
} 

Dưới đây là một phiên bản chung chung hơn, nhưng nó vẫn chỉ hoạt động cho các sự kiện với Action chữ ký:

public Task WaitFirstEvent(
    Action<Action> subscribe, 
    Action<Action> unsubscribe, 
    CancellationToken cancellationToken) 
{ 
    var tcs = new TaskCompletionSource<object>(); 
    Action handler = null; 
    var registration = cancellationToken.Register(() => 
    { 
     unsubscribe(handler); 
     tcs.TrySetCanceled(); 
    }); 
    handler =() => 
    { 
     unsubscribe(handler); 
     registration.Dispose(); 
     tcs.TrySetResult(null); 
    }; 
    subscribe(handler); 
    return tcs.Task; 
} 

Bạn có thể sử dụng nó như thế này:

await WaitFirstEvent(
     handler => foo.MyEvent += handler, 
     handler => foo.MyEvent -= handler, 
     cancellationToken); 

Nếu bạn muốn nó để làm việc với chữ ký sự kiện khác (ví dụ EventHandler), bạn sẽ phải tạo ra quá tải riêng biệt. Tôi không nghĩ rằng có một cách dễ dàng để làm cho nó làm việc cho bất kỳ chữ ký, đặc biệt là kể từ khi số lượng các thông số không phải lúc nào cũng giống nhau.