From 3ef1abf2a6e64351b5266ac2a6986737b505c466 Mon Sep 17 00:00:00 2001 From: hatayama Date: Thu, 9 Jul 2026 07:42:54 +0900 Subject: [PATCH 1/2] Give the bridge server factory explicit file ownership Move the existing public factory unchanged into its matching infrastructure file so the remaining server source can focus on listener and session lifecycle responsibilities. --- .../UnityCliLoopBridgeServer.cs | 46 ---------------- ...UnityCliLoopBridgeServerInstanceFactory.cs | 53 +++++++++++++++++++ ...CliLoopBridgeServerInstanceFactory.cs.meta | 11 ++++ 3 files changed, 64 insertions(+), 46 deletions(-) create mode 100644 Packages/src/Editor/Infrastructure/UnityCliLoopBridgeServerInstanceFactory.cs create mode 100644 Packages/src/Editor/Infrastructure/UnityCliLoopBridgeServerInstanceFactory.cs.meta diff --git a/Packages/src/Editor/Infrastructure/UnityCliLoopBridgeServer.cs b/Packages/src/Editor/Infrastructure/UnityCliLoopBridgeServer.cs index e0575d39c3..684c0bafb3 100644 --- a/Packages/src/Editor/Infrastructure/UnityCliLoopBridgeServer.cs +++ b/Packages/src/Editor/Infrastructure/UnityCliLoopBridgeServer.cs @@ -14,52 +14,6 @@ namespace io.github.hatayama.UnityCliLoop.Infrastructure { - /// - /// Creates Unity CLI Loop Bridge Server Instance instances with the dependencies required by this module. - /// - public sealed class UnityCliLoopBridgeServerInstanceFactory : - IUnityCliLoopServerInstanceFactory, - IUnityCliLoopServerLifecycleSource - { - public event Action ServerLoopExited; - private readonly IDomainReloadDetectionService _domainReloadDetectionService; - private readonly UnityCliLoopToolRegistrarService _toolRegistrarService; - - internal UnityCliLoopBridgeServerInstanceFactory( - IDomainReloadDetectionService domainReloadDetectionService, - UnityCliLoopToolRegistrarService toolRegistrarService) - { - System.Diagnostics.Debug.Assert(domainReloadDetectionService != null, "domainReloadDetectionService must not be null"); - System.Diagnostics.Debug.Assert(toolRegistrarService != null, "toolRegistrarService must not be null"); - - _domainReloadDetectionService = domainReloadDetectionService - ?? throw new ArgumentNullException(nameof(domainReloadDetectionService)); - _toolRegistrarService = toolRegistrarService - ?? throw new ArgumentNullException(nameof(toolRegistrarService)); - } - - public IUnityCliLoopServerInstance Create() - { - UnityCliLoopBridgeHeartbeatService heartbeatService = new(); - UnityCliLoopBridgeClientDisconnectMonitor clientDisconnectMonitor = new(); - UnityCliLoopExecutionRouter executionRouter = new(_toolRegistrarService); - JsonRpcRequestProcessor jsonRpcRequestProcessor = new(executionRouter); - UnityCliLoopBridgeServer server = new( - _domainReloadDetectionService, - jsonRpcRequestProcessor, - heartbeatService, - clientDisconnectMonitor); - server.ServerLoopExited += NotifyServerLoopExited; - - return server; - } - - private void NotifyServerLoopExited() - { - ServerLoopExited?.Invoke(); - } - } - /// /// Unity CLI bridge server. /// Accepts project-local CLI connections and handles JSON-RPC 2.0 communication. diff --git a/Packages/src/Editor/Infrastructure/UnityCliLoopBridgeServerInstanceFactory.cs b/Packages/src/Editor/Infrastructure/UnityCliLoopBridgeServerInstanceFactory.cs new file mode 100644 index 0000000000..38319074d6 --- /dev/null +++ b/Packages/src/Editor/Infrastructure/UnityCliLoopBridgeServerInstanceFactory.cs @@ -0,0 +1,53 @@ +using System; + +using io.github.hatayama.UnityCliLoop.Application; +using io.github.hatayama.UnityCliLoop.Domain; + +namespace io.github.hatayama.UnityCliLoop.Infrastructure +{ + /// + /// Creates Unity CLI Loop Bridge Server Instance instances with the dependencies required by this module. + /// + public sealed class UnityCliLoopBridgeServerInstanceFactory : + IUnityCliLoopServerInstanceFactory, + IUnityCliLoopServerLifecycleSource + { + public event Action ServerLoopExited; + private readonly IDomainReloadDetectionService _domainReloadDetectionService; + private readonly UnityCliLoopToolRegistrarService _toolRegistrarService; + + internal UnityCliLoopBridgeServerInstanceFactory( + IDomainReloadDetectionService domainReloadDetectionService, + UnityCliLoopToolRegistrarService toolRegistrarService) + { + System.Diagnostics.Debug.Assert(domainReloadDetectionService != null, "domainReloadDetectionService must not be null"); + System.Diagnostics.Debug.Assert(toolRegistrarService != null, "toolRegistrarService must not be null"); + + _domainReloadDetectionService = domainReloadDetectionService + ?? throw new ArgumentNullException(nameof(domainReloadDetectionService)); + _toolRegistrarService = toolRegistrarService + ?? throw new ArgumentNullException(nameof(toolRegistrarService)); + } + + public IUnityCliLoopServerInstance Create() + { + UnityCliLoopBridgeHeartbeatService heartbeatService = new(); + UnityCliLoopBridgeClientDisconnectMonitor clientDisconnectMonitor = new(); + UnityCliLoopExecutionRouter executionRouter = new(_toolRegistrarService); + JsonRpcRequestProcessor jsonRpcRequestProcessor = new(executionRouter); + UnityCliLoopBridgeServer server = new( + _domainReloadDetectionService, + jsonRpcRequestProcessor, + heartbeatService, + clientDisconnectMonitor); + server.ServerLoopExited += NotifyServerLoopExited; + + return server; + } + + private void NotifyServerLoopExited() + { + ServerLoopExited?.Invoke(); + } + } +} diff --git a/Packages/src/Editor/Infrastructure/UnityCliLoopBridgeServerInstanceFactory.cs.meta b/Packages/src/Editor/Infrastructure/UnityCliLoopBridgeServerInstanceFactory.cs.meta new file mode 100644 index 0000000000..ff040005b4 --- /dev/null +++ b/Packages/src/Editor/Infrastructure/UnityCliLoopBridgeServerInstanceFactory.cs.meta @@ -0,0 +1,11 @@ +fileFormatVersion: 2 +guid: 620d1d36ae77447990654fa185c55822 +MonoImporter: + externalObjects: {} + serializedVersion: 2 + defaultReferences: [] + executionOrder: 0 + icon: {instanceID: 0} + userData: + assetBundleName: + assetBundleVariant: From d5ab49387abdd97c3ac424ecdb016ddb54137e05 Mon Sep 17 00:00:00 2001 From: hatayama Date: Thu, 9 Jul 2026 07:46:23 +0900 Subject: [PATCH 2/2] Move bridge client sessions into a dedicated owner Give one instance manager ownership of connected streams, handler tasks, request framing coordination, and session cleanup. Keep the server responsible for listener acceptance, server cancellation, shutdown orchestration, and lifecycle events while preserving every handshake and task transition. --- .../UnityCliLoopBridgeClientSessionManager.cs | 358 ++++++++++++++++++ ...yCliLoopBridgeClientSessionManager.cs.meta | 11 + .../UnityCliLoopBridgeServer.cs | 345 +---------------- 3 files changed, 382 insertions(+), 332 deletions(-) create mode 100644 Packages/src/Editor/Infrastructure/UnityCliLoopBridgeClientSessionManager.cs create mode 100644 Packages/src/Editor/Infrastructure/UnityCliLoopBridgeClientSessionManager.cs.meta diff --git a/Packages/src/Editor/Infrastructure/UnityCliLoopBridgeClientSessionManager.cs b/Packages/src/Editor/Infrastructure/UnityCliLoopBridgeClientSessionManager.cs new file mode 100644 index 0000000000..26a42bcba9 --- /dev/null +++ b/Packages/src/Editor/Infrastructure/UnityCliLoopBridgeClientSessionManager.cs @@ -0,0 +1,358 @@ +using System; +using System.Collections.Concurrent; +using System.Collections.Generic; +using System.IO; +using System.Linq; +using System.Net.Sockets; +using System.Threading; +using System.Threading.Tasks; +using UnityEngine; + +using io.github.hatayama.UnityCliLoop.Application; +using io.github.hatayama.UnityCliLoop.ToolContracts; + +namespace io.github.hatayama.UnityCliLoop.Infrastructure +{ + /// + /// Owns connected client streams, handler tasks, and per-request project IPC session processing. + /// + internal sealed class UnityCliLoopBridgeClientSessionManager + { + private readonly JsonRpcRequestProcessor _jsonRpcRequestProcessor; + private readonly UnityCliLoopBridgeHeartbeatService _heartbeatService; + private readonly UnityCliLoopBridgeClientDisconnectMonitor _clientDisconnectMonitor; + + // HResult error codes for normal disconnection detection + private static readonly HashSet NormalDisconnectionHResults = new() + { + unchecked((int)0x800703E3), // ERROR_OPERATION_ABORTED + unchecked((int)0x80070040), // ERROR_NETNAME_DELETED + unchecked((int)0x80072745), // ERROR_CONNECTION_ABORTED + unchecked((int)0x80072746) // ERROR_CONNECTION_RESET + }; + + private readonly ConcurrentDictionary _clientStreams = new(); + private readonly ConcurrentDictionary _clientTasks = new(); + private int _nextClientTaskId; + + internal UnityCliLoopBridgeClientSessionManager( + JsonRpcRequestProcessor jsonRpcRequestProcessor, + UnityCliLoopBridgeHeartbeatService heartbeatService, + UnityCliLoopBridgeClientDisconnectMonitor clientDisconnectMonitor) + { + System.Diagnostics.Debug.Assert(jsonRpcRequestProcessor != null, "jsonRpcRequestProcessor must not be null"); + System.Diagnostics.Debug.Assert(heartbeatService != null, "heartbeatService must not be null"); + System.Diagnostics.Debug.Assert(clientDisconnectMonitor != null, "clientDisconnectMonitor must not be null"); + + _jsonRpcRequestProcessor = jsonRpcRequestProcessor; + _heartbeatService = heartbeatService; + _clientDisconnectMonitor = clientDisconnectMonitor; + } + + /// + /// Explicitly disconnect all connected clients + /// This ensures CLI clients receive proper close events + /// + internal void DisconnectAllClients() + { + if (_clientStreams.IsEmpty) + { + return; + } + + List clientsToRemove = new(); + + foreach (KeyValuePair client in _clientStreams) + { + try + { + if (client.Value != null && client.Value.CanWrite) + { + client.Value.Close(); + } + clientsToRemove.Add(client.Key); + } + catch (Exception ex) + { + VibeLogger.LogWarning( + "client_disconnect_failed", + $"Error disconnecting client {client.Key}: {ex.Message}"); + clientsToRemove.Add(client.Key); // Remove even if disconnect failed + } + } + + // Remove all clients from the connected clients list + foreach (string clientKey in clientsToRemove) + { + _clientStreams.TryRemove(clientKey, out _); + } + } + + internal void StartClientHandler(BridgeClientConnection client, CancellationToken cancellationToken) + { + int taskId = Interlocked.Increment(ref _nextClientTaskId); + Task clientTask = Task.Run(() => HandleClientAsync(client, cancellationToken)); + _clientTasks.TryAdd(taskId, clientTask); + clientTask.ContinueWith( + task => + { + _clientTasks.TryRemove(taskId, out _); + if (task.IsFaulted && task.Exception != null) + { + // HandleClientAsync catches expected failures itself, so a faulted task is + // an anomaly worth surfacing in the console for non-debug installs. + Debug.LogError( + $"[{UnityCliLoopConstants.PROJECT_NAME}] Client handler task faulted: {task.Exception.GetBaseException()}"); + } + }, + CancellationToken.None, + TaskContinuationOptions.ExecuteSynchronously, + TaskScheduler.Default); + } + + internal Task[] GetActiveClientTasks() + { + return _clientTasks.Values.ToArray(); + } + + /// + /// Handles communication with the client using Content-Length framing. + /// + private async Task HandleClientAsync(BridgeClientConnection client, CancellationToken cancellationToken) + { + string clientKey = client.Endpoint; + + // Initialize new components for Content-Length framing + DynamicBufferManager bufferManager = null; + MessageReassembler messageReassembler = null; + + try + { + using (client) + using (Stream stream = client.Stream) + { + + // Check for existing connection from same endpoint and close it + if (_clientStreams.TryRemove(clientKey, out Stream existingStream)) + { + existingStream?.Close(); + } + + _clientStreams.TryAdd(clientKey, stream); + + // Initialize new framing components + bufferManager = new DynamicBufferManager(); + messageReassembler = new MessageReassembler(bufferManager); + + // Not disposed deliberately: a heartbeat writer could still be releasing it + // during teardown, and an undisposed SemaphoreSlim holds no unmanaged state. + SemaphoreSlim streamWriteLock = new(1, 1); + + // Start with initial buffer size + byte[] buffer = bufferManager.GetBuffer(BufferConfig.INITIAL_BUFFER_SIZE); + + while (!cancellationToken.IsCancellationRequested) + { + int bytesRead = await stream.ReadAsync(buffer, 0, buffer.Length, cancellationToken); + + if (bytesRead == 0) + { + break; // Client disconnected. + } + + // Add received data to message reassembler + messageReassembler.AddData(buffer, bytesRead); + + // Extract any complete messages + string[] completeJsonMessages = messageReassembler.ExtractCompleteMessages(); + + foreach (string requestJson in completeJsonMessages) + { + if (string.IsNullOrWhiteSpace(requestJson)) continue; + + await ProcessRequestFrameAsync(client, stream, streamWriteLock, requestJson, cancellationToken); + } + + // Why: false means the reassembler was disposed and can no longer make + // progress; closing the connection lets the CLI observe the disconnect + // immediately instead of waiting silently for its own timeout. + // (Corrupted framing state throws from ValidateState and is handled below.) + if (!messageReassembler.ValidateState()) + { + break; + } + } + } + } + catch (ThreadAbortException) + { + // Treat as normal behavior if a domain reload is in progress. + // No need to log thread aborts during domain reload + } + catch (OperationCanceledException) + { + // Normal cancellation during server shutdown or domain reload + // No logging needed as this is expected behavior during Unity Editor operations + } + catch (IOException ex) + { + // I/O errors are usually normal disconnections - only log as info instead of warning + if (IsNormalDisconnectionException(ex)) + { + // Log normal disconnections as info level + } + } + catch (Exception ex) + { + // Expected failures (cancellation, reload aborts, normal disconnects) are filtered + // above, so anything reaching here means a CLI session died abnormally and the + // console must say so even on installs where VibeLogger is compiled out. + Debug.LogError( + $"[{UnityCliLoopConstants.PROJECT_NAME}] Client session for {clientKey} failed: {ex}"); + } + finally + { + // Dispose of framing components + try + { + messageReassembler?.Dispose(); + bufferManager?.Dispose(); + } + catch (Exception ex) + { + VibeLogger.LogWarning( + "client_dispose_failed", + $"Error during client disposal: {ex.Message}"); + } + + _clientStreams.TryRemove(clientKey, out _); + + client.Dispose(); + } + } + + private async Task ProcessRequestFrameAsync( + BridgeClientConnection client, + Stream stream, + SemaphoreSlim streamWriteLock, + string requestJson, + CancellationToken serverCancellationToken) + { + using (CancellationTokenSource requestCancellationTokenSource = + CancellationTokenSource.CreateLinkedTokenSource(serverCancellationToken)) + { + Task clientDisconnectMonitorTask = null; + Task heartbeatTask = null; + CancellationTokenSource heartbeatCancellationSource = null; + try + { + string responseJson = await _jsonRpcRequestProcessor.ProcessRequestWithEarlyResponseAsync( + requestJson, + requestCancellationTokenSource.Token, + async (responseJsonValue, cancelOnClientDisconnect, createHeartbeatJson) => + { + await UnityCliLoopBridgeResponseWriter.WriteJsonResponseLockedAsync( + stream, streamWriteLock, responseJsonValue, serverCancellationToken); + + if (createHeartbeatJson != null) + { + heartbeatCancellationSource = CancellationTokenSource.CreateLinkedTokenSource( + requestCancellationTokenSource.Token); + // Why: the heartbeat token must govern the write as well; with the + // server token an in-flight write could ignore StopHeartbeatsAsync + // and stall the final response behind a slow client. + CancellationToken heartbeatToken = heartbeatCancellationSource.Token; + heartbeatTask = _heartbeatService.SendHeartbeatsAsync( + createHeartbeatJson, + heartbeatJson => UnityCliLoopBridgeResponseWriter.WriteJsonResponseLockedAsync( + stream, streamWriteLock, heartbeatJson, heartbeatToken), + TimeSpan.FromSeconds(UnityCliLoopServerConfig.HEARTBEAT_INTERVAL_SECONDS), + heartbeatToken); + } + + if (!cancelOnClientDisconnect) + { + return; + } + + clientDisconnectMonitorTask = + _clientDisconnectMonitor.MonitorClientDisconnectAsync( + client, + requestCancellationTokenSource); + }); + + // Stop heartbeats before the final response so no heartbeat frame can be + // queued after the response the CLI stops reading at. + await _heartbeatService.StopHeartbeatsAsync(heartbeatTask, heartbeatCancellationSource); + heartbeatTask = null; + + await UnityCliLoopBridgeResponseWriter.WriteJsonResponseLockedAsync( + stream, streamWriteLock, responseJson, serverCancellationToken); + } + finally + { + await _heartbeatService.StopHeartbeatsAsync(heartbeatTask, heartbeatCancellationSource); + heartbeatCancellationSource?.Dispose(); + await _clientDisconnectMonitor.StopClientDisconnectMonitorAsync( + clientDisconnectMonitorTask, + requestCancellationTokenSource); + } + } + } + + /// + /// Determines if the given exception represents a normal client disconnection. + /// + /// The exception to evaluate + /// True if the exception represents a normal disconnection, false otherwise + private static bool IsNormalDisconnectionException(Exception ex) + { + switch (ex) + { + case SocketException sockEx: + return sockEx.SocketErrorCode is SocketError.ConnectionReset or + SocketError.ConnectionAborted or + SocketError.OperationAborted or + SocketError.Shutdown or + SocketError.NotConnected; + + case ObjectDisposedException: + return true; + + case IOException ioEx when ioEx.InnerException is SocketException innerSockEx: + return innerSockEx.SocketErrorCode is SocketError.ConnectionReset or + SocketError.ConnectionAborted or + SocketError.OperationAborted or + SocketError.Shutdown or + SocketError.NotConnected; + + case IOException ioEx: + // Check HResult codes for common disconnection scenarios + return NormalDisconnectionHResults.Contains(ioEx.HResult) || + IsNormalDisconnectionByInnerException(ioEx); + + default: + return false; + } + } + + /// + /// Recursively checks inner exceptions for normal disconnection scenarios + /// + /// The exception to check + /// True if any inner exception indicates a normal disconnection + private static bool IsNormalDisconnectionByInnerException(Exception ex) + { + Exception innerEx = ex.InnerException; + while (innerEx != null) + { + if (IsNormalDisconnectionException(innerEx)) + { + return true; + } + innerEx = innerEx.InnerException; + } + return false; + } + } +} diff --git a/Packages/src/Editor/Infrastructure/UnityCliLoopBridgeClientSessionManager.cs.meta b/Packages/src/Editor/Infrastructure/UnityCliLoopBridgeClientSessionManager.cs.meta new file mode 100644 index 0000000000..28b69480dc --- /dev/null +++ b/Packages/src/Editor/Infrastructure/UnityCliLoopBridgeClientSessionManager.cs.meta @@ -0,0 +1,11 @@ +fileFormatVersion: 2 +guid: e62d177cf2934c958c6f86185877f86b +MonoImporter: + externalObjects: {} + serializedVersion: 2 + defaultReferences: [] + executionOrder: 0 + icon: {instanceID: 0} + userData: + assetBundleName: + assetBundleVariant: diff --git a/Packages/src/Editor/Infrastructure/UnityCliLoopBridgeServer.cs b/Packages/src/Editor/Infrastructure/UnityCliLoopBridgeServer.cs index 684c0bafb3..d3894afe97 100644 --- a/Packages/src/Editor/Infrastructure/UnityCliLoopBridgeServer.cs +++ b/Packages/src/Editor/Infrastructure/UnityCliLoopBridgeServer.cs @@ -1,7 +1,5 @@ using System; using System.Collections.Generic; -using System.Collections.Concurrent; -using System.IO; using System.Linq; using System.Net.Sockets; using System.Threading; @@ -24,18 +22,7 @@ public class UnityCliLoopBridgeServer : IUnityCliLoopServerInstance // Subscribers must marshal to main thread before accessing Unity APIs. public event Action ServerLoopExited; private readonly IDomainReloadDetectionService _domainReloadDetectionService; - private readonly JsonRpcRequestProcessor _jsonRpcRequestProcessor; - private readonly UnityCliLoopBridgeHeartbeatService _heartbeatService; - private readonly UnityCliLoopBridgeClientDisconnectMonitor _clientDisconnectMonitor; - - // HResult error codes for normal disconnection detection - private static readonly HashSet NormalDisconnectionHResults = new() - { - unchecked((int)0x800703E3), // ERROR_OPERATION_ABORTED - unchecked((int)0x80070040), // ERROR_NETNAME_DELETED - unchecked((int)0x80072745), // ERROR_CONNECTION_ABORTED - unchecked((int)0x80072746) // ERROR_CONNECTION_RESET - }; + private readonly UnityCliLoopBridgeClientSessionManager _clientSessionManager; private IBridgeTransportListener _transportListener; private CancellationTokenSource _cancellationTokenSource; @@ -46,10 +33,6 @@ public class UnityCliLoopBridgeServer : IUnityCliLoopServerInstance // Guard against concurrent cleanup from ServerLoopAsync finally + external disposal private int _unexpectedExitCleanupStarted = 0; - private readonly ConcurrentDictionary _clientStreams = new(); - private readonly ConcurrentDictionary _clientTasks = new(); - private int _nextClientTaskId; - internal UnityCliLoopBridgeServer( IDomainReloadDetectionService domainReloadDetectionService, JsonRpcRequestProcessor jsonRpcRequestProcessor, @@ -63,12 +46,16 @@ internal UnityCliLoopBridgeServer( _domainReloadDetectionService = domainReloadDetectionService ?? throw new ArgumentNullException(nameof(domainReloadDetectionService)); - _jsonRpcRequestProcessor = jsonRpcRequestProcessor + JsonRpcRequestProcessor validatedJsonRpcRequestProcessor = jsonRpcRequestProcessor ?? throw new ArgumentNullException(nameof(jsonRpcRequestProcessor)); - _heartbeatService = heartbeatService + UnityCliLoopBridgeHeartbeatService validatedHeartbeatService = heartbeatService ?? throw new ArgumentNullException(nameof(heartbeatService)); - _clientDisconnectMonitor = clientDisconnectMonitor + UnityCliLoopBridgeClientDisconnectMonitor validatedClientDisconnectMonitor = clientDisconnectMonitor ?? throw new ArgumentNullException(nameof(clientDisconnectMonitor)); + _clientSessionManager = new UnityCliLoopBridgeClientSessionManager( + validatedJsonRpcRequestProcessor, + validatedHeartbeatService, + validatedClientDisconnectMonitor); } /// @@ -140,7 +127,7 @@ public void StopServer() cancellationTokenSource?.Cancel(); // Explicitly disconnect all connected clients before stopping the server - DisconnectAllClients(); + _clientSessionManager.DisconnectAllClients(); try { @@ -152,7 +139,7 @@ public void StopServer() } Task serverTask = _serverTask; - Task[] clientTasks = GetActiveClientTasks(); + Task[] clientTasks = _clientSessionManager.GetActiveClientTasks(); _serverTask = null; DisposeCancellationSourceAfterServerTaskAsync( serverTask, @@ -210,45 +197,6 @@ private static Task[] BuildShutdownWaitTasks(Task serverTask, Task[] clientTasks return tasks.ToArray(); } - /// - /// Explicitly disconnect all connected clients - /// This ensures CLI clients receive proper close events - /// - private void DisconnectAllClients() - { - if (_clientStreams.IsEmpty) - { - return; - } - - List clientsToRemove = new(); - - foreach (KeyValuePair client in _clientStreams) - { - try - { - if (client.Value != null && client.Value.CanWrite) - { - client.Value.Close(); - } - clientsToRemove.Add(client.Key); - } - catch (Exception ex) - { - VibeLogger.LogWarning( - "client_disconnect_failed", - $"Error disconnecting client {client.Key}: {ex.Message}"); - clientsToRemove.Add(client.Key); // Remove even if disconnect failed - } - } - - // Remove all clients from the connected clients list - foreach (string clientKey in clientsToRemove) - { - _clientStreams.TryRemove(clientKey, out _); - } - } - /// /// StopServer() guards on _isRunning==true, but by the time this runs _isRunning may already /// be false or the normal shutdown path may race with the finally block. @@ -263,7 +211,7 @@ private void CleanupAfterUnexpectedLoopExit() return; } - DisconnectAllClients(); + _clientSessionManager.DisconnectAllClients(); CancellationTokenSource cancellationTokenSource = TakeCancellationTokenSource(); cancellationTokenSource?.Cancel(); @@ -294,7 +242,7 @@ private async Task ServerLoopAsync(CancellationToken cancellationToken) BridgeClientConnection client = await AcceptClientAsync(_transportListener, cancellationToken); if (client != null) { - StartClientHandler(client, cancellationToken); + _clientSessionManager.StartClientHandler(client, cancellationToken); } } catch (ObjectDisposedException) @@ -362,33 +310,6 @@ private async Task ServerLoopAsync(CancellationToken cancellationToken) } } - private void StartClientHandler(BridgeClientConnection client, CancellationToken cancellationToken) - { - int taskId = Interlocked.Increment(ref _nextClientTaskId); - Task clientTask = Task.Run(() => HandleClientAsync(client, cancellationToken)); - _clientTasks.TryAdd(taskId, clientTask); - clientTask.ContinueWith( - task => - { - _clientTasks.TryRemove(taskId, out _); - if (task.IsFaulted && task.Exception != null) - { - // HandleClientAsync catches expected failures itself, so a faulted task is - // an anomaly worth surfacing in the console for non-debug installs. - Debug.LogError( - $"[{UnityCliLoopConstants.PROJECT_NAME}] Client handler task faulted: {task.Exception.GetBaseException()}"); - } - }, - CancellationToken.None, - TaskContinuationOptions.ExecuteSynchronously, - TaskScheduler.Default); - } - - private Task[] GetActiveClientTasks() - { - return _clientTasks.Values.ToArray(); - } - private async Task AcceptClientAsync(IBridgeTransportListener listener, CancellationToken cancellationToken) { try @@ -412,246 +333,6 @@ private async Task AcceptClientAsync(IBridgeTransportLis } } - /// - /// Handles communication with the client using Content-Length framing. - /// - private async Task HandleClientAsync(BridgeClientConnection client, CancellationToken cancellationToken) - { - string clientKey = client.Endpoint; - - // Initialize new components for Content-Length framing - DynamicBufferManager bufferManager = null; - MessageReassembler messageReassembler = null; - - try - { - using (client) - using (Stream stream = client.Stream) - { - - // Check for existing connection from same endpoint and close it - if (_clientStreams.TryRemove(clientKey, out Stream existingStream)) - { - existingStream?.Close(); - } - - _clientStreams.TryAdd(clientKey, stream); - - // Initialize new framing components - bufferManager = new DynamicBufferManager(); - messageReassembler = new MessageReassembler(bufferManager); - - // Not disposed deliberately: a heartbeat writer could still be releasing it - // during teardown, and an undisposed SemaphoreSlim holds no unmanaged state. - SemaphoreSlim streamWriteLock = new(1, 1); - - // Start with initial buffer size - byte[] buffer = bufferManager.GetBuffer(BufferConfig.INITIAL_BUFFER_SIZE); - - while (!cancellationToken.IsCancellationRequested) - { - int bytesRead = await stream.ReadAsync(buffer, 0, buffer.Length, cancellationToken); - - if (bytesRead == 0) - { - break; // Client disconnected. - } - - // Add received data to message reassembler - messageReassembler.AddData(buffer, bytesRead); - - // Extract any complete messages - string[] completeJsonMessages = messageReassembler.ExtractCompleteMessages(); - - foreach (string requestJson in completeJsonMessages) - { - if (string.IsNullOrWhiteSpace(requestJson)) continue; - - await ProcessRequestFrameAsync(client, stream, streamWriteLock, requestJson, cancellationToken); - } - - // Why: false means the reassembler was disposed and can no longer make - // progress; closing the connection lets the CLI observe the disconnect - // immediately instead of waiting silently for its own timeout. - // (Corrupted framing state throws from ValidateState and is handled below.) - if (!messageReassembler.ValidateState()) - { - break; - } - } - } - } - catch (ThreadAbortException) - { - // Treat as normal behavior if a domain reload is in progress. - // No need to log thread aborts during domain reload - } - catch (OperationCanceledException) - { - // Normal cancellation during server shutdown or domain reload - // No logging needed as this is expected behavior during Unity Editor operations - } - catch (IOException ex) - { - // I/O errors are usually normal disconnections - only log as info instead of warning - if (IsNormalDisconnectionException(ex)) - { - // Log normal disconnections as info level - } - } - catch (Exception ex) - { - // Expected failures (cancellation, reload aborts, normal disconnects) are filtered - // above, so anything reaching here means a CLI session died abnormally and the - // console must say so even on installs where VibeLogger is compiled out. - Debug.LogError( - $"[{UnityCliLoopConstants.PROJECT_NAME}] Client session for {clientKey} failed: {ex}"); - } - finally - { - // Dispose of framing components - try - { - messageReassembler?.Dispose(); - bufferManager?.Dispose(); - } - catch (Exception ex) - { - VibeLogger.LogWarning( - "client_dispose_failed", - $"Error during client disposal: {ex.Message}"); - } - - _clientStreams.TryRemove(clientKey, out _); - - client.Dispose(); - } - } - - private async Task ProcessRequestFrameAsync( - BridgeClientConnection client, - Stream stream, - SemaphoreSlim streamWriteLock, - string requestJson, - CancellationToken serverCancellationToken) - { - using (CancellationTokenSource requestCancellationTokenSource = - CancellationTokenSource.CreateLinkedTokenSource(serverCancellationToken)) - { - Task clientDisconnectMonitorTask = null; - Task heartbeatTask = null; - CancellationTokenSource heartbeatCancellationSource = null; - try - { - string responseJson = await _jsonRpcRequestProcessor.ProcessRequestWithEarlyResponseAsync( - requestJson, - requestCancellationTokenSource.Token, - async (responseJsonValue, cancelOnClientDisconnect, createHeartbeatJson) => - { - await UnityCliLoopBridgeResponseWriter.WriteJsonResponseLockedAsync( - stream, streamWriteLock, responseJsonValue, serverCancellationToken); - - if (createHeartbeatJson != null) - { - heartbeatCancellationSource = CancellationTokenSource.CreateLinkedTokenSource( - requestCancellationTokenSource.Token); - // Why: the heartbeat token must govern the write as well; with the - // server token an in-flight write could ignore StopHeartbeatsAsync - // and stall the final response behind a slow client. - CancellationToken heartbeatToken = heartbeatCancellationSource.Token; - heartbeatTask = _heartbeatService.SendHeartbeatsAsync( - createHeartbeatJson, - heartbeatJson => UnityCliLoopBridgeResponseWriter.WriteJsonResponseLockedAsync( - stream, streamWriteLock, heartbeatJson, heartbeatToken), - TimeSpan.FromSeconds(UnityCliLoopServerConfig.HEARTBEAT_INTERVAL_SECONDS), - heartbeatToken); - } - - if (!cancelOnClientDisconnect) - { - return; - } - - clientDisconnectMonitorTask = - _clientDisconnectMonitor.MonitorClientDisconnectAsync( - client, - requestCancellationTokenSource); - }); - - // Stop heartbeats before the final response so no heartbeat frame can be - // queued after the response the CLI stops reading at. - await _heartbeatService.StopHeartbeatsAsync(heartbeatTask, heartbeatCancellationSource); - heartbeatTask = null; - - await UnityCliLoopBridgeResponseWriter.WriteJsonResponseLockedAsync( - stream, streamWriteLock, responseJson, serverCancellationToken); - } - finally - { - await _heartbeatService.StopHeartbeatsAsync(heartbeatTask, heartbeatCancellationSource); - heartbeatCancellationSource?.Dispose(); - await _clientDisconnectMonitor.StopClientDisconnectMonitorAsync( - clientDisconnectMonitorTask, - requestCancellationTokenSource); - } - } - } - - /// - /// Determines if the given exception represents a normal client disconnection. - /// - /// The exception to evaluate - /// True if the exception represents a normal disconnection, false otherwise - private static bool IsNormalDisconnectionException(Exception ex) - { - switch (ex) - { - case SocketException sockEx: - return sockEx.SocketErrorCode is SocketError.ConnectionReset or - SocketError.ConnectionAborted or - SocketError.OperationAborted or - SocketError.Shutdown or - SocketError.NotConnected; - - case ObjectDisposedException: - return true; - - case IOException ioEx when ioEx.InnerException is SocketException innerSockEx: - return innerSockEx.SocketErrorCode is SocketError.ConnectionReset or - SocketError.ConnectionAborted or - SocketError.OperationAborted or - SocketError.Shutdown or - SocketError.NotConnected; - - case IOException ioEx: - // Check HResult codes for common disconnection scenarios - return NormalDisconnectionHResults.Contains(ioEx.HResult) || - IsNormalDisconnectionByInnerException(ioEx); - - default: - return false; - } - } - - /// - /// Recursively checks inner exceptions for normal disconnection scenarios - /// - /// The exception to check - /// True if any inner exception indicates a normal disconnection - private static bool IsNormalDisconnectionByInnerException(Exception ex) - { - Exception innerEx = ex.InnerException; - while (innerEx != null) - { - if (IsNormalDisconnectionException(innerEx)) - { - return true; - } - innerEx = innerEx.InnerException; - } - return false; - } - /// /// Releases resources. /// @@ -664,4 +345,4 @@ public void Dispose() _serverTask = null; } } -} +}