using System.Collections.Concurrent; using System.Threading; using System.Threading.Tasks; using Microsoft.Extensions.Logging; namespace store { public class OrdersEventBroker { private ConcurrentDictionary _entries; private ILogger _logger; public OrdersEventBroker(ILogger logger) { _entries = new ConcurrentDictionary(); _logger = logger; } public async Task GetOrderConfirmationAsync(string orderId, CancellationToken cancellationToken) { _logger.LogInformation("Waiting for confirmation of order {OrderId}", orderId); var entry = new Entry(orderId); using (cancellationToken.Register(Cancel)) { _entries.TryAdd(orderId, entry); var state = await entry.Completion.Task; _entries.TryRemove(orderId, out _); _logger.LogInformation("Order {OrderId} has been processed", orderId); return state; } void Cancel() { _logger.LogInformation("Canceling subscription {OrderId}", orderId); _entries.TryRemove(orderId, out _); entry.Completion.TrySetCanceled(cancellationToken); } } public void Complete(OrderConfirmation result) { if (_entries.TryGetValue(result.OrderId, out var entry)) { _logger.LogInformation("Processing order {OrderId}", result.OrderId); entry.Completion.TrySetResult(result); } } private class Entry { public Entry(string orderId) { OrderId = orderId; Completion = new TaskCompletionSource(); } public string OrderId { get; } public TaskCompletionSource Completion { get; } } } }