EonaCat/EonaCat.Connections
1
0
Fork 0
Code Issues Pull Requests Packages Projects Releases Wiki Activity

Update EonaCat.Connections/Processors/JsonDataProcessor.cs

Browse Source
This commit is contained in:
EonaCat
2026-01-27 06:15:22 +00:00
parent b38514f0d1
commit 94119081bb
1 changed files with 252 additions and 329 deletions
Download Patch File Download Diff File Expand all files Collapse all files

225
EonaCat.Connections/Processors/JsonDataProcessor.cs
View File

@@ -1,38 +1,23 @@
using EonaCat.Json;
using Heijmans.Connector.Models;
using System;
using System.Text;
namespace EonaCat.Connections.Processors
{
public class ProcessedJsonMessage<TData>
{
public TData Data { get; set; }
public string RawData { get; set; } = string.Empty;
public string ClientName { get; set; } = string.Empty;
public string ClientEndpoint { get; set; } = string.Empty;
}
public class ProcessedTextMessage
{
public string Text { get; set; } = string.Empty;
public string ClientName { get; set; } = string.Empty;
}
public sealed class JsonDataProcessor<TData> : IDisposable
{
private readonly StringBuilder _buffer = new StringBuilder();
private readonly object _syncLock = new object();
private bool _isDisposed;
private readonly StringBuilder _buffer = new StringBuilder(4096);
private readonly object _sync = new object();
private bool _disposed;
// 30 MB default max buffer size
public int MaxAllowedBufferSize { get; set; } = 30 * 1024 * 1024;
public int MaxMessagesPerBatch { get; set; } = 200;
public string ClientName { get; }
// Diagnostics
public long TotalBytesProcessed { get; private set; }
public long TotalChunksReceived { get; private set; }
// Events
public event EventHandler<ProcessedJsonMessage<TData>>? OnProcessMessage;
public event EventHandler<ProcessedTextMessage>? OnProcessTextMessage;
public event EventHandler<Exception>? OnMessageError;
@@ -42,96 +27,48 @@ namespace EonaCat.Connections.Processors
ClientName = Guid.NewGuid().ToString();
}
public void Process(string jsonString, string? clientName = null, string? endpoint = null)
public void Process(string data, string? client = null, string? endpoint = null)
{
ThrowIfDisposed();
if (string.IsNullOrEmpty(jsonString))
if (string.IsNullOrEmpty(data))
{
return;
}
TotalChunksReceived++;
TotalBytesProcessed += Encoding.UTF8.GetByteCount(jsonString);
TotalBytesProcessed += Encoding.UTF8.GetByteCount(data);
ProcessInternal(jsonString, clientName, endpoint);
ProcessInternal(data, client ?? ClientName, endpoint);
}
public void Process(DataReceivedEventArgs e, string? clientName = null)
private void ProcessInternal(string data, string client, string? endpoint)
{
ThrowIfDisposed();
if (e == null)
lock (_sync)
{
return;
}
_buffer.Append(data);
string dataString;
if (e.IsBinary)
int processed = 0;
while (processed < MaxMessagesPerBatch &&
TryExtract(out int start, out int length, out bool isText))
{
if (e.Data == null || e.Data.Length == 0)
if (isText)
{
return;
}
dataString = Encoding.UTF8.GetString(e.Data);
}
else
{
dataString = e.StringData;
}
if (string.IsNullOrWhiteSpace(dataString))
{
return;
}
string client = e.Nickname ?? clientName ?? ClientName;
TotalChunksReceived++;
TotalBytesProcessed += Encoding.UTF8.GetByteCount(dataString);
ProcessInternal(dataString, client, e.RemoteEndPoint?.ToString());
}
private void ProcessInternal(string jsonString, string? clientName, string? endpoint)
{
string client = clientName ?? ClientName;
lock (_syncLock)
{
_buffer.Append(jsonString);
int processedCount = 0;
while (processedCount < MaxMessagesPerBatch &&
TryExtractFullJson(out int fullJsonStart, out int fullJsonLength,
out int textStart, out int textLength))
{
// Check if we have any leading text to process
if (textLength > 0)
{
string text = _buffer.ToString(textStart, textLength);
var text = _buffer.ToString(start, length);
OnProcessTextMessage?.Invoke(this, new ProcessedTextMessage
{
Text = text,
ClientName = client
});
// Remove the processed text immediately
_buffer.Remove(textStart, textLength);
processedCount++;
continue;
}
// Process the full JSON
if (fullJsonLength > 0)
else
{
string json = _buffer.ToString(fullJsonStart, fullJsonLength);
var json = _buffer.ToString(start, length);
try
{
var deserialized = JsonHelper.ToObject<TData>(json);
var obj = JsonHelper.ToObject<TData>(json);
OnProcessMessage?.Invoke(this, new ProcessedJsonMessage<TData>
{
Data = deserialized,
Data = obj,
RawData = json,
ClientName = client,
ClientEndpoint = endpoint ?? string.Empty
@@ -139,102 +76,87 @@ namespace EonaCat.Connections.Processors
}
catch (Exception ex)
{
OnMessageError?.Invoke(this, new Exception(
$"Failed to deserialize JSON for client {client}", ex));
}
// Remove processed JSON immediately
_buffer.Remove(fullJsonStart, fullJsonLength);
processedCount++;
OnMessageError?.Invoke(this,
new Exception($"Failed to parse JSON for {client}", ex));
}
}
// Prevent buffer overflow
Consume(start, length);
processed++;
}
if (_buffer.Length > MaxAllowedBufferSize)
{
OnMessageError?.Invoke(this, new Exception(
$"Buffer overflow for client {client}. Current size: {_buffer.Length / 1024.0 / 1024.0:F2} MB. " +
$"Max allowed: {MaxAllowedBufferSize / 1024.0 / 1024.0:F2} MB. Clearing buffer."));
OnMessageError?.Invoke(this,
new Exception($"Buffer exceeded {MaxAllowedBufferSize} bytes for client {client}. Discarding."));
_buffer.Clear();
_buffer.Capacity = 4096;
}
}
}
private bool TryExtractFullJson(
out int fullJsonStart,
out int fullJsonLength,
out int textStart,
out int textLength)
private bool TryExtract(out int start, out int length, out bool isText)
{
fullJsonStart = fullJsonLength = textStart = textLength = 0;
start = length = 0;
isText = false;
if (_buffer.Length == 0)
{
return false;
}
var span = _buffer.ToString().AsSpan();
int pos = 0;
// Skip leading whitespace
while (pos < _buffer.Length && char.IsWhiteSpace(_buffer[pos]))
while (pos < span.Length && char.IsWhiteSpace(span[pos]))
{
pos++;
}
if (pos >= _buffer.Length)
if (pos >= span.Length)
{
return false;
}
// Json needs to start with { or [
if (_buffer[pos] != '{' && _buffer[pos] != '[')
char c = span[pos];
if (c != '{' && c != '[')
{
textStart = pos;
while (pos < _buffer.Length && _buffer[pos] != '{' && _buffer[pos] != '[')
isText = true;
start = pos;
while (pos < span.Length && span[pos] != '{' && span[pos] != '[')
{
pos++;
}
textLength = pos - textStart;
length = pos - start;
return true;
}
// Check if we also have a json end token
fullJsonStart = pos;
fullJsonLength = FindJsonTokenEnd(_buffer, pos) - pos;
if (fullJsonLength <= 0)
start = pos;
length = FindJsonEnd(span, pos) - pos;
if (length <= 0)
{
// partial JSON
return false;
}
isText = false;
return true;
}
private static int FindJsonTokenEnd(StringBuilder buffer, int start)
private static int FindJsonEnd(ReadOnlySpan<char> span, int start)
{
if (start >= buffer.Length)
{
return 0;
}
char open = span[start];
char close = open == '{' ? '}' : ']';
int i = start;
char firstChar = buffer[start];
if (firstChar == '{' || firstChar == '[')
{
char open = firstChar;
char close = firstChar == '{' ? '}' : ']';
int depth = 1;
bool inString = false;
bool escape = false;
i++;
while (i < buffer.Length)
for (int i = start + 1; i < span.Length; i++)
{
char c = buffer[i];
char c = span[i];
if (inString)
{
@@ -261,46 +183,38 @@ namespace EonaCat.Connections.Processors
{
depth++;
}
else if (c == close)
{
depth--;
if (depth == 0)
else if (c == close && --depth == 0)
{
return i + 1;
}
}
}
i++;
}
// partial JSON
return 0;
}
// only objects/arrays supported as JSON start
return 0;
}
private void Consume(int start, int length)
{
_buffer.Remove(start, length);
private void ThrowIfDisposed()
if (_buffer.Capacity > 1024 * 1024 && _buffer.Length < _buffer.Capacity / 2)
{
if (_isDisposed)
{
throw new ObjectDisposedException(nameof(JsonDataProcessor<TData>));
_buffer.Capacity = Math.Max(_buffer.Length, 4096);
}
}
public void ClearBuffer()
{
lock (_syncLock)
lock (_sync)
{
_buffer.Clear();
_buffer.Capacity = 4096;
}
}
public void ResetStatistics()
{
lock (_syncLock)
lock (_sync)
{
TotalBytesProcessed = 0;
TotalChunksReceived = 0;
@@ -309,21 +223,30 @@ namespace EonaCat.Connections.Processors
public void Dispose()
{
if (_isDisposed)
if (_disposed)
{
return;
}
_isDisposed = true;
_disposed = true;
lock (_syncLock)
lock (_sync)
{
_buffer.Clear();
_buffer.Capacity = 0;
}
OnProcessMessage = null;
OnProcessTextMessage = null;
OnMessageError = null;
}
private void ThrowIfDisposed()
{
if (_disposed)
{
throw new ObjectDisposedException(nameof(JsonDataProcessor<TData>));
}
}
}
}