858 lines
36 KiB
C#
858 lines
36 KiB
C#
#nullable enable
|
|
|
|
using System;
|
|
using System.Collections.Concurrent;
|
|
using System.Collections.Generic;
|
|
using System.Linq;
|
|
using System.Threading;
|
|
using Cysharp.Threading.Tasks;
|
|
|
|
namespace ShrinkNetwork
|
|
{
|
|
public sealed class ShrinkNetworkService
|
|
{
|
|
private sealed class PendingRequest
|
|
{
|
|
public long SessionId;
|
|
public Type ResponseType = null!;
|
|
public UniTaskCompletionSource<object?> CompletionSource = null!;
|
|
}
|
|
|
|
private readonly ConcurrentDictionary<long, ShrinkNetworkSession> _sessions = new();
|
|
private readonly ConcurrentDictionary<ShrinkRequestToken, PendingRequest> _pendingRequests = new();
|
|
private int _requestTokenGenerator;
|
|
private IShrinkNetworkTransport? _transport;
|
|
private long _sessionsConnected;
|
|
private long _sessionsDisconnected;
|
|
private long _packetsSent;
|
|
private long _packetsReceived;
|
|
private long _bytesSent;
|
|
private long _bytesReceived;
|
|
private long _rpcStarted;
|
|
private long _rpcCompleted;
|
|
private long _rpcTimedOut;
|
|
private long _rpcCanceled;
|
|
private long _rpcFailed;
|
|
private long _protocolViolations;
|
|
private long _authRejectedCount;
|
|
private long _permissionDeniedCount;
|
|
private long _handlerExceptionCount;
|
|
private long _unknownOpcodeCount;
|
|
private long _dispatchMissCount;
|
|
private long _serializationErrorCount;
|
|
private long _dispatchQueueRejectedCount;
|
|
private IShrinkNetworkDispatchScheduler _dispatchScheduler;
|
|
|
|
public ShrinkNetworkService()
|
|
: this(new ShrinkJsonNetworkSerializer(), new ShrinkNetworkMessageRegistry(), new ShrinkNetworkRouter())
|
|
{
|
|
}
|
|
|
|
public ShrinkNetworkService(IShrinkNetworkSerializer serializer, ShrinkNetworkMessageRegistry messageRegistry,
|
|
ShrinkNetworkRouter router, IShrinkNetworkDispatchScheduler? dispatchScheduler = null)
|
|
{
|
|
Serializer = serializer ?? throw new ArgumentNullException(nameof(serializer));
|
|
MessageRegistry = messageRegistry ?? throw new ArgumentNullException(nameof(messageRegistry));
|
|
Router = router ?? throw new ArgumentNullException(nameof(router));
|
|
_dispatchScheduler = dispatchScheduler ?? ShrinkNetworkDispatchSchedulers.Inline;
|
|
}
|
|
|
|
public IShrinkNetworkSerializer Serializer { get; }
|
|
public ShrinkNetworkMessageRegistry MessageRegistry { get; }
|
|
public ShrinkNetworkRouter Router { get; }
|
|
public IReadOnlyDictionary<long, ShrinkNetworkSession> Sessions => _sessions;
|
|
public int MinProtocolVersion { get; set; } = ShrinkNetworkProtocol.CurrentProtocolVersion;
|
|
public int MaxProtocolVersion { get; set; } = ShrinkNetworkProtocol.CurrentProtocolVersion;
|
|
public int MinSchemaVersion { get; set; } = ShrinkNetworkProtocol.CurrentSchemaVersion;
|
|
public int MaxSchemaVersion { get; set; } = ShrinkNetworkProtocol.CurrentSchemaVersion;
|
|
public bool DisconnectOnProtocolViolation { get; set; } = true;
|
|
public Func<ShrinkNetworkSession, ShrinkNetworkPacket, ShrinkIncomingPacketValidationResult?>? IncomingPacketValidator { get; set; }
|
|
|
|
public IShrinkNetworkDispatchScheduler DispatchScheduler
|
|
{
|
|
get => _dispatchScheduler;
|
|
set => _dispatchScheduler = value ?? throw new ArgumentNullException(nameof(value));
|
|
}
|
|
|
|
public event Action<ShrinkNetworkSession>? OnSessionConnected;
|
|
public event Action<ShrinkNetworkSession>? OnSessionDisconnected;
|
|
|
|
public void BindTransport(IShrinkNetworkTransport? transport)
|
|
{
|
|
if (_transport != null)
|
|
_transport.OnEvent -= OnTransportEvent;
|
|
|
|
_transport = transport;
|
|
|
|
if (_transport != null)
|
|
{
|
|
_transport.OnEvent += OnTransportEvent;
|
|
if (!_transport.IsStarted)
|
|
_transport.Start();
|
|
return;
|
|
}
|
|
}
|
|
|
|
public void RegisterMessage<TMessage>(int opcode, string? route = null) where TMessage : IShrinkNetworkMessage
|
|
=> MessageRegistry.Register<TMessage>(opcode, route);
|
|
|
|
public void RegisterMessage(Type messageType, int opcode, string? route = null)
|
|
=> MessageRegistry.Register(messageType, opcode, route);
|
|
|
|
public void RegisterHandler<TMessage>(Func<ShrinkNetworkContext, TMessage, UniTask> handler,
|
|
ShrinkNetworkPermissionRequirement requirement = default)
|
|
where TMessage : IShrinkNetworkMessage
|
|
=> Router.RegisterHandler(handler, requirement);
|
|
|
|
public void RegisterHandler(Type messageType, Func<ShrinkNetworkContext, object, UniTask> handler,
|
|
ShrinkNetworkPermissionRequirement requirement = default)
|
|
=> Router.RegisterHandler(messageType, handler, requirement);
|
|
|
|
public void RegisterRequestHandler<TRequest, TResponse>(Func<ShrinkNetworkContext, TRequest, UniTask<TResponse>> handler,
|
|
ShrinkNetworkPermissionRequirement requirement = default)
|
|
where TRequest : IShrinkNetworkRequest
|
|
where TResponse : class, IShrinkNetworkResponse
|
|
=> Router.RegisterRequestHandler(handler, requirement);
|
|
|
|
public void RegisterRequestHandler(Type requestType, Type responseType,
|
|
Func<ShrinkNetworkContext, object, UniTask<object?>> handler,
|
|
ShrinkNetworkPermissionRequirement requirement = default)
|
|
=> Router.RegisterRequestHandler(requestType, responseType, handler, requirement);
|
|
|
|
public void RegisterRpc<TRequest, TResponse>(Func<ShrinkNetworkContext, TRequest, UniTask<TResponse>> handler,
|
|
ShrinkNetworkPermissionRequirement requirement = default)
|
|
where TRequest : IShrinkNetworkRequest
|
|
where TResponse : class, IShrinkNetworkResponse
|
|
=> RegisterRequestHandler(handler, requirement);
|
|
|
|
public void AutoRegisterAttributedMessages()
|
|
=> ShrinkNetworkRegHelper.RegisterAttributedMessages(this);
|
|
|
|
public void AutoRegisterStaticHandlers()
|
|
=> ShrinkNetworkRegHelper.RegisterStaticHandlers(this);
|
|
|
|
public void RegisterHandlers(object target)
|
|
=> ShrinkNetworkRegHelper.RegisterHandlers(this, target);
|
|
|
|
public void AutoRegisterAll()
|
|
{
|
|
AutoRegisterAttributedMessages();
|
|
AutoRegisterStaticHandlers();
|
|
}
|
|
|
|
public ShrinkNetworkServiceDiagnosticsSnapshot GetDiagnosticsSnapshot()
|
|
{
|
|
return new ShrinkNetworkServiceDiagnosticsSnapshot
|
|
{
|
|
CurrentSessions = _sessions.Count,
|
|
SessionsConnected = Volatile.Read(ref _sessionsConnected),
|
|
SessionsDisconnected = Volatile.Read(ref _sessionsDisconnected),
|
|
PacketsSent = Volatile.Read(ref _packetsSent),
|
|
PacketsReceived = Volatile.Read(ref _packetsReceived),
|
|
BytesSent = Volatile.Read(ref _bytesSent),
|
|
BytesReceived = Volatile.Read(ref _bytesReceived),
|
|
RpcStarted = Volatile.Read(ref _rpcStarted),
|
|
RpcCompleted = Volatile.Read(ref _rpcCompleted),
|
|
RpcTimedOut = Volatile.Read(ref _rpcTimedOut),
|
|
RpcCanceled = Volatile.Read(ref _rpcCanceled),
|
|
RpcFailed = Volatile.Read(ref _rpcFailed),
|
|
ProtocolViolations = Volatile.Read(ref _protocolViolations),
|
|
AuthRejectedCount = Volatile.Read(ref _authRejectedCount),
|
|
PermissionDeniedCount = Volatile.Read(ref _permissionDeniedCount),
|
|
HandlerExceptionCount = Volatile.Read(ref _handlerExceptionCount),
|
|
UnknownOpcodeCount = Volatile.Read(ref _unknownOpcodeCount),
|
|
DispatchMissCount = Volatile.Read(ref _dispatchMissCount),
|
|
SerializationErrorCount = Volatile.Read(ref _serializationErrorCount),
|
|
DispatchQueueRejectedCount = Volatile.Read(ref _dispatchQueueRejectedCount)
|
|
};
|
|
}
|
|
|
|
public UniTask SendAsync<TMessage>(ShrinkNetworkSession session, TMessage message, string? route = null)
|
|
where TMessage : IShrinkNetworkMessage
|
|
=> SendInternalAsync(session, message, ShrinkNetworkPacketKind.Message, ShrinkRequestToken.Default, route);
|
|
|
|
public UniTask SendAsync(ShrinkNetworkSession session, IShrinkNetworkMessage message, string? route = null)
|
|
{
|
|
if (message == null)
|
|
throw new ArgumentNullException(nameof(message));
|
|
|
|
return SendInternalAsync(session, message, message.GetType(), ShrinkNetworkPacketKind.Message, ShrinkRequestToken.Default, route);
|
|
}
|
|
|
|
public UniTask NotifyAsync<TMessage>(ShrinkNetworkSession session, TMessage message, string? route = null)
|
|
where TMessage : IShrinkNetworkMessage
|
|
=> SendAsync(session, message, route);
|
|
|
|
public UniTask NotifyAsync(ShrinkNetworkSession session, IShrinkNetworkMessage message, string? route = null)
|
|
=> SendAsync(session, message, route);
|
|
|
|
/// <summary>
|
|
/// Sends a message whose payload has already been serialized. The payload is
|
|
/// treated as immutable and can be reused for multiple sessions.
|
|
/// </summary>
|
|
public UniTask SendSerializedAsync(ShrinkNetworkSession session, Type messageType, byte[] payload,
|
|
string? route = null)
|
|
{
|
|
return SendPacketAsync(session, messageType, ShrinkNetworkPacketKind.Message,
|
|
ShrinkRequestToken.Default, route, payload);
|
|
}
|
|
|
|
public UniTask<TResponse> CallAsync<TRequest, TResponse>(ShrinkNetworkSession session, TRequest request, string? route = null)
|
|
where TRequest : IShrinkNetworkRequest
|
|
where TResponse : class, IShrinkNetworkResponse
|
|
=> CallAsync<TRequest, TResponse>(session, request, new ShrinkRpcCallOptions { RouteOverride = route });
|
|
|
|
public async UniTask<TResponse> CallAsync<TRequest, TResponse>(ShrinkNetworkSession session, TRequest request,
|
|
ShrinkRpcCallOptions options)
|
|
where TRequest : IShrinkNetworkRequest
|
|
where TResponse : class, IShrinkNetworkResponse
|
|
{
|
|
if (session == null)
|
|
throw new ArgumentNullException(nameof(session));
|
|
|
|
Interlocked.Increment(ref _rpcStarted);
|
|
var requestToken = options?.RequestTokenOverride ?? new ShrinkRequestToken(Interlocked.Increment(ref _requestTokenGenerator));
|
|
var pending = new PendingRequest
|
|
{
|
|
SessionId = session.SessionId,
|
|
ResponseType = typeof(TResponse),
|
|
CompletionSource = new UniTaskCompletionSource<object?>()
|
|
};
|
|
if (!_pendingRequests.TryAdd(requestToken, pending))
|
|
{
|
|
throw new InvalidOperationException(
|
|
$"Pending request token already exists. RequestToken={requestToken}");
|
|
}
|
|
|
|
try
|
|
{
|
|
await SendInternalAsync(session, request, ShrinkNetworkPacketKind.Request, requestToken, options?.RouteOverride);
|
|
var response = await WaitForPendingResponse<TResponse>(requestToken, pending, options);
|
|
EnsureResponseSucceeded(response);
|
|
Interlocked.Increment(ref _rpcCompleted);
|
|
return response;
|
|
}
|
|
catch
|
|
{
|
|
_pendingRequests.TryRemove(requestToken, out _);
|
|
throw;
|
|
}
|
|
}
|
|
|
|
public UniTask<TResponse> RpcAsync<TRequest, TResponse>(ShrinkNetworkSession session, TRequest request, string? route = null)
|
|
where TRequest : IShrinkNetworkRequest
|
|
where TResponse : class, IShrinkNetworkResponse
|
|
=> CallAsync<TRequest, TResponse>(session, request, route);
|
|
|
|
public UniTask<TResponse> RpcAsync<TRequest, TResponse>(ShrinkNetworkSession session, TRequest request,
|
|
ShrinkRpcCallOptions options)
|
|
where TRequest : IShrinkNetworkRequest
|
|
where TResponse : class, IShrinkNetworkResponse
|
|
=> CallAsync<TRequest, TResponse>(session, request, options);
|
|
|
|
internal UniTask SendResponseAsync<TResponse>(ShrinkNetworkSession session, TResponse response, ShrinkRequestToken requestToken, string? route = null)
|
|
where TResponse : class, IShrinkNetworkResponse
|
|
=> SendInternalAsync(session, response, typeof(TResponse), ShrinkNetworkPacketKind.Response, requestToken, route);
|
|
|
|
internal UniTask SendResponseAsync(ShrinkNetworkSession session, IShrinkNetworkResponse response, Type responseType,
|
|
ShrinkRequestToken requestToken, string? route = null)
|
|
=> SendInternalAsync(session, response, responseType, ShrinkNetworkPacketKind.Response, requestToken, route);
|
|
|
|
private UniTask SendInternalAsync<TMessage>(ShrinkNetworkSession session, TMessage message,
|
|
ShrinkNetworkPacketKind kind, ShrinkRequestToken requestToken, string? route)
|
|
where TMessage : IShrinkNetworkMessage
|
|
=> SendInternalAsync(session, message, typeof(TMessage), kind, requestToken, route);
|
|
|
|
private UniTask SendInternalAsync(ShrinkNetworkSession session, IShrinkNetworkMessage message, Type messageType,
|
|
ShrinkNetworkPacketKind kind, ShrinkRequestToken requestToken, string? route)
|
|
{
|
|
if (session == null)
|
|
throw new ArgumentNullException(nameof(session));
|
|
if (message == null)
|
|
throw new ArgumentNullException(nameof(message));
|
|
if (messageType == null)
|
|
throw new ArgumentNullException(nameof(messageType));
|
|
|
|
return SendPacketAsync(session, messageType, kind, requestToken, route, Serializer.Serialize(message));
|
|
}
|
|
|
|
private async UniTask SendPacketAsync(ShrinkNetworkSession session, Type messageType,
|
|
ShrinkNetworkPacketKind kind, ShrinkRequestToken requestToken, string? route, byte[] payload)
|
|
{
|
|
if (session == null)
|
|
throw new ArgumentNullException(nameof(session));
|
|
if (messageType == null)
|
|
throw new ArgumentNullException(nameof(messageType));
|
|
if (payload == null)
|
|
throw new ArgumentNullException(nameof(payload));
|
|
|
|
var transport = _transport ?? throw new InvalidOperationException("Transport is not bound.");
|
|
var meta = MessageRegistry.GetMeta(messageType);
|
|
var packet = new ShrinkNetworkPacket
|
|
{
|
|
Opcode = meta.Opcode,
|
|
RequestToken = requestToken,
|
|
SessionToken = session.SessionToken,
|
|
SessionTokenExpiresAtUnixTimeSeconds = session.SessionTokenExpiresAtUtc?.ToUnixTimeSeconds() ?? 0,
|
|
Route = string.IsNullOrWhiteSpace(route) ? meta.Route : route.Trim(),
|
|
Kind = kind,
|
|
Payload = payload
|
|
};
|
|
|
|
var packetData = Serializer.Serialize(packet);
|
|
Interlocked.Increment(ref _packetsSent);
|
|
Interlocked.Add(ref _bytesSent, packetData.Length);
|
|
if (transport is IShrinkNetworkAsyncTransport asyncTransport)
|
|
{
|
|
await asyncTransport.SendAsync(session.SessionId, packetData);
|
|
return;
|
|
}
|
|
|
|
transport.Send(session.SessionId, packetData);
|
|
}
|
|
|
|
private void OnTransportEvent(ShrinkNetworkTransportEvent evt)
|
|
{
|
|
if (evt == null)
|
|
return;
|
|
|
|
// 连接生命周期必须在传输回调返回前提交，否则快速退役/重激活时，
|
|
// 旧的断开事件可能晚于新的连接事件完成，观察者会读到过期会话视图。
|
|
if (evt.Type == ShrinkNetworkTransportEventType.Connected)
|
|
{
|
|
HandleConnected(evt);
|
|
return;
|
|
}
|
|
|
|
if (evt.Type == ShrinkNetworkTransportEventType.Disconnected)
|
|
{
|
|
HandleDisconnected(evt);
|
|
return;
|
|
}
|
|
|
|
ScheduleTransportEventAsync(evt).Forget();
|
|
}
|
|
|
|
private async UniTaskVoid ScheduleTransportEventAsync(ShrinkNetworkTransportEvent evt)
|
|
{
|
|
try
|
|
{
|
|
var scheduled = await DispatchScheduler.ScheduleAsync(() => HandleTransportEventAsync(evt));
|
|
if (!scheduled)
|
|
Interlocked.Increment(ref _dispatchQueueRejectedCount);
|
|
}
|
|
catch (Exception ex)
|
|
{
|
|
ShrinkNetworkLogger.Exception(ex);
|
|
ShrinkNetworkLogger.Error($"[ShrinkNetwork] Transport event dispatch failed: {ex.Message}");
|
|
}
|
|
}
|
|
|
|
private UniTask HandleTransportEventAsync(ShrinkNetworkTransportEvent evt)
|
|
{
|
|
switch (evt.Type)
|
|
{
|
|
case ShrinkNetworkTransportEventType.Connected:
|
|
HandleConnected(evt);
|
|
return UniTask.CompletedTask;
|
|
case ShrinkNetworkTransportEventType.Disconnected:
|
|
HandleDisconnected(evt);
|
|
return UniTask.CompletedTask;
|
|
case ShrinkNetworkTransportEventType.Packet:
|
|
return HandlePacketAsync(evt);
|
|
default:
|
|
return UniTask.CompletedTask;
|
|
}
|
|
}
|
|
|
|
private void HandleConnected(ShrinkNetworkTransportEvent evt)
|
|
{
|
|
var session = _sessions.AddOrUpdate(evt.SessionId,
|
|
_ => new ShrinkNetworkSession(evt.SessionId, evt.RemoteAddress, this),
|
|
(_, existing) =>
|
|
{
|
|
existing.RemoteAddress = evt.RemoteAddress;
|
|
return existing;
|
|
});
|
|
|
|
Interlocked.Increment(ref _sessionsConnected);
|
|
OnSessionConnected?.Invoke(session);
|
|
}
|
|
|
|
private void HandleDisconnected(ShrinkNetworkTransportEvent evt)
|
|
{
|
|
if (_sessions.TryRemove(evt.SessionId, out var session))
|
|
{
|
|
FailPendingRequestsForSession(evt.SessionId);
|
|
Interlocked.Increment(ref _sessionsDisconnected);
|
|
OnSessionDisconnected?.Invoke(session);
|
|
}
|
|
}
|
|
|
|
private async UniTask HandlePacketAsync(ShrinkNetworkTransportEvent evt)
|
|
{
|
|
try
|
|
{
|
|
Interlocked.Increment(ref _packetsReceived);
|
|
Interlocked.Add(ref _bytesReceived, evt.PacketData.Length);
|
|
var packet = Serializer.Deserialize<ShrinkNetworkPacket>(evt.PacketData);
|
|
|
|
if (!_sessions.TryGetValue(evt.SessionId, out var session))
|
|
{
|
|
session = _sessions.GetOrAdd(evt.SessionId,
|
|
id => new ShrinkNetworkSession(id, evt.RemoteAddress, this));
|
|
}
|
|
|
|
if (!ValidatePacketCompatibility(packet, evt.SessionId))
|
|
return;
|
|
|
|
ApplySessionTokenFromPacket(session, packet);
|
|
|
|
if (packet.Kind == ShrinkNetworkPacketKind.Response)
|
|
{
|
|
HandleResponse(packet);
|
|
return;
|
|
}
|
|
|
|
if (!ValidateIncomingPacket(session, packet))
|
|
return;
|
|
|
|
if (!MessageRegistry.TryGetMeta(packet.Opcode, out var meta))
|
|
{
|
|
Interlocked.Increment(ref _unknownOpcodeCount);
|
|
ShrinkNetworkLogger.Warn($"[ShrinkNetwork] Unregistered opcode: {packet.Opcode}");
|
|
return;
|
|
}
|
|
|
|
var resolvedMeta = meta!;
|
|
var message = Serializer.Deserialize(packet.Payload, resolvedMeta.MessageType);
|
|
if (message == null)
|
|
{
|
|
ShrinkNetworkLogger.Warn($"[ShrinkNetwork] Failed to deserialize message for opcode {packet.Opcode}.");
|
|
Interlocked.Increment(ref _serializationErrorCount);
|
|
return;
|
|
}
|
|
|
|
var context = new ShrinkNetworkContext(this, session, packet);
|
|
|
|
var handled = await Router.DispatchAsync(context, message, resolvedMeta.MessageType);
|
|
if (!handled)
|
|
{
|
|
Interlocked.Increment(ref _dispatchMissCount);
|
|
ShrinkNetworkLogger.Warn($"[ShrinkNetwork] No handler found for {resolvedMeta.MessageType.FullName}");
|
|
}
|
|
}
|
|
catch (Exception ex)
|
|
{
|
|
Interlocked.Increment(ref _serializationErrorCount);
|
|
ShrinkNetworkLogger.Exception(ex);
|
|
ShrinkNetworkLogger.Error($"[ShrinkNetwork] Packet handling failed: {ex.Message}");
|
|
}
|
|
}
|
|
|
|
private void HandleResponse(ShrinkNetworkPacket packet)
|
|
{
|
|
if (!_pendingRequests.TryRemove(packet.RequestToken, out var pending))
|
|
{
|
|
ShrinkNetworkLogger.Warn($"[ShrinkNetwork] Pending request not found. RequestToken={packet.RequestToken}");
|
|
return;
|
|
}
|
|
|
|
try
|
|
{
|
|
var response = Serializer.Deserialize(packet.Payload, pending.ResponseType);
|
|
pending.CompletionSource.TrySetResult(response);
|
|
}
|
|
catch (Exception ex)
|
|
{
|
|
pending.CompletionSource.TrySetException(ex);
|
|
}
|
|
}
|
|
|
|
private async UniTask<TResponse> WaitForPendingResponse<TResponse>(ShrinkRequestToken requestToken, PendingRequest pending,
|
|
ShrinkRpcCallOptions? options)
|
|
where TResponse : class, IShrinkNetworkResponse
|
|
{
|
|
var timeoutMs = options?.TimeoutMs ?? 10000;
|
|
var externalToken = options?.CancellationToken ?? default;
|
|
var debugLabel = string.IsNullOrWhiteSpace(options?.DebugLabel) ? string.Empty : options.DebugLabel.Trim();
|
|
|
|
using var timeoutCts = timeoutMs > 0 ? new CancellationTokenSource(timeoutMs) : null;
|
|
using var linkedCts = CreateLinkedTokenSource(externalToken, timeoutCts?.Token ?? default);
|
|
|
|
try
|
|
{
|
|
var boxed = await pending.CompletionSource.Task.AttachExternalCancellation(linkedCts?.Token ?? default);
|
|
if (boxed is not TResponse response)
|
|
throw new ShrinkRpcException(ShrinkRpcErrorCode.InvalidResponse, BuildRpcMessage("RPC returned an invalid response.", requestToken, debugLabel));
|
|
|
|
return response;
|
|
}
|
|
catch (OperationCanceledException) when (externalToken.IsCancellationRequested)
|
|
{
|
|
_pendingRequests.TryRemove(requestToken, out _);
|
|
Interlocked.Increment(ref _rpcCanceled);
|
|
Interlocked.Increment(ref _rpcFailed);
|
|
throw new ShrinkRpcException(ShrinkRpcErrorCode.Canceled, BuildRpcMessage("RPC call was canceled.", requestToken, debugLabel));
|
|
}
|
|
catch (OperationCanceledException)
|
|
{
|
|
_pendingRequests.TryRemove(requestToken, out _);
|
|
Interlocked.Increment(ref _rpcTimedOut);
|
|
Interlocked.Increment(ref _rpcFailed);
|
|
throw new ShrinkRpcException(ShrinkRpcErrorCode.Timeout, BuildRpcMessage("RPC call timed out.", requestToken, debugLabel));
|
|
}
|
|
}
|
|
|
|
private void FailPendingRequestsForSession(long sessionId)
|
|
{
|
|
var pendingEntries = _pendingRequests
|
|
.Where(pair => pair.Value.SessionId == sessionId)
|
|
.ToArray();
|
|
if (pendingEntries.Length == 0)
|
|
return;
|
|
|
|
foreach (var pendingEntry in pendingEntries)
|
|
{
|
|
if (_pendingRequests.TryRemove(pendingEntry.Key, out var pending))
|
|
{
|
|
pending.CompletionSource.TrySetException(
|
|
new ShrinkRpcException(ShrinkRpcErrorCode.ConnectionClosed,
|
|
$"RPC peer disconnected. RequestToken={pendingEntry.Key}, SessionId={sessionId}"));
|
|
}
|
|
}
|
|
}
|
|
|
|
private static string BuildRpcMessage(string message, ShrinkRequestToken requestToken, string debugLabel)
|
|
{
|
|
if (string.IsNullOrWhiteSpace(debugLabel))
|
|
return $"{message} RequestToken={requestToken}";
|
|
|
|
return $"{message} RequestToken={requestToken}, Label={debugLabel}";
|
|
}
|
|
|
|
private static CancellationTokenSource? CreateLinkedTokenSource(CancellationToken left, CancellationToken right)
|
|
{
|
|
if (left.CanBeCanceled && right.CanBeCanceled)
|
|
return CancellationTokenSource.CreateLinkedTokenSource(left, right);
|
|
|
|
if (left.CanBeCanceled)
|
|
return CancellationTokenSource.CreateLinkedTokenSource(left);
|
|
|
|
if (right.CanBeCanceled)
|
|
return CancellationTokenSource.CreateLinkedTokenSource(right);
|
|
|
|
return null;
|
|
}
|
|
|
|
private static void EnsureResponseSucceeded<TResponse>(TResponse response)
|
|
where TResponse : class, IShrinkNetworkResponse
|
|
{
|
|
if (response == null)
|
|
throw new ShrinkRpcException(ShrinkRpcErrorCode.InvalidResponse, "RPC returned a null response.");
|
|
|
|
if (response.ErrorCode != 0)
|
|
throw new ShrinkRpcException(response.ErrorCode,
|
|
string.IsNullOrWhiteSpace(response.ErrorMessage) ? "RPC call failed." : response.ErrorMessage);
|
|
}
|
|
|
|
internal void ReportPermissionDenied(Type messageType)
|
|
{
|
|
Interlocked.Increment(ref _permissionDeniedCount);
|
|
}
|
|
|
|
internal void ReportHandlerException(Type messageType, Exception ex)
|
|
{
|
|
Interlocked.Increment(ref _handlerExceptionCount);
|
|
Interlocked.Increment(ref _rpcFailed);
|
|
}
|
|
|
|
private bool ValidatePacketCompatibility(ShrinkNetworkPacket packet, long sessionId)
|
|
{
|
|
var protocolAllowed = packet.ProtocolVersion >= MinProtocolVersion && packet.ProtocolVersion <= MaxProtocolVersion;
|
|
var schemaAllowed = packet.SchemaVersion >= MinSchemaVersion && packet.SchemaVersion <= MaxSchemaVersion;
|
|
if (protocolAllowed && schemaAllowed)
|
|
return true;
|
|
|
|
Interlocked.Increment(ref _protocolViolations);
|
|
var reason =
|
|
$"protocol/schema mismatch. Protocol={packet.ProtocolVersion}, Schema={packet.SchemaVersion}, AllowedProtocol={MinProtocolVersion}-{MaxProtocolVersion}, AllowedSchema={MinSchemaVersion}-{MaxSchemaVersion}";
|
|
ShrinkNetworkLogger.Warn($"[ShrinkNetwork] Session {sessionId} rejected: {reason}");
|
|
if (DisconnectOnProtocolViolation && _transport is IShrinkNetworkSessionControlTransport sessionControl)
|
|
sessionControl.DisconnectSession(sessionId, reason);
|
|
return false;
|
|
}
|
|
|
|
private bool ValidateIncomingPacket(ShrinkNetworkSession session, ShrinkNetworkPacket packet)
|
|
{
|
|
var validator = IncomingPacketValidator;
|
|
if (validator == null)
|
|
return true;
|
|
|
|
var result = validator(session, packet);
|
|
if (result == null || result.IsAllowed)
|
|
return true;
|
|
|
|
Interlocked.Increment(ref _authRejectedCount);
|
|
var reason = string.IsNullOrWhiteSpace(result.Reason)
|
|
? "incoming packet rejected by validator."
|
|
: result.Reason.Trim();
|
|
ShrinkNetworkLogger.Warn($"[ShrinkNetwork] Session {session.SessionId} rejected: {reason}");
|
|
if (result.DisconnectSession && _transport is IShrinkNetworkSessionControlTransport sessionControl)
|
|
sessionControl.DisconnectSession(session.SessionId, reason);
|
|
return false;
|
|
}
|
|
|
|
private static void ApplySessionTokenFromPacket(ShrinkNetworkSession session, ShrinkNetworkPacket packet)
|
|
{
|
|
if (string.IsNullOrWhiteSpace(packet.SessionToken))
|
|
return;
|
|
|
|
var expiresAtUtc = packet.SessionTokenExpiresAtUnixTimeSeconds > 0
|
|
? DateTimeOffset.FromUnixTimeSeconds(packet.SessionTokenExpiresAtUnixTimeSeconds)
|
|
: (DateTimeOffset?)null;
|
|
session.SetSessionToken(packet.SessionToken, expiresAtUtc);
|
|
}
|
|
|
|
}
|
|
|
|
public interface IShrinkNetworkDispatchScheduler
|
|
{
|
|
/// <summary>
|
|
/// Schedules a transport event. The result is false when the scheduler
|
|
/// rejects or drops the work because its bounded queue is full.
|
|
/// </summary>
|
|
UniTask<bool> ScheduleAsync(Func<UniTask> callback);
|
|
}
|
|
|
|
public static class ShrinkNetworkDispatchSchedulers
|
|
{
|
|
public static IShrinkNetworkDispatchScheduler Inline { get; } =
|
|
new ShrinkNetworkInlineDispatchScheduler();
|
|
}
|
|
|
|
public sealed class ShrinkNetworkInlineDispatchScheduler : IShrinkNetworkDispatchScheduler
|
|
{
|
|
public async UniTask<bool> ScheduleAsync(Func<UniTask> callback)
|
|
{
|
|
if (callback == null)
|
|
throw new ArgumentNullException(nameof(callback));
|
|
|
|
await callback();
|
|
return true;
|
|
}
|
|
}
|
|
|
|
public enum ShrinkNetworkDispatchOverflowPolicy
|
|
{
|
|
Reject = 0,
|
|
DropNewest = 1,
|
|
DropOldest = 2
|
|
}
|
|
|
|
/// <summary>
|
|
/// A caller-pumped, bounded dispatch queue. Unity can pump it from Update
|
|
/// while a dedicated server can keep the default inline scheduler.
|
|
/// </summary>
|
|
public sealed class ShrinkNetworkDispatchQueue : IShrinkNetworkDispatchScheduler, IDisposable
|
|
{
|
|
private sealed class WorkItem
|
|
{
|
|
public Func<UniTask> Callback = null!;
|
|
public UniTaskCompletionSource<bool> Completion = null!;
|
|
}
|
|
|
|
private readonly ConcurrentQueue<WorkItem> _queue = new();
|
|
private readonly object _lifecycleLock = new();
|
|
private readonly int _capacity;
|
|
private readonly ShrinkNetworkDispatchOverflowPolicy _overflowPolicy;
|
|
private int _queuedCount;
|
|
private int _pumping;
|
|
private int _disposed;
|
|
private long _rejectedCount;
|
|
private long _droppedCount;
|
|
|
|
public ShrinkNetworkDispatchQueue(int capacity,
|
|
ShrinkNetworkDispatchOverflowPolicy overflowPolicy = ShrinkNetworkDispatchOverflowPolicy.Reject)
|
|
{
|
|
if (capacity <= 0)
|
|
throw new ArgumentOutOfRangeException(nameof(capacity));
|
|
|
|
_capacity = capacity;
|
|
_overflowPolicy = overflowPolicy;
|
|
}
|
|
|
|
public int Capacity => _capacity;
|
|
public int PendingCount => Volatile.Read(ref _queuedCount);
|
|
public long RejectedCount => Volatile.Read(ref _rejectedCount);
|
|
public long DroppedCount => Volatile.Read(ref _droppedCount);
|
|
|
|
public UniTask<bool> ScheduleAsync(Func<UniTask> callback)
|
|
{
|
|
if (callback == null)
|
|
throw new ArgumentNullException(nameof(callback));
|
|
if (Volatile.Read(ref _disposed) != 0)
|
|
return UniTask.FromException<bool>(new ObjectDisposedException(nameof(ShrinkNetworkDispatchQueue)));
|
|
|
|
var item = new WorkItem
|
|
{
|
|
Callback = callback,
|
|
Completion = new UniTaskCompletionSource<bool>()
|
|
};
|
|
|
|
while (true)
|
|
{
|
|
if (Volatile.Read(ref _queuedCount) >= _capacity)
|
|
{
|
|
switch (_overflowPolicy)
|
|
{
|
|
case ShrinkNetworkDispatchOverflowPolicy.Reject:
|
|
Interlocked.Increment(ref _rejectedCount);
|
|
return UniTask.FromResult(false);
|
|
case ShrinkNetworkDispatchOverflowPolicy.DropNewest:
|
|
Interlocked.Increment(ref _droppedCount);
|
|
return UniTask.FromResult(false);
|
|
case ShrinkNetworkDispatchOverflowPolicy.DropOldest:
|
|
if (_queue.TryDequeue(out var dropped))
|
|
{
|
|
Interlocked.Decrement(ref _queuedCount);
|
|
Interlocked.Increment(ref _droppedCount);
|
|
dropped.Completion.TrySetResult(false);
|
|
continue;
|
|
}
|
|
|
|
Thread.Yield();
|
|
continue;
|
|
default:
|
|
throw new ArgumentOutOfRangeException();
|
|
}
|
|
}
|
|
|
|
var currentCount = Volatile.Read(ref _queuedCount);
|
|
if (currentCount >= _capacity ||
|
|
Interlocked.CompareExchange(ref _queuedCount, currentCount + 1, currentCount) != currentCount)
|
|
{
|
|
continue;
|
|
}
|
|
|
|
lock (_lifecycleLock)
|
|
{
|
|
if (Volatile.Read(ref _disposed) != 0)
|
|
{
|
|
Interlocked.Decrement(ref _queuedCount);
|
|
item.Completion.TrySetResult(false);
|
|
return item.Completion.Task;
|
|
}
|
|
|
|
_queue.Enqueue(item);
|
|
return item.Completion.Task;
|
|
}
|
|
}
|
|
}
|
|
|
|
public UniTask<int> PumpAsync(int maxItems)
|
|
{
|
|
if (maxItems <= 0)
|
|
throw new ArgumentOutOfRangeException(nameof(maxItems));
|
|
if (Interlocked.Exchange(ref _pumping, 1) == 1)
|
|
return UniTask.FromResult(0);
|
|
|
|
return PumpCoreAsync(maxItems);
|
|
}
|
|
|
|
public void Dispose()
|
|
{
|
|
lock (_lifecycleLock)
|
|
{
|
|
if (Interlocked.Exchange(ref _disposed, 1) != 0)
|
|
return;
|
|
|
|
while (_queue.TryDequeue(out var item))
|
|
{
|
|
Interlocked.Decrement(ref _queuedCount);
|
|
item.Completion.TrySetResult(false);
|
|
}
|
|
}
|
|
}
|
|
|
|
private async UniTask<int> PumpCoreAsync(int maxItems)
|
|
{
|
|
var processed = 0;
|
|
try
|
|
{
|
|
while (processed < maxItems && _queue.TryDequeue(out var item))
|
|
{
|
|
Interlocked.Decrement(ref _queuedCount);
|
|
await ExecuteItemAsync(item);
|
|
processed++;
|
|
}
|
|
|
|
return processed;
|
|
}
|
|
finally
|
|
{
|
|
Volatile.Write(ref _pumping, 0);
|
|
}
|
|
}
|
|
|
|
private static async UniTask ExecuteItemAsync(WorkItem item)
|
|
{
|
|
try
|
|
{
|
|
await item.Callback();
|
|
item.Completion.TrySetResult(true);
|
|
}
|
|
catch (Exception ex)
|
|
{
|
|
item.Completion.TrySetException(ex);
|
|
}
|
|
}
|
|
}
|
|
|
|
public sealed class ShrinkNetworkServiceDiagnosticsSnapshot
|
|
{
|
|
public int CurrentSessions { get; set; }
|
|
public long SessionsConnected { get; set; }
|
|
public long SessionsDisconnected { get; set; }
|
|
public long PacketsSent { get; set; }
|
|
public long PacketsReceived { get; set; }
|
|
public long BytesSent { get; set; }
|
|
public long BytesReceived { get; set; }
|
|
public long RpcStarted { get; set; }
|
|
public long RpcCompleted { get; set; }
|
|
public long RpcTimedOut { get; set; }
|
|
public long RpcCanceled { get; set; }
|
|
public long RpcFailed { get; set; }
|
|
public long ProtocolViolations { get; set; }
|
|
public long AuthRejectedCount { get; set; }
|
|
public long PermissionDeniedCount { get; set; }
|
|
public long HandlerExceptionCount { get; set; }
|
|
public long UnknownOpcodeCount { get; set; }
|
|
public long DispatchMissCount { get; set; }
|
|
public long SerializationErrorCount { get; set; }
|
|
public long DispatchQueueRejectedCount { get; set; }
|
|
}
|
|
|
|
public sealed class ShrinkIncomingPacketValidationResult
|
|
{
|
|
public static readonly ShrinkIncomingPacketValidationResult Allow = new()
|
|
{
|
|
IsAllowed = true
|
|
};
|
|
|
|
public bool IsAllowed { get; set; }
|
|
public bool DisconnectSession { get; set; }
|
|
public string Reason { get; set; } = string.Empty;
|
|
|
|
public static ShrinkIncomingPacketValidationResult Reject(string reason, bool disconnectSession = true)
|
|
{
|
|
return new ShrinkIncomingPacketValidationResult
|
|
{
|
|
IsAllowed = false,
|
|
DisconnectSession = disconnectSession,
|
|
Reason = reason ?? string.Empty
|
|
};
|
|
}
|
|
}
|
|
}
|