#nullable enable using System; using System.Collections.Concurrent; using System.Collections.Generic; using System.Threading; using System.Threading.Tasks; namespace ShrinkEventBus { public delegate ValueTask ShrinkAsyncEventHandler(TEvent eventData, CancellationToken cancellationToken) where TEvent : IShrinkEvent; public interface IShrinkEventBus : IDisposable { ShrinkBusKey Key { get; } ShrinkBusOptions Options { get; } ShrinkPostResult Post(in TEvent eventData) where TEvent : IShrinkEvent; ValueTask PostAsync(TEvent eventData, CancellationToken cancellationToken = default) where TEvent : IShrinkEvent; IDisposable Attach(object target); } public interface IShrinkBusResolver { IShrinkEventBus GetBus(ShrinkBusKey key); bool TryGetBus(ShrinkBusKey key, out IShrinkEventBus bus); } public interface IShrinkGeneratedSubscriber { IDisposable AttachGenerated(IShrinkBusResolver resolver, ShrinkBusKey? defaultBus = null); } public sealed class ShrinkEventBinding : IDisposable { private readonly List _items = new List(); private bool _disposed; public void Add(IDisposable item) { if (item == null) throw new ArgumentNullException(nameof(item)); if (_disposed) { item.Dispose(); throw new ObjectDisposedException(nameof(ShrinkEventBinding)); } _items.Add(item); } public void Dispose() { if (_disposed) return; _disposed = true; for (var i = _items.Count - 1; i >= 0; i--) _items[i].Dispose(); _items.Clear(); } } public static class ShrinkStaticBindingRegistry { internal sealed class Entry { public long Id; public ShrinkBusKey Key; public Func Factory = null!; } private static readonly object Gate = new object(); private static readonly List Entries = new List(); private static readonly List> Hosts = new List>(); private static long _nextId; public static void Register(ShrinkBusKey key, Func factory) { if (factory == null) throw new ArgumentNullException(nameof(factory)); ShrinkEventBusHost[] hosts; Entry entry; lock (Gate) { entry = new Entry { Id = Interlocked.Increment(ref _nextId), Key = key, Factory = factory }; Entries.Add(entry); hosts = LiveHostsLocked(); } for (var i = 0; i < hosts.Length; i++) hosts[i].TryAttachStatic(entry); } internal static void TrackHost(ShrinkEventBusHost host) { lock (Gate) { LiveHostsLocked(); Hosts.Add(new WeakReference(host)); } } internal static Entry[] GetEntries(ShrinkBusKey key) { lock (Gate) return Entries.FindAll(item => item.Key == key).ToArray(); } private static ShrinkEventBusHost[] LiveHostsLocked() { var result = new List(Hosts.Count); for (var i = Hosts.Count - 1; i >= 0; i--) { if (Hosts[i].TryGetTarget(out var host)) result.Add(host); else Hosts.RemoveAt(i); } return result.ToArray(); } } public sealed class ShrinkEventBusHost : IShrinkBusResolver, IDisposable { private readonly ConcurrentDictionary _buses = new ConcurrentDictionary(); private readonly object _staticGate = new object(); private readonly Dictionary _staticBindings = new Dictionary(); private bool _disposed; public ShrinkEventBusHost() { ShrinkStaticBindingRegistry.TrackHost(this); } public IShrinkEventBus CreateBus(ShrinkBusKey key, ShrinkBusOptions options) { if (_disposed) throw new ObjectDisposedException(nameof(ShrinkEventBusHost)); var bus = new DotNetShrinkEventBus(key, options); if (!_buses.TryAdd(key, bus)) { bus.Dispose(); throw new InvalidOperationException($"Bus '{key}' is already registered."); } var entries = ShrinkStaticBindingRegistry.GetEntries(key); for (var i = 0; i < entries.Length; i++) TryAttachStatic(entries[i]); return bus; } public IShrinkEventBus GetBus(ShrinkBusKey key) { if (_buses.TryGetValue(key, out var bus)) return bus; throw new KeyNotFoundException($"Bus '{key}' is not registered."); } public bool TryGetBus(ShrinkBusKey key, out IShrinkEventBus bus) => _buses.TryGetValue(key, out bus!); public IDisposable Attach(object target, ShrinkBusKey? defaultBus = null) { if (target is not IShrinkGeneratedSubscriber generated) throw new InvalidOperationException( $"Type {target?.GetType().FullName ?? ""} has no generated event binding."); return generated.AttachGenerated(this, defaultBus); } public void Dispose() { if (_disposed) return; _disposed = true; lock (_staticGate) { foreach (var binding in _staticBindings.Values) binding.Dispose(); _staticBindings.Clear(); } foreach (var bus in _buses.Values) bus.Dispose(); _buses.Clear(); } internal void TryAttachStatic(ShrinkStaticBindingRegistry.Entry entry) { lock (_staticGate) { if (_disposed || _staticBindings.ContainsKey(entry.Id) || !_buses.ContainsKey(entry.Key)) return; _staticBindings.Add(entry.Id, entry.Factory(this)); } } } public static class ShrinkGeneratedBinding { public static IDisposable Subscribe(IShrinkBusResolver resolver, ShrinkBusKey? defaultBus, string? configuredBus, object? owner, Action handler, ShrinkEventPriority priority, int numericPriority, bool receiveCanceled) where TEvent : IShrinkEvent { return Resolve(resolver, defaultBus, configuredBus).Subscribe( owner, handler, priority, numericPriority, receiveCanceled); } public static IDisposable SubscribeAsync(IShrinkBusResolver resolver, ShrinkBusKey? defaultBus, string? configuredBus, object? owner, ShrinkAsyncEventHandler handler, ShrinkEventPriority priority, int numericPriority, bool receiveCanceled) where TEvent : IShrinkEvent { return Resolve(resolver, defaultBus, configuredBus).SubscribeAsync( owner, handler, priority, numericPriority, receiveCanceled); } private static DotNetShrinkEventBus Resolve(IShrinkBusResolver resolver, ShrinkBusKey? defaultBus, string? configuredBus) { var key = string.IsNullOrWhiteSpace(configuredBus) ? defaultBus ?? ShrinkBusKey.Game : ShrinkBusKey.Parse(configuredBus); return resolver.GetBus(key) as DotNetShrinkEventBus ?? throw new InvalidOperationException("Generated bindings require the built-in bus implementation."); } } internal sealed class DotNetShrinkEventBus : IShrinkEventBus { private readonly ConcurrentDictionary _channels = new ConcurrentDictionary(); private readonly IDotNetScheduler _scheduler; private long _nextSubscriptionId; private bool _disposed; public DotNetShrinkEventBus(ShrinkBusKey key, ShrinkBusOptions options) { Key = key; Options = (options ?? throw new ArgumentNullException(nameof(options))).CloneValidated(); _scheduler = DotNetSchedulerFactory.Create(key.Id, Options); } public ShrinkBusKey Key { get; } public ShrinkBusOptions Options { get; } public ShrinkPostResult Post(in TEvent eventData) where TEvent : IShrinkEvent { if (_disposed) return ShrinkPostResult.Rejected(ShrinkPostFailure.BusStopped); var copied = eventData; if (_scheduler.IsOnSchedulerThread) return Channel().Post(copied); return _scheduler.TryPost(() => Channel().Post(copied)) ? new ShrinkPostResult(true, false, false) : ShrinkPostResult.Rejected(ShrinkPostFailure.QueueFull); } public async ValueTask PostAsync(TEvent eventData, CancellationToken cancellationToken = default) where TEvent : IShrinkEvent { if (_disposed) return ShrinkPostResult.Rejected(ShrinkPostFailure.BusStopped); ShrinkPostResult result = default; try { await _scheduler.PostAsync(async () => { result = await Channel().PostAsync(eventData, Options.DispatchMode, cancellationToken) .ConfigureAwait(false); }, cancellationToken).ConfigureAwait(false); return result; } catch (QueueFullException) { return ShrinkPostResult.Rejected(ShrinkPostFailure.QueueFull); } catch (OperationCanceledException) when (cancellationToken.IsCancellationRequested) { return new ShrinkPostResult(true, false, false, ShrinkPostFailure.Canceled); } } public IDisposable Attach(object target) { if (target is not IShrinkGeneratedSubscriber generated) throw new InvalidOperationException( $"Type {target?.GetType().FullName ?? ""} has no generated event binding."); return generated.AttachGenerated(new SingleBusResolver(this), Key); } public IDisposable Subscribe(object? owner, Action handler, ShrinkEventPriority priority, int numericPriority, bool receiveCanceled) where TEvent : IShrinkEvent { return Channel().Add(Interlocked.Increment(ref _nextSubscriptionId), owner, handler, null, priority, numericPriority, receiveCanceled); } public IDisposable SubscribeAsync(object? owner, ShrinkAsyncEventHandler handler, ShrinkEventPriority priority, int numericPriority, bool receiveCanceled) where TEvent : IShrinkEvent { return Channel().Add(Interlocked.Increment(ref _nextSubscriptionId), owner, null, handler, priority, numericPriority, receiveCanceled); } public void Dispose() { if (_disposed) return; _disposed = true; _scheduler.Dispose(); foreach (var channel in _channels.Values) channel.Clear(); _channels.Clear(); } private EventChannel Channel() where TEvent : IShrinkEvent => (EventChannel)_channels.GetOrAdd(typeof(TEvent), _ => new EventChannel()); private sealed class SingleBusResolver : IShrinkBusResolver { private readonly IShrinkEventBus _bus; public SingleBusResolver(IShrinkEventBus bus) => _bus = bus; public IShrinkEventBus GetBus(ShrinkBusKey key) => key == _bus.Key ? _bus : throw new KeyNotFoundException($"Bus '{key}' is not available."); public bool TryGetBus(ShrinkBusKey key, out IShrinkEventBus bus) { bus = _bus; return key == _bus.Key; } } } internal interface IEventChannel { void Clear(); } internal sealed class EventChannel : IEventChannel where TEvent : IShrinkEvent { private sealed class Entry { public long Id = 0; public object? Owner; public Action? Sync; public ShrinkAsyncEventHandler? Async; public ShrinkEventPriority Priority; public int NumericPriority; public bool ReceiveCanceled; } private readonly object _gate = new object(); private readonly List _entries = new List(); private Entry[] _snapshot = Array.Empty(); public IDisposable Add(long id, object? owner, Action? sync, ShrinkAsyncEventHandler? asyncHandler, ShrinkEventPriority priority, int numericPriority, bool receiveCanceled) { lock (_gate) { _entries.Add(new Entry { Id = id, Owner = owner, Sync = sync, Async = asyncHandler, Priority = priority, NumericPriority = numericPriority, ReceiveCanceled = receiveCanceled }); Rebuild(); } return new Subscription(this, id); } public ShrinkPostResult Post(TEvent eventData) { var handled = false; var snapshot = _snapshot; for (var i = 0; i < snapshot.Length; i++) { var entry = snapshot[i]; if (IsCanceled(eventData) && !entry.ReceiveCanceled) continue; if (entry.Sync != null) entry.Sync(eventData); else if (entry.Async != null) _ = entry.Async(eventData, CancellationToken.None).AsTask(); else continue; handled = true; } return ShrinkPostResult.Completed(handled, IsCanceled(eventData)); } public async ValueTask PostAsync(TEvent eventData, ShrinkDispatchMode mode, CancellationToken cancellationToken) { var handled = false; var snapshot = _snapshot; if (mode == ShrinkDispatchMode.Parallel) { var tasks = new List(); for (var i = 0; i < snapshot.Length; i++) { var entry = snapshot[i]; if (IsCanceled(eventData) && !entry.ReceiveCanceled) continue; if (entry.Sync != null) entry.Sync(eventData); else if (entry.Async != null) tasks.Add(entry.Async(eventData, cancellationToken).AsTask()); else continue; handled = true; } if (tasks.Count > 0) await Task.WhenAll(tasks).ConfigureAwait(false); } else { for (var i = 0; i < snapshot.Length; i++) { cancellationToken.ThrowIfCancellationRequested(); var entry = snapshot[i]; if (IsCanceled(eventData) && !entry.ReceiveCanceled) continue; if (entry.Sync != null) entry.Sync(eventData); else if (entry.Async != null) await entry.Async(eventData, cancellationToken).ConfigureAwait(false); else continue; handled = true; } } return ShrinkPostResult.Completed(handled, IsCanceled(eventData)); } public void Clear() { lock (_gate) { _entries.Clear(); _snapshot = Array.Empty(); } } private void Remove(long id) { lock (_gate) { _entries.RemoveAll(item => item.Id == id); Rebuild(); } } private void Rebuild() { _entries.Sort((left, right) => { var leftValue = left.NumericPriority == 0 ? (int)left.Priority * 1000 : -left.NumericPriority; var rightValue = right.NumericPriority == 0 ? (int)right.Priority * 1000 : -right.NumericPriority; var result = leftValue.CompareTo(rightValue); return result != 0 ? result : left.Id.CompareTo(right.Id); }); _snapshot = _entries.ToArray(); } private static bool IsCanceled(TEvent value) => value is IShrinkCancelableEvent cancelable && cancelable.IsCanceled; private sealed class Subscription : IDisposable { private EventChannel? _owner; private readonly long _id; public Subscription(EventChannel owner, long id) { _owner = owner; _id = id; } public void Dispose() => Interlocked.Exchange(ref _owner, null)?.Remove(_id); } } }