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}";
}
}
}