Наблюдаемое событие - как ждать всех асинхронных подписок на завершение ⇐ C#

[Anonymous]

Допустим, у меня есть диспетчер Singleton с недвижимостью, который должен использоваться несколькими компонентами. Компоненты предоставляют асинхронные обработчики для мероприятия по вспомогательству. Цель состоит в том, чтобы дождаться завершения всех обработчиков. < /P>
Вот пример кода: < /p>
public class Dispatcher
{
private readonly Subject _submitSubject = new();

public IObservable OnSubmit { get; }

private Dispatcher()
{
OnSubmit = _submitSubject.AsObservable().Publish().RefCount();
}

public static Dispatcher Instance { get; } = new();

public async Task SubmitAsync(int id)
{
_submitSubject.OnNext(id);
// await ???

// Here is the place where I would like to await the completion of all the subscribers
// Unfortunately I get here before the async handlers are completed since Rx.net isn't async by default
// https://github.com/dotnet/reactive#asyncrxnet
}
}

// Blazor componentA and other similar components
@code {
[Inject] private Dispatcher Dispatcher { get; set; } = Dispatcher.Instance;

protected override void OnInitialized()
{
Dispatcher.OnSubmit
.Select(id => Observable.FromAsync(() => HandleSubmitAsync(id)))
.Subscribe();
}

private async Task HandleSubmitAsync(int id)
{
await Task.Delay(new Random().Next(1000, 3000)); // Here will be real async operation
}

}
< /code>
Вот как я решил проблему, но я не уверен, что это лучший подход для ее обработки. < /p>
public sealed class ObservableEvent(T args)
{
private readonly ConcurrentQueue _handlers = new();

public T Args { get; private set; } = args;

public void RegisterHandler(Task handler)
{
ArgumentNullException.ThrowIfNull(handler);
_handlers.Enqueue(handler);
}

public async Task WhenAll() => await Task.WhenAll(_handlers);
}

public class Dispatcher
{
private readonly Subject _submitSubject = new();

public IObservable OnSubmit { get; }

private Dispatcher()
{
OnSubmit = _submitSubject.AsObservable().Publish().RefCount();
}

public static Dispatcher Instance { get; } = new();

public async Task SubmitAsync(int id)
{
ObservableEvent observableEvent = new(new TabActionContext(tab, Id));
_submitSubject.OnNext(observableEvent);
await observableEvent.WhenAll();
}
}

// Blazor componentA and other similar components
@code {
[Inject] private Dispatcher Dispatcher { get; set; } = Dispatcher.Instance;

protected override void OnInitialized()
{
Dispatcher.OnSubmit
.Subscribe(@event => @event.RegisterHandler(HandleSubmitAsync(@event.Args)));
}

private async Task HandleSubmitAsync(TabActionContext context)
{
await Task.Delay(new Random().Next(1000, 3000)); // Here will be real async operation
}
}


Подробнее здесь: https://stackoverflow.com/questions/794 ... -to-finish