using System; using System.Collections.Generic; using System.Text; namespace io.github.hatayama.uLoopMCP { /// /// Reassembles fragmented TCP messages using Content-Length framing. /// Handles partial frames and maintains state across multiple data chunks. /// public class MessageReassembler : IDisposable { private readonly DynamicBufferManager _bufferManager; private byte[] _assemblyBuffer; private int _currentDataLength = 0; private bool _disposed = false; // Frame parsing state private int _expectedContentLength = -1; private int _headerLength = -1; private bool _headerParsed = false; // Statistics private int _totalMessagesReassembled = 0; private int _totalDataChunksProcessed = 0; /// /// Initializes a new instance of the MessageReassembler. /// /// The buffer manager to use for memory management public MessageReassembler(DynamicBufferManager bufferManager = null) { this._bufferManager = bufferManager ?? new DynamicBufferManager(); this._assemblyBuffer = this._bufferManager.GetBuffer(BufferConfig.INITIAL_BUFFER_SIZE); } /// /// Gets whether there is incomplete data waiting for more chunks. /// public bool HasIncompleteData => _currentDataLength > 0; /// /// Gets the current amount of data in the assembly buffer. /// public int CurrentDataLength => _currentDataLength; /// /// Gets whether the header has been parsed and we're waiting for content. /// public bool IsWaitingForContent => _headerParsed && _expectedContentLength > 0; /// /// Adds new data to the reassembly buffer. /// /// The data chunk to add /// The length of valid data in the chunk public void AddData(byte[] data, int length) { if (_disposed) { throw new ObjectDisposedException(nameof(MessageReassembler)); } if (data == null || length <= 0) { return; } if (length > data.Length) { throw new ArgumentException("Length cannot exceed data array size", nameof(length)); } _totalDataChunksProcessed++; // Ensure we have enough space in the assembly buffer int requiredSize = _currentDataLength + length; if (requiredSize > _assemblyBuffer.Length) { _bufferManager.ResizeBuffer(ref _assemblyBuffer, _currentDataLength, requiredSize); } // Copy new data to assembly buffer Array.Copy(data, 0, _assemblyBuffer, _currentDataLength, length); _currentDataLength += length; } /// /// Attempts to extract complete messages from the assembly buffer. /// /// Array of complete JSON messages, or empty array if none available public string[] ExtractCompleteMessages() { if (_disposed) { throw new ObjectDisposedException(nameof(MessageReassembler)); } var completeMessages = new List(); while (_currentDataLength > 0) { string extractedMessage = TryExtractSingleMessage(); if (extractedMessage == null) { // No complete message available break; } completeMessages.Add(extractedMessage); _totalMessagesReassembled++; } return completeMessages.ToArray(); } /// /// Attempts to extract a single complete message from the buffer. /// /// The extracted JSON message, or null if not complete private string TryExtractSingleMessage() { // If we haven't parsed the header yet, try to parse it if (!_headerParsed) { if (!TryParseHeader()) { // Header not complete yet return null; } } // Early detection of abnormal states - throw exceptions immediately if (_expectedContentLength < 0) { throw new InvalidOperationException($"Invalid Content-Length value: {_expectedContentLength}. Message framing is corrupted."); } if (_headerLength < 0) { throw new InvalidOperationException($"Invalid header length: {_headerLength}. Message framing is corrupted."); } if (_expectedContentLength > BufferConfig.MAX_MESSAGE_SIZE) { throw new InvalidOperationException($"Content-Length {_expectedContentLength} exceeds maximum message size {BufferConfig.MAX_MESSAGE_SIZE}."); } // Check if we have the complete message int expectedTotalLength = _headerLength + _expectedContentLength; if (_currentDataLength < expectedTotalLength) { // Message not complete yet return null; } // Validate buffer bounds before extraction if (_headerLength + _expectedContentLength > _currentDataLength) { throw new InvalidOperationException($"Buffer underflow: trying to read {_expectedContentLength} bytes at offset {_headerLength}, but only {_currentDataLength} bytes available."); } // Extract the JSON content with proper error handling string jsonContent; try { jsonContent = Encoding.UTF8.GetString(_assemblyBuffer, _headerLength, _expectedContentLength); } catch (ArgumentException ex) { throw new InvalidOperationException($"Invalid UTF-8 sequence detected in message content at offset {_headerLength}, length {_expectedContentLength}: {ex.Message}", ex); } // Log before resetting state // Remove the processed message from the buffer RemoveProcessedData(expectedTotalLength); // Reset parsing state for next message ResetParsingState(); return jsonContent; } /// /// Attempts to parse the Content-Length header from the current buffer. /// /// True if header was successfully parsed, false otherwise private bool TryParseHeader() { var frameParser = new FrameParser(); bool parseResult = frameParser.TryParseFrame(_assemblyBuffer, _currentDataLength, out _expectedContentLength, out _headerLength); if (parseResult) { // Early validation of parsed values if (_expectedContentLength < 0) { throw new InvalidOperationException($"FrameParser returned invalid Content-Length: {_expectedContentLength}"); } if (_headerLength < 0) { throw new InvalidOperationException($"FrameParser returned invalid header length: {_headerLength}"); } _headerParsed = true; } return parseResult; } /// /// Removes processed data from the beginning of the assembly buffer. /// /// Number of bytes to remove from the beginning private void RemoveProcessedData(int bytesToRemove) { if (bytesToRemove <= 0 || bytesToRemove > _currentDataLength) { return; } // Move remaining data to the beginning of the buffer int remainingDataLength = _currentDataLength - bytesToRemove; if (remainingDataLength > 0) { Array.Copy(_assemblyBuffer, bytesToRemove, _assemblyBuffer, 0, remainingDataLength); } _currentDataLength = remainingDataLength; } /// /// Resets the parsing state for the next message. /// private void ResetParsingState() { _headerParsed = false; _expectedContentLength = -1; _headerLength = -1; } /// /// Clears all data from the assembly buffer. /// Useful for error recovery or connection reset. /// public void Clear() { if (_disposed) { return; } _currentDataLength = 0; ResetParsingState(); } /// /// Gets statistics about the reassembler's performance. /// /// Statistics about message reassembly public MessageReassemblerStats GetStats() { return new MessageReassemblerStats { TotalMessagesReassembled = _totalMessagesReassembled, TotalDataChunksProcessed = _totalDataChunksProcessed, CurrentBufferSize = _assemblyBuffer?.Length ?? 0, CurrentDataLength = _currentDataLength, HasIncompleteData = HasIncompleteData, IsWaitingForContent = IsWaitingForContent, ExpectedContentLength = _expectedContentLength, HeaderLength = _headerLength }; } /// /// Validates the current state of the reassembler. /// /// True if state is valid, false if cleanup is needed public bool ValidateState() { if (_disposed) { return false; } // Check for reasonable buffer size - throw exception instead of silently clearing if (_currentDataLength > BufferConfig.MAX_MESSAGE_SIZE) { throw new InvalidOperationException($"Buffer size {_currentDataLength} exceeds maximum message size {BufferConfig.MAX_MESSAGE_SIZE}. Data corruption detected."); } // Check for stale incomplete data - throw exception for corrupted state if (_headerParsed && _expectedContentLength > 0) { int expectedTotalLength = _headerLength + _expectedContentLength; if (expectedTotalLength > BufferConfig.MAX_MESSAGE_SIZE) { throw new InvalidOperationException($"Expected message size {expectedTotalLength} exceeds maximum {BufferConfig.MAX_MESSAGE_SIZE}. Message framing is corrupted."); } } // Validate consistency of parsing state if (_headerParsed && (_expectedContentLength < 0 || _headerLength < 0)) { throw new InvalidOperationException($"Inconsistent parsing state: headerParsed={_headerParsed}, expectedContentLength={_expectedContentLength}, headerLength={_headerLength}"); } return true; } /// /// Gets a preview of the current buffer content for debugging. /// /// Maximum length of preview /// String representation of buffer content public string GetBufferPreview(int maxLength = 100) { if (_disposed || _currentDataLength == 0) { return string.Empty; } // Safe UTF-8 decoding with proper error handling int previewLength = Math.Min(_currentDataLength, maxLength); string preview; try { preview = Encoding.UTF8.GetString(_assemblyBuffer, 0, previewLength); } catch (ArgumentException ex) { throw new InvalidOperationException($"Invalid UTF-8 sequence detected in buffer preview (length {previewLength}): {ex.Message}", ex); } if (_currentDataLength > maxLength) { preview += "..."; } return preview; } /// /// Releases all resources used by the MessageReassembler. /// public void Dispose() { if (!_disposed) { if (_assemblyBuffer != null) { _bufferManager.ReturnBuffer(_assemblyBuffer); _assemblyBuffer = null; } _bufferManager?.Dispose(); _disposed = true; } } } /// /// Statistics about message reassembler performance and state. /// public class MessageReassemblerStats { /// /// Total number of complete messages reassembled. /// public int TotalMessagesReassembled { get; set; } /// /// Total number of data chunks processed. /// public int TotalDataChunksProcessed { get; set; } /// /// Current size of the assembly buffer. /// public int CurrentBufferSize { get; set; } /// /// Current amount of data in the buffer. /// public int CurrentDataLength { get; set; } /// /// Whether there is incomplete data in the buffer. /// public bool HasIncompleteData { get; set; } /// /// Whether the reassembler is waiting for content after parsing header. /// public bool IsWaitingForContent { get; set; } /// /// Expected content length from parsed header. /// public int ExpectedContentLength { get; set; } /// /// Length of the parsed header. /// public int HeaderLength { get; set; } /// /// Returns a formatted string representation of the statistics. /// public override string ToString() { return $"MessageReassembler Stats: Messages={TotalMessagesReassembled}, " + $"Chunks={TotalDataChunksProcessed}, BufferSize={CurrentDataLength}/{CurrentBufferSize}, " + $"Incomplete={HasIncompleteData}, WaitingForContent={IsWaitingForContent}"; } } }