552 lines
21 KiB
C#
552 lines
21 KiB
C#
#nullable enable
|
|
using System;
|
|
using System.Collections.Concurrent;
|
|
using System.Collections.Generic;
|
|
using System.Linq;
|
|
using System.Reflection;
|
|
using System.Threading;
|
|
using Cysharp.Threading.Tasks;
|
|
using ShrinkEventBus;
|
|
using UnityEngine;
|
|
|
|
namespace ShrinkNetwork.Integration
|
|
{
|
|
public static class ShrinkNetworkEventBusBridge
|
|
{
|
|
private sealed class ServiceRegistration
|
|
{
|
|
public ShrinkNetworkService Service = null!;
|
|
public Func<ShrinkNetworkSession, EventBase, bool>? SessionFilter;
|
|
public Dictionary<string, long> AppliedDeltaVersions { get; } = new();
|
|
}
|
|
|
|
private static readonly object SyncRoot = new();
|
|
private static readonly Dictionary<ShrinkNetworkService, ServiceRegistration> RegisteredServices = new();
|
|
private static readonly AsyncLocal<int> SuppressForwardDepth = new();
|
|
private static readonly ConcurrentDictionary<Type, Func<EventBase, UniTask>> IncomingDispatchers = new();
|
|
private static readonly ConcurrentDictionary<Type, Func<ShrinkNetworkContext, EventBase, UniTask<object?>>> IncomingRequestDispatchers = new();
|
|
private static readonly MethodInfo DispatchIncomingGenericMethod =
|
|
typeof(ShrinkNetworkEventBusBridge).GetMethod(nameof(DispatchIncomingGenericAsync),
|
|
BindingFlags.NonPublic | BindingFlags.Static)!;
|
|
private static readonly MethodInfo DispatchIncomingRequestGenericMethod =
|
|
typeof(ShrinkNetworkEventBusBridge).GetMethod(nameof(DispatchIncomingRequestGenericAsync),
|
|
BindingFlags.NonPublic | BindingFlags.Static)!;
|
|
|
|
private static Dictionary<Type, ShrinkNetworkMessageAttribute> _networkEventTypes = new();
|
|
private static bool _initialized;
|
|
|
|
[RuntimeInitializeOnLoadMethod(RuntimeInitializeLoadType.SubsystemRegistration)]
|
|
private static void ResetStaticState()
|
|
{
|
|
EventBus.OnEventTriggered -= HandleLocalEventTriggered;
|
|
lock (SyncRoot)
|
|
{
|
|
RegisteredServices.Clear();
|
|
FallbackDeltaVersions.Clear();
|
|
}
|
|
|
|
IncomingDispatchers.Clear();
|
|
IncomingRequestDispatchers.Clear();
|
|
_networkEventTypes = new Dictionary<Type, ShrinkNetworkMessageAttribute>();
|
|
_initialized = false;
|
|
}
|
|
|
|
private static void EnsureBridgeInitialized()
|
|
{
|
|
if (_initialized)
|
|
return;
|
|
|
|
_initialized = true;
|
|
RefreshNetworkEventTypes();
|
|
EventBus.OnEventTriggered += HandleLocalEventTriggered;
|
|
}
|
|
|
|
public static void RegisterService(ShrinkNetworkService service,
|
|
Func<ShrinkNetworkSession, EventBase, bool>? sessionFilter = null)
|
|
{
|
|
if (service == null)
|
|
throw new ArgumentNullException(nameof(service));
|
|
|
|
EnsureBridgeInitialized();
|
|
RefreshNetworkEventTypes();
|
|
|
|
lock (SyncRoot)
|
|
{
|
|
if (RegisteredServices.TryGetValue(service, out var existing))
|
|
{
|
|
existing.SessionFilter = sessionFilter;
|
|
return;
|
|
}
|
|
|
|
RegisteredServices.Add(service, new ServiceRegistration
|
|
{
|
|
Service = service,
|
|
SessionFilter = sessionFilter
|
|
});
|
|
}
|
|
|
|
RegisterInboundBridgeHandlers(service);
|
|
}
|
|
|
|
public static void UnregisterService(ShrinkNetworkService service)
|
|
{
|
|
if (service == null)
|
|
return;
|
|
|
|
lock (SyncRoot)
|
|
{
|
|
RegisteredServices.Remove(service);
|
|
}
|
|
}
|
|
|
|
public static void RefreshBindings()
|
|
{
|
|
RefreshNetworkEventTypes();
|
|
|
|
List<ShrinkNetworkService> services;
|
|
lock (SyncRoot)
|
|
{
|
|
services = RegisteredServices.Keys.ToList();
|
|
}
|
|
|
|
foreach (var service in services)
|
|
{
|
|
RegisterInboundBridgeHandlers(service);
|
|
}
|
|
}
|
|
|
|
public static UniTask PublishAsync<TEvent>(ShrinkNetworkSession session, TEvent eventArgs, string? route = null)
|
|
where TEvent : EventBase, IShrinkNetworkMessage
|
|
{
|
|
if (session == null)
|
|
throw new ArgumentNullException(nameof(session));
|
|
if (eventArgs == null)
|
|
throw new ArgumentNullException(nameof(eventArgs));
|
|
|
|
EnsureNetworkEventType(typeof(TEvent));
|
|
EnsurePublishableEventType(typeof(TEvent));
|
|
return session.SendAsync(eventArgs, route);
|
|
}
|
|
|
|
public static async UniTask PublishAsync<TEvent>(IEnumerable<ShrinkNetworkSession> sessions, TEvent eventArgs,
|
|
string? route = null)
|
|
where TEvent : EventBase, IShrinkNetworkMessage
|
|
{
|
|
if (sessions == null)
|
|
throw new ArgumentNullException(nameof(sessions));
|
|
if (eventArgs == null)
|
|
throw new ArgumentNullException(nameof(eventArgs));
|
|
|
|
EnsureNetworkEventType(typeof(TEvent));
|
|
EnsurePublishableEventType(typeof(TEvent));
|
|
|
|
var groupedSessions = sessions
|
|
.Where(session => session != null)
|
|
.GroupBy(session => session.Service);
|
|
|
|
foreach (var serviceGroup in groupedSessions)
|
|
{
|
|
var sessionList = serviceGroup.ToList();
|
|
if (sessionList.Count == 0)
|
|
continue;
|
|
|
|
var payload = serviceGroup.Key.Serializer.Serialize(eventArgs);
|
|
|
|
foreach (var session in sessionList)
|
|
await serviceGroup.Key.SendSerializedAsync(session, typeof(TEvent), payload, route);
|
|
}
|
|
}
|
|
|
|
public static async UniTask<ShrinkNetworkEventResultResponse> RequestResultAsync<TEvent>(ShrinkNetworkSession session,
|
|
TEvent eventArgs, ShrinkRpcCallOptions? options = null)
|
|
where TEvent : EventBase, IShrinkNetworkRequest
|
|
{
|
|
if (session == null)
|
|
throw new ArgumentNullException(nameof(session));
|
|
if (eventArgs == null)
|
|
throw new ArgumentNullException(nameof(eventArgs));
|
|
|
|
EnsureResultRequestEventType(typeof(TEvent));
|
|
|
|
options ??= new ShrinkRpcCallOptions();
|
|
if (string.IsNullOrWhiteSpace(options.DebugLabel))
|
|
options.DebugLabel = typeof(TEvent).Name;
|
|
|
|
var response = await session.RpcAsync<TEvent, ShrinkNetworkEventResultResponse>(eventArgs,
|
|
options);
|
|
|
|
if (response.IsSuccess)
|
|
{
|
|
eventArgs.SetResult(response.Result);
|
|
if (eventArgs.IsCancelable)
|
|
eventArgs.SetCanceled(response.IsCanceled);
|
|
}
|
|
|
|
return response;
|
|
}
|
|
|
|
private static void HandleLocalEventTriggered(EventBase eventArgs, Type eventType)
|
|
{
|
|
if (eventArgs == null || eventType == null)
|
|
return;
|
|
if (SuppressForwardDepth.Value > 0)
|
|
return;
|
|
if (!TryGetNetworkEventMeta(eventType, out _))
|
|
return;
|
|
if (eventArgs is IShrinkNetworkRequest)
|
|
return;
|
|
if (eventArgs is not IShrinkNetworkMessage networkMessage)
|
|
return;
|
|
|
|
ForwardEventAsync(eventArgs, eventType, networkMessage).Forget();
|
|
}
|
|
|
|
private static async UniTaskVoid ForwardEventAsync(EventBase eventArgs, Type eventType,
|
|
IShrinkNetworkMessage networkMessage)
|
|
{
|
|
List<ServiceRegistration> registrations;
|
|
lock (SyncRoot)
|
|
{
|
|
registrations = RegisteredServices.Values.ToList();
|
|
}
|
|
|
|
var batches = new List<(ServiceRegistration Registration, List<ShrinkNetworkSession> Sessions, byte[] Payload)>();
|
|
foreach (var registration in registrations)
|
|
{
|
|
var sessions = registration.Service.Sessions.Values
|
|
.Where(session => registration.SessionFilter == null || registration.SessionFilter(session, eventArgs))
|
|
.ToList();
|
|
if (sessions.Count == 0)
|
|
continue;
|
|
|
|
byte[] payload;
|
|
try
|
|
{
|
|
// Event payload is immutable after serialization and can be
|
|
// reused for every session owned by this service.
|
|
payload = registration.Service.Serializer.Serialize(networkMessage);
|
|
}
|
|
catch (Exception ex)
|
|
{
|
|
Debug.LogException(ex);
|
|
Debug.LogError(
|
|
$"[ShrinkNetwork.Integration] 序列化网络事件失败: {eventArgs.GetType().FullName}");
|
|
continue;
|
|
}
|
|
|
|
// Complete every serializer pass before the first await. EventBase
|
|
// instances may be returned to EventPool immediately after the
|
|
// synchronous trigger returns.
|
|
batches.Add((registration, sessions, payload));
|
|
}
|
|
|
|
foreach (var batch in batches)
|
|
{
|
|
foreach (var session in batch.Sessions)
|
|
{
|
|
try
|
|
{
|
|
await batch.Registration.Service.SendSerializedAsync(session, eventType, batch.Payload);
|
|
}
|
|
catch (Exception ex)
|
|
{
|
|
Debug.LogException(ex);
|
|
Debug.LogError(
|
|
$"[ShrinkNetwork.Integration] 转发网络事件失败: {eventArgs.GetType().FullName}, Session={session.SessionId}");
|
|
}
|
|
}
|
|
}
|
|
}
|
|
|
|
private static void RegisterInboundBridgeHandlers(ShrinkNetworkService service)
|
|
{
|
|
foreach (var pair in _networkEventTypes)
|
|
{
|
|
if (!service.MessageRegistry.TryGetMeta(pair.Key, out _))
|
|
{
|
|
service.RegisterMessage(pair.Key, pair.Value.Opcode, pair.Value.Route);
|
|
}
|
|
|
|
if (!service.MessageRegistry.TryGetMeta(typeof(ShrinkNetworkEventResultResponse), out _))
|
|
{
|
|
var responseAttribute =
|
|
typeof(ShrinkNetworkEventResultResponse).GetCustomAttribute<ShrinkNetworkMessageAttribute>(false);
|
|
if (responseAttribute != null)
|
|
{
|
|
service.RegisterMessage(typeof(ShrinkNetworkEventResultResponse), responseAttribute.Opcode,
|
|
responseAttribute.Route);
|
|
}
|
|
}
|
|
|
|
var isRequestEvent = typeof(IShrinkNetworkRequest).IsAssignableFrom(pair.Key);
|
|
|
|
try
|
|
{
|
|
if (isRequestEvent)
|
|
{
|
|
service.RegisterRequestHandler(pair.Key, typeof(ShrinkNetworkEventResultResponse),
|
|
DispatchIncomingRequestAsync);
|
|
}
|
|
else
|
|
{
|
|
service.RegisterHandler(pair.Key, DispatchIncomingAsync);
|
|
}
|
|
}
|
|
catch (InvalidOperationException)
|
|
{
|
|
// 同一个 service 上该事件类型只能绑定一个处理器。已存在时认为调用方自行接管。
|
|
}
|
|
}
|
|
}
|
|
|
|
private static UniTask DispatchIncomingAsync(ShrinkNetworkContext context, object message)
|
|
{
|
|
if (message is not EventBase eventArgs)
|
|
return UniTask.CompletedTask;
|
|
|
|
var dispatcher = IncomingDispatchers.GetOrAdd(eventArgs.GetType(), CreateIncomingDispatcher);
|
|
return dispatcher(eventArgs);
|
|
}
|
|
|
|
private static Func<EventBase, UniTask> CreateIncomingDispatcher(Type eventType)
|
|
{
|
|
var closedMethod = DispatchIncomingGenericMethod.MakeGenericMethod(eventType);
|
|
return (Func<EventBase, UniTask>)closedMethod.CreateDelegate(typeof(Func<EventBase, UniTask>));
|
|
}
|
|
|
|
private static UniTask<object?> DispatchIncomingRequestAsync(ShrinkNetworkContext context, object message)
|
|
{
|
|
if (message is not EventBase eventArgs)
|
|
{
|
|
return UniTask.FromResult<object?>(new ShrinkNetworkEventResultResponse
|
|
{
|
|
ErrorCode = ShrinkRpcErrorCode.InvalidResponse,
|
|
ErrorMessage = "Incoming network event request is not an EventBase."
|
|
});
|
|
}
|
|
|
|
var dispatcher = IncomingRequestDispatchers.GetOrAdd(eventArgs.GetType(), CreateIncomingRequestDispatcher);
|
|
return dispatcher(context, eventArgs);
|
|
}
|
|
|
|
private static Func<ShrinkNetworkContext, EventBase, UniTask<object?>> CreateIncomingRequestDispatcher(Type eventType)
|
|
{
|
|
var closedMethod = DispatchIncomingRequestGenericMethod.MakeGenericMethod(eventType);
|
|
return (Func<ShrinkNetworkContext, EventBase, UniTask<object?>>)closedMethod.CreateDelegate(
|
|
typeof(Func<ShrinkNetworkContext, EventBase, UniTask<object?>>));
|
|
}
|
|
|
|
private static async UniTask DispatchIncomingGenericAsync<TEvent>(EventBase eventArgs)
|
|
where TEvent : EventBase
|
|
{
|
|
SuppressForwardDepth.Value++;
|
|
try
|
|
{
|
|
if (TryGetDeltaEvent(eventArgs, typeof(TEvent), out var deltaEvent) &&
|
|
!TryMarkIncomingDelta(typeof(TEvent), deltaEvent, null))
|
|
{
|
|
return;
|
|
}
|
|
|
|
await EventBus.TriggerEventAsync((TEvent)eventArgs);
|
|
}
|
|
finally
|
|
{
|
|
SuppressForwardDepth.Value--;
|
|
}
|
|
}
|
|
|
|
private static async UniTask<object?> DispatchIncomingRequestGenericAsync<TEvent>(ShrinkNetworkContext context,
|
|
EventBase eventArgs)
|
|
where TEvent : EventBase
|
|
{
|
|
var typedEvent = (TEvent)eventArgs;
|
|
|
|
SuppressForwardDepth.Value++;
|
|
try
|
|
{
|
|
if (TryGetDeltaEvent(typedEvent, typeof(TEvent), out var deltaEvent) &&
|
|
!TryMarkIncomingDelta(typeof(TEvent), deltaEvent, context))
|
|
{
|
|
return BuildResultResponse(typedEvent);
|
|
}
|
|
|
|
await EventBus.TriggerEventAsync(typedEvent);
|
|
return BuildResultResponse(typedEvent);
|
|
}
|
|
catch (Exception ex)
|
|
{
|
|
Debug.LogException(ex);
|
|
return new ShrinkNetworkEventResultResponse
|
|
{
|
|
ErrorCode = ShrinkRpcErrorCode.HandlerException,
|
|
ErrorMessage = ex.Message,
|
|
Result = typedEvent.HasResult ? typedEvent.Result : EventResult.DEFAULT,
|
|
IsCanceled = typedEvent.IsCancelable && typedEvent.IsCanceled
|
|
};
|
|
}
|
|
finally
|
|
{
|
|
SuppressForwardDepth.Value--;
|
|
}
|
|
}
|
|
|
|
private static void RefreshNetworkEventTypes()
|
|
{
|
|
var eventTypes = new Dictionary<Type, ShrinkNetworkMessageAttribute>();
|
|
|
|
foreach (var assembly in AppDomain.CurrentDomain.GetAssemblies())
|
|
{
|
|
Type[] types;
|
|
try
|
|
{
|
|
types = assembly.GetTypes();
|
|
}
|
|
catch
|
|
{
|
|
continue;
|
|
}
|
|
|
|
foreach (var type in types)
|
|
{
|
|
if (type == null || type.IsAbstract)
|
|
continue;
|
|
if (!typeof(EventBase).IsAssignableFrom(type))
|
|
continue;
|
|
if (!typeof(IShrinkNetworkMessage).IsAssignableFrom(type))
|
|
continue;
|
|
if (type.GetCustomAttribute<ShrinkNetworkEventAttribute>(false) == null)
|
|
continue;
|
|
|
|
var messageAttr = type.GetCustomAttribute<ShrinkNetworkMessageAttribute>(false);
|
|
if (messageAttr == null)
|
|
{
|
|
Debug.LogWarning(
|
|
$"[ShrinkNetwork.Integration] {type.FullName} 标记了 [ShrinkNetworkEvent]，但缺少 [ShrinkNetworkMessage]，已忽略。");
|
|
continue;
|
|
}
|
|
|
|
eventTypes[type] = messageAttr;
|
|
}
|
|
}
|
|
|
|
_networkEventTypes = eventTypes;
|
|
}
|
|
|
|
private static bool TryGetNetworkEventMeta(Type eventType, out ShrinkNetworkMessageAttribute? messageAttribute)
|
|
{
|
|
if (eventType == null)
|
|
{
|
|
messageAttribute = null;
|
|
return false;
|
|
}
|
|
|
|
return _networkEventTypes.TryGetValue(eventType, out messageAttribute);
|
|
}
|
|
|
|
private static void EnsureNetworkEventType(Type eventType)
|
|
{
|
|
if (!TryGetNetworkEventMeta(eventType, out _))
|
|
throw new InvalidOperationException(
|
|
$"Event type {eventType.FullName} must implement EventBase + IShrinkNetworkMessage and declare [ShrinkNetworkEvent] + [ShrinkNetworkMessage].");
|
|
}
|
|
|
|
private static void EnsurePublishableEventType(Type eventType)
|
|
{
|
|
if (typeof(IShrinkNetworkRequest).IsAssignableFrom(eventType))
|
|
throw new InvalidOperationException(
|
|
$"Event type {eventType.FullName} is a request event. Use RequestResultAsync instead of PublishAsync.");
|
|
}
|
|
|
|
private static void EnsureResultRequestEventType(Type eventType)
|
|
{
|
|
EnsureNetworkEventType(eventType);
|
|
|
|
if (!typeof(IShrinkNetworkRequest).IsAssignableFrom(eventType))
|
|
throw new InvalidOperationException(
|
|
$"Event type {eventType.FullName} must implement IShrinkNetworkRequest to use RequestResultAsync.");
|
|
if (eventType.GetCustomAttribute<HasResultAttribute>(false) == null)
|
|
throw new InvalidOperationException(
|
|
$"Event type {eventType.FullName} must declare [HasResult] to use RequestResultAsync.");
|
|
}
|
|
|
|
private static ShrinkNetworkEventResultResponse BuildResultResponse(EventBase eventArgs)
|
|
{
|
|
return new ShrinkNetworkEventResultResponse
|
|
{
|
|
Result = eventArgs.HasResult ? eventArgs.Result : EventResult.DEFAULT,
|
|
IsCanceled = eventArgs.IsCancelable && eventArgs.IsCanceled
|
|
};
|
|
}
|
|
|
|
private static bool TryGetDeltaEvent(EventBase eventArgs, Type eventType, out IShrinkNetworkDeltaEvent deltaEvent)
|
|
{
|
|
if (eventArgs is IShrinkNetworkDeltaEvent eventDelta)
|
|
{
|
|
deltaEvent = eventDelta;
|
|
return true;
|
|
}
|
|
|
|
deltaEvent = null!;
|
|
return false;
|
|
}
|
|
|
|
private static bool TryMarkIncomingDelta(Type eventType, IShrinkNetworkDeltaEvent deltaEvent,
|
|
ShrinkNetworkContext? context)
|
|
{
|
|
if (deltaEvent == null)
|
|
return true;
|
|
|
|
var scopeKey = BuildDeltaScopeKey(eventType, deltaEvent, context);
|
|
|
|
lock (SyncRoot)
|
|
{
|
|
if (context != null && RegisteredServices.TryGetValue(context.Service, out var serviceRegistration))
|
|
{
|
|
return TryMarkIncomingDeltaCore(serviceRegistration.AppliedDeltaVersions, scopeKey,
|
|
deltaEvent.DeltaVersion);
|
|
}
|
|
|
|
return TryMarkIncomingDeltaCore(FallbackDeltaVersions, scopeKey, deltaEvent.DeltaVersion);
|
|
}
|
|
}
|
|
|
|
private static bool TryMarkIncomingDeltaCore(Dictionary<string, long> versionMap, string scopeKey, long deltaVersion)
|
|
{
|
|
if (versionMap.TryGetValue(scopeKey, out var existingVersion) && deltaVersion <= existingVersion)
|
|
return false;
|
|
|
|
versionMap[scopeKey] = deltaVersion;
|
|
return true;
|
|
}
|
|
|
|
private static string BuildDeltaScopeKey(Type eventType, IShrinkNetworkDeltaEvent deltaEvent,
|
|
ShrinkNetworkContext? context)
|
|
{
|
|
var sessionId = context?.Session?.SessionId ?? 0;
|
|
var key = string.IsNullOrWhiteSpace(deltaEvent.DeltaKey) ? "__default" : deltaEvent.DeltaKey.Trim();
|
|
return $"{sessionId}:{eventType.FullName}:{key}";
|
|
}
|
|
|
|
private static readonly Dictionary<string, long> FallbackDeltaVersions = new();
|
|
}
|
|
|
|
/// <summary>
|
|
/// Switches network callbacks to Unity's main thread before entering EventBus
|
|
/// handlers. Use ShrinkNetworkDispatchQueue when a hard per-frame budget is
|
|
/// required instead of an unbounded PlayerLoop backlog.
|
|
/// </summary>
|
|
public sealed class ShrinkNetworkUnityMainThreadDispatchScheduler : IShrinkNetworkDispatchScheduler
|
|
{
|
|
public async UniTask<bool> ScheduleAsync(Func<UniTask> callback)
|
|
{
|
|
if (callback == null)
|
|
throw new ArgumentNullException(nameof(callback));
|
|
|
|
await UniTask.SwitchToMainThread();
|
|
await callback();
|
|
return true;
|
|
}
|
|
}
|
|
}
|