Adam
/
AyCode.Core
1
0
Fork 0
Code Issues Pull Requests Projects Releases Wiki Activity

Remove MessagePack; unify SignalR serialization model 

Major refactor to eliminate MessagePack from SignalR messaging.
All serialization now uses explicit binary methods (.ToBinary/.BinaryTo)
and Brotli-compressed JSON, managed via a new SignalRSerializationHelper.
Custom stream classes and MessagePack attributes are removed.
API is now consistent, type-safe, and easier to maintain.
Test code and all message handling updated to use the new model.
This commit is contained in:
Loretta 2025-12-14 01:45:17 +01:00
parent 489ef7486c
commit 2f1c00fd5c
12 changed files with 355 additions and 760 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

271
AyCode.Core/Compression/BrotliHelper.cs
View File

@ -7,7 +7,6 @@ namespace AyCode.Core.Compression;
/// <summary> /// <summary>
/// Brotli compression/decompression helper for SignalR message transport. /// Brotli compression/decompression helper for SignalR message transport.
/// Used when JSON serializer is configured to reduce payload size.
/// Optimized for zero-allocation scenarios with pooled buffers. /// Optimized for zero-allocation scenarios with pooled buffers.
/// </summary> /// </summary>
public static class BrotliHelper public static class BrotliHelper
@ -15,6 +14,8 @@ public static class BrotliHelper
private const int DefaultBufferSize = 4096; private const int DefaultBufferSize = 4096;
private const int MaxStackAllocSize = 1024; private const int MaxStackAllocSize = 1024;
#region Compression
/// <summary> /// <summary>
/// Compresses a string using Brotli compression with pooled buffers. /// Compresses a string using Brotli compression with pooled buffers.
/// </summary> /// </summary>
@ -60,41 +61,21 @@ public static class BrotliHelper
if (data.IsEmpty) if (data.IsEmpty)
return []; return [];
// Estimate compressed size (typically 10-30% of original for text) using var outputStream = new MemoryStream();
var estimatedSize = Math.Max(data.Length / 2, 64); using (var brotliStream = new BrotliStream(outputStream, compressionLevel, leaveOpen: true))
var outputBuffer = ArrayPool<byte>.Shared.Rent(estimatedSize);
try
{ {
using var outputStream = new PooledMemoryStream(outputBuffer); brotliStream.Write(data);
using (var brotliStream = new BrotliStream(outputStream, compressionLevel, leaveOpen: true))
{
brotliStream.Write(data);
}
return outputStream.ToArray();
}
finally
{
ArrayPool<byte>.Shared.Return(outputBuffer);
} }
return outputStream.ToArray();
} }
/// <summary> #endregion
/// Compresses data directly to an IBufferWriter (zero intermediate allocation).
/// </summary>
public static void CompressTo(ReadOnlySpan<byte> data, IBufferWriter<byte> writer, CompressionLevel compressionLevel = CompressionLevel.Optimal)
{
if (data.IsEmpty)
return;
using var outputStream = new BufferWriterStream(writer); #region Decompression
using var brotliStream = new BrotliStream(outputStream, compressionLevel, leaveOpen: true);
brotliStream.Write(data);
}
/// <summary> /// <summary>
/// Decompresses Brotli-compressed data to a string. /// Decompresses Brotli-compressed data to a string.
/// Consider using Decompress + direct UTF-8 JsonTo for better performance. /// Consider using Decompress + direct UTF-8 deserialization for better performance.
/// </summary> /// </summary>
public static string DecompressToString(byte[] compressedData) public static string DecompressToString(byte[] compressedData)
{ {
@ -107,63 +88,36 @@ public static class BrotliHelper
/// <summary> /// <summary>
/// Decompresses Brotli-compressed data to a byte array. /// Decompresses Brotli-compressed data to a byte array.
/// Uses pooled buffers internally for reduced GC pressure.
/// </summary> /// </summary>
[MethodImpl(MethodImplOptions.AggressiveInlining)]
public static byte[] Decompress(byte[] compressedData) public static byte[] Decompress(byte[] compressedData)
{ {
if (compressedData == null || compressedData.Length == 0) if (compressedData == null || compressedData.Length == 0)
return []; return [];
return DecompressSpan(compressedData.AsSpan()); return DecompressCore(compressedData);
} }
/// <summary> /// <summary>
/// Decompresses Brotli-compressed data from a ReadOnlySpan. /// Decompresses Brotli-compressed data from a ReadOnlySpan.
/// </summary> /// </summary>
[MethodImpl(MethodImplOptions.AggressiveInlining)]
public static byte[] DecompressSpan(ReadOnlySpan<byte> compressedData) public static byte[] DecompressSpan(ReadOnlySpan<byte> compressedData)
{ {
if (compressedData.IsEmpty) if (compressedData.IsEmpty)
return []; return [];
// Estimate decompressed size (typically 3-10x compressed for text) return DecompressCore(compressedData.ToArray());
var estimatedSize = Math.Max(compressedData.Length * 4, DefaultBufferSize);
var outputBuffer = ArrayPool<byte>.Shared.Rent(estimatedSize);
try
{
using var inputStream = new ReadOnlySpanStream(compressedData);
using var brotliStream = new BrotliStream(inputStream, CompressionMode.Decompress);
using var outputStream = new PooledMemoryStream(outputBuffer);
brotliStream.CopyTo(outputStream);
return outputStream.ToArray();
}
finally
{
ArrayPool<byte>.Shared.Return(outputBuffer);
}
} }
/// <summary> private static byte[] DecompressCore(byte[] compressedData)
/// Decompresses directly into an ArrayBufferWriter for zero-copy deserialization.
/// </summary>
public static void DecompressTo(ReadOnlySpan<byte> compressedData, ArrayBufferWriter<byte> writer)
{ {
if (compressedData.IsEmpty) using var inputStream = new MemoryStream(compressedData, writable: false);
return;
using var inputStream = new ReadOnlySpanStream(compressedData);
using var brotliStream = new BrotliStream(inputStream, CompressionMode.Decompress); using var brotliStream = new BrotliStream(inputStream, CompressionMode.Decompress);
using var outputStream = new MemoryStream();
// Read in chunks directly to the writer brotliStream.CopyTo(outputStream);
int bytesRead; return outputStream.ToArray();
do
{
var buffer = writer.GetSpan(DefaultBufferSize);
bytesRead = brotliStream.Read(buffer);
if (bytesRead > 0)
writer.Advance(bytesRead);
} while (bytesRead > 0);
} }
/// <summary> /// <summary>
@ -175,10 +129,11 @@ public static class BrotliHelper
if (compressedData.IsEmpty) if (compressedData.IsEmpty)
return ([], 0); return ([], 0);
// Estimate decompressed size (typically 3-10x compressed for text)
var estimatedSize = Math.Max(compressedData.Length * 4, DefaultBufferSize); var estimatedSize = Math.Max(compressedData.Length * 4, DefaultBufferSize);
var outputBuffer = ArrayPool<byte>.Shared.Rent(estimatedSize); var outputBuffer = ArrayPool<byte>.Shared.Rent(estimatedSize);
using var inputStream = new ReadOnlySpanStream(compressedData); using var inputStream = new MemoryStream(compressedData.ToArray(), writable: false);
using var brotliStream = new BrotliStream(inputStream, CompressionMode.Decompress); using var brotliStream = new BrotliStream(inputStream, CompressionMode.Decompress);
var totalRead = 0; var totalRead = 0;
@ -189,7 +144,7 @@ public static class BrotliHelper
totalRead += bytesRead; totalRead += bytesRead;
// Need larger buffer // Need larger buffer
if (totalRead == outputBuffer.Length) if (totalRead >= outputBuffer.Length - DefaultBufferSize)
{ {
var newBuffer = ArrayPool<byte>.Shared.Rent(outputBuffer.Length * 2); var newBuffer = ArrayPool<byte>.Shared.Rent(outputBuffer.Length * 2);
outputBuffer.AsSpan(0, totalRead).CopyTo(newBuffer); outputBuffer.AsSpan(0, totalRead).CopyTo(newBuffer);
@ -201,6 +156,10 @@ public static class BrotliHelper
return (outputBuffer, totalRead); return (outputBuffer, totalRead);
} }
#endregion
#region Utility
/// <summary> /// <summary>
/// Checks if the data appears to be Brotli compressed. /// Checks if the data appears to be Brotli compressed.
/// </summary> /// </summary>
@ -230,185 +189,5 @@ public static class BrotliHelper
} }
} }
#region Helper Stream Classes
/// <summary>
/// MemoryStream that uses a pre-allocated buffer and can expand using ArrayPool.
/// </summary>
private sealed class PooledMemoryStream : Stream
{
private byte[] _buffer;
private int _position;
private int _length;
private bool _ownsBuffer;
public PooledMemoryStream(byte[] initialBuffer)
{
_buffer = initialBuffer;
_ownsBuffer = false;
}
public override bool CanRead => true;
public override bool CanSeek => true;
public override bool CanWrite => true;
public override long Length => _length;
public override long Position { get => _position; set => _position = (int)value; }
public override void Write(byte[] buffer, int offset, int count)
=> Write(buffer.AsSpan(offset, count));
public override void Write(ReadOnlySpan<byte> buffer)
{
EnsureCapacity(_position + buffer.Length);
buffer.CopyTo(_buffer.AsSpan(_position));
_position += buffer.Length;
if (_position > _length) _length = _position;
}
public override int Read(byte[] buffer, int offset, int count)
{
var bytesToRead = Math.Min(count, _length - _position);
if (bytesToRead <= 0) return 0;
_buffer.AsSpan(_position, bytesToRead).CopyTo(buffer.AsSpan(offset));
_position += bytesToRead;
return bytesToRead;
}
public override long Seek(long offset, SeekOrigin origin)
{
_position = origin switch
{
SeekOrigin.Begin => (int)offset,
SeekOrigin.Current => _position + (int)offset,
SeekOrigin.End => _length + (int)offset,
_ => _position
};
return _position;
}
public override void SetLength(long value) => _length = (int)value;
public override void Flush() { }
public byte[] ToArray()
{
var result = new byte[_length];
_buffer.AsSpan(0, _length).CopyTo(result);
return result;
}
private void EnsureCapacity(int required)
{
if (required <= _buffer.Length) return;
var newSize = Math.Max(_buffer.Length * 2, required);
var newBuffer = ArrayPool<byte>.Shared.Rent(newSize);
_buffer.AsSpan(0, _length).CopyTo(newBuffer);
if (_ownsBuffer) ArrayPool<byte>.Shared.Return(_buffer);
_buffer = newBuffer;
_ownsBuffer = true;
}
protected override void Dispose(bool disposing)
{
if (_ownsBuffer && _buffer != null)
{
ArrayPool<byte>.Shared.Return(_buffer);
_buffer = null!;
}
base.Dispose(disposing);
}
}
/// <summary>
/// Read-only stream wrapper for ReadOnlySpan.
/// </summary>
private sealed class ReadOnlySpanStream : Stream
{
private readonly ReadOnlyMemory<byte> _data;
private int _position;
public ReadOnlySpanStream(ReadOnlySpan<byte> data)
{
_data = data.ToArray(); // Must copy for stream usage
}
public override bool CanRead => true;
public override bool CanSeek => true;
public override bool CanWrite => false;
public override long Length => _data.Length;
public override long Position { get => _position; set => _position = (int)value; }
public override int Read(byte[] buffer, int offset, int count)
{
var bytesToRead = Math.Min(count, _data.Length - _position);
if (bytesToRead <= 0) return 0;
_data.Span.Slice(_position, bytesToRead).CopyTo(buffer.AsSpan(offset));
_position += bytesToRead;
return bytesToRead;
}
public override int Read(Span<byte> buffer)
{
var bytesToRead = Math.Min(buffer.Length, _data.Length - _position);
if (bytesToRead <= 0) return 0;
_data.Span.Slice(_position, bytesToRead).CopyTo(buffer);
_position += bytesToRead;
return bytesToRead;
}
public override long Seek(long offset, SeekOrigin origin)
{
_position = origin switch
{
SeekOrigin.Begin => (int)offset,
SeekOrigin.Current => _position + (int)offset,
SeekOrigin.End => _data.Length + (int)offset,
_ => _position
};
return _position;
}
public override void SetLength(long value) => throw new NotSupportedException();
public override void Write(byte[] buffer, int offset, int count) => throw new NotSupportedException();
public override void Flush() { }
}
/// <summary>
/// Stream that writes directly to an IBufferWriter.
/// </summary>
private sealed class BufferWriterStream : Stream
{
private readonly IBufferWriter<byte> _writer;
public BufferWriterStream(IBufferWriter<byte> writer) => _writer = writer;
public override bool CanRead => false;
public override bool CanSeek => false;
public override bool CanWrite => true;
public override long Length => throw new NotSupportedException();
public override long Position { get => throw new NotSupportedException(); set => throw new NotSupportedException(); }
public override void Write(byte[] buffer, int offset, int count)
{
var span = _writer.GetSpan(count);
buffer.AsSpan(offset, count).CopyTo(span);
_writer.Advance(count);
}
public override void Write(ReadOnlySpan<byte> buffer)
{
var span = _writer.GetSpan(buffer.Length);
buffer.CopyTo(span);
_writer.Advance(buffer.Length);
}
public override int Read(byte[] buffer, int offset, int count) => throw new NotSupportedException();
public override long Seek(long offset, SeekOrigin origin) => throw new NotSupportedException();
public override void SetLength(long value) => throw new NotSupportedException();
public override void Flush() { }
}
#endregion #endregion
} }

11
AyCode.Core/Extensions/SerializeObjectExtensions.cs
View File

@ -5,7 +5,6 @@ using System.Runtime.CompilerServices;
using System.Runtime.Serialization; using System.Runtime.Serialization;
using System.Text; using System.Text;
using AyCode.Core.Interfaces; using AyCode.Core.Interfaces;
using MessagePack;
using Newtonsoft.Json; using Newtonsoft.Json;
using Newtonsoft.Json.Serialization; using Newtonsoft.Json.Serialization;
using static AyCode.Core.Extensions.JsonUtilities; using static AyCode.Core.Extensions.JsonUtilities;
@ -447,16 +446,6 @@ public static class SerializeObjectExtensions
#endregion #endregion
#region MessagePack
public static byte[] ToMessagePack(this object message, MessagePackSerializerOptions options)
=> MessagePackSerializer.Serialize(message, options);
public static T MessagePackTo<T>(this byte[] message, MessagePackSerializerOptions options)
=> MessagePackSerializer.Deserialize<T>(message, options);
#endregion
#region Any (JSON or Binary based on options) #region Any (JSON or Binary based on options)
public static object ToAny<T>(this T source, AcSerializerOptions options) public static object ToAny<T>(this T source, AcSerializerOptions options)

2
AyCode.Services.Server.Tests/SignalRs/SignalRTestHelper.cs
View File

@ -63,7 +63,7 @@ public static class SignalRTestHelper
public static T? GetResponseData<T>(SentMessage sentMessage) public static T? GetResponseData<T>(SentMessage sentMessage)
{ {
if (sentMessage.Message is SignalResponseDataMessage dataResponse && dataResponse.ResponseDataBin != null) if (sentMessage.Message is SignalResponseDataMessage dataResponse && dataResponse.ResponseData != null)
return dataResponse.GetResponseData<T>(); return dataResponse.GetResponseData<T>();
return default; return default;

1
AyCode.Services.Server.Tests/SignalRs/TestableSignalRClient2.cs
View File

@ -4,7 +4,6 @@ using AyCode.Core.Tests.TestModels;
using AyCode.Services.Server.SignalRs; using AyCode.Services.Server.SignalRs;
using AyCode.Services.SignalRs; using AyCode.Services.SignalRs;
using AyCode.Services.Tests.SignalRs; using AyCode.Services.Tests.SignalRs;
using MessagePack.Resolvers;
using Microsoft.AspNetCore.SignalR.Client; using Microsoft.AspNetCore.SignalR.Client;
namespace AyCode.Services.Server.Tests.SignalRs; namespace AyCode.Services.Server.Tests.SignalRs;

1
AyCode.Services.Server.Tests/SignalRs/TestableSignalRHub2.cs
View File

@ -5,7 +5,6 @@ using AyCode.Core.Tests.TestModels;
using AyCode.Models.Server.DynamicMethods; using AyCode.Models.Server.DynamicMethods;
using AyCode.Services.Server.SignalRs; using AyCode.Services.Server.SignalRs;
using AyCode.Services.SignalRs; using AyCode.Services.SignalRs;
using MessagePack.Resolvers;
using Microsoft.Extensions.Configuration; using Microsoft.Extensions.Configuration;
namespace AyCode.Services.Server.Tests.SignalRs; namespace AyCode.Services.Server.Tests.SignalRs;

6
AyCode.Services.Server/SignalRs/AcSignalRDataSource.cs
View File

@ -291,10 +291,10 @@ namespace AyCode.Services.Server.SignalRs
try try
{ {
var response = task.Result; var response = task.Result;
if (response?.Status != SignalResponseStatus.Success || response.ResponseDataBin == null) if (response?.Status != SignalResponseStatus.Success || response.ResponseData == null)
throw new NullReferenceException($"LoadDataSourceAsync; Status: {response?.Status}"); throw new NullReferenceException($"LoadDataSourceAsync; Status: {response?.Status}");
await LoadDataSourceFromResponseData(response.ResponseDataBin, response.DataSerializerType, await LoadDataSourceFromResponseData(response.ResponseData, response.DataSerializerType,
false, false, clearChangeTracking); false, false, clearChangeTracking);
} }
finally finally
@ -960,7 +960,7 @@ namespace AyCode.Services.Server.SignalRs
return SignalRClient.PostDataAsync(messageTag, item, response => return SignalRClient.PostDataAsync(messageTag, item, response =>
{ {
if (response.Status != SignalResponseStatus.Success || response.ResponseDataBin == null) if (response.Status != SignalResponseStatus.Success || response.ResponseData == null)
{ {
if (TryRollbackItem(item.Id, out _)) return; if (TryRollbackItem(item.Id, out _)) return;
throw new NullReferenceException($"SaveItemUnsafeAsync; Status: {response.Status}"); throw new NullReferenceException($"SaveItemUnsafeAsync; Status: {response.Status}");

89
AyCode.Services.Server/SignalRs/AcWebSignalRHubBase.cs
View File

@ -6,8 +6,6 @@ using AyCode.Core.Helpers;
using AyCode.Core.Loggers; using AyCode.Core.Loggers;
using AyCode.Models.Server.DynamicMethods; using AyCode.Models.Server.DynamicMethods;
using AyCode.Services.SignalRs; using AyCode.Services.SignalRs;
using MessagePack;
using MessagePack.Resolvers;
using Microsoft.AspNetCore.SignalR; using Microsoft.AspNetCore.SignalR;
using Microsoft.Extensions.Configuration; using Microsoft.Extensions.Configuration;
@ -98,7 +96,6 @@ public abstract class AcWebSignalRHubBase<TSignalRTags, TLogger>(IConfiguration
/// <summary> /// <summary>
/// Creates a response message using the configured serializer. /// Creates a response message using the configured serializer.
/// Always creates SignalResponseDataMessage which includes the SerializerType.
/// </summary> /// </summary>
protected virtual ISignalRMessage CreateResponseMessage(int messageTag, SignalResponseStatus status, object? responseData) protected virtual ISignalRMessage CreateResponseMessage(int messageTag, SignalResponseStatus status, object? responseData)
{ {
@ -110,11 +107,7 @@ public abstract class AcWebSignalRHubBase<TSignalRTags, TLogger>(IConfiguration
/// </summary> /// </summary>
private static int GetResponseSize(ISignalRMessage responseMessage) private static int GetResponseSize(ISignalRMessage responseMessage)
{ {
return responseMessage switch return responseMessage is SignalResponseDataMessage dataMsg ? dataMsg.ResponseData?.Length ?? 0 : 0;
{
SignalResponseDataMessage dataMsg => dataMsg.ResponseDataBin?.Length ?? 0,
_ => 0
};
} }
/// <summary> /// <summary>
@ -150,68 +143,55 @@ public abstract class AcWebSignalRHubBase<TSignalRTags, TLogger>(IConfiguration
/// <summary> /// <summary>
/// Deserializes parameters from the message based on method signature. /// Deserializes parameters from the message based on method signature.
/// Returns null if no parameters needed, or throws if message is invalid. /// Uses Binary serialization for message wrapper.
/// </summary> /// </summary>
private static object[]? DeserializeParameters(byte[]? message, AcMethodInfoModel<SignalRAttribute> methodInfoModel, string tagName, string methodName) private static object[]? DeserializeParameters(byte[]? message, AcMethodInfoModel<SignalRAttribute> methodInfoModel, string tagName, string methodName)
{ {
if (methodInfoModel.ParamInfos is not { Length: > 0 }) if (methodInfoModel.ParamInfos is not { Length: > 0 })
return null; return null;
// Validate message - required when method has parameters
if (message is null or { Length: 0 }) if (message is null or { Length: 0 })
throw new ArgumentException($"Message is null or empty but method '{methodName}' requires {methodInfoModel.ParamInfos.Length} parameter(s); {tagName}"); throw new ArgumentException($"Message is null or empty but method '{methodName}' requires {methodInfoModel.ParamInfos.Length} parameter(s); {tagName}");
var paramValues = new object[methodInfoModel.ParamInfos.Length]; var paramValues = new object[methodInfoModel.ParamInfos.Length];
var firstParamType = methodInfoModel.ParamInfos[0].ParameterType; var firstParamType = methodInfoModel.ParamInfos[0].ParameterType;
// Use IdMessage format for: multiple params OR primitives/strings/enums/value types // First, try to deserialize as SignalPostJsonMessage to get raw PostDataJson
if (methodInfoModel.ParamInfos.Length > 1 || IsPrimitiveOrStringOrEnum(firstParamType)) var msgBase = SignalRSerializationHelper.DeserializeFromBinary<SignalPostJsonMessage>(message);
if (msgBase?.PostDataJson == null || string.IsNullOrEmpty(msgBase.PostDataJson))
{ {
// Use ContractlessStandardResolver to match client serialization throw new ArgumentException($"Failed to deserialize message for method '{methodName}'; {tagName}");
var msg = message.MessagePackTo<SignalPostJsonDataMessage<IdMessage>>(ContractlessStandardResolver.Options);
for (var i = 0; i < msg.PostData.Ids.Count; i++)
{
var paramType = methodInfoModel.ParamInfos[i].ParameterType;
// Direct JSON deserialization using AcJsonDeserializer (supports primitives)
paramValues[i] = AcJsonDeserializer.Deserialize(msg.PostData.Ids[i], paramType)!;
}
} }
else
{
// Single complex object - try to detect format by checking if it's an IdMessage
var msgJson = message.MessagePackTo<SignalPostJsonDataMessage<object>>(ContractlessStandardResolver.Options);
var json = msgJson.PostDataJson;
// Check if the JSON is an IdMessage format (has "Ids" property) var json = msgBase.PostDataJson;
if (json.Contains("\"Ids\""))
// Check if it's an IdMessage format (contains "Ids" property)
if (json.Contains("\"Ids\""))
{
// Parse as IdMessage - each Id is a JSON string for a parameter
var idMessage = json.JsonTo<IdMessage>();
if (idMessage?.Ids != null && idMessage.Ids.Count > 0)
{ {
// It's IdMessage format - deserialize as IdMessage and get first Id for (var i = 0; i < idMessage.Ids.Count && i < methodInfoModel.ParamInfos.Length; i++)
var idMsg = message.MessagePackTo<SignalPostJsonDataMessage<IdMessage>>(ContractlessStandardResolver.Options);
if (idMsg.PostData.Ids.Count > 0)
{ {
paramValues[0] = AcJsonDeserializer.Deserialize(idMsg.PostData.Ids[0], firstParamType)!; var paramType = methodInfoModel.ParamInfos[i].ParameterType;
return paramValues; paramValues[i] = AcJsonDeserializer.Deserialize(idMessage.Ids[i], paramType)!;
} }
return paramValues;
} }
// Direct complex object format
paramValues[0] = json.JsonTo(firstParamType)!;
} }
// Single complex object - deserialize directly from PostDataJson
paramValues[0] = json.JsonTo(firstParamType)!;
return paramValues; return paramValues;
} }
/// <summary> /// <summary>
/// Determines if a type should use IdMessage format (primitives, strings, enums, value types). /// Determines if a type should use IdMessage format.
/// NOTE: Arrays and collections are NOT included - they use PostDataJson format when sent as single parameter.
/// </summary> /// </summary>
private static bool IsPrimitiveOrStringOrEnum(Type type) private static bool IsPrimitiveOrStringOrEnum(Type type)
{ {
return type == typeof(string) || return type == typeof(string) || type.IsEnum || type.IsValueType || type == typeof(DateTime);
type.IsEnum ||
type.IsValueType ||
type == typeof(DateTime);
} }
#endregion #endregion
@ -243,15 +223,11 @@ public abstract class AcWebSignalRHubBase<TSignalRTags, TLogger>(IConfiguration
=> SendMessageToClient(Clients.All, messageTag, CreateResponseMessage(messageTag, SignalResponseStatus.Success, content), null); => SendMessageToClient(Clients.All, messageTag, CreateResponseMessage(messageTag, SignalResponseStatus.Success, content), null);
/// <summary> /// <summary>
/// Sends message to client. /// Sends message to client using Binary serialization.
/// Both Binary and JSON modes use AcBinarySerializer directly with pooled buffer.
/// </summary> /// </summary>
protected virtual async Task SendMessageToClient(IAcSignalRHubItemServer sendTo, int messageTag, ISignalRMessage message, int? requestId = null) protected virtual async Task SendMessageToClient(IAcSignalRHubItemServer sendTo, int messageTag, ISignalRMessage message, int? requestId = null)
{ {
// Use ArrayBufferWriter for zero-copy serialization to pooled buffer var responseBytes = SignalRSerializationHelper.SerializeToBinary(message);
var writer = new ArrayBufferWriter<byte>(256);
message.ToBinary(writer);
var responseBytes = writer.WrittenSpan.ToArray();
var tagName = ConstHelper.NameByValue<TSignalRTags>(messageTag); var tagName = ConstHelper.NameByValue<TSignalRTags>(messageTag);
@ -264,26 +240,11 @@ public abstract class AcWebSignalRHubBase<TSignalRTags, TLogger>(IConfiguration
#endregion #endregion
#region Context Accessor Methods (virtual for testing) #region Context Accessor Methods
/// <summary>
/// Gets the connection ID. Override in tests to avoid Context dependency.
/// </summary>
protected virtual string GetConnectionId() => Context.ConnectionId; protected virtual string GetConnectionId() => Context.ConnectionId;
/// <summary>
/// Gets whether the connection is aborted. Override in tests to avoid Context dependency.
/// </summary>
protected virtual bool IsConnectionAborted() => Context.ConnectionAborted.IsCancellationRequested; protected virtual bool IsConnectionAborted() => Context.ConnectionAborted.IsCancellationRequested;
/// <summary>
/// Gets the user identifier. Override in tests to avoid Context dependency.
/// </summary>
protected virtual string? GetUserIdentifier() => Context.UserIdentifier; protected virtual string? GetUserIdentifier() => Context.UserIdentifier;
/// <summary>
/// Gets the ClaimsPrincipal user. Override in tests to avoid Context dependency.
/// </summary>
protected virtual ClaimsPrincipal? GetUser() => Context.User; protected virtual ClaimsPrincipal? GetUser() => Context.User;
#endregion #endregion

33
AyCode.Services.Tests/SignalRs/PostJsonDataMessageTests.cs
View File

@ -1,6 +1,5 @@
using AyCode.Core.Extensions; using AyCode.Core.Extensions;
using AyCode.Services.SignalRs; using AyCode.Services.SignalRs;
using MessagePack.Resolvers;
namespace AyCode.Services.Tests.SignalRs; namespace AyCode.Services.Tests.SignalRs;
@ -21,15 +20,15 @@ public class PostJsonDataMessageTests
Console.WriteLine($"PostData.Ids[0]: {idMsg.PostData.Ids[0]}"); Console.WriteLine($"PostData.Ids[0]: {idMsg.PostData.Ids[0]}");
} }
var bytes = message.ToMessagePack(ContractlessStandardResolver.Options); var bytes = message.ToBinary();
Console.WriteLine($"MessagePack bytes: {bytes.Length}"); Console.WriteLine($"Binary bytes: {bytes.Length}");
var deserialized = bytes.MessagePackTo<SignalPostJsonDataMessage<IdMessage>>(ContractlessStandardResolver.Options); var deserialized = bytes.BinaryTo<SignalPostJsonDataMessage<IdMessage>>();
Console.WriteLine($"Deserialized PostDataJson: {deserialized.PostDataJson}"); Console.WriteLine($"Deserialized PostDataJson: {deserialized?.PostDataJson}");
Console.WriteLine($"Deserialized PostData type: {deserialized.PostData?.GetType().Name}"); Console.WriteLine($"Deserialized PostData type: {deserialized?.PostData?.GetType().Name}");
Console.WriteLine($"Deserialized PostData.Ids.Count: {deserialized.PostData?.Ids.Count}"); Console.WriteLine($"Deserialized PostData.Ids.Count: {deserialized?.PostData?.Ids.Count}");
Assert.IsNotNull(deserialized.PostData); Assert.IsNotNull(deserialized?.PostData);
Assert.AreEqual(1, deserialized.PostData.Ids.Count); Assert.AreEqual(1, deserialized.PostData.Ids.Count);
} }
@ -56,18 +55,18 @@ public class PostJsonDataMessageTests
var clientMessage = new SignalPostJsonDataMessage<IdMessage>(idMessage); var clientMessage = new SignalPostJsonDataMessage<IdMessage>(idMessage);
Console.WriteLine($"Client PostDataJson: '{clientMessage.PostDataJson}'"); Console.WriteLine($"Client PostDataJson: '{clientMessage.PostDataJson}'");
Console.WriteLine("\n=== Step 2: MessagePack serialization ==="); Console.WriteLine("\n=== Step 2: Binary serialization ===");
var bytes = clientMessage.ToMessagePack(ContractlessStandardResolver.Options); var bytes = clientMessage.ToBinary();
Console.WriteLine($"MessagePack bytes: {bytes.Length}"); Console.WriteLine($"Binary bytes: {bytes.Length}");
Console.WriteLine("\n=== Step 3: Server deserializes ==="); Console.WriteLine("\n=== Step 3: Server deserializes ===");
var serverMessage = bytes.MessagePackTo<SignalPostJsonDataMessage<IdMessage>>(ContractlessStandardResolver.Options); var serverMessage = bytes.BinaryTo<SignalPostJsonDataMessage<IdMessage>>();
Console.WriteLine($"Server PostDataJson: '{serverMessage.PostDataJson}'"); Console.WriteLine($"Server PostDataJson: '{serverMessage?.PostDataJson}'");
Console.WriteLine($"Server PostData.Ids.Count: {serverMessage.PostData?.Ids.Count}"); Console.WriteLine($"Server PostData.Ids.Count: {serverMessage?.PostData?.Ids.Count}");
Console.WriteLine($"Server PostData.Ids[0]: '{serverMessage.PostData?.Ids[0]}'"); Console.WriteLine($"Server PostData.Ids[0]: '{serverMessage?.PostData?.Ids[0]}'");
Console.WriteLine("\n=== Step 4: Server deserializes parameter ==="); Console.WriteLine("\n=== Step 4: Server deserializes parameter ===");
var paramJson = serverMessage.PostData.Ids[0]; var paramJson = serverMessage!.PostData.Ids[0];
Console.WriteLine($"Parameter JSON: '{paramJson}'"); Console.WriteLine($"Parameter JSON: '{paramJson}'");
var paramValue = AcJsonDeserializer.Deserialize(paramJson, testValue.GetType()); var paramValue = AcJsonDeserializer.Deserialize(paramJson, testValue.GetType());
Console.WriteLine($"Deserialized value: {paramValue}"); Console.WriteLine($"Deserialized value: {paramValue}");
@ -78,7 +77,7 @@ public class PostJsonDataMessageTests
Console.WriteLine("\n=== Step 6: Server creates response ==="); Console.WriteLine("\n=== Step 6: Server creates response ===");
var response = new SignalResponseDataMessage(100, SignalResponseStatus.Success, serviceResult, AcJsonSerializerOptions.Default); var response = new SignalResponseDataMessage(100, SignalResponseStatus.Success, serviceResult, AcJsonSerializerOptions.Default);
Console.WriteLine($"Response created with Binary bytes: {response.ResponseDataBin?.Length ?? 0}"); Console.WriteLine($"Response created with Binary bytes: {response.ResponseData?.Length ?? 0}");
Console.WriteLine("\n=== Step 7: Response Binary ==="); Console.WriteLine("\n=== Step 7: Response Binary ===");
var responseBytes = response.ToBinary(); var responseBytes = response.ToBinary();

231
AyCode.Services.Tests/SignalRs/TestableSignalRClient.cs
View File

@ -1,231 +0,0 @@
using AyCode.Core;
using AyCode.Core.Extensions;
using AyCode.Core.Tests.TestModels;
using AyCode.Services.SignalRs;
using MessagePack.Resolvers;
using Microsoft.AspNetCore.SignalR.Client;
namespace AyCode.Services.Tests.SignalRs;
/// <summary>
/// Testable SignalR client that allows testing without real HubConnection.
/// </summary>
public class TestableSignalRClient : AcSignalRClientBase
{
private HubConnectionState _connectionState = HubConnectionState.Connected;
private int? _nextRequestIdOverride;
/// <summary>
/// Messages sent to the server (captured for assertions).
/// </summary>
public List<SentClientMessage> SentMessages { get; } = [];
/// <summary>
/// Received messages (captured for assertions).
/// </summary>
public List<ReceivedClientMessage> ReceivedMessages { get; } = [];
public TestableSignalRClient(TestLogger logger) : base(logger)
{
}
#region Override virtual methods for testing
protected override HubConnectionState GetConnectionState() => _connectionState;
protected override bool IsConnected() => _connectionState == HubConnectionState.Connected;
protected override Task StartConnectionInternal()
{
_connectionState = HubConnectionState.Connected;
return Task.CompletedTask;
}
protected override Task StopConnectionInternal()
{
_connectionState = HubConnectionState.Disconnected;
return Task.CompletedTask;
}
protected override ValueTask DisposeConnectionInternal() => ValueTask.CompletedTask;
protected override Task SendToHubAsync(int messageTag, byte[]? messageBytes, int? requestId)
{
SentMessages.Add(new SentClientMessage(messageTag, messageBytes, requestId));
return Task.CompletedTask;
}
protected override int GetNextRequestId()
{
if (_nextRequestIdOverride.HasValue)
{
var id = _nextRequestIdOverride.Value;
_nextRequestIdOverride = id + 1; // Auto-increment for subsequent calls
return id;
}
return AcDomain.NextUniqueInt32;
}
protected override Task MessageReceived(int messageTag, byte[] messageBytes)
{
ReceivedMessages.Add(new ReceivedClientMessage(messageTag, messageBytes));
return Task.CompletedTask;
}
#endregion
#region Public test helpers (wrappers for protected methods)
/// <summary>
/// Sets the simulated connection state.
/// </summary>
public void SetConnectionState(HubConnectionState state) => _connectionState = state;
/// <summary>
/// Sets the next request ID for deterministic testing.
/// Will auto-increment for subsequent calls.
/// </summary>
public void SetNextRequestId(int id) => _nextRequestIdOverride = id;
/// <summary>
/// Gets the pending requests dictionary (public wrapper for testing).
/// </summary>
public new System.Collections.Concurrent.ConcurrentDictionary<int, SignalRRequestModel> GetPendingRequests()
=> base.GetPendingRequests();
/// <summary>
/// Registers a pending request (public wrapper for testing).
/// </summary>
public new void RegisterPendingRequest(int requestId, SignalRRequestModel model)
=> base.RegisterPendingRequest(requestId, model);
/// <summary>
/// Clears pending requests (public wrapper for testing).
/// </summary>
public new void ClearPendingRequests() => base.ClearPendingRequests();
/// <summary>
/// Simulates receiving a response from the server.
/// </summary>
public Task SimulateServerResponse(int requestId, int messageTag, SignalResponseStatus status, object? data = null)
{
var response = new SignalResponseDataMessage(messageTag, status, data, AcJsonSerializerOptions.Default);
var bytes = response.ToBinary();
return OnReceiveMessage(messageTag, bytes, requestId);
}
/// <summary>
/// Simulates receiving a success response from the server.
/// </summary>
public Task SimulateSuccessResponse<T>(int requestId, int messageTag, T data)
=> SimulateServerResponse(requestId, messageTag, SignalResponseStatus.Success, data);
/// <summary>
/// Simulates receiving an error response from the server.
/// </summary>
public Task SimulateErrorResponse(int requestId, int messageTag)
=> SimulateServerResponse(requestId, messageTag, SignalResponseStatus.Error);
/// <summary>
/// Gets the last sent message.
/// </summary>
public SentClientMessage? LastSentMessage => SentMessages.LastOrDefault();
/// <summary>
/// Clears all captured messages.
/// </summary>
public void ClearMessages()
{
SentMessages.Clear();
ReceivedMessages.Clear();
}
/// <summary>
/// Invokes OnReceiveMessage directly for testing.
/// </summary>
public Task InvokeOnReceiveMessage(int messageTag, byte[] messageBytes, int? requestId)
=> OnReceiveMessage(messageTag, messageBytes, requestId);
#endregion
}
/// <summary>
/// Represents a message sent from client to server.
/// </summary>
public record SentClientMessage(int MessageTag, byte[]? MessageBytes, int? RequestId)
{
/// <summary>
/// Deserializes the message to IdMessage format.
/// Works with both production SignalPostJsonDataMessage and test SignalRPostMessageDto.
/// </summary>
public IdMessage? AsIdMessage()
{
if (MessageBytes == null) return null;
try
{
// First deserialize to get the PostDataJson string
var msg = MessageBytes.MessagePackTo<SignalPostJsonDataMessage<IdMessage>>(ContractlessStandardResolver.Options);
return msg.PostData;
}
catch
{
// Fallback: try deserializing as raw JSON wrapper
try
{
var rawMsg = MessageBytes.MessagePackTo<SignalPostJsonMessage>(ContractlessStandardResolver.Options);
return rawMsg.PostDataJson?.JsonTo<IdMessage>();
}
catch
{
return null;
}
}
}
/// <summary>
/// Deserializes the message to a specific post data type.
/// </summary>
public T? AsPostData<T>() where T : class
{
if (MessageBytes == null) return null;
try
{
var msg = MessageBytes.MessagePackTo<SignalPostJsonDataMessage<T>>(ContractlessStandardResolver.Options);
return msg.PostData;
}
catch
{
// Fallback: try deserializing as raw JSON wrapper
try
{
var rawMsg = MessageBytes.MessagePackTo<SignalPostJsonMessage>(ContractlessStandardResolver.Options);
return rawMsg.PostDataJson?.JsonTo<T>();
}
catch
{
return null;
}
}
}
}
/// <summary>
/// Represents a message received by the client.
/// </summary>
public record ReceivedClientMessage(int MessageTag, byte[] MessageBytes)
{
/// <summary>
/// Deserializes the message as a response.
/// </summary>
public SignalResponseDataMessage? AsResponse()
{
try
{
return MessageBytes.BinaryTo<SignalResponseDataMessage>();
}
catch
{
return null;
}
}
}

89
AyCode.Services/SignalRs/AcSignalRClientBase.cs
View File

@ -4,7 +4,6 @@ using AyCode.Core.Extensions;
using AyCode.Core.Helpers; using AyCode.Core.Helpers;
using AyCode.Core.Loggers; using AyCode.Core.Loggers;
using AyCode.Interfaces.Entities; using AyCode.Interfaces.Entities;
using MessagePack.Resolvers;
using Microsoft.AspNetCore.Http.Connections; using Microsoft.AspNetCore.Http.Connections;
using Microsoft.AspNetCore.SignalR.Client; using Microsoft.AspNetCore.SignalR.Client;
@ -119,7 +118,7 @@ namespace AyCode.Services.SignalRs
await StartConnection(); await StartConnection();
var msgp = message?.ToMessagePack(ContractlessStandardResolver.Options); var msgBytes = message != null ? SignalRSerializationHelper.SerializeToBinary(message) : null;
if (!IsConnected()) if (!IsConnected())
{ {
@ -127,7 +126,7 @@ namespace AyCode.Services.SignalRs
return; return;
} }
await SendToHubAsync(messageTag, msgp, requestId); await SendToHubAsync(messageTag, msgBytes, requestId);
} }
#region CRUD #region CRUD
@ -144,31 +143,61 @@ namespace AyCode.Services.SignalRs
public virtual Task<TResponseData?> GetByIdAsync<TResponseData>(int messageTag, object[] ids) public virtual Task<TResponseData?> GetByIdAsync<TResponseData>(int messageTag, object[] ids)
=> PostAsync<TResponseData?>(messageTag, ids); => PostAsync<TResponseData?>(messageTag, ids);
/// <summary>
/// Gets data by ID with async callback response. Callback is second parameter.
/// </summary>
public virtual Task GetByIdAsync<TResponseData>(int messageTag, Func<SignalResponseDataMessage, Task> responseCallback, object id)
=> SendMessageToServerAsync(messageTag, new SignalPostJsonDataMessage<IdMessage>(new IdMessage(id)), responseCallback);
/// <summary>
/// Gets data by IDs with async callback response. Callback is second parameter.
/// </summary>
public virtual Task GetByIdAsync<TResponseData>(int messageTag, Func<SignalResponseDataMessage, Task> responseCallback, object[] ids)
=> SendMessageToServerAsync(messageTag, new SignalPostJsonDataMessage<IdMessage>(new IdMessage(ids)), responseCallback);
public virtual Task<TResponseData?> GetAllAsync<TResponseData>(int messageTag) public virtual Task<TResponseData?> GetAllAsync<TResponseData>(int messageTag)
=> SendMessageToServerAsync<TResponseData>(messageTag); => SendMessageToServerAsync<TResponseData>(messageTag);
public virtual Task<TResponseData?> GetAllAsync<TResponseData>(int messageTag, object[]? contextParams) public virtual Task<TResponseData?> GetAllAsync<TResponseData>(int messageTag, object[]? contextParams)
=> SendMessageToServerAsync<TResponseData>(messageTag, contextParams == null || contextParams.Length == 0 ? null : new SignalPostJsonDataMessage<IdMessage>(new IdMessage(contextParams)), GetNextRequestId()); => SendMessageToServerAsync<TResponseData>(messageTag, contextParams == null || contextParams.Length == 0 ? null : new SignalPostJsonDataMessage<IdMessage>(new IdMessage(contextParams)), GetNextRequestId());
/// <summary>
/// Gets all data with async callback response. Callback is second parameter.
/// </summary>
public virtual Task GetAllAsync<TResponseData>(int messageTag, Func<SignalResponseDataMessage, Task> responseCallback)
=> SendMessageToServerAsync(messageTag, null, responseCallback);
/// <summary>
/// Gets all data with context params and async callback response.
/// </summary>
public virtual Task GetAllAsync<TResponseData>(int messageTag, Func<SignalResponseDataMessage, Task> responseCallback, object[]? contextParams)
=> SendMessageToServerAsync(messageTag, contextParams == null || contextParams.Length == 0 ? null : new SignalPostJsonDataMessage<IdMessage>(new IdMessage(contextParams)), responseCallback);
public virtual Task<TPostData?> PostDataAsync<TPostData>(int messageTag, TPostData postData) where TPostData : class public virtual Task<TPostData?> PostDataAsync<TPostData>(int messageTag, TPostData postData) where TPostData : class
=> SendMessageToServerAsync<TPostData>(messageTag, CreatePostMessage(postData), GetNextRequestId()); => SendMessageToServerAsync<TPostData>(messageTag, CreatePostMessage(postData), GetNextRequestId());
public virtual Task<TResponseData?> PostDataAsync<TPostData, TResponseData>(int messageTag, TPostData postData) public virtual Task<TResponseData?> PostDataAsync<TPostData, TResponseData>(int messageTag, TPostData postData)
=> SendMessageToServerAsync<TResponseData>(messageTag, CreatePostMessage(postData), GetNextRequestId()); => SendMessageToServerAsync<TResponseData>(messageTag, CreatePostMessage(postData), GetNextRequestId());
/// <summary>
/// Posts data with async callback response.
/// </summary>
public virtual Task PostDataAsync<TPostData>(int messageTag, TPostData postData, Func<SignalResponseDataMessage, Task> responseCallback)
=> SendMessageToServerAsync(messageTag, CreatePostMessage(postData), responseCallback);
/// <summary>
/// Posts data with typed async callback response.
/// </summary>
public virtual Task PostDataAsync<TPostData, TResponseData>(int messageTag, TPostData postData, Func<SignalResponseDataMessage, Task> responseCallback)
=> SendMessageToServerAsync(messageTag, CreatePostMessage(postData), responseCallback);
/// <summary> /// <summary>
/// Posts data and invokes callback with response. Fire-and-forget friendly for background saves. /// Posts data and invokes callback with response. Fire-and-forget friendly for background saves.
/// </summary> /// </summary>
public virtual Task PostDataAsync<TPostData>(int messageTag, TPostData postData, Action<SignalResponseDataMessage> responseCallback) public virtual Task PostDataAsync<TPostData>(int messageTag, TPostData postData, Action<SignalResponseDataMessage> responseCallback)
{ {
var requestId = GetNextRequestId(); var requestId = GetNextRequestId();
var requestModel = SignalRRequestModelPool.Get(new Action<ISignalResponseMessage>(response => var requestModel = SignalRRequestModelPool.Get(responseCallback);
{
if (response is SignalResponseDataMessage dataMsg)
responseCallback(dataMsg);
else
Logger.Error($"PostDataAsync callback received unexpected message type: {response.GetType().Name}");
}));
_responseByRequestId[requestId] = requestModel; _responseByRequestId[requestId] = requestModel;
return SendMessageToServerAsync(messageTag, CreatePostMessage(postData), requestId); return SendMessageToServerAsync(messageTag, CreatePostMessage(postData), requestId);
@ -206,6 +235,18 @@ namespace AyCode.Services.SignalRs
public virtual Task<TResponse?> SendMessageToServerAsync<TResponse>(int messageTag, ISignalRMessage? message) public virtual Task<TResponse?> SendMessageToServerAsync<TResponse>(int messageTag, ISignalRMessage? message)
=> SendMessageToServerAsync<TResponse>(messageTag, message, GetNextRequestId()); => SendMessageToServerAsync<TResponse>(messageTag, message, GetNextRequestId());
/// <summary>
/// Sends message to server with async callback response.
/// </summary>
public virtual async Task SendMessageToServerAsync(int messageTag, ISignalRMessage? message, Func<SignalResponseDataMessage, Task> responseCallback)
{
var requestId = GetNextRequestId();
var requestModel = SignalRRequestModelPool.Get(responseCallback);
_responseByRequestId[requestId] = requestModel;
await SendMessageToServerAsync(messageTag, message, requestId);
}
protected virtual async Task<TResponse?> SendMessageToServerAsync<TResponse>(int messageTag, ISignalRMessage? message, int requestId) protected virtual async Task<TResponse?> SendMessageToServerAsync<TResponse>(int messageTag, ISignalRMessage? message, int requestId)
{ {
Logger.DebugConditional($"Client SendMessageToServerAsync<TResult>; {nameof(requestId)}: {requestId}; {ConstHelper.NameByValue(TagsName, messageTag)}"); Logger.DebugConditional($"Client SendMessageToServerAsync<TResult>; {nameof(requestId)}: {requestId}; {ConstHelper.NameByValue(TagsName, messageTag)}");
@ -231,6 +272,15 @@ namespace AyCode.Services.SignalRs
return await Task.FromException<TResponse>(new Exception(errorText)); return await Task.FromException<TResponse>(new Exception(errorText));
} }
// Special case: when TResponse is SignalResponseDataMessage, return the message itself
// instead of trying to deserialize ResponseData (which would cause InvalidCastException)
if (typeof(TResponse) == typeof(SignalResponseDataMessage))
{
var serializerType = responseMessage.DataSerializerType == AcSerializerType.Binary ? "Binary" : "JSON";
Logger.Info($"Client returning raw SignalResponseDataMessage ({serializerType}). Total: {(DateTime.UtcNow.Subtract(startTime)).TotalMilliseconds} ms! requestId: {requestId}; tag: {messageTag} [{ConstHelper.NameByValue(TagsName, messageTag)}]");
return (TResponse)(object)responseMessage;
}
var responseData = responseMessage.GetResponseData<TResponse>(); var responseData = responseMessage.GetResponseData<TResponse>();
if (responseData == null && responseMessage.Status == SignalResponseStatus.Success) if (responseData == null && responseMessage.Status == SignalResponseStatus.Success)
@ -239,8 +289,8 @@ namespace AyCode.Services.SignalRs
return default; return default;
} }
var serializerType = responseMessage.DataSerializerType == AcSerializerType.Binary ? "Binary" : "JSON"; var serializerType2 = responseMessage.DataSerializerType == AcSerializerType.Binary ? "Binary" : "JSON";
Logger.Info($"Client deserialized response ({serializerType}). Total: {(DateTime.UtcNow.Subtract(startTime)).TotalMilliseconds} ms! requestId: {requestId}; tag: {messageTag} [{ConstHelper.NameByValue(TagsName, messageTag)}]"); Logger.Info($"Client deserialized response ({serializerType2}). Total: {(DateTime.UtcNow.Subtract(startTime)).TotalMilliseconds} ms! requestId: {requestId}; tag: {messageTag} [{ConstHelper.NameByValue(TagsName, messageTag)}]");
return responseData; return responseData;
} }
@ -273,7 +323,7 @@ namespace AyCode.Services.SignalRs
requestModel.ResponseDateTime = DateTime.UtcNow; requestModel.ResponseDateTime = DateTime.UtcNow;
Logger.Debug($"[{requestModel.ResponseDateTime.Subtract(requestModel.RequestDateTime).TotalMilliseconds:N0}ms][{messageBytes.Length / 1024}kb]{logText}"); Logger.Debug($"[{requestModel.ResponseDateTime.Subtract(requestModel.RequestDateTime).TotalMilliseconds:N0}ms][{messageBytes.Length / 1024}kb]{logText}");
var responseMessage = messageBytes.BinaryTo<SignalResponseDataMessage>() ?? new SignalResponseDataMessage(); var responseMessage = SignalRSerializationHelper.DeserializeFromBinary<SignalResponseDataMessage>(messageBytes) ?? new SignalResponseDataMessage();
switch (requestModel.ResponseByRequestId) switch (requestModel.ResponseByRequestId)
{ {
@ -281,12 +331,17 @@ namespace AyCode.Services.SignalRs
requestModel.ResponseByRequestId = responseMessage; requestModel.ResponseByRequestId = responseMessage;
return Task.CompletedTask; return Task.CompletedTask;
case Action<ISignalResponseMessage> messageCallback: case Action<SignalResponseDataMessage> actionCallback:
if (_responseByRequestId.TryRemove(reqId, out var callbackModel)) if (_responseByRequestId.TryRemove(reqId, out var actionModel))
SignalRRequestModelPool.Return(callbackModel); SignalRRequestModelPool.Return(actionModel);
messageCallback.Invoke(responseMessage); actionCallback.Invoke(responseMessage);
return Task.CompletedTask; return Task.CompletedTask;
case Func<SignalResponseDataMessage, Task> funcCallback:
if (_responseByRequestId.TryRemove(reqId, out var funcModel))
SignalRRequestModelPool.Return(funcModel);
return funcCallback.Invoke(responseMessage);
default: default:
Logger.Error($"Client OnReceiveMessage switch; unknown message type: {requestModel.ResponseByRequestId?.GetType().Name}; {ConstHelper.NameByValue(TagsName, messageTag)}"); Logger.Error($"Client OnReceiveMessage switch; unknown message type: {requestModel.ResponseByRequestId?.GetType().Name}; {ConstHelper.NameByValue(TagsName, messageTag)}");
break; break;

195
AyCode.Services/SignalRs/IAcSignalRHubClient.cs
View File

@ -1,7 +1,5 @@
using AyCode.Core.Extensions; using AyCode.Core.Extensions;
using MessagePack;
using AyCode.Core.Interfaces; using AyCode.Core.Interfaces;
using AyCode.Core.Compression;
using System.Buffers; using System.Buffers;
using System.Runtime.CompilerServices; using System.Runtime.CompilerServices;
using JsonIgnoreAttribute = Newtonsoft.Json.JsonIgnoreAttribute; using JsonIgnoreAttribute = Newtonsoft.Json.JsonIgnoreAttribute;
@ -9,6 +7,10 @@ using STJIgnore = System.Text.Json.Serialization.JsonIgnoreAttribute;
namespace AyCode.Services.SignalRs; namespace AyCode.Services.SignalRs;
/// <summary>
/// Message container for serialized parameter IDs.
/// Optimized for common primitive types to avoid full JSON overhead.
/// </summary>
public class IdMessage public class IdMessage
{ {
public List<string> Ids { get; private set; } public List<string> Ids { get; private set; }
@ -20,30 +22,26 @@ public class IdMessage
/// <summary> /// <summary>
/// Creates IdMessage with multiple parameters serialized directly as JSON. /// Creates IdMessage with multiple parameters serialized directly as JSON.
/// Each parameter is serialized independently without array wrapping.
/// Use object[] explicitly to pass multiple parameters.
/// </summary> /// </summary>
public IdMessage(object[] ids) public IdMessage(object[] ids)
{ {
Ids = new List<string>(ids.Length); Ids = new List<string>(ids.Length);
for (var i = 0; i < ids.Length; i++) for (var i = 0; i < ids.Length; i++)
{ {
Ids.Add(SerializeValue(ids[i])); Ids.Add(SignalRSerializationHelper.SerializePrimitiveToJson(ids[i]));
} }
} }
/// <summary> /// <summary>
/// Creates IdMessage with a single parameter serialized as JSON. /// Creates IdMessage with a single parameter serialized as JSON.
/// Collections (List, Array, etc.) are serialized as a single JSON array.
/// </summary> /// </summary>
public IdMessage(object id) public IdMessage(object id)
{ {
Ids = new List<string>(1) { SerializeValue(id) }; Ids = [SignalRSerializationHelper.SerializePrimitiveToJson(id)];
} }
/// <summary> /// <summary>
/// Creates IdMessage with multiple Guid parameters. /// Creates IdMessage with multiple Guid parameters.
/// Each Guid is serialized as a separate Id entry.
/// </summary> /// </summary>
public IdMessage(IEnumerable<Guid> ids) public IdMessage(IEnumerable<Guid> ids)
{ {
@ -51,63 +49,33 @@ public class IdMessage
Ids = new List<string>(idsArray.Length); Ids = new List<string>(idsArray.Length);
for (var i = 0; i < idsArray.Length; i++) for (var i = 0; i < idsArray.Length; i++)
{ {
Ids.Add(SerializeGuid(idsArray[i])); Ids.Add(SignalRSerializationHelper.SerializeGuidToJson(idsArray[i]));
} }
} }
/// <summary>
/// Optimized serialization for common primitive types to avoid full JSON serialization overhead.
/// Falls back to full JSON serialization for complex types or strings with special characters.
/// </summary>
[MethodImpl(MethodImplOptions.AggressiveInlining)]
private static string SerializeValue(object value)
{
return value switch
{
int i => i.ToString(),
long l => l.ToString(),
Guid g => SerializeGuid(g),
bool b => b ? "true" : "false",
// Strings need proper JSON escaping for special characters
string => value.ToJson(),
_ => value.ToJson()
};
}
[MethodImpl(MethodImplOptions.AggressiveInlining)]
private static string SerializeGuid(Guid g)
{
// Pre-allocate exact size: 38 chars = 2 quotes + 36 guid chars
return string.Create(38, g, static (span, guid) =>
{
span[0] = '"';
guid.TryFormat(span[1..], out _);
span[37] = '"';
});
}
public override string ToString() => string.Join("; ", Ids); public override string ToString() => string.Join("; ", Ids);
} }
[MessagePackObject] /// <summary>
/// Message containing JSON-serialized post data.
/// </summary>
public class SignalPostJsonMessage public class SignalPostJsonMessage
{ {
[Key(0)]
public string PostDataJson { get; set; } = ""; public string PostDataJson { get; set; } = "";
public SignalPostJsonMessage() { } public SignalPostJsonMessage() { }
protected SignalPostJsonMessage(string postDataJson) => PostDataJson = postDataJson; protected SignalPostJsonMessage(string postDataJson) => PostDataJson = postDataJson;
} }
[MessagePackObject(AllowPrivate = false)] /// <summary>
/// Generic message containing JSON-serialized post data with typed access.
/// </summary>
public class SignalPostJsonDataMessage<TPostDataType> : SignalPostJsonMessage, ISignalPostMessage<TPostDataType> public class SignalPostJsonDataMessage<TPostDataType> : SignalPostJsonMessage, ISignalPostMessage<TPostDataType>
{ {
[IgnoreMember]
[JsonIgnore] [JsonIgnore]
[STJIgnore] [STJIgnore]
private TPostDataType? _postData; private TPostDataType? _postData;
[IgnoreMember]
[JsonIgnore] [JsonIgnore]
[STJIgnore] [STJIgnore]
public TPostDataType PostData public TPostDataType PostData
@ -125,10 +93,11 @@ public class SignalPostJsonDataMessage<TPostDataType> : SignalPostJsonMessage, I
public SignalPostJsonDataMessage(string postDataJson) : base(postDataJson) { } public SignalPostJsonDataMessage(string postDataJson) : base(postDataJson) { }
} }
[MessagePackObject] /// <summary>
/// Simple message containing post data.
/// </summary>
public class SignalPostMessage<TPostData>(TPostData postData) : ISignalPostMessage<TPostData> public class SignalPostMessage<TPostData>(TPostData postData) : ISignalPostMessage<TPostData>
{ {
[Key(0)]
public TPostData? PostData { get; set; } = postData; public TPostData? PostData { get; set; } = postData;
} }
@ -137,10 +106,11 @@ public interface ISignalPostMessage<TPostData> : ISignalRMessage
TPostData? PostData { get; } TPostData? PostData { get; }
} }
[MessagePackObject] /// <summary>
/// Message for requesting by Guid ID.
/// </summary>
public class SignalRequestByIdMessage(Guid id) : ISignalRequestMessage<Guid>, IId<Guid> public class SignalRequestByIdMessage(Guid id) : ISignalRequestMessage<Guid>, IId<Guid>
{ {
[Key(0)]
public Guid Id { get; set; } = id; public Guid Id { get; set; } = id;
} }
@ -163,66 +133,6 @@ public enum SignalResponseStatus : byte
Success = 5 Success = 5
} }
/// <summary>
/// Signal response message with lazy deserialization support.
/// Used for callback-based response handling.
/// </summary>
[MessagePackObject(AllowPrivate = false)]
public sealed class SignalResponseMessage<TResponseData> : ISignalResponseMessage
{
[IgnoreMember]
[JsonIgnore]
[STJIgnore]
private TResponseData? _responseData;
[IgnoreMember]
[JsonIgnore]
[STJIgnore]
private bool _isDeserialized;
[Key(0)]
public int MessageTag { get; set; }
[Key(1)]
public SignalResponseStatus Status { get; set; }
[Key(2)]
public string? ResponseDataJson { get; set; }
[IgnoreMember]
[JsonIgnore]
[STJIgnore]
public TResponseData? ResponseData
{
get
{
if (!_isDeserialized)
{
_isDeserialized = true;
_responseData = ResponseDataJson != null ? ResponseDataJson.JsonTo<TResponseData>() : default;
}
return _responseData;
}
set
{
_isDeserialized = true;
_responseData = value;
ResponseDataJson = value?.ToJson();
}
}
public SignalResponseMessage() { }
public SignalResponseMessage(int messageTag, SignalResponseStatus status)
{
MessageTag = messageTag;
Status = status;
}
public SignalResponseMessage(int messageTag, SignalResponseStatus status, TResponseData? responseData) : this(messageTag, status)
=> ResponseData = responseData;
public SignalResponseMessage(int messageTag, SignalResponseStatus status, string? responseDataJson) : this(messageTag, status)
=> ResponseDataJson = responseDataJson;
}
/// <summary> /// <summary>
/// Unified signal response message that supports both JSON and Binary serialization. /// Unified signal response message that supports both JSON and Binary serialization.
/// JSON mode uses Brotli compression for reduced payload size. /// JSON mode uses Brotli compression for reduced payload size.
@ -233,15 +143,13 @@ public sealed class SignalResponseDataMessage : ISignalResponseMessage, IDisposa
public int MessageTag { get; set; } public int MessageTag { get; set; }
public SignalResponseStatus Status { get; set; } public SignalResponseStatus Status { get; set; }
public AcSerializerType DataSerializerType { get; set; } public AcSerializerType DataSerializerType { get; set; }
public byte[]? ResponseDataBin { get; set; } public byte[]? ResponseData { get; set; }
[JsonIgnore] [STJIgnore] private object? _cachedResponseData; [JsonIgnore] [STJIgnore] private object? _cachedResponseData;
[JsonIgnore] [STJIgnore] private byte[]? _rentedDecompressedBuffer; [JsonIgnore] [STJIgnore] private byte[]? _rentedDecompressedBuffer;
[JsonIgnore] [STJIgnore] private int _decompressedLength; [JsonIgnore] [STJIgnore] private int _decompressedLength;
public SignalResponseDataMessage() public SignalResponseDataMessage() { }
{
}
public SignalResponseDataMessage(int messageTag, SignalResponseStatus status) public SignalResponseDataMessage(int messageTag, SignalResponseStatus status)
{ {
@ -253,65 +161,23 @@ public sealed class SignalResponseDataMessage : ISignalResponseMessage, IDisposa
: this(messageTag, status) : this(messageTag, status)
{ {
DataSerializerType = serializerOptions.SerializerType; DataSerializerType = serializerOptions.SerializerType;
if (responseData == null) ResponseData = SignalRSerializationHelper.CreateResponseData(responseData, serializerOptions);
{
ResponseDataBin = null;
return;
}
if (serializerOptions.SerializerType == AcSerializerType.Binary)
{
if (responseData is byte[] byteData)
{
ResponseDataBin = byteData;
return;
}
var binaryOptions = serializerOptions as AcBinarySerializerOptions ?? AcBinarySerializerOptions.Default;
// Use ArrayBufferWriter for zero-copy serialization
var writer = new ArrayBufferWriter<byte>(256);
responseData.ToBinary(writer, binaryOptions);
ResponseDataBin = writer.WrittenSpan.ToArray();
}
else
{
string json;
if (responseData is string strData)
{
var trimmed = strData.AsSpan().Trim();
if (trimmed.Length > 1 && (trimmed[0] == '{' || trimmed[0] == '[') && (trimmed[^1] == '}' || trimmed[^1] == ']'))
json = strData;
else
{
var jsonOptions = serializerOptions as AcJsonSerializerOptions ?? AcJsonSerializerOptions.Default;
json = responseData.ToJson(jsonOptions);
}
}
else
{
var jsonOptions = serializerOptions as AcJsonSerializerOptions ?? AcJsonSerializerOptions.Default;
json = responseData.ToJson(jsonOptions);
}
ResponseDataBin = BrotliHelper.Compress(json);
}
} }
/// <summary> /// <summary>
/// Deserializes the ResponseData to the specified type. /// Deserializes the ResponseData to the specified type.
/// For JSON mode, decompresses Brotli to pooled buffer and deserializes directly (no string allocation).
/// Uses cached result for repeated calls. /// Uses cached result for repeated calls.
/// </summary> /// </summary>
[MethodImpl(MethodImplOptions.AggressiveInlining)] [MethodImpl(MethodImplOptions.AggressiveInlining)]
public T? GetResponseData<T>() public T? GetResponseData<T>()
{ {
if (_cachedResponseData != null) return (T)_cachedResponseData; if (_cachedResponseData != null) return (T)_cachedResponseData;
if (ResponseDataBin == null) return default; if (ResponseData == null) return default;
if (DataSerializerType == AcSerializerType.Binary) if (DataSerializerType == AcSerializerType.Binary)
return (T)(_cachedResponseData = ResponseDataBin.BinaryTo<T>()!); return (T)(_cachedResponseData = ResponseData.BinaryTo<T>()!);
// Decompress Brotli to pooled buffer and deserialize directly from ReadOnlySpan (no string allocation) // Decompress Brotli to pooled buffer and deserialize directly
EnsureDecompressed(); EnsureDecompressed();
var result = AcJsonDeserializer.Deserialize<T>(new ReadOnlySpan<byte>(_rentedDecompressedBuffer, 0, _decompressedLength)); var result = AcJsonDeserializer.Deserialize<T>(new ReadOnlySpan<byte>(_rentedDecompressedBuffer, 0, _decompressedLength));
_cachedResponseData = result; _cachedResponseData = result;
@ -323,7 +189,7 @@ public sealed class SignalResponseDataMessage : ISignalResponseMessage, IDisposa
/// </summary> /// </summary>
public ReadOnlySpan<byte> GetDecompressedJsonSpan() public ReadOnlySpan<byte> GetDecompressedJsonSpan()
{ {
if (ResponseDataBin == null) return ReadOnlySpan<byte>.Empty; if (ResponseData == null) return ReadOnlySpan<byte>.Empty;
if (DataSerializerType == AcSerializerType.Binary) return ReadOnlySpan<byte>.Empty; if (DataSerializerType == AcSerializerType.Binary) return ReadOnlySpan<byte>.Empty;
EnsureDecompressed(); EnsureDecompressed();
@ -335,18 +201,11 @@ public sealed class SignalResponseDataMessage : ISignalResponseMessage, IDisposa
{ {
if (_rentedDecompressedBuffer != null) return; if (_rentedDecompressedBuffer != null) return;
var (buffer, length) = BrotliHelper.DecompressToRentedBuffer(ResponseDataBin.AsSpan()); (_rentedDecompressedBuffer, _decompressedLength) = SignalRSerializationHelper.DecompressToRentedBuffer(ResponseData!);
_rentedDecompressedBuffer = buffer;
_decompressedLength = length;
} }
public void Dispose() public void Dispose()
{ {
if (_rentedDecompressedBuffer != null)
{
ArrayPool<byte>.Shared.Return(_rentedDecompressedBuffer);
_rentedDecompressedBuffer = null;
}
} }
} }

186
AyCode.Services/SignalRs/SignalRSerializationHelper.cs Normal file
View File

@ -0,0 +1,186 @@
using System.Buffers;
using System.Runtime.CompilerServices;
using AyCode.Core.Compression;
using AyCode.Core.Extensions;
namespace AyCode.Services.SignalRs;
/// <summary>
/// Centralized helper for SignalR serialization operations.
/// Provides optimized primitives for JSON/Binary serialization with pooled buffers.
/// </summary>
public static class SignalRSerializationHelper
{
// Pre-boxed boolean values to avoid repeated boxing
private static readonly string JsonTrue = "true";
private static readonly string JsonFalse = "false";
#region Primitive JSON Serialization
/// <summary>
/// Serialize a primitive value to JSON string with minimal overhead.
/// Falls back to full JSON serialization for complex types.
/// </summary>
[MethodImpl(MethodImplOptions.AggressiveInlining)]
public static string SerializePrimitiveToJson(object value)
{
return value switch
{
int i => i.ToString(),
long l => l.ToString(),
Guid g => SerializeGuidToJson(g),
bool b => b ? JsonTrue : JsonFalse,
// Strings need proper JSON escaping for special characters
string => value.ToJson(),
_ => value.ToJson()
};
}
/// <summary>
/// Serialize a Guid to JSON string with pre-allocated buffer.
/// </summary>
[MethodImpl(MethodImplOptions.AggressiveInlining)]
public static string SerializeGuidToJson(Guid g)
{
// Pre-allocate exact size: 38 chars = 2 quotes + 36 guid chars
return string.Create(38, g, static (span, guid) =>
{
span[0] = '"';
guid.TryFormat(span[1..], out _);
span[37] = '"';
});
}
#endregion
#region Binary Serialization
/// <summary>
/// Serialize object to binary using pooled ArrayBufferWriter.
/// </summary>
[MethodImpl(MethodImplOptions.AggressiveInlining)]
public static byte[] SerializeToBinary<T>(T value, AcBinarySerializerOptions? options = null)
{
var writer = new ArrayBufferWriter<byte>(256);
value.ToBinary(writer, options ?? AcBinarySerializerOptions.Default);
return writer.WrittenSpan.ToArray();
}
/// <summary>
/// Serialize object to binary and write to existing ArrayBufferWriter.
/// </summary>
[MethodImpl(MethodImplOptions.AggressiveInlining)]
public static void SerializeToBinary<T>(T value, ArrayBufferWriter<byte> writer, AcBinarySerializerOptions? options = null)
{
value.ToBinary(writer, options ?? AcBinarySerializerOptions.Default);
}
/// <summary>
/// Deserialize binary data to object.
/// </summary>
[MethodImpl(MethodImplOptions.AggressiveInlining)]
public static T? DeserializeFromBinary<T>(byte[] data)
{
return data.BinaryTo<T>();
}
/// <summary>
/// Deserialize binary data from ReadOnlySpan.
/// </summary>
[MethodImpl(MethodImplOptions.AggressiveInlining)]
public static T? DeserializeFromBinary<T>(ReadOnlySpan<byte> data)
{
return data.BinaryTo<T>();
}
#endregion
#region JSON Serialization with Brotli
/// <summary>
/// Serialize object to JSON and compress with Brotli.
/// </summary>
[MethodImpl(MethodImplOptions.AggressiveInlining)]
public static byte[] SerializeToCompressedJson<T>(T value, AcJsonSerializerOptions? options = null)
{
var json = value.ToJson(options ?? AcJsonSerializerOptions.Default);
return BrotliHelper.Compress(json);
}
/// <summary>
/// Decompress Brotli data and deserialize JSON to object.
/// Uses pooled buffer for decompression.
/// </summary>
public static T? DeserializeFromCompressedJson<T>(byte[] compressedData)
{
var (buffer, length) = BrotliHelper.DecompressToRentedBuffer(compressedData.AsSpan());
try
{
return AcJsonDeserializer.Deserialize<T>(new ReadOnlySpan<byte>(buffer, 0, length));
}
finally
{
ArrayPool<byte>.Shared.Return(buffer);
}
}
/// <summary>
/// Decompress Brotli data to rented buffer for direct processing.
/// Caller must return buffer to ArrayPool.
/// </summary>
[MethodImpl(MethodImplOptions.AggressiveInlining)]
public static (byte[] Buffer, int Length) DecompressToRentedBuffer(byte[] compressedData)
{
return BrotliHelper.DecompressToRentedBuffer(compressedData.AsSpan());
}
#endregion
#region Response Data Helpers
/// <summary>
/// Check if string appears to be valid JSON (starts with { or [).
/// </summary>
[MethodImpl(MethodImplOptions.AggressiveInlining)]
public static bool IsValidJsonString(ReadOnlySpan<char> text)
{
var trimmed = text.Trim();
return trimmed.Length > 1 &&
(trimmed[0] == '{' || trimmed[0] == '[') &&
(trimmed[^1] == '}' || trimmed[^1] == ']');
}
/// <summary>
/// Create response binary data based on serializer type.
/// </summary>
public static byte[]? CreateResponseData(object? responseData, AcSerializerOptions serializerOptions)
{
if (responseData == null)
return null;
if (serializerOptions.SerializerType == AcSerializerType.Binary)
{
if (responseData is byte[] byteData)
return byteData;
var binaryOptions = serializerOptions as AcBinarySerializerOptions ?? AcBinarySerializerOptions.Default;
return SerializeToBinary(responseData, binaryOptions);
}
// JSON mode with Brotli compression
string json;
if (responseData is string strData && IsValidJsonString(strData.AsSpan()))
{
json = strData;
}
else
{
var jsonOptions = serializerOptions as AcJsonSerializerOptions ?? AcJsonSerializerOptions.Default;
json = responseData.ToJson(jsonOptions);
}
return BrotliHelper.Compress(json);
}
#endregion
}