Beobachtbares Ereignis - So warten Sie auf alle asynchronen Abonnements, die fertig sind ⇐ C#

[Anonymous]

Nehmen wir an, ich habe einen Singleton -Dispatcher mit einer Immobilie OnSubmit, die von mehreren Komponenten konsumiert werden sollte. Die Komponenten bieten asynchronisierte Handler für das OnSubmit -Ereignis. Das Ziel ist es, auf die Fertigstellung aller Handler zu warten.

Code: Select all

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>
Hier ist das Problem, wie ich das [url=viewtopic.php?t=11587]Problem[/url] gelöst habe, aber ich bin mir nicht sicher, ob es der beste Ansatz ist, um damit umzugehen. < /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
}
}