c# - 使用从未发生的 `FirstAsync` 观察者任务避免资源泄漏

Question

我有一些软件使用基于事件的网络协议(protocol)进行控制，它使用 IObservable<Event>用于处理绑定(bind)消息。

在许多情况下，发送的消息需要特定的响应(或顺序，例如报告进度)。为了不至于错过响应，我们提前使用 FirstAsync 设置了一项任务。和 ToTask ，但是如果任务从未完成，这些似乎会泄漏。

也不允许简单的写evtTask在using阻止，因为不允许尝试处理未完成的任务。

var jobUuid = Guid.NewGuid();
var evtTask = Events.FirstAsync((x) => x.Action == Action.JobComplete && x.JobUuid == jobUuid).ToTask();
// e.g. if this throws without ever sending the message
await SendMessage($"job {jobUuid} download {url}");
var evt = await evtTask;
if (evt.Success)
{
    ...
}

库是否为这个用例提供了一种在离开范围时取消订阅的简单方法？

var jobUuid = Guid.NewGuid();
using(var evtTask = Events.FirstAsync((x) => x.Action == Action.JobComplete && x.JobUuid == jobUuid)
    .ToDisposableTask())) // Some method like this
{
    // e.g. if this throws without ever sending the message
    await SendMessage($"job {jobUuid} download {url}");
    var evt = await evtTask;
    if (evt.Success)
    {
        ...
    }
} // Get rid of the FirstAsync task if leave here before it completes for any reason

Answer 1

Disposing Task 无济于事，因为它没有任何用处(在大多数情况下，包括这种情况)。不过，取消任务会有帮助。取消处理由 ToTask 创建的基础订阅，因此解决了这个“泄漏”。

所以它可以像这样:

Task<Event> evtTask;
using (var cts = new CancellationTokenSource()) {
    evtTask = Events.FirstAsync((x) => x.Action == Action.JobComplete && x.JobUuid == jobUuid)
             .ToTask(cts.Token);
    // e.g. if this throws without ever sending the message
    try {
        await SendMessage($"job {jobUuid} download {url}");
    }
    catch {
        cts.Cancel(); // disposes subscription
        throw;
    }
}
var evt = await evtTask;
if (evt.Success)
{
    ...
}

当然你可以用一些更方便的形式包装它(比如扩展方法)。例如:

public static class ObservableExtensions {
    public static CancellableTaskWrapper<T> ToCancellableTask<T>(this IObservable<T> source) {
        return new CancellableTaskWrapper<T>(source);
    }

    public class CancellableTaskWrapper<T> : IDisposable
    {
        private readonly CancellationTokenSource _cts;
        public CancellableTaskWrapper(IObservable<T> source)
        {
            _cts = new CancellationTokenSource();
            Task = source.ToTask(_cts.Token);
        }

        public Task<T> Task { get; }

        public void Dispose()
        {
            _cts.Cancel();
            _cts.Dispose();
        }
    }
}

然后它变得接近你想要的:

var jobUuid = Guid.NewGuid();
using (var evtTask = Events.FirstAsync((x) => x.Action == Action.JobComplete && x.JobUuid == jobUuid).ToCancellableTask()) {
    await SendMessage($"job {jobUuid} download {url}");
    var evt = await evtTask.Task;
    if (evt.Success) {
        ...
    }
}