Switch SignalR payloads to ArrayPool-backed SignalData
Major protocol refactor: all byte[] payloads in SignalR hub/client interfaces, plumbing, and DTOs are now wrapped in SignalData, a disposable, ArrayPool-backed type with Span access. Introduces AyCodeBinaryHubProtocol (derived from AcBinaryHubProtocol) to rent pooled buffers for SignalData on receive. All message signatures, diagnostics, and serialization logic updated. Documentation and tests revised to reflect SignalData usage. Enables zero-copy, low-GC, high-performance binary messaging for large payloads.
This commit is contained in:
parent
3b7007002a
commit
d147398698
|
|
@ -212,7 +212,7 @@ public class BenchmarkSignalRClient : AcSignalRClientBase, IAcSignalRHubItemServ
|
|||
public TResponse? GetAllSync<TResponse>(int tag)
|
||||
=> GetAllAsync<TResponse>(tag).GetAwaiter().GetResult();
|
||||
|
||||
protected override Task MessageReceived(int messageTag, SignalParams signalParams, byte[] data) => Task.CompletedTask;
|
||||
protected override Task MessageReceived(int messageTag, SignalParams signalParams, SignalData data) => Task.CompletedTask;
|
||||
protected override HubConnectionState GetConnectionState() => HubConnectionState.Connected;
|
||||
protected override bool IsConnected() => true;
|
||||
protected override Task StartConnectionInternal() => Task.CompletedTask;
|
||||
|
|
@ -221,7 +221,7 @@ public class BenchmarkSignalRClient : AcSignalRClientBase, IAcSignalRHubItemServ
|
|||
|
||||
protected override async Task SendToHubAsync(int messageTag, int? requestId, SignalParams signalParams, byte[]? messageBytes)
|
||||
{
|
||||
await _hub.OnReceiveMessage(messageTag, requestId, signalParams, messageBytes ?? []);
|
||||
await _hub.OnReceiveMessage(messageTag, requestId, signalParams, new SignalData(messageBytes ?? []));
|
||||
}
|
||||
}
|
||||
|
||||
|
|
|
|||
|
|
@ -29,7 +29,7 @@ public class TestableSignalRClient2 : AcSignalRClientBase, IAcSignalRHubItemServ
|
|||
|
||||
#region Override virtual methods for testing
|
||||
|
||||
protected override async Task MessageReceived(int messageTag, SignalParams signalParams, byte[] data)
|
||||
protected override async Task MessageReceived(int messageTag, SignalParams signalParams, SignalData data)
|
||||
{
|
||||
throw new NotImplementedException();
|
||||
}
|
||||
|
|
@ -54,7 +54,7 @@ public class TestableSignalRClient2 : AcSignalRClientBase, IAcSignalRHubItemServ
|
|||
|
||||
protected override async Task SendToHubAsync(int messageTag, int? requestId, SignalParams signalParams, byte[]? messageBytes)
|
||||
{
|
||||
await _signalRHub.OnReceiveMessage(messageTag, requestId, signalParams, messageBytes ?? []);
|
||||
await _signalRHub.OnReceiveMessage(messageTag, requestId, signalParams, new SignalData(messageBytes ?? []));
|
||||
}
|
||||
#endregion
|
||||
|
||||
|
|
|
|||
|
|
@ -288,11 +288,11 @@ namespace AyCode.Services.Server.SignalRs
|
|||
BeginSync();
|
||||
// Request SignalResponseDataMessage directly to avoid deserializing ResponseData
|
||||
return SignalRClient.GetAllAsync<SignalResponseDataMessage>(SignalRCrudTags.GetAllMessageTag, GetContextParams())
|
||||
.ContinueWith(async task =>
|
||||
.ContinueWith(async responseTask =>
|
||||
{
|
||||
try
|
||||
{
|
||||
var response = task.Result;
|
||||
var response = await responseTask;
|
||||
if (response?.Status != SignalResponseStatus.Success || response.ResponseData == null)
|
||||
throw new NullReferenceException($"LoadDataSourceAsync; Status: {response?.Status}");
|
||||
|
||||
|
|
@ -309,7 +309,7 @@ namespace AyCode.Services.Server.SignalRs
|
|||
/// <summary>
|
||||
/// Loads data source directly from ResponseData byte[], avoiding double deserialization.
|
||||
/// </summary>
|
||||
public async Task LoadDataSourceFromResponseData(byte[] responseData, AcSerializerType serializerType,
|
||||
public async Task LoadDataSourceFromResponseData(SignalData responseData, AcSerializerType serializerType,
|
||||
bool refreshDataFromDbAsync = false, bool setSourceToWorkingReferenceList = false, bool clearChangeTracking = true)
|
||||
{
|
||||
await _asyncLock.WaitAsync();
|
||||
|
|
@ -325,7 +325,7 @@ namespace AyCode.Services.Server.SignalRs
|
|||
observable.BeginUpdate();
|
||||
try
|
||||
{
|
||||
responseData.BinaryToMerge(InnerList);
|
||||
responseData.Span.BinaryToMerge(InnerList);
|
||||
}
|
||||
finally
|
||||
{
|
||||
|
|
@ -334,13 +334,13 @@ namespace AyCode.Services.Server.SignalRs
|
|||
}
|
||||
else
|
||||
{
|
||||
responseData.BinaryTo(InnerList);
|
||||
responseData.Span.BinaryTo(InnerList);
|
||||
}
|
||||
}
|
||||
else
|
||||
{
|
||||
// JSON mode - decompress GZip first
|
||||
var json = GzipHelper.DecompressToString(responseData);
|
||||
// JSON mode - decompress GZip first (no span overload for DecompressToString)
|
||||
var json = GzipHelper.DecompressToString(responseData.ToArray());
|
||||
if (InnerList is IAcObservableCollection observable)
|
||||
{
|
||||
observable.PopulateFromJson(json);
|
||||
|
|
@ -356,9 +356,9 @@ namespace AyCode.Services.Server.SignalRs
|
|||
// Deserialize to new list and set as reference
|
||||
TIList? fromSource;
|
||||
if (serializerType == AcSerializerType.Binary)
|
||||
fromSource = responseData.BinaryTo<TIList>();
|
||||
fromSource = responseData.Span.BinaryTo<TIList>();
|
||||
else
|
||||
fromSource = GzipHelper.DecompressToString(responseData).JsonTo<TIList>();
|
||||
fromSource = GzipHelper.DecompressToString(responseData.ToArray()).JsonTo<TIList>();
|
||||
|
||||
if (fromSource != null)
|
||||
{
|
||||
|
|
|
|||
|
|
@ -15,7 +15,8 @@ public abstract class AcSignalRSendToClientService<TSignalRHub, TSignalRTags, TL
|
|||
|
||||
protected virtual async Task SendMessageToClient(IAcSignalRHubItemServer sendTo, int messageTag, object? content)
|
||||
{
|
||||
var responseData = SignalRSerializationHelper.CreateResponseData(content, AcBinarySerializerOptions.Default) ?? [];
|
||||
var responseBytes = SignalRSerializationHelper.CreateResponseData(content, AcBinarySerializerOptions.Default) ?? [];
|
||||
var responseData = new SignalData(responseBytes);
|
||||
var signalParams = new SignalParams
|
||||
{
|
||||
Status = SignalResponseStatus.Success,
|
||||
|
|
|
|||
|
|
@ -64,7 +64,7 @@ public abstract class AcWebSignalRHubBase<TSignalRTags, TLogger>(IConfiguration
|
|||
|
||||
#region Message Processing
|
||||
|
||||
public virtual Task OnReceiveMessage(int messageTag, int? requestId, SignalParams signalParams, byte[] data)
|
||||
public virtual Task OnReceiveMessage(int messageTag, int? requestId, SignalParams signalParams, SignalData data)
|
||||
{
|
||||
return ProcessOnReceiveMessage(messageTag, signalParams, requestId, null);
|
||||
}
|
||||
|
|
@ -194,9 +194,9 @@ public abstract class AcWebSignalRHubBase<TSignalRTags, TLogger>(IConfiguration
|
|||
}
|
||||
|
||||
// Log binary diagnostics if enabled
|
||||
if (EnableBinaryDiagnostics && responseMessage is SignalResponseDataMessage dataMsg && dataMsg.ResponseData != null)
|
||||
if (EnableBinaryDiagnostics && responseMessage is SignalResponseDataMessage dataMsg && dataMsg.ResponseData is { IsEmpty: false })
|
||||
{
|
||||
LogResponseDataDiagnostics(messageTag, tagName, requestId, dataMsg.ResponseData);
|
||||
LogResponseDataDiagnostics(messageTag, tagName, requestId, dataMsg.ResponseData.ToArray());
|
||||
}
|
||||
|
||||
await ResponseToCaller(messageTag, responseMessage, requestId);
|
||||
|
|
@ -490,7 +490,7 @@ public abstract class AcWebSignalRHubBase<TSignalRTags, TLogger>(IConfiguration
|
|||
Status = responseMessage.Status,
|
||||
DataSerializerType = responseMessage.DataSerializerType
|
||||
};
|
||||
var responseData = responseMessage.ResponseData ?? [];
|
||||
var responseData = responseMessage.ResponseData ?? new SignalData([]);
|
||||
|
||||
var tagName = ConstHelper.NameByValue<TSignalRTags>(messageTag);
|
||||
|
||||
|
|
|
|||
|
|
@ -8,7 +8,7 @@ Server-side SignalR hub infrastructure: method dispatch, session management, bro
|
|||
## Server Processing
|
||||
|
||||
```
|
||||
6. OnReceiveMessage(tag, requestId, signalParams, data)
|
||||
6. OnReceiveMessage(tag, requestId, signalParams, SignalData data)
|
||||
7. Extract parameterBytes from signalParams.Parameters
|
||||
8. DynamicMethodRegistry.GetMethodByMessageTag(tag) <- ConcurrentDictionary lookup
|
||||
9. signalParams.GetParameterValues(paramInfos):
|
||||
|
|
@ -20,8 +20,8 @@ Server-side SignalR hub infrastructure: method dispatch, session management, bro
|
|||
10. MethodInfo.InvokeMethod(instance, params) <- unwraps Task/ValueTask
|
||||
11. CreateResponseMessage(tag, Success, result) <- Binary serialize payload -> byte[]
|
||||
12. SendMessageToClient(caller, tag, message, requestId):
|
||||
|- Extract signalParams { Status, DataSerializerType } + responseData byte[] from message
|
||||
'- caller.OnReceiveMessage(tag, requestId, signalParams, responseData)
|
||||
|- Extract signalParams { Status, DataSerializerType } + SignalData from message
|
||||
'- caller.OnReceiveMessage(tag, requestId, signalParams, SignalData)
|
||||
(metadata + payload as separate args -- no envelope serialization)
|
||||
13. If SendToOtherClientType != None:
|
||||
'- SendMessageToOthers(sendToOtherClientTag, result) <- uses sendToOtherClientTag, not messageTag
|
||||
|
|
@ -34,7 +34,7 @@ See also: `AyCode.Models.Server/DynamicMethods/README.md`
|
|||
### Server-Side Lookup
|
||||
|
||||
```
|
||||
1. OnReceiveMessage(tag=100, requestId, signalParams, data)
|
||||
1. OnReceiveMessage(tag=100, requestId, signalParams, SignalData data)
|
||||
|
||||
2. DynamicMethodRegistry.GetMethodByMessageTag(100)
|
||||
|- Check static ConcurrentDictionary<int, (Type, AcMethodInfoModel)?> cache
|
||||
|
|
@ -83,7 +83,7 @@ ConcurrentDictionary<TSessionItemId, TSessionItem> Sessions
|
|||
| `SendMessageToUser(userId)` | User (all connections) |
|
||||
| `SendMessageToUsers(userIds)` | Multiple users |
|
||||
|
||||
All messages serialized to `byte[]` payload + `SignalParams` metadata (Parameters=null for server->client push) -> sent as separate hub arguments via `OnReceiveMessage` (no envelope wrapping).
|
||||
All messages serialized to `SignalData` payload + `SignalParams` metadata (Parameters=null for server->client push) -> sent as separate hub arguments via `OnReceiveMessage` (no envelope wrapping). Server wraps `byte[]` in non-pooled `SignalData`; client receives as `ArrayPool`-backed `SignalData` via `AyCodeBinaryHubProtocol`.
|
||||
|
||||
## Hub Events
|
||||
|
||||
|
|
@ -96,7 +96,7 @@ Enable with `AcWebSignalRHubBase.EnableBinaryDiagnostics = true`.
|
|||
|
||||
Logs: hex dump (500 byte sample), header parsing (version, marker), property count + names via VarUInt reading.
|
||||
|
||||
`SignalResponseDataMessage.DiagnosticLogger` -- per-response logging: target type info, property list, inheritance chain, hex dump.
|
||||
`SignalResponseDataMessage.DiagnosticLogger` -- per-response logging: target type info, property list, inheritance chain, hex dump. Uses `SignalData.Span` for zero-alloc diagnostics.
|
||||
|
||||
## Key Source Files
|
||||
|
||||
|
|
|
|||
|
|
@ -42,12 +42,17 @@ public class PostJsonDataMessageTests
|
|||
Assert.IsNotNull(serverParams);
|
||||
Assert.AreEqual(testValue, serverParams![0]);
|
||||
|
||||
// Response round-trip
|
||||
// Response round-trip (SignalResponseDataMessage is in-memory DTO, not serialized as envelope on wire)
|
||||
var serviceResult = $"{serverParams[0]}";
|
||||
var response = new SignalResponseDataMessage(100, SignalResponseStatus.Success, serviceResult, AyCode.Core.Serializers.Binaries.AcBinarySerializerOptions.Default);
|
||||
var responseBytes = response.ToBinary();
|
||||
var clientResponse = responseBytes.BinaryTo<SignalResponseDataMessage>();
|
||||
var finalResult = clientResponse?.GetResponseData<string>();
|
||||
var responseData = SignalRSerializationHelper.CreateResponseData(serviceResult, AyCode.Core.Serializers.Binaries.AcBinarySerializerOptions.Default);
|
||||
var clientResponse = new SignalResponseDataMessage
|
||||
{
|
||||
MessageTag = 100,
|
||||
Status = SignalResponseStatus.Success,
|
||||
DataSerializerType = AyCode.Core.Serializers.AcSerializerType.Binary,
|
||||
ResponseData = responseData != null ? new SignalData(responseData) : null
|
||||
};
|
||||
var finalResult = clientResponse.GetResponseData<string>();
|
||||
Assert.AreEqual(testValue.ToString(), finalResult);
|
||||
}
|
||||
|
||||
|
|
|
|||
|
|
@ -28,7 +28,7 @@ namespace AyCode.Services.SignalRs;
|
|||
/// on the hot path. Argument payloads are serialized directly to the pipe
|
||||
/// via AcBinarySerializer (zero-copy). Length prefixes are patched in-place.
|
||||
/// </summary>
|
||||
public sealed class AcBinaryHubProtocol : IHubProtocol
|
||||
public class AcBinaryHubProtocol : IHubProtocol
|
||||
{
|
||||
private const int LengthPrefixSize = 4;
|
||||
|
||||
|
|
@ -43,7 +43,7 @@ public sealed class AcBinaryHubProtocol : IHubProtocol
|
|||
private const byte MsgAck = 8;
|
||||
private const byte MsgSequence = 9;
|
||||
|
||||
private volatile AcBinarySerializerOptions _options;
|
||||
protected volatile AcBinarySerializerOptions _options;
|
||||
|
||||
public AcBinaryHubProtocol() : this(AcBinarySerializerOptions.Default) { }
|
||||
|
||||
|
|
@ -416,6 +416,18 @@ public sealed class AcBinaryHubProtocol : IHubProtocol
|
|||
return;
|
||||
}
|
||||
|
||||
if (value is SignalData signalData)
|
||||
{
|
||||
// SignalData fast-path: same wire format as byte[], reads from Span
|
||||
var span = signalData.Span;
|
||||
var argPayload = 1 + VarUIntSize((uint)span.Length) + span.Length;
|
||||
bw.WriteRaw(argPayload);
|
||||
bw.WriteByte(BinaryTypeCode.ByteArray);
|
||||
bw.WriteVarUInt((uint)span.Length);
|
||||
bw.WriteBytes(span);
|
||||
return;
|
||||
}
|
||||
|
||||
// Flush BWO to pipe, then serialize directly to the pipe via AcBinarySerializer
|
||||
bw.FlushAndReset();
|
||||
|
||||
|
|
@ -469,12 +481,26 @@ public sealed class AcBinaryHubProtocol : IHubProtocol
|
|||
{
|
||||
var byteReader = new SpanReader(argSpan.Slice(1));
|
||||
var len = (int)byteReader.ReadVarUInt();
|
||||
return byteReader.ReadSpan(len).ToArray();
|
||||
var payloadSpan = byteReader.ReadSpan(len);
|
||||
// Skip virtual dispatch for plain byte[] (most common case — SignalParams.Parameters).
|
||||
// Only call virtual hook when targetType is not byte[] (e.g. SignalData).
|
||||
return targetType == typeof(byte[]) || targetType == typeof(object)
|
||||
? payloadSpan.ToArray()
|
||||
: CreateByteArrayResult(payloadSpan, targetType);
|
||||
}
|
||||
|
||||
return AcBinaryDeserializer.Deserialize(argSpan, targetType, _options);
|
||||
}
|
||||
|
||||
/// <summary>
|
||||
/// Hook for derived protocols to customize byte[] argument creation.
|
||||
/// Called from the byte[] fast-path (ByteArray wire marker 0x44).
|
||||
/// Base implementation: allocates new byte[] via .ToArray().
|
||||
/// Override to use ArrayPool, return SignalData, etc.
|
||||
/// </summary>
|
||||
protected virtual object CreateByteArrayResult(ReadOnlySpan<byte> data, Type targetType)
|
||||
=> data.ToArray();
|
||||
|
||||
#endregion
|
||||
|
||||
#region Framing Helpers
|
||||
|
|
|
|||
|
|
@ -21,7 +21,7 @@ namespace AyCode.Services.SignalRs
|
|||
protected readonly HubConnection? HubConnection;
|
||||
protected readonly AcLoggerBase Logger;
|
||||
|
||||
protected abstract Task MessageReceived(int messageTag, SignalParams signalParams, byte[] data);
|
||||
protected abstract Task MessageReceived(int messageTag, SignalParams signalParams, SignalData data);
|
||||
|
||||
public int MsDelay = 25;
|
||||
public int MsFirstDelay = 50;
|
||||
|
|
@ -64,13 +64,13 @@ namespace AyCode.Services.SignalRs
|
|||
|
||||
if (useAcBinaryProtocol)
|
||||
{
|
||||
hubBuilder.Services.AddSingleton<IHubProtocol, AcBinaryHubProtocol>();
|
||||
hubBuilder.Services.AddSingleton<IHubProtocol, AyCodeBinaryHubProtocol>();
|
||||
}
|
||||
|
||||
HubConnection = hubBuilder.Build();
|
||||
|
||||
HubConnection.Closed += HubConnection_Closed;
|
||||
_ = HubConnection.On<int, int?, SignalParams, byte[]>(nameof(IAcSignalRHubClient.OnReceiveMessage), OnReceiveMessage);
|
||||
_ = HubConnection.On<int, int?, SignalParams, SignalData>(nameof(IAcSignalRHubClient.OnReceiveMessage), OnReceiveMessage);
|
||||
}
|
||||
|
||||
protected AcSignalRClientBase(AcLoggerBase logger)
|
||||
|
|
@ -416,11 +416,11 @@ namespace AyCode.Services.SignalRs
|
|||
|
||||
protected virtual int GetNextRequestId() => AcDomain.NextUniqueInt32;
|
||||
|
||||
public virtual Task OnReceiveMessage(int messageTag, int? requestId, SignalParams signalParams, byte[] data)
|
||||
public virtual Task OnReceiveMessage(int messageTag, int? requestId, SignalParams signalParams, SignalData data)
|
||||
{
|
||||
var logText = $"Client OnReceiveMessage; {nameof(requestId)}: {requestId}; {ConstHelper.NameByValue(TagsName, messageTag)}";
|
||||
|
||||
if (data.Length == 0) Logger.Warning($"data.Length == 0! {logText}");
|
||||
if (data.IsEmpty) Logger.Warning($"data.IsEmpty! {logText}");
|
||||
|
||||
try
|
||||
{
|
||||
|
|
@ -431,7 +431,7 @@ namespace AyCode.Services.SignalRs
|
|||
Logger.Debug($"[{requestModel.ResponseDateTime.Subtract(requestModel.RequestDateTime).TotalMilliseconds:N0}ms][{data.Length / 1024}kb]{logText}");
|
||||
|
||||
// Diagnostic logging for binary deserialization debugging
|
||||
LogBinaryDiagnostics(messageTag, data, requestId);
|
||||
LogBinaryDiagnostics(messageTag, data);
|
||||
|
||||
// No envelope deserialization — construct directly from params + data
|
||||
var responseMessage = new SignalResponseDataMessage
|
||||
|
|
@ -475,9 +475,9 @@ namespace AyCode.Services.SignalRs
|
|||
catch (Exception ex)
|
||||
{
|
||||
// Enhanced error logging with binary diagnostics
|
||||
if (data.Length > 0)
|
||||
if (!data.IsEmpty)
|
||||
{
|
||||
LogBinaryDiagnosticsOnError(messageTag, data, requestId, ex);
|
||||
LogBinaryDiagnosticsOnError(messageTag, data, ex);
|
||||
}
|
||||
|
||||
if (requestId.HasValue && _responseByRequestId.TryRemove(requestId.Value, out var exModel))
|
||||
|
|
@ -499,36 +499,37 @@ namespace AyCode.Services.SignalRs
|
|||
/// <summary>
|
||||
/// Logs binary diagnostics for debugging serialization issues.
|
||||
/// </summary>
|
||||
private void LogBinaryDiagnostics(int messageTag, byte[] messageBytes, int? requestId)
|
||||
private void LogBinaryDiagnostics(int messageTag, SignalData data)
|
||||
{
|
||||
if (!EnableBinaryDiagnostics || messageBytes.Length == 0) return;
|
||||
if (!EnableBinaryDiagnostics || data.IsEmpty) return;
|
||||
|
||||
try
|
||||
{
|
||||
var hexDump = Convert.ToHexString(messageBytes.AsSpan(0, Math.Min(500, messageBytes.Length)));
|
||||
Logger.Info($"=== BINARY DIAGNOSTICS === Tag: {messageTag}; RequestId: {requestId}; Length: {messageBytes.Length}");
|
||||
var span = data.Span;
|
||||
var hexDump = Convert.ToHexString(span[..Math.Min(500, span.Length)]);
|
||||
Logger.Info($"=== BINARY DIAGNOSTICS === Tag: {messageTag}; Length: {data.Length}");
|
||||
Logger.Info($"HEX (first 500 bytes): {hexDump}");
|
||||
|
||||
// Parse header info
|
||||
if (messageBytes.Length >= 3)
|
||||
if (span.Length >= 3)
|
||||
{
|
||||
var version = messageBytes[0];
|
||||
var marker = messageBytes[1];
|
||||
var version = span[0];
|
||||
var marker = span[1];
|
||||
Logger.Info($"Version: {version}; Marker: 0x{marker:X2}");
|
||||
|
||||
if ((marker & 0x10) != 0 && messageBytes.Length > 2)
|
||||
if ((marker & 0x10) != 0 && span.Length > 2)
|
||||
{
|
||||
var propCount = messageBytes[2];
|
||||
var propCount = span[2];
|
||||
Logger.Info($"Header property count: {propCount}");
|
||||
|
||||
// Parse first 10 property names
|
||||
var pos = 3;
|
||||
for (int i = 0; i < Math.Min((int)propCount, 10) && pos < messageBytes.Length; i++)
|
||||
for (int i = 0; i < Math.Min((int)propCount, 10) && pos < span.Length; i++)
|
||||
{
|
||||
var strLen = messageBytes[pos++];
|
||||
if (pos + strLen <= messageBytes.Length)
|
||||
var strLen = span[pos++];
|
||||
if (pos + strLen <= span.Length)
|
||||
{
|
||||
var propName = System.Text.Encoding.UTF8.GetString(messageBytes, pos, strLen);
|
||||
var propName = System.Text.Encoding.UTF8.GetString(span.Slice(pos, strLen));
|
||||
pos += strLen;
|
||||
Logger.Info($" [{i}]: '{propName}'");
|
||||
}
|
||||
|
|
@ -545,37 +546,38 @@ namespace AyCode.Services.SignalRs
|
|||
/// <summary>
|
||||
/// Logs binary diagnostics when an error occurs during deserialization.
|
||||
/// </summary>
|
||||
private void LogBinaryDiagnosticsOnError(int messageTag, byte[] messageBytes, int? requestId, Exception error)
|
||||
private void LogBinaryDiagnosticsOnError(int messageTag, SignalData data, Exception error)
|
||||
{
|
||||
try
|
||||
{
|
||||
var span = data.Span;
|
||||
Logger.Error($"=== BINARY DESERIALIZATION ERROR ===");
|
||||
Logger.Error($"Tag: {messageTag}; RequestId: {requestId}; Length: {messageBytes.Length}");
|
||||
Logger.Error($"Tag: {messageTag}; Length: {data.Length}");
|
||||
Logger.Error($"Error: {error.Message}");
|
||||
|
||||
var hexDump = Convert.ToHexString(messageBytes.AsSpan(0, Math.Min(1000, messageBytes.Length)));
|
||||
var hexDump = Convert.ToHexString(span[..Math.Min(1000, span.Length)]);
|
||||
Logger.Error($"HEX (first 1000 bytes): {hexDump}");
|
||||
|
||||
// Parse header info
|
||||
if (messageBytes.Length >= 3)
|
||||
if (span.Length >= 3)
|
||||
{
|
||||
var version = messageBytes[0];
|
||||
var marker = messageBytes[1];
|
||||
var version = span[0];
|
||||
var marker = span[1];
|
||||
Logger.Error($"Version: {version}; Marker: 0x{marker:X2}");
|
||||
|
||||
if ((marker & 0x10) != 0 && messageBytes.Length > 2)
|
||||
if ((marker & 0x10) != 0 && span.Length > 2)
|
||||
{
|
||||
var propCount = messageBytes[2];
|
||||
var propCount = span[2];
|
||||
Logger.Error($"Header property count: {propCount}");
|
||||
|
||||
// Parse ALL property names
|
||||
var pos = 3;
|
||||
for (int i = 0; i < propCount && pos < messageBytes.Length; i++)
|
||||
for (int i = 0; i < propCount && pos < span.Length; i++)
|
||||
{
|
||||
var strLen = messageBytes[pos++];
|
||||
if (pos + strLen <= messageBytes.Length)
|
||||
var strLen = span[pos++];
|
||||
if (pos + strLen <= span.Length)
|
||||
{
|
||||
var propName = System.Text.Encoding.UTF8.GetString(messageBytes, pos, strLen);
|
||||
var propName = System.Text.Encoding.UTF8.GetString(span.Slice(pos, strLen));
|
||||
pos += strLen;
|
||||
Logger.Error($" Header[{i}]: '{propName}'");
|
||||
}
|
||||
|
|
|
|||
|
|
@ -0,0 +1,27 @@
|
|||
using System.Buffers;
|
||||
using AyCode.Core.Serializers.Binaries;
|
||||
|
||||
namespace AyCode.Services.SignalRs;
|
||||
|
||||
/// <summary>
|
||||
/// Project-specific binary protocol. Uses ArrayPool for byte[] arguments
|
||||
/// when the target type is SignalData (client receive path optimization).
|
||||
/// Register this in PluginNopStartup.cs and AcSignalRClientBase instead of AcBinaryHubProtocol.
|
||||
/// </summary>
|
||||
public class AyCodeBinaryHubProtocol : AcBinaryHubProtocol
|
||||
{
|
||||
public AyCodeBinaryHubProtocol() { }
|
||||
public AyCodeBinaryHubProtocol(AcBinarySerializerOptions options) : base(options) { }
|
||||
|
||||
protected override object CreateByteArrayResult(ReadOnlySpan<byte> data, Type targetType)
|
||||
{
|
||||
if (targetType == typeof(SignalData))
|
||||
{
|
||||
var rented = ArrayPool<byte>.Shared.Rent(data.Length);
|
||||
data.CopyTo(rented);
|
||||
return new SignalData(rented, data.Length, isRented: true);
|
||||
}
|
||||
|
||||
return base.CreateByteArrayResult(data, targetType);
|
||||
}
|
||||
}
|
||||
|
|
@ -3,5 +3,5 @@
|
|||
public interface IAcSignalRHubBase
|
||||
{
|
||||
//Task OnRequestMessage(int messageTag, int requestId);
|
||||
Task OnReceiveMessage(int messageTag, int? requestId, SignalParams signalParams, byte[] data);
|
||||
Task OnReceiveMessage(int messageTag, int? requestId, SignalParams signalParams, SignalData data);
|
||||
}
|
||||
|
|
@ -2,6 +2,7 @@
|
|||
using AyCode.Core.Interfaces;
|
||||
using System.Buffers;
|
||||
using System.Runtime.CompilerServices;
|
||||
using AyCode.Core.Serializers.Binaries;
|
||||
using AyCode.Core.Serializers.Jsons;
|
||||
using JsonIgnoreAttribute = Newtonsoft.Json.JsonIgnoreAttribute;
|
||||
using STJIgnore = System.Text.Json.Serialization.JsonIgnoreAttribute;
|
||||
|
|
@ -150,7 +151,7 @@ public sealed class SignalResponseDataMessage : ISignalResponseMessage, IDisposa
|
|||
public int MessageTag { get; set; }
|
||||
public SignalResponseStatus Status { get; set; }
|
||||
public AcSerializerType DataSerializerType { get; set; }
|
||||
public byte[]? ResponseData { get; set; }
|
||||
public SignalData? ResponseData { get; set; }
|
||||
|
||||
[JsonIgnore] [STJIgnore] private object? _cachedResponseData;
|
||||
[JsonIgnore] [STJIgnore] private byte[]? _rentedDecompressedBuffer;
|
||||
|
|
@ -175,7 +176,8 @@ public sealed class SignalResponseDataMessage : ISignalResponseMessage, IDisposa
|
|||
: this(messageTag, status)
|
||||
{
|
||||
DataSerializerType = serializerOptions.SerializerType;
|
||||
ResponseData = SignalRSerializationHelper.CreateResponseData(responseData, serializerOptions);
|
||||
var bytes = SignalRSerializationHelper.CreateResponseData(responseData, serializerOptions);
|
||||
ResponseData = bytes != null ? new SignalData(bytes) : null;
|
||||
}
|
||||
|
||||
/// <summary>
|
||||
|
|
@ -185,7 +187,7 @@ public sealed class SignalResponseDataMessage : ISignalResponseMessage, IDisposa
|
|||
public T? GetResponseData<T>()
|
||||
{
|
||||
if (_cachedResponseData != null) return (T)_cachedResponseData;
|
||||
if (ResponseData == null) return default;
|
||||
if (ResponseData == null || ResponseData.IsEmpty) return default;
|
||||
|
||||
try
|
||||
{
|
||||
|
|
@ -194,7 +196,7 @@ public sealed class SignalResponseDataMessage : ISignalResponseMessage, IDisposa
|
|||
// Log diagnostics if enabled
|
||||
LogResponseDataDiagnostics<T>();
|
||||
|
||||
return (T)(_cachedResponseData = ResponseData.BinaryTo<T>()!);
|
||||
return (T)(_cachedResponseData = AcBinaryDeserializer.Deserialize<T>(ResponseData.Span)!);
|
||||
}
|
||||
|
||||
// Decompress GZip to pooled buffer and deserialize directly
|
||||
|
|
@ -267,10 +269,10 @@ public sealed class SignalResponseDataMessage : ISignalResponseMessage, IDisposa
|
|||
}
|
||||
|
||||
DiagnosticLogger($"ResponseData.Length: {ResponseData.Length}");
|
||||
DiagnosticLogger($"HEX (first 500 bytes): {Convert.ToHexString(ResponseData.AsSpan(0, Math.Min(500, ResponseData.Length)))}");
|
||||
DiagnosticLogger($"HEX (first 500 bytes): {Convert.ToHexString(ResponseData.Span[..Math.Min(500, ResponseData.Length)])}");
|
||||
|
||||
// Parse header with VarInt support
|
||||
LogBinaryHeader(ResponseData);
|
||||
LogBinaryHeader(ResponseData.Span);
|
||||
}
|
||||
catch (Exception ex)
|
||||
{
|
||||
|
|
@ -297,7 +299,7 @@ public sealed class SignalResponseDataMessage : ISignalResponseMessage, IDisposa
|
|||
}
|
||||
}
|
||||
|
||||
private static void LogBinaryHeader(byte[] data)
|
||||
private static void LogBinaryHeader(ReadOnlySpan<byte> data)
|
||||
{
|
||||
if (DiagnosticLogger == null || data.Length < 3) return;
|
||||
|
||||
|
|
@ -314,7 +316,7 @@ public sealed class SignalResponseDataMessage : ISignalResponseMessage, IDisposa
|
|||
|
||||
// Read property count as VarUInt
|
||||
var pos = 2;
|
||||
var (propCount, bytesRead) = ReadVarUIntFromSpan(data.AsSpan(pos));
|
||||
var (propCount, bytesRead) = ReadVarUIntFromSpan(data[pos..]);
|
||||
pos += bytesRead;
|
||||
|
||||
DiagnosticLogger($"Header Property Count: {propCount}");
|
||||
|
|
@ -322,12 +324,12 @@ public sealed class SignalResponseDataMessage : ISignalResponseMessage, IDisposa
|
|||
for (var i = 0; i < (int)propCount && pos < data.Length; i++)
|
||||
{
|
||||
// Read string length as VarUInt
|
||||
var (strLen, strLenBytes) = ReadVarUIntFromSpan(data.AsSpan(pos));
|
||||
var (strLen, strLenBytes) = ReadVarUIntFromSpan(data[pos..]);
|
||||
pos += strLenBytes;
|
||||
|
||||
if (pos + (int)strLen <= data.Length)
|
||||
{
|
||||
var propName = System.Text.Encoding.UTF8.GetString(data, pos, (int)strLen);
|
||||
var propName = System.Text.Encoding.UTF8.GetString(data.Slice(pos, (int)strLen));
|
||||
pos += (int)strLen;
|
||||
DiagnosticLogger($" Header[{i}]: '{propName}'");
|
||||
}
|
||||
|
|
@ -393,10 +395,10 @@ public sealed class SignalResponseDataMessage : ISignalResponseMessage, IDisposa
|
|||
}
|
||||
|
||||
DiagnosticLogger($"ResponseData.Length: {ResponseData.Length}");
|
||||
DiagnosticLogger($"HEX (first 1000 bytes): {Convert.ToHexString(ResponseData.AsSpan(0, Math.Min(1000, ResponseData.Length)))}");
|
||||
DiagnosticLogger($"HEX (first 1000 bytes): {Convert.ToHexString(ResponseData.Span[..Math.Min(1000, ResponseData.Length)])}");
|
||||
|
||||
// Parse header
|
||||
LogBinaryHeader(ResponseData);
|
||||
LogBinaryHeader(ResponseData.Span);
|
||||
|
||||
// Log inner exception if present
|
||||
if (error.InnerException != null)
|
||||
|
|
@ -418,7 +420,7 @@ public sealed class SignalResponseDataMessage : ISignalResponseMessage, IDisposa
|
|||
/// </summary>
|
||||
public ReadOnlySpan<byte> GetDecompressedJsonSpan()
|
||||
{
|
||||
if (ResponseData == null) return ReadOnlySpan<byte>.Empty;
|
||||
if (ResponseData == null || ResponseData.IsEmpty) return ReadOnlySpan<byte>.Empty;
|
||||
if (DataSerializerType == AcSerializerType.Binary) return ReadOnlySpan<byte>.Empty;
|
||||
|
||||
EnsureDecompressed();
|
||||
|
|
@ -430,11 +432,17 @@ public sealed class SignalResponseDataMessage : ISignalResponseMessage, IDisposa
|
|||
{
|
||||
if (_rentedDecompressedBuffer != null) return;
|
||||
|
||||
(_rentedDecompressedBuffer, _decompressedLength) = SignalRSerializationHelper.DecompressToRentedBuffer(ResponseData!);
|
||||
(_rentedDecompressedBuffer, _decompressedLength) = AyCode.Core.Compression.GzipHelper.DecompressToRentedBuffer(ResponseData!.Span);
|
||||
}
|
||||
|
||||
public void Dispose()
|
||||
{
|
||||
ResponseData?.Dispose();
|
||||
if (_rentedDecompressedBuffer != null)
|
||||
{
|
||||
ArrayPool<byte>.Shared.Return(_rentedDecompressedBuffer, clearArray: true);
|
||||
_rentedDecompressedBuffer = null;
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
|
|
|
|||
|
|
@ -8,13 +8,15 @@ Custom binary SignalR protocol, client infrastructure, message tagging, and seri
|
|||
## Key Files
|
||||
|
||||
### Protocol
|
||||
- **`AcBinaryHubProtocol.cs`** — Custom `IHubProtocol` replacing JSON+Base64 with `AcBinarySerializer`. Handles all 9 SignalR message types (Invocation, StreamItem, Completion, Ping, Close, etc.). Uses `BufferWriterBinaryOutput` standalone mode for zero-copy writes to the SignalR pipe. `byte[]` fast-path bypasses the serializer entirely. Inner `SpanReader` ref struct for zero-alloc parsing.
|
||||
- **`AcBinaryHubProtocol.cs`** — Unsealed base `IHubProtocol` replacing JSON+Base64 with `AcBinarySerializer`. Handles all 9 SignalR message types. Uses `BufferWriterBinaryOutput` standalone mode for zero-copy writes. `byte[]`/`SignalData` fast-path bypasses serializer. `CreateByteArrayResult` virtual hook for derived protocols. Inner `SpanReader` ref struct for zero-alloc parsing.
|
||||
- **`AyCodeBinaryHubProtocol.cs`** — Derived protocol. Overrides `CreateByteArrayResult` to rent from `ArrayPool` when `targetType == typeof(SignalData)`. Register this instead of `AcBinaryHubProtocol`.
|
||||
- **`SignalData.cs`** — `IDisposable` wrapper for `byte[]` with optional `ArrayPool` lifecycle. `Span` for zero-copy access, `Dispose()` returns rented buffer. Created by `AyCodeBinaryHubProtocol` (pooled) or directly from `byte[]` (server send).
|
||||
|
||||
### Client
|
||||
- **`AcSignalRClientBase.cs`** — Abstract SignalR client managing `HubConnection`, request/response tracking via pooled `SignalRRequestModel`. Methods: `SendMessageToServerAsync<TResponse>()`, CRUD helpers (Post, Get, GetAll, GetAllInto). Configurable timeouts.
|
||||
- **`IAcSignalRHubClient.cs`** — Client interface + `SignalResponseDataMessage` (sealed, supports JSON/Binary with GZip, caching, diagnostics).
|
||||
- **`IAcSignalRHubBase.cs`** — Base hub interface: `OnReceiveMessage(int messageTag, int? requestId, SignalParams signalParams, byte[] data)`.
|
||||
- **`ISignalParams.cs`** — `ISignalParams` base interface + `SignalParams` (Status, DataSerializerType, Parameters `byte[][]?`). Metadata travels as separate hub argument (AcBinary serialized), payload `byte[]` uses protocol fast-path (zero-copy). Parameters and data are independent — both nullable in any direction.
|
||||
- **`IAcSignalRHubClient.cs`** — Client interface + `SignalResponseDataMessage` (sealed, `ResponseData` is `SignalData?`, supports JSON/Binary with GZip, caching, diagnostics, `Dispose()` returns buffers to ArrayPool).
|
||||
- **`IAcSignalRHubBase.cs`** — Base hub interface: `OnReceiveMessage(int messageTag, int? requestId, SignalParams signalParams, SignalData data)`.
|
||||
- **`ISignalParams.cs`** — `ISignalParams` base interface + `SignalParams` (Status, DataSerializerType, Parameters `byte[]?`). Metadata travels as separate hub argument (AcBinary serialized), payload `SignalData` uses protocol fast-path (ArrayPool-backed). Parameters and data are independent — both nullable in any direction.
|
||||
|
||||
### Message Tagging
|
||||
- **`SignalMessageTagAttribute.cs`** — Three attributes: `TagAttribute` (base, int messageTag), `SignalRAttribute` (server method routing + client notification), `SignalRSendToClientAttribute` (client-side receive).
|
||||
|
|
|
|||
|
|
@ -0,0 +1,52 @@
|
|||
using System.Buffers;
|
||||
|
||||
namespace AyCode.Services.SignalRs;
|
||||
|
||||
/// <summary>
|
||||
/// Wrapper for byte[] response data with optional ArrayPool lifecycle.
|
||||
/// Created by AyCodeBinaryHubProtocol for pooled buffers,
|
||||
/// or directly from byte[] for non-pooled data (server send path).
|
||||
/// Consumer must Dispose() to return rented buffer.
|
||||
/// Supports future AsyncEnumerable streaming (per-chunk lifecycle).
|
||||
/// </summary>
|
||||
public sealed class SignalData : IDisposable
|
||||
{
|
||||
private byte[]? _buffer;
|
||||
private readonly int _length;
|
||||
private readonly bool _isRented;
|
||||
|
||||
/// <summary>Pooled buffer from ArrayPool (rented, length >= actual data).</summary>
|
||||
public SignalData(byte[] rentedBuffer, int length, bool isRented)
|
||||
{
|
||||
_buffer = rentedBuffer;
|
||||
_length = length;
|
||||
_isRented = isRented;
|
||||
}
|
||||
|
||||
/// <summary>Non-pooled byte[] (server send, direct creation).</summary>
|
||||
public SignalData(byte[] data)
|
||||
{
|
||||
_buffer = data;
|
||||
_length = data?.Length ?? 0;
|
||||
_isRented = false;
|
||||
}
|
||||
|
||||
public ReadOnlySpan<byte> Span => _buffer.AsSpan(0, _length);
|
||||
public int Length => _length;
|
||||
public bool IsEmpty => _length == 0 || _buffer == null;
|
||||
|
||||
/// <summary>
|
||||
/// Returns a copy as byte[]. Use only when a byte[] is absolutely required.
|
||||
/// Prefer Span for zero-copy access.
|
||||
/// </summary>
|
||||
public byte[] ToArray() => Span.ToArray();
|
||||
|
||||
public void Dispose()
|
||||
{
|
||||
if (_isRented && _buffer != null)
|
||||
{
|
||||
ArrayPool<byte>.Shared.Return(_buffer, clearArray: true);
|
||||
_buffer = null;
|
||||
}
|
||||
}
|
||||
}
|
||||
|
|
@ -15,13 +15,13 @@ Client ◄──OnReceiveMessage(tag, requestId, signalParams, data)── Serve
|
|||
```
|
||||
|
||||
Tag (int) determines server method. All calls go through `OnReceiveMessage`.
|
||||
Metadata (`SignalParams`) and payload (`byte[]`) travel as **separate hub arguments** — the `byte[]` uses the protocol's zero-copy fast-path, metadata is AcBinary serialized normally.
|
||||
Metadata (`SignalParams`) and payload (`SignalData`) travel as **separate hub arguments** — `SignalData` wraps pooled `byte[]` from `ArrayPool` via `AyCodeBinaryHubProtocol` (zero-copy fast-path), metadata is AcBinary serialized normally.
|
||||
|
||||
```
|
||||
Client: Server:
|
||||
AcSignalRClientBase AcWebSignalRHubBase<TTags, TLogger>
|
||||
├─ HubConnection (WebSocket) ├─ Hub<IAcSignalRHubItemServer>
|
||||
├─ AcBinaryHubProtocol ├─ DynamicMethodRegistry
|
||||
├─ AyCodeBinaryHubProtocol ├─ DynamicMethodRegistry
|
||||
├─ Pending request tracking ├─ Parameter deserialization
|
||||
└─ Response callbacks └─ Broadcast to other clients
|
||||
```
|
||||
|
|
@ -68,9 +68,11 @@ CRUD helpers (`PostAsync`, `GetByIdAsync`, `GetAllAsync`, `PostDataAsync`) are g
|
|||
|
||||
## Wire Protocol
|
||||
|
||||
### AcBinaryHubProtocol
|
||||
### AcBinaryHubProtocol / AyCodeBinaryHubProtocol
|
||||
|
||||
Custom `IHubProtocol` (`"acbinary"`), replaces JSON. Zero-copy via `BufferWriterBinaryOutput` standalone mode. `byte[]` args bypass serializer.
|
||||
Custom `IHubProtocol` (`"acbinary"`), replaces JSON. Zero-copy via `BufferWriterBinaryOutput` standalone mode. `byte[]` and `SignalData` args bypass serializer.
|
||||
|
||||
`AcBinaryHubProtocol` is the base (unsealed, generic). `AyCodeBinaryHubProtocol` derives from it and uses `ArrayPool` for `SignalData` arguments — the `CreateByteArrayResult` hook rents from pool instead of `.ToArray()`. Register `AyCodeBinaryHubProtocol` in both client and server.
|
||||
|
||||
> Wire format, argument framing, dual BWO pattern, length prefix patching: `SIGNALR_BINARY_PROTOCOL.md`
|
||||
|
||||
|
|
@ -89,18 +91,20 @@ Typed access via methods (PostDataJson pattern):
|
|||
- **Server**: `GetParameterValues(ParameterInfo[])` — unpacks `byte[]` → `byte[][]` → per-element `BinaryTo(targetType)`
|
||||
- Protocol never sees `byte[][]` — only `byte[]`.
|
||||
|
||||
`byte[] data` (separate hub argument, protocol fast-path, zero-copy).
|
||||
`SignalData data` (separate hub argument, protocol fast-path, ArrayPool-backed via `AyCodeBinaryHubProtocol`).
|
||||
|
||||
`SignalData` wraps pooled `byte[]` with `IDisposable` lifecycle. Consumer accesses via `Span` (zero-copy) or `ToArray()` (copy, rare). `Dispose()` returns rented buffer to `ArrayPool` with `clearArray: true`.
|
||||
|
||||
`Parameters` and `data` are **independent** — both can be null or filled in any direction (SignalR is bidirectional).
|
||||
|
||||
| Combination | Parameters | data | Example |
|
||||
|------------|-----------|------|---------|
|
||||
| Request | `byte[]` (packed params) | null | client calls server method |
|
||||
| Response | null | response payload | server returns result |
|
||||
| Request + data | `byte[]` | `byte[]` | client responds to server with data |
|
||||
| Signal | null | null | ping, status change, broadcast trigger |
|
||||
| Request | `byte[]` (packed params) | null/empty | client calls server method |
|
||||
| Response | null | SignalData (response payload) | server returns result |
|
||||
| Request + data | `byte[]` | SignalData | client responds to server with data |
|
||||
| Signal | null | null/empty | ping, status change, broadcast trigger |
|
||||
|
||||
`SignalResponseDataMessage` remains as **internal DTO** for callback routing — constructed in-memory from `signalParams` + `data`, never serialized as envelope on wire. `GetResponseData<T>()` dispatches on `DataSerializerType`: Binary → `BinaryTo<T>()`, JsonGZip → decompress → `JsonTo<T>()`.
|
||||
`SignalResponseDataMessage` remains as **internal DTO** for callback routing — constructed in-memory from `signalParams` + `data`, never serialized as envelope on wire. `ResponseData` is `SignalData?`. `GetResponseData<T>()` dispatches on `DataSerializerType`: Binary → `AcBinaryDeserializer.Deserialize<T>(Span)`, JsonGZip → decompress → `JsonTo<T>()`. `Dispose()` returns both SignalData and JSON decompression buffers to ArrayPool.
|
||||
|
||||
## Request/Response Flow
|
||||
|
||||
|
|
@ -112,7 +116,7 @@ Typed access via methods (PostDataJson pattern):
|
|||
Each param ToBinary() → byte[][] → ToBinary() → byte[] (single wire blob)
|
||||
3. SignalParams { Status = Success, Parameters = byte[] }
|
||||
4. HubConnection.SendAsync("OnReceiveMessage", tag, requestId, signalParams, null)
|
||||
5. AcBinaryHubProtocol frames on wire (signalParams via AcBinary, data = null for requests)
|
||||
5. AyCodeBinaryHubProtocol frames on wire (signalParams via AcBinary, data = null for requests)
|
||||
```
|
||||
|
||||
### Server → Client
|
||||
|
|
@ -120,13 +124,13 @@ Typed access via methods (PostDataJson pattern):
|
|||
```
|
||||
OnReceiveMessage(tag, requestId, signalParams, data)
|
||||
├─ Construct SignalResponseDataMessage in-memory (no envelope deser):
|
||||
│ └─ { Status, DataSerializerType, ResponseData } from signalParams + data
|
||||
│ └─ { Status, DataSerializerType, ResponseData (SignalData) } from signalParams + data
|
||||
├─ Matching requestId in pending dict:
|
||||
│ ├─ Route: null→sync wait, Action→invoke, Func<Task>→await
|
||||
│ └─ GetResponseData<T>(): dispatches on DataSerializerType
|
||||
│ Binary→BinaryTo<T>(), JsonGZip→Decompress→JsonTo<T>()
|
||||
│ Binary→Deserialize<T>(Span), JsonGZip→Decompress→JsonTo<T>()
|
||||
└─ No match (broadcast):
|
||||
└─ abstract MessageReceived(tag, signalParams, data).Forget()
|
||||
└─ abstract MessageReceived(tag, signalParams, SignalData data).Forget()
|
||||
```
|
||||
|
||||
Request pooling: `SignalRRequestModel` via `SignalRRequestModelPool` (ObjectPool + IResettable).
|
||||
|
|
@ -173,7 +177,9 @@ Type-guided deserialization — each parameter is individually serialized/deseri
|
|||
| Component | Path |
|
||||
|-----------|------|
|
||||
| Client base | `SignalRs/AcSignalRClientBase.cs` |
|
||||
| Binary protocol | `SignalRs/AcBinaryHubProtocol.cs` |
|
||||
| Binary protocol (base) | `SignalRs/AcBinaryHubProtocol.cs` |
|
||||
| Binary protocol (derived) | `SignalRs/AyCodeBinaryHubProtocol.cs` |
|
||||
| Signal data wrapper | `SignalRs/SignalData.cs` |
|
||||
| Tag attributes | `SignalRs/SignalMessageTagAttribute.cs` |
|
||||
| Base tags | `SignalRs/AcSignalRTags.cs` |
|
||||
| CRUD tags | `SignalRs/SignalRCrudTags.cs` |
|
||||
|
|
|
|||
|
|
@ -1,6 +1,6 @@
|
|||
# SignalR Binary Protocol
|
||||
|
||||
`AcBinaryHubProtocol` — custom `IHubProtocol` (name: `"acbinary"`) replacing SignalR JSON+Base64 with `AcBinarySerializer`.
|
||||
`AcBinaryHubProtocol` (unsealed base) — custom `IHubProtocol` (name: `"acbinary"`) replacing SignalR JSON+Base64 with `AcBinarySerializer`. `AyCodeBinaryHubProtocol` (derived) adds `ArrayPool`-backed `SignalData` creation via `CreateByteArrayResult` hook.
|
||||
|
||||
> Architecture (tag system, dispatch, request/response): `SIGNALR.md`
|
||||
> Output writers (cached chunk, buffer states, chunk sizing): `AyCode.Core/docs/BINARY_WRITERS.md`
|
||||
|
|
@ -84,7 +84,9 @@ When argument is `byte[]`, bypasses serializer:
|
|||
2. INT32 prefix written with actual value (no patching)
|
||||
3. `BinaryTypeCode.ByteArray(68)` + VarUInt length + raw bytes via BWO
|
||||
|
||||
Read side mirrors: if first byte is `ByteArray(0x44)`, deserializer bypassed → direct `SpanReader`. Detection is **wire-format only** (no targetType check) — ByteArray marker is unambiguous since no AcBinary object starts with 0x44 (version=1).
|
||||
Read side mirrors: if first byte is `ByteArray(0x44)`, deserializer bypassed → direct `SpanReader` → `CreateByteArrayResult(span, targetType)`. Base returns `data.ToArray()`. `AyCodeBinaryHubProtocol` overrides: if `targetType == typeof(SignalData)`, rents from `ArrayPool` and returns `SignalData(rented, length, isRented: true)`. Detection is **wire-format only** (no targetType check for the marker) — ByteArray marker is unambiguous since no AcBinary object starts with 0x44 (version=1).
|
||||
|
||||
Write side: `WriteArgument` handles both `byte[]` and `SignalData` via the same ByteArray wire format. `SignalData.Span` is written directly — same marker + VarUInt length + raw bytes.
|
||||
|
||||
## Read Path
|
||||
|
||||
|
|
@ -103,4 +105,4 @@ Read side mirrors: if first byte is `ByteArray(0x44)`, deserializer bypassed →
|
|||
| `Options` | `AcBinarySerializerOptions.Default` | Serializer options (volatile, runtime-replaceable) |
|
||||
| `BufferWriterChunkSize` | 65536 | Chunk size for both BWOs |
|
||||
|
||||
**Source:** `AyCode.Services/SignalRs/AcBinaryHubProtocol.cs`
|
||||
**Source:** `AyCode.Services/SignalRs/AcBinaryHubProtocol.cs` (base), `AyCode.Services/SignalRs/AyCodeBinaryHubProtocol.cs` (derived, ArrayPool)
|
||||
|
|
|
|||
|
|
@ -36,7 +36,7 @@ AyCode.Services ← AyCode.Services.Server
|
|||
- **AyCode.Services.Server** — Server-side: SignalR hub with custom binary protocol, email (SendGrid), JWT auth.
|
||||
- **AyCode.Models.Server/DynamicMethods** — Reflection-based tag→method dispatch used by the SignalR hub.
|
||||
|
||||
> **SignalR Dispatch:** Both directions use a single method `OnReceiveMessage(int messageTag, byte[] messageBytes, int? requestId)` with integer tag-based routing instead of standard Hub methods. See `AyCode.Services/docs/SIGNALR.md` for full details.
|
||||
> **SignalR Dispatch:** Both directions use a single method `OnReceiveMessage(int messageTag, int? requestId, SignalParams signalParams, SignalData data)` with integer tag-based routing instead of standard Hub methods. `SignalData` wraps `ArrayPool`-backed byte buffers for zero-alloc receive path. See `AyCode.Services/docs/SIGNALR.md` for full details.
|
||||
|
||||
### Server Extensions
|
||||
- **AyCode.Core.Server**, **AyCode.Interfaces.Server**, **AyCode.Entities.Server**, **AyCode.Models.Server** — Server-only additions that don't belong in shared code.
|
||||
|
|
|
|||
|
|
@ -26,10 +26,10 @@
|
|||
|
||||
See `AyCode.Services/docs/SIGNALR.md` for full architecture documentation.
|
||||
|
||||
- **Single dispatch method** — all communication goes through `OnReceiveMessage(int messageTag, byte[] messageBytes, int? requestId)`. Do not add new hub methods.
|
||||
- **Single dispatch method** — all communication goes through `OnReceiveMessage(int messageTag, int? requestId, SignalParams signalParams, SignalData data)`. Do not add new hub methods.
|
||||
- **Tag-based routing** — associate methods with integer tags via `[SignalR(tag)]` (server) or `[SignalRSendToClient(tag)]` (client). Tags must be unique across the entire system.
|
||||
- **CRUD bundles** — entities use `SignalRCrudTags(getAllTag, getItemTag, addTag, updateTag, removeTag)` with 5 independent tag integers. Tags must be unique across the system. See `AyCode.Services.Server/docs/SIGNALR_DATASOURCE.md`.
|
||||
- **Binary protocol** — `AcBinaryHubProtocol` is the transport protocol. Responses use pure Binary serialization.
|
||||
- **Binary protocol** — `AyCodeBinaryHubProtocol` (derived from `AcBinaryHubProtocol`) is the transport protocol. Uses `ArrayPool`-backed `SignalData` for response payload. Responses use pure Binary serialization.
|
||||
|
||||
### ⚠️ Temporary: JSON-in-Binary Request Parameters
|
||||
|
||||
|
|
|
|||
|
|
@ -70,18 +70,20 @@ For full architecture see `AyCode.Services/docs/SIGNALR.md`.
|
|||
|
||||
| Term | Definition |
|
||||
|---|---|
|
||||
| **OnReceiveMessage** | The single SignalR method for all communication. Signature: `(int messageTag, int? requestId, SignalParams signalParams, byte[] data)`. Metadata and payload are separate hub arguments — `byte[]` uses protocol zero-copy fast-path. `SignalParams.Parameters` carries packed method params as `byte[]`. `data` carries response payload. Both are independent and nullable in any direction. |
|
||||
| **OnReceiveMessage** | The single SignalR method for all communication. Signature: `(int messageTag, int? requestId, SignalParams signalParams, SignalData data)`. Metadata and payload are separate hub arguments — `SignalData` uses protocol zero-copy fast-path with `ArrayPool` backing. `SignalParams.Parameters` carries packed method params as `byte[]`. `data` carries response payload. Both are independent and nullable in any direction. |
|
||||
| **SignalParams** | Metadata sent alongside message payload as separate hub argument. Contains `Status`, `DataSerializerType`, and `Parameters` (`byte[]?` — packed `byte[][]` as single blob, protocol fast-path). Typed access via `SetParameterValues(object[])` / `GetParameterValues(ParameterInfo[])` — PostDataJson pattern. `[AcBinarySerializable]`. Never null — only `Parameters` inside is nullable. |
|
||||
| **Message Tag** | Integer identifier mapping to a method via `[SignalR(tag)]` or `[SignalRSendToClient(tag)]` attributes. |
|
||||
| **DynamicMethodRegistry** | Resolves message tags to `MethodInfo` at runtime. Static `ConcurrentDictionary` cache with lazy scan on miss. |
|
||||
| **SignalRCrudTags** | Sealed class bundling 5 independent tag integers (getAllTag, getItemTag, addTag, updateTag, removeTag) for entity CRUD. See `AyCode.Services.Server/docs/SIGNALR_DATASOURCE.md`. |
|
||||
| **AcBinaryHubProtocol** | Custom `IHubProtocol` replacing SignalR's JSON+Base64 with `AcBinarySerializer`. Protocol name: `"acbinary"`. Uses `BufferWriterBinaryOutput` for zero-copy writes to the SignalR pipe. |
|
||||
| **SignalResponseDataMessage** | Internal DTO for callback routing (not serialized on wire). Constructed in-memory from `signalParams` + `data`. Supports Binary or JSON+GZip via `GetResponseData<T>()`. |
|
||||
| **AcBinaryHubProtocol** | Unsealed base `IHubProtocol` replacing SignalR's JSON+Base64 with `AcBinarySerializer`. Protocol name: `"acbinary"`. Uses `BufferWriterBinaryOutput` for zero-copy writes. `CreateByteArrayResult` virtual hook for derived protocols. |
|
||||
| **AyCodeBinaryHubProtocol** | Derived protocol. Overrides `CreateByteArrayResult`: when `targetType == typeof(SignalData)`, rents from `ArrayPool` instead of `.ToArray()`. Register this in both client and server. |
|
||||
| **SignalData** | `IDisposable` wrapper for `byte[]` response data with optional `ArrayPool` lifecycle. `Span` for zero-copy read access, `ToArray()` for copy (rare). `Dispose()` returns rented buffer with `clearArray: true`. Created by `AyCodeBinaryHubProtocol` (pooled) or directly from `byte[]` (server send, non-pooled). |
|
||||
| **SignalResponseDataMessage** | Internal DTO for callback routing (not serialized on wire). Constructed in-memory from `signalParams` + `SignalData`. `ResponseData` is `SignalData?`. `GetResponseData<T>()` deserializes from `Span`. `Dispose()` returns both SignalData and JSON decompression buffers to ArrayPool. |
|
||||
| **SignalPostJsonDataMessage** | OBSOLETE — removed. Legacy: serialized params to JSON inside Binary envelope. |
|
||||
| **AcSignalRDataSource** | Generic real-time `IList<T>` with change tracking, CRUD via SignalRCrudTags, binary merge, rollback, sync state. |
|
||||
| **TrackingItem** | Wraps a modified DataSource item with `TrackingState` (Add/Update/Remove) + `OriginalValue` for rollback. |
|
||||
| **SendToClientType** | Enum controlling broadcast scope: None, Others, Caller, All. |
|
||||
| **AcWebSignalRHubBase** | Abstract server hub. Receives `OnReceiveMessage`, dispatches via DynamicMethodRegistry, responds/broadcasts via `SendMessageToClient` (metadata + payload as separate args, no envelope). |
|
||||
| **AcWebSignalRHubBase** | Abstract server hub. Receives `OnReceiveMessage` (with `SignalData`), dispatches via DynamicMethodRegistry, responds/broadcasts via `SendMessageToClient` (metadata + `SignalData` payload as separate args, no envelope). |
|
||||
| **AcSignalRClientBase** | Abstract client. Manages `HubConnection`, request/response correlation via `requestId`, pooled `SignalRRequestModel`. |
|
||||
| **AcSessionService** | `ConcurrentDictionary`-based session tracker for connected SignalR clients. |
|
||||
| **AcSignalRSendToClientService** | Server-push service: `SendMessageToAllClients`, `SendMessageToConnection`, `SendMessageToUser`. |
|
||||
|
|
|
|||
Loading…
Reference in New Issue