From 901ad18453dd20e27240cf003846645199e699fd Mon Sep 17 00:00:00 2001 From: cneicy Date: Fri, 6 Nov 2026 19:13:18 +0800 Subject: [PATCH] feat: update Godot integration and coordinated package dependencies --- CHANGELOG.md | 13 + DotNet~/ShrinkSDK.EventBus.csproj | 4 +- README.md | 27 +- Runtime/EventBus.cs | 59 +++- Runtime/ShrinkBusSchedulers.cs | 141 +++++++--- Runtime/ShrinkEventBus.Runtime.asmdef | 3 +- Runtime/ShrinkEventBusInstance.cs | 38 ++- Runtime/ShrinkEventBusRuntime.cs | 26 +- Runtime/ShrinkEventContracts.cs | 4 +- Runtime/ShrinkEventDiagnostic.cs | 37 +++ Runtime/ShrinkEventDiagnostic.cs.meta | 11 + Tests/ShrinkEventBus.Tests.asmdef | 1 + .../ShrinkEventBusDeferredDiagnosticsTests.cs | 251 ++++++++++++++++++ ...nkEventBusDeferredDiagnosticsTests.cs.meta | 11 + package.json | 5 +- 15 files changed, 576 insertions(+), 55 deletions(-) create mode 100644 Runtime/ShrinkEventDiagnostic.cs create mode 100644 Runtime/ShrinkEventDiagnostic.cs.meta create mode 100644 Tests/ShrinkEventBusDeferredDiagnosticsTests.cs create mode 100644 Tests/ShrinkEventBusDeferredDiagnosticsTests.cs.meta diff --git a/CHANGELOG.md b/CHANGELOG.md index c22aa56..9cc4b71 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -2,6 +2,19 @@ 本文件记录 `ShrinkEventBus` 在当前工作区中的包内变更。 +## [2.3.0] - 2026-11-06 + +### Added + +- 新增 `EventBus.PostDeferred(bus, event)`，把 MainThread Bus 的事件推迟到下一次主线程泵送；重入发布不会在同一次泵送中立即执行。 +- 新增公开的 `EventBus.ObserveDiagnostics(...)` 与 `ShrinkEventDiagnostic`，供编辑器和外部工具读取分发时间、Bus、调度器、线程、耗时、结果与同步标记，不再依赖私有事件或反射。 +- `ShrinkPostFailure` 新增 `DeferredUnsupported`，用于 `Inline`、`DedicatedThread`、`TaskPool` 或宿主没有延后派发能力的情况。 + +### Changed + +- Runtime Abstractions 0.2.0 新增 `IShrinkDeferredMainThreadDispatcher`；Unity PlayerLoop 与共享 `SynchronizationContext` 调度器实现该接口。 +- 诊断观察者逐个隔离异常，单个观察者失败不会阻断业务 handler 或其他观察者。 + ## [2.2.3] - 2026-10-03 - Unity 可独立通过 Git URL 安装，必装依赖仅保留 UniTask 和官方 Mono.Cecil。 diff --git a/DotNet~/ShrinkSDK.EventBus.csproj b/DotNet~/ShrinkSDK.EventBus.csproj index 67c10a5..d54d74a 100644 --- a/DotNet~/ShrinkSDK.EventBus.csproj +++ b/DotNet~/ShrinkSDK.EventBus.csproj @@ -5,7 +5,7 @@ ShrinkEventBus.Runtime ShrinkEventBus ShrinkSDK.EventBus - 2.2.2 + 2.3.1 ShrinkSDK typed event bus runtime. false @@ -14,6 +14,6 @@ Exclude="..\Runtime\ShrinkMonoEventScope.cs" /> - + diff --git a/README.md b/README.md index 673e7e5..d5af685 100644 --- a/README.md +++ b/README.md @@ -20,7 +20,7 @@ https://github.com/cneicy/ShrinkEventBus.git | 已安装的共享包 | 行为 | |---|---| | 无 | 使用包内调度接口和 EventBus 专用 ILPostProcessor | -| Runtime Abstractions 0.1.0+ | 额外启用共享主线程调度器适配程序集 | +| Runtime Abstractions 0.2.0+ | 额外启用共享主线程调度器适配程序集，并支持显式延后发布 | | Shared CodeGen 0.2.1+ | 停用包内织入，交给共享 CodeGen 统一生成订阅 | | 两者都有 | 同时启用上述两项集成 | @@ -199,6 +199,29 @@ Burst Job 只写 unmanaged 事实事件到 NativeQueue；Playback 在托管/ECS `ShrinkBusOptions` 配置队列容量、溢出策略、最大并发、关闭排空和超时。关键业务默认使用 `Reject`；`DropNewest/DropOldest` 只适用于明确允许丢失的数据；`Wait` 只由 `PostAsync` 使用。 +## 延后发布与诊断 + +MainThread Bus 可以用 `PostDeferred` 把事件推迟到下一次主线程泵送： + +```csharp +var result = EventBus.PostDeferred(gameBus, new PlayerReadyEvent()); +if (!result.Accepted) + Debug.LogWarning($"deferred post rejected: {result.Failure}"); +``` + +延后发布只接受 `MainThread` 调度且宿主实现了 `IShrinkDeferredMainThreadDispatcher` 的 Bus；其他调度器返回 `DeferredUnsupported`，不会退化成同步分发。队列仍遵守 `QueueCapacity` 与 `OverflowPolicy`，重入发布的事件留到下一次泵送。 + +需要外部工具观察分发时使用公开诊断订阅： + +```csharp +using var subscription = EventBus.ObserveDiagnostics(diagnostic => +{ + Debug.Log($"{diagnostic.BusKey} {diagnostic.EventType.Name} {diagnostic.Result.Failure}"); +}); +``` + +`ShrinkEventDiagnostic` 提供时间、事件类型、Bus、调度器、分发模式、线程、耗时、发布结果和同步标记，不暴露事件对象。观察者抛出的异常会记录并隔离，不影响业务分发或其他观察者。 + ## UI Toolkit 调试器 菜单：`ShrinkSDK/事件总线/事件查看器`。 @@ -207,6 +230,8 @@ Burst Job 只写 unmanaged 事实事件到 NativeQueue；Playback 在托管/ECS 它只观察静态 `EventBus` 宿主管理的 Bus，不捕获独立 `ShrinkEventBusHost`；性能基准仍使用独立 Host，避免 UI 采样进入被测热路径。 +Godot 项目安装 `ShrinkSDK.EventBus.Godot` 后可在编辑器中使用同一套诊断数据的独立调试窗口。 + ## Benchmark 实现位于 `Benchmark/ShrinkEventBusBenchmark.cs`，入口： diff --git a/Runtime/EventBus.cs b/Runtime/EventBus.cs index caea5f3..0b3f4b3 100644 --- a/Runtime/EventBus.cs +++ b/Runtime/EventBus.cs @@ -133,6 +133,30 @@ namespace ShrinkEventBus CancellationToken cancellationToken = default) where TEvent : IShrinkEvent => DefaultBus.PostAsync(eventData, cancellationToken); + public static ShrinkPostResult PostDeferred(IShrinkEventBus bus, TEvent eventData) + where TEvent : IShrinkEvent + { + if (bus == null) + throw new ArgumentNullException(nameof(bus)); + return bus is ShrinkEventBusInstance instance + ? instance.PostDeferred(in eventData) + : ShrinkPostResult.Rejected(ShrinkPostFailure.DeferredUnsupported); + } + + public static IDisposable ObserveDiagnostics(Action observer) + { + if (observer == null) + throw new ArgumentNullException(nameof(observer)); + + void Handler(ShrinkEventTrace trace) => observer(new ShrinkEventDiagnostic( + trace.TimestampUtc, trace.EventType, trace.BusKey, trace.Scheduler, + trace.DispatchMode, trace.ThreadId, trace.ElapsedTimestampTicks, + trace.Result, trace.IsAsync)); + + DetailedPosted += Handler; + return new DiagnosticSubscription(Handler); + } + public static IDisposable Attach(object target, ShrinkBusKey? defaultBus = null) { if (target == null) @@ -191,17 +215,38 @@ namespace ShrinkEventBus var observers = Volatile.Read(ref _detailedPosted); if (observers == null) return; - try - { - observers(trace); - } - catch (Exception exception) + var invocationList = observers.GetInvocationList(); + for (var i = 0; i < invocationList.Length; i++) { + try + { + ((Action)invocationList[i])(trace); + } + catch (Exception exception) + { #if UNITY_5_3_OR_NEWER - UnityEngine.Debug.LogException(exception); + UnityEngine.Debug.LogException(exception); #else - System.Diagnostics.Trace.TraceError(exception.ToString()); + System.Diagnostics.Trace.TraceError(exception.ToString()); #endif + } + } + } + + private sealed class DiagnosticSubscription : IDisposable + { + private Action? _handler; + + public DiagnosticSubscription(Action handler) + { + _handler = handler; + } + + public void Dispose() + { + var handler = Interlocked.Exchange(ref _handler, null); + if (handler != null) + DetailedPosted -= handler; } } } diff --git a/Runtime/ShrinkBusSchedulers.cs b/Runtime/ShrinkBusSchedulers.cs index 5435d5b..98dcb15 100644 --- a/Runtime/ShrinkBusSchedulers.cs +++ b/Runtime/ShrinkBusSchedulers.cs @@ -1,6 +1,7 @@ #nullable enable using System; +using System.Buffers; using System.Collections.Concurrent; using System.Diagnostics; using System.Threading; @@ -17,11 +18,22 @@ namespace ShrinkEventBus internal interface IShrinkBusScheduler : IDisposable { bool IsOnSchedulerThread { get; } + bool SupportsDeferred { get; } bool TryPost(Action action); + ShrinkDeferredPostStatus TryPostDeferred(Action action); UniTask PostAsync(Func action, CancellationToken cancellationToken); UniTask ShutdownAsync(bool drain, TimeSpan timeout); } + internal enum ShrinkDeferredPostStatus + { + Accepted, + QueueFull, + SchedulerRejected, + Unsupported, + Stopped + } + internal sealed class ShrinkQueueFullException : InvalidOperationException { public ShrinkQueueFullException(string schedulerName) @@ -81,6 +93,7 @@ namespace ShrinkEventBus } public bool IsEmpty => _queue.IsEmpty; + public int Count => _queue.Count; public bool TryEnqueue(ShrinkSchedulerWorkItem item) { @@ -192,6 +205,7 @@ namespace ShrinkEventBus internal sealed class ShrinkInlineScheduler : IShrinkBusScheduler { public bool IsOnSchedulerThread => true; + public bool SupportsDeferred => false; public bool TryPost(Action action) { @@ -199,6 +213,9 @@ namespace ShrinkEventBus return true; } + public ShrinkDeferredPostStatus TryPostDeferred(Action action) => + ShrinkDeferredPostStatus.Unsupported; + public UniTask PostAsync(Func action, CancellationToken cancellationToken) { cancellationToken.ThrowIfCancellationRequested(); @@ -213,36 +230,53 @@ namespace ShrinkEventBus { private readonly ShrinkSchedulerQueue _queue; private readonly MainThreadDispatcherContract _dispatcher; + private readonly ShrinkSDK.Runtime.IShrinkDeferredMainThreadDispatcher? _deferredDispatcher; + private readonly object _gate = new(); private int _pumpScheduled; private int _disposed; + private bool _discardPending; public ShrinkMainThreadScheduler(string name, ShrinkBusOptions options) { _queue = new ShrinkSchedulerQueue(name, options); _dispatcher = ShrinkEventBusRuntime.MainThreadDispatcher; + _deferredDispatcher = _dispatcher as ShrinkSDK.Runtime.IShrinkDeferredMainThreadDispatcher; } public bool IsOnSchedulerThread => _dispatcher.IsMainThread; + public bool SupportsDeferred => _deferredDispatcher != null; public bool TryPost(Action action) { - if (Volatile.Read(ref _disposed) != 0) - return false; + if (Volatile.Read(ref _disposed) != 0) return false; if (IsOnSchedulerThread) { action(); return true; } + return Enqueue(action) == ShrinkDeferredPostStatus.Accepted; + } - var item = new ShrinkSchedulerWorkItem(() => + public ShrinkDeferredPostStatus TryPostDeferred(Action action) + { + if (action == null) throw new ArgumentNullException(nameof(action)); + if (_deferredDispatcher == null) return ShrinkDeferredPostStatus.Unsupported; + return Enqueue(action); + } + + private ShrinkDeferredPostStatus Enqueue(Action action) + { + lock (_gate) { - action(); - return UniTask.CompletedTask; - }, awaitable: false); - if (!_queue.TryEnqueue(item)) - return false; - SchedulePump(); - return true; + if (_disposed != 0) return ShrinkDeferredPostStatus.Stopped; + var item = new ShrinkSchedulerWorkItem(() => + { + action(); + return UniTask.CompletedTask; + }, awaitable: false); + if (!_queue.TryEnqueue(item)) return ShrinkDeferredPostStatus.QueueFull; + return SchedulePump() ? ShrinkDeferredPostStatus.Accepted : ShrinkDeferredPostStatus.SchedulerRejected; + } } public async UniTask PostAsync(Func action, CancellationToken cancellationToken) @@ -255,19 +289,24 @@ namespace ShrinkEventBus await action(); return; } - var item = new ShrinkSchedulerWorkItem(action, awaitable: true); await _queue.EnqueueAsync(item, cancellationToken); - SchedulePump(); + lock (_gate) + { + if (_disposed != 0) _queue.DropPending(); + else SchedulePump(); + } await item.Completion.AttachExternalCancellation(cancellationToken); } public async UniTask ShutdownAsync(bool drain, TimeSpan timeout) { - Interlocked.Exchange(ref _disposed, 1); - if (!drain) - _queue.DropPending(); - else + lock (_gate) + { + _disposed = 1; + if (!drain) { _discardPending = true; _queue.DropPending(); } + } + if (drain) await ShrinkSchedulerShutdown.WaitUntilDrainedAsync( _queue, () => Volatile.Read(ref _pumpScheduled) != 0, timeout); _queue.Dispose(); @@ -275,34 +314,66 @@ namespace ShrinkEventBus public void Dispose() { - if (Interlocked.Exchange(ref _disposed, 1) != 0) - return; - _queue.Dispose(); + lock (_gate) + { + if (_disposed != 0) return; + _disposed = 1; + _discardPending = true; + _queue.Dispose(); + } } - private void SchedulePump() + // All queued work shares one FIFO and capacity. Prefer the deferred dispatcher + // so a normal worker-thread post cannot pull deferred events into an inline pump. + private bool SchedulePump() { - if (Interlocked.Exchange(ref _pumpScheduled, 1) != 0) - return; - if (_dispatcher.TryPost(() => PumpAsync().Forget())) - return; - Interlocked.Exchange(ref _pumpScheduled, 0); - ShrinkEventDiagnostics.LogException(new InvalidOperationException( - "The configured main-thread dispatcher rejected an EventBus pump.")); + if (_pumpScheduled != 0) return true; + _pumpScheduled = 1; + bool accepted; + try + { + accepted = _deferredDispatcher != null + ? _deferredDispatcher.TryPostDeferred(() => PumpAsync().Forget()) + : _dispatcher.TryPost(() => PumpAsync().Forget()); + } + catch + { + _pumpScheduled = 0; + _queue.DropPending(); + throw; + } + if (accepted) return true; + _pumpScheduled = 0; + _queue.DropPending(); + return false; } private async UniTask PumpAsync() { + ShrinkSchedulerWorkItem[] batch; + var count = 0; + lock (_gate) + { + var capacity = _queue.Count; + batch = ArrayPool.Shared.Rent(Math.Max(1, capacity)); + while (count < capacity && _queue.TryDequeue(out var item)) batch[count++] = item; + } try { - while (_queue.TryDequeue(out var item)) - await ShrinkSchedulerWorkItemRunner.RunAsync(item); + for (var i = 0; i < count; i++) + { + if (_discardPending) batch[i].Drop(nameof(ShrinkMainThreadScheduler)); + else await ShrinkSchedulerWorkItemRunner.RunAsync(batch[i]); + } } finally { - Interlocked.Exchange(ref _pumpScheduled, 0); - if (!_queue.IsEmpty && Volatile.Read(ref _disposed) == 0) - SchedulePump(); + ArrayPool.Shared.Return(batch, clearArray: true); + lock (_gate) + { + _pumpScheduled = 0; + if (!_queue.IsEmpty) SchedulePump(); + } } } } @@ -321,6 +392,7 @@ namespace ShrinkEventBus } public bool IsOnSchedulerThread => false; + public bool SupportsDeferred => false; public bool TryPost(Action action) { @@ -337,6 +409,8 @@ namespace ShrinkEventBus return true; } + public ShrinkDeferredPostStatus TryPostDeferred(Action action) => ShrinkDeferredPostStatus.Unsupported; + public async UniTask PostAsync(Func action, CancellationToken cancellationToken) { if (Volatile.Read(ref _disposed) != 0) @@ -417,6 +491,7 @@ namespace ShrinkEventBus } public bool IsOnSchedulerThread => Thread.CurrentThread.ManagedThreadId == Volatile.Read(ref _threadId); + public bool SupportsDeferred => false; public bool TryPost(Action action) { @@ -439,6 +514,8 @@ namespace ShrinkEventBus return true; } + public ShrinkDeferredPostStatus TryPostDeferred(Action action) => ShrinkDeferredPostStatus.Unsupported; + public async UniTask PostAsync(Func action, CancellationToken cancellationToken) { if (Volatile.Read(ref _disposed) != 0) diff --git a/Runtime/ShrinkEventBus.Runtime.asmdef b/Runtime/ShrinkEventBus.Runtime.asmdef index 513bd1d..d6c2660 100644 --- a/Runtime/ShrinkEventBus.Runtime.asmdef +++ b/Runtime/ShrinkEventBus.Runtime.asmdef @@ -2,7 +2,8 @@ "name": "ShrinkEventBus.Runtime", "rootNamespace": "ShrinkEventBus", "references": [ - "UniTask" + "UniTask", + "ShrinkRuntime.Abstractions" ], "includePlatforms": [], "excludePlatforms": [], diff --git a/Runtime/ShrinkEventBusInstance.cs b/Runtime/ShrinkEventBusInstance.cs index 5465f2a..a028972 100644 --- a/Runtime/ShrinkEventBusInstance.cs +++ b/Runtime/ShrinkEventBusInstance.cs @@ -87,6 +87,30 @@ namespace ShrinkEventBus return PostScheduled(in eventData); } + internal ShrinkPostResult PostDeferred(in TEvent eventData) where TEvent : IShrinkEvent + { + if (!typeof(TEvent).IsValueType && eventData is null) + throw new ArgumentNullException(nameof(eventData)); + if (Volatile.Read(ref _disposed) != 0) + return TraceWithoutDispatch(in eventData, + ShrinkPostResult.Rejected(ShrinkPostFailure.BusStopped), false); + if (Options.Scheduler != ShrinkBusSchedulerKind.MainThread || !_scheduler.SupportsDeferred) + return TraceWithoutDispatch(in eventData, + ShrinkPostResult.Rejected(ShrinkPostFailure.DeferredUnsupported), false); + + var queuedEvent = eventData; + var status = _scheduler.TryPostDeferred(() => Dispatch(in queuedEvent)); + if (status == ShrinkDeferredPostStatus.Accepted) return new ShrinkPostResult(true, false, false); + var failure = status switch + { + ShrinkDeferredPostStatus.Unsupported => ShrinkPostFailure.DeferredUnsupported, + ShrinkDeferredPostStatus.Stopped => ShrinkPostFailure.BusStopped, + ShrinkDeferredPostStatus.SchedulerRejected => ShrinkPostFailure.SchedulerRejected, + _ => ShrinkPostFailure.QueueFull + }; + return TraceWithoutDispatch(in eventData, ShrinkPostResult.Rejected(failure), false); + } + private ShrinkPostResult PostScheduled(in TEvent eventData) where TEvent : IShrinkEvent { var queuedEvent = eventData; @@ -207,7 +231,7 @@ namespace ShrinkEventBus [MethodImpl(MethodImplOptions.AggressiveInlining)] protected ShrinkPostResult Dispatch(in TEvent eventData) where TEvent : IShrinkEvent { - if (_postObserver != null && EventBus.HasDetailedPostedObservers) + if (EventBus.HasDetailedPostedObservers) return DispatchObserved(in eventData); var channel = ShrinkEventChannelSlots.Get(_busSlot); @@ -222,7 +246,7 @@ namespace ShrinkEventBus protected async UniTask DispatchAsync(TEvent eventData, CancellationToken cancellationToken) where TEvent : IShrinkEvent { - if (_postObserver != null && EventBus.HasDetailedPostedObservers) + if (EventBus.HasDetailedPostedObservers) return await DispatchAsyncObserved(eventData, cancellationToken); var channel = ShrinkEventChannelSlots.Get(_busSlot); @@ -258,8 +282,8 @@ namespace ShrinkEventBus } var elapsedTicks = Stopwatch.GetTimestamp() - startedAt; - if (EventBus.HasPostedObservers) - _postObserver!(eventData, typeof(TEvent), Key); + if (_postObserver != null && EventBus.HasPostedObservers) + _postObserver(eventData, typeof(TEvent), Key); EventBus.NotifyDetailedPosted(new ShrinkEventTrace(DateTime.UtcNow, typeof(TEvent), Key, Options, threadId, elapsedTicks, result, false)); return result; @@ -269,7 +293,7 @@ namespace ShrinkEventBus protected ShrinkPostResult TraceWithoutDispatch(in TEvent eventData, ShrinkPostResult result, bool isAsync) where TEvent : IShrinkEvent { - if (_postObserver != null && EventBus.HasDetailedPostedObservers) + if (EventBus.HasDetailedPostedObservers) { EventBus.NotifyDetailedPosted(new ShrinkEventTrace(DateTime.UtcNow, typeof(TEvent), Key, Options, Thread.CurrentThread.ManagedThreadId, 0L, result, isAsync)); @@ -309,8 +333,8 @@ namespace ShrinkEventBus } var elapsedTicks = Stopwatch.GetTimestamp() - startedAt; - if (EventBus.HasPostedObservers) - _postObserver!(eventData, typeof(TEvent), Key); + if (_postObserver != null && EventBus.HasPostedObservers) + _postObserver(eventData, typeof(TEvent), Key); EventBus.NotifyDetailedPosted(new ShrinkEventTrace(DateTime.UtcNow, typeof(TEvent), Key, Options, threadId, elapsedTicks, result, true)); return result; diff --git a/Runtime/ShrinkEventBusRuntime.cs b/Runtime/ShrinkEventBusRuntime.cs index 756c0c6..464b511 100644 --- a/Runtime/ShrinkEventBusRuntime.cs +++ b/Runtime/ShrinkEventBusRuntime.cs @@ -37,7 +37,8 @@ namespace ShrinkEventBus } #if UNITY_5_3_OR_NEWER - internal sealed class ShrinkUnityMainThreadDispatcher : MainThreadDispatcherContract + internal sealed class ShrinkUnityMainThreadDispatcher : MainThreadDispatcherContract, + ShrinkSDK.Runtime.IShrinkDeferredMainThreadDispatcher { public bool IsMainThread => Cysharp.Threading.Tasks.PlayerLoopHelper.IsMainThread; @@ -50,9 +51,20 @@ namespace ShrinkEventBus action); return true; } + + public bool TryPostDeferred(Action action) + { + if (action == null) + throw new ArgumentNullException(nameof(action)); + Cysharp.Threading.Tasks.PlayerLoopHelper.AddContinuation( + Cysharp.Threading.Tasks.PlayerLoopTiming.Update, + action); + return true; + } } #else - internal sealed class ShrinkSynchronizationContextDispatcher : MainThreadDispatcherContract + internal sealed class ShrinkSynchronizationContextDispatcher : MainThreadDispatcherContract, + ShrinkSDK.Runtime.IShrinkDeferredMainThreadDispatcher { private readonly SynchronizationContext? _context; private readonly int _threadId; @@ -79,6 +91,16 @@ namespace ShrinkEventBus _context.Post(static state => ((Action)state!).Invoke(), action); return true; } + + public bool TryPostDeferred(Action action) + { + if (action == null) + throw new ArgumentNullException(nameof(action)); + if (_context == null) + return false; + _context.Post(static state => ((Action)state!).Invoke(), action); + return true; + } } #endif } diff --git a/Runtime/ShrinkEventContracts.cs b/Runtime/ShrinkEventContracts.cs index 0fe5cab..0a3dd0b 100644 --- a/Runtime/ShrinkEventContracts.cs +++ b/Runtime/ShrinkEventContracts.cs @@ -61,7 +61,9 @@ namespace ShrinkEventBus QueueFull = 2, Canceled = 3, HandlerException = 4, - InvalidEvent = 5 + InvalidEvent = 5, + DeferredUnsupported = 6, + SchedulerRejected = 7 } public readonly struct ShrinkPostResult diff --git a/Runtime/ShrinkEventDiagnostic.cs b/Runtime/ShrinkEventDiagnostic.cs new file mode 100644 index 0000000..0ddab20 --- /dev/null +++ b/Runtime/ShrinkEventDiagnostic.cs @@ -0,0 +1,37 @@ +#nullable enable + +using System; +using System.Diagnostics; + +namespace ShrinkEventBus +{ + public readonly struct ShrinkEventDiagnostic + { + internal ShrinkEventDiagnostic(DateTime timestampUtc, Type eventType, ShrinkBusKey busKey, + ShrinkBusSchedulerKind scheduler, ShrinkDispatchMode dispatchMode, int threadId, + long elapsedTimestampTicks, ShrinkPostResult result, bool isAsync) + { + TimestampUtc = timestampUtc; + EventType = eventType; + BusKey = busKey; + Scheduler = scheduler; + DispatchMode = dispatchMode; + ThreadId = threadId; + ElapsedTimestampTicks = elapsedTimestampTicks; + Result = result; + IsAsync = isAsync; + } + + public DateTime TimestampUtc { get; } + public Type EventType { get; } + public ShrinkBusKey BusKey { get; } + public ShrinkBusSchedulerKind Scheduler { get; } + public ShrinkDispatchMode DispatchMode { get; } + public int ThreadId { get; } + public long ElapsedTimestampTicks { get; } + public double ElapsedMilliseconds => ElapsedTimestampTicks * 1000d / Stopwatch.Frequency; + public TimeSpan Elapsed => TimeSpan.FromSeconds(ElapsedTimestampTicks / (double)Stopwatch.Frequency); + public ShrinkPostResult Result { get; } + public bool IsAsync { get; } + } +} diff --git a/Runtime/ShrinkEventDiagnostic.cs.meta b/Runtime/ShrinkEventDiagnostic.cs.meta new file mode 100644 index 0000000..ee5ff09 --- /dev/null +++ b/Runtime/ShrinkEventDiagnostic.cs.meta @@ -0,0 +1,11 @@ +fileFormatVersion: 2 +guid: 0dd120bc74194af29d7cbd1f8c80c141 +MonoImporter: + externalObjects: {} + serializedVersion: 2 + defaultReferences: [] + executionOrder: 0 + icon: {instanceID: 0} + userData: + assetBundleName: + assetBundleVariant: diff --git a/Tests/ShrinkEventBus.Tests.asmdef b/Tests/ShrinkEventBus.Tests.asmdef index 69004b2..c670588 100644 --- a/Tests/ShrinkEventBus.Tests.asmdef +++ b/Tests/ShrinkEventBus.Tests.asmdef @@ -3,6 +3,7 @@ "rootNamespace": "ShrinkEventBus.Tests", "references": [ "ShrinkEventBus.Runtime", + "ShrinkRuntime.Abstractions", "UniTask", "UnityEngine.TestRunner", "UnityEditor.TestRunner" diff --git a/Tests/ShrinkEventBusDeferredDiagnosticsTests.cs b/Tests/ShrinkEventBusDeferredDiagnosticsTests.cs new file mode 100644 index 0000000..bac92a3 --- /dev/null +++ b/Tests/ShrinkEventBusDeferredDiagnosticsTests.cs @@ -0,0 +1,251 @@ +using System; +using System.Collections.Generic; +using NUnit.Framework; +using UnityEngine; +using UnityEngine.TestTools; + +namespace ShrinkEventBus.Tests +{ + [TestFixture] + public class ShrinkEventBusDeferredDiagnosticsTests + { + private IShrinkEventBusMainThreadDispatcher _originalDispatcher; + + [SetUp] + public void SetUp() + { + _originalDispatcher = ShrinkEventBusRuntime.MainThreadDispatcher; + } + + [TearDown] + public void TearDown() + { + ShrinkEventBusRuntime.ConfigureMainThreadDispatcher(_originalDispatcher); + } + + [Test] + public void ObserveDiagnostics_ReportsMetadataAndUnsubscribes() + { + var diagnostics = new List(); + using var host = new ShrinkEventBusHost(); + var key = new ShrinkBusKey("diagnostics", Guid.NewGuid().ToString("N")); + var bus = host.CreateBus(key, ShrinkBusOptions.Inline()); + + var subscription = EventBus.ObserveDiagnostics(diagnostics.Add); + bus.Post(new DeferredEvent(1)); + subscription.Dispose(); + bus.Post(new DeferredEvent(2)); + + Assert.AreEqual(1, diagnostics.Count); + Assert.AreEqual(typeof(DeferredEvent), diagnostics[0].EventType); + Assert.AreEqual(key, diagnostics[0].BusKey); + Assert.IsTrue(diagnostics[0].Result.Accepted); + Assert.IsFalse(diagnostics[0].IsAsync); + Assert.GreaterOrEqual(diagnostics[0].ElapsedTimestampTicks, 0L); + } + + [Test] + public void ObserveDiagnostics_ThrowingObserverDoesNotBlockBusinessOrOtherObservers() + { + var calls = 0; + var secondObserverCalls = 0; + using var host = new ShrinkEventBusHost(); + var key = new ShrinkBusKey("diagnostics", Guid.NewGuid().ToString("N")); + var bus = host.CreateBus(key, ShrinkBusOptions.Inline()); + using var binding = ShrinkGeneratedBinding.Subscribe(host, key, string.Empty, + this, _ => calls++, ShrinkEventPriority.Normal, 0, false); + using var first = EventBus.ObserveDiagnostics(_ => + throw new InvalidOperationException("observer failure")); + using var second = EventBus.ObserveDiagnostics(_ => secondObserverCalls++); + + var previous = LogAssert.ignoreFailingMessages; + LogAssert.ignoreFailingMessages = true; + try + { + Assert.DoesNotThrow(() => bus.Post(new DeferredEvent(3))); + } + finally + { + LogAssert.ignoreFailingMessages = previous; + } + + Assert.AreEqual(1, calls); + Assert.AreEqual(1, secondObserverCalls); + } + + [Test] + public void PostDeferred_DoesNotDispatchBeforePump() + { + var dispatcher = new ManualDispatcher(); + ShrinkEventBusRuntime.ConfigureMainThreadDispatcher(dispatcher); + using var host = new ShrinkEventBusHost(); + var key = new ShrinkBusKey("deferred", Guid.NewGuid().ToString("N")); + var bus = host.CreateBus(key, new ShrinkBusOptions + { + Scheduler = ShrinkBusSchedulerKind.MainThread, + QueueCapacity = 8 + }); + var calls = 0; + using var binding = ShrinkGeneratedBinding.Subscribe(host, key, string.Empty, + this, _ => calls++, ShrinkEventPriority.Normal, 0, false); + + var result = EventBus.PostDeferred(bus, new DeferredEvent(4)); + + Assert.IsTrue(result.Accepted); + Assert.IsFalse(result.Handled); + Assert.AreEqual(0, calls); + Assert.AreEqual(1, dispatcher.Count); + + dispatcher.PumpAll(); + Assert.AreEqual(1, calls); + } + + [Test] + public void PostDeferred_ReentrantPostWaitsForNextPump() + { + var dispatcher = new ManualDispatcher(); + ShrinkEventBusRuntime.ConfigureMainThreadDispatcher(dispatcher); + using var host = new ShrinkEventBusHost(); + var key = new ShrinkBusKey("deferred", Guid.NewGuid().ToString("N")); + var bus = host.CreateBus(key, new ShrinkBusOptions + { + Scheduler = ShrinkBusSchedulerKind.MainThread, + QueueCapacity = 8 + }); + var calls = 0; + using var binding = ShrinkGeneratedBinding.Subscribe(host, key, string.Empty, + this, _ => + { + calls++; + if (calls == 1) + EventBus.PostDeferred(bus, new DeferredEvent(5)); + }, ShrinkEventPriority.Normal, 0, false); + + EventBus.PostDeferred(bus, new DeferredEvent(4)); + dispatcher.PumpAll(); + Assert.AreEqual(1, calls); + Assert.AreEqual(1, dispatcher.Count); + + dispatcher.PumpAll(); + Assert.AreEqual(2, calls); + } + + [Test] + public void PostDeferred_DropOldestKeepsNewest() + { + var dispatcher = new ManualDispatcher(); + ShrinkEventBusRuntime.ConfigureMainThreadDispatcher(dispatcher); + using var host = new ShrinkEventBusHost(); + var key = new ShrinkBusKey("deferred", Guid.NewGuid().ToString("N")); + var bus = host.CreateBus(key, new ShrinkBusOptions + { + Scheduler = ShrinkBusSchedulerKind.MainThread, + QueueCapacity = 1, + OverflowPolicy = ShrinkQueueOverflowPolicy.DropOldest + }); + var values = new List(); + using var binding = ShrinkGeneratedBinding.Subscribe(host, key, string.Empty, + this, evt => values.Add(evt.Value), ShrinkEventPriority.Normal, 0, false); + + EventBus.PostDeferred(bus, new DeferredEvent(1)); + EventBus.PostDeferred(bus, new DeferredEvent(2)); + dispatcher.PumpAll(); + + CollectionAssert.AreEqual(new[] { 2 }, values); + } + + [Test] + public void PostDeferred_InlineBusFailsWithoutDispatch() + { + using var host = new ShrinkEventBusHost(); + var key = new ShrinkBusKey("deferred", Guid.NewGuid().ToString("N")); + var bus = host.CreateBus(key, ShrinkBusOptions.Inline()); + var calls = 0; + using var binding = ShrinkGeneratedBinding.Subscribe(host, key, string.Empty, + this, _ => calls++, ShrinkEventPriority.Normal, 0, false); + + var result = EventBus.PostDeferred(bus, new DeferredEvent(6)); + + Assert.AreEqual(ShrinkPostFailure.DeferredUnsupported, result.Failure); + Assert.AreEqual(0, calls); + } + + [Test] + public void PostDeferred_WithoutDeferredDispatcherFailsWithoutDispatch() + { + ShrinkEventBusRuntime.ConfigureMainThreadDispatcher(new NonDeferredDispatcher()); + using var host = new ShrinkEventBusHost(); + var key = new ShrinkBusKey("deferred", Guid.NewGuid().ToString("N")); + var bus = host.CreateBus(key, new ShrinkBusOptions + { + Scheduler = ShrinkBusSchedulerKind.MainThread + }); + var calls = 0; + using var binding = ShrinkGeneratedBinding.Subscribe(host, key, string.Empty, + this, _ => calls++, ShrinkEventPriority.Normal, 0, false); + + var result = EventBus.PostDeferred(bus, new DeferredEvent(7)); + + Assert.AreEqual(ShrinkPostFailure.DeferredUnsupported, result.Failure); + Assert.AreEqual(0, calls); + } + + [Test] + public void PostDeferred_DisposedBusRejects() + { + var dispatcher = new ManualDispatcher(); + ShrinkEventBusRuntime.ConfigureMainThreadDispatcher(dispatcher); + using var host = new ShrinkEventBusHost(); + var key = new ShrinkBusKey("deferred", Guid.NewGuid().ToString("N")); + var bus = host.CreateBus(key, new ShrinkBusOptions + { + Scheduler = ShrinkBusSchedulerKind.MainThread + }); + bus.Dispose(); + + var result = EventBus.PostDeferred(bus, new DeferredEvent(8)); + + Assert.AreEqual(ShrinkPostFailure.BusStopped, result.Failure); + } + + private readonly struct DeferredEvent : IShrinkEvent + { + public DeferredEvent(int value) => Value = value; + public int Value { get; } + } + + private sealed class ManualDispatcher : IShrinkEventBusMainThreadDispatcher, + ShrinkSDK.Runtime.IShrinkDeferredMainThreadDispatcher + { + private readonly Queue _queue = new(); + + public bool IsMainThread => true; + public int Count => _queue.Count; + + public bool TryPost(Action action) + { + _queue.Enqueue(action); + return true; + } + + public bool TryPostDeferred(Action action) + { + _queue.Enqueue(action); + return true; + } + + public void PumpAll() + { + var count = _queue.Count; + for (var i = 0; i < count; i++) + _queue.Dequeue()(); + } + } + + private sealed class NonDeferredDispatcher : IShrinkEventBusMainThreadDispatcher + { + public bool IsMainThread => true; + public bool TryPost(Action action) => true; + } + } +} diff --git a/Tests/ShrinkEventBusDeferredDiagnosticsTests.cs.meta b/Tests/ShrinkEventBusDeferredDiagnosticsTests.cs.meta new file mode 100644 index 0000000..3039e5f --- /dev/null +++ b/Tests/ShrinkEventBusDeferredDiagnosticsTests.cs.meta @@ -0,0 +1,11 @@ +fileFormatVersion: 2 +guid: 070c5e5975c64a39badcb0ebb04c2902 +MonoImporter: + externalObjects: {} + serializedVersion: 2 + defaultReferences: [] + executionOrder: 0 + icon: {instanceID: 0} + userData: + assetBundleName: + assetBundleVariant: diff --git a/package.json b/package.json index d98d4e4..28fc84c 100644 --- a/package.json +++ b/package.json @@ -1,6 +1,6 @@ { "name": "com.cneicy.shrink-eventbus", - "version": "2.2.3", + "version": "2.3.1", "displayName": "ShrinkEventBus", "description": "多 Bus、特性强类型注册、UniTask 调度与低分配发布的 Unity 事件总线。", "unity": "2022.3", @@ -9,7 +9,8 @@ "licensesUrl": "https://github.com/cneicy/ShrinkEventBus/blob/main/LICENSE", "dependencies": { "com.cysharp.unitask": "2.5.10", - "com.unity.nuget.mono-cecil": "1.11.4" + "com.unity.nuget.mono-cecil": "1.11.4", + "com.cneicy.shrink-runtime-abstractions": "0.2.0" }, "keywords": [ "eventbus",