281 lines
15 KiB
C#
281 lines
15 KiB
C#
using System.Collections.Concurrent;
|
|
using AyCode.Core;
|
|
using AyCode.Core.Extensions;
|
|
using AyCode.Core.Helpers;
|
|
using AyCode.Core.Loggers;
|
|
using AyCode.Interfaces.Entities;
|
|
using AyCode.Services.Loggers;
|
|
using AyCode.Services.SignalRs;
|
|
using MessagePack.Resolvers;
|
|
using Microsoft.AspNetCore.SignalR.Client;
|
|
|
|
namespace AyCode.Blazor.Components.Services
|
|
{
|
|
public abstract class AcSignalRClientBase : IAcSignalRHubClient
|
|
{
|
|
private readonly ConcurrentDictionary<int, SignalRRequestModel> _responseByRequestId = new();
|
|
|
|
protected readonly HubConnection HubConnection;
|
|
protected readonly AcLoggerBase Logger;
|
|
|
|
public event Action<int, byte[], int?> OnMessageReceived = null!;
|
|
//public event Action<int, int> OnMessageRequested;
|
|
|
|
public int Timeout = 10000;
|
|
private const string TagsName = "SignalRTags";
|
|
|
|
protected AcSignalRClientBase(string fullHubName, AcLoggerBase logger)
|
|
{
|
|
Logger = logger;
|
|
|
|
HubConnection = new HubConnectionBuilder()
|
|
.WithUrl(fullHubName)
|
|
//.AddMessagePackProtocol(options => {
|
|
// options.SerializerOptions = MessagePackSerializerOptions.Standard
|
|
// .WithResolver(MessagePack.Resolvers.StandardResolver.Instance)
|
|
// .WithSecurity(MessagePackSecurity.UntrustedData)
|
|
// .WithCompression(MessagePackCompression.Lz4Block)
|
|
// .WithCompressionMinLength(256);})
|
|
.Build();
|
|
|
|
HubConnection.Closed += HubConnection_Closed;
|
|
|
|
_ = HubConnection.On<int, byte[], int?>(nameof(IAcSignalRHubClient.OnReceiveMessage), OnReceiveMessage);
|
|
//_ = HubConnection.On<int, int>(nameof(IAcSignalRHubClient.OnRequestMessage), OnRequestMessage);
|
|
|
|
HubConnection.StartAsync().Forget();
|
|
}
|
|
|
|
private Task HubConnection_Closed(Exception? arg)
|
|
{
|
|
if (_responseByRequestId.IsEmpty) Logger.DebugConditional($"Client HubConnection_Closed");
|
|
else Logger.Warning($"Client HubConnection_Closed; {nameof(_responseByRequestId)} count: {_responseByRequestId.Count}");
|
|
|
|
_responseByRequestId.Clear();
|
|
return Task.CompletedTask;
|
|
}
|
|
|
|
public async Task StartConnection()
|
|
{
|
|
if (HubConnection.State == HubConnectionState.Disconnected)
|
|
await HubConnection.StartAsync();
|
|
|
|
if (HubConnection.State != HubConnectionState.Connected)
|
|
await TaskHelper.WaitToAsync(() => HubConnection.State == HubConnectionState.Connected, Timeout, 10, 25);
|
|
}
|
|
|
|
public async Task StopConnection()
|
|
{
|
|
await HubConnection.StopAsync();
|
|
await HubConnection.DisposeAsync();
|
|
}
|
|
|
|
public virtual Task SendMessageToServerAsync(int messageTag)
|
|
=> SendMessageToServerAsync(messageTag, null, AcDomain.NextUniqueInt32);
|
|
|
|
public virtual Task SendMessageToServerAsync(int messageTag, ISignalRMessage? message, int? requestId)
|
|
{
|
|
Logger.DebugConditional($"Client SendMessageToServerAsync; {nameof(requestId)}: {requestId}; {ConstHelper.NameByValue(TagsName, messageTag)}");
|
|
|
|
return StartConnection().ContinueWith(_ =>
|
|
{
|
|
var msgp = message?.ToMessagePack(ContractlessStandardResolver.Options);
|
|
return HubConnection.SendAsync(nameof(IAcSignalRHubClient.OnReceiveMessage), messageTag, msgp, requestId);
|
|
});
|
|
}
|
|
|
|
#region CRUD
|
|
public virtual Task<TResponseData?> GetByIdAsync<TResponseData>(int messageTag, object id) //where TResponseData : class
|
|
=> SendMessageToServerAsync<TResponseData>(messageTag, new SignalPostJsonDataMessage<IdMessage>(new IdMessage(id)), AcDomain.NextUniqueInt32);
|
|
public virtual Task GetByIdAsync<TResponseData>(int messageTag, Func<ISignalResponseMessage<TResponseData?>, Task> responseCallback, object id)
|
|
=> SendMessageToServerAsync(messageTag, new SignalPostJsonDataMessage<IdMessage>(new IdMessage(id)), responseCallback);
|
|
|
|
public virtual Task<TResponseData?> GetByIdAsync<TResponseData>(int messageTag, object[] ids) //where TResponseData : class
|
|
=> SendMessageToServerAsync<TResponseData>(messageTag, new SignalPostJsonDataMessage<IdMessage>(new IdMessage(ids)), AcDomain.NextUniqueInt32);
|
|
public virtual Task GetByIdAsync<TResponseData>(int messageTag, Func<ISignalResponseMessage<TResponseData?>, Task> responseCallback, object[] ids)
|
|
=> SendMessageToServerAsync(messageTag, new SignalPostJsonDataMessage<IdMessage>(new IdMessage(ids)), responseCallback);
|
|
|
|
public virtual Task<TResponseData?> GetAllAsync<TResponseData>(int messageTag) //where TResponseData : class
|
|
=> SendMessageToServerAsync<TResponseData>(messageTag);
|
|
public virtual Task GetAllAsync<TResponseData>(int messageTag, Func<ISignalResponseMessage<TResponseData?>, Task> responseCallback)
|
|
=> SendMessageToServerAsync(messageTag, null, responseCallback);
|
|
|
|
public virtual Task GetAllAsync<TResponseData>(int messageTag, Func<ISignalResponseMessage<TResponseData?>, Task> responseCallback, object[]? contextParams)
|
|
=> SendMessageToServerAsync(messageTag, (contextParams == null || contextParams.Length == 0 ? null : new SignalPostJsonDataMessage<IdMessage>(new IdMessage(contextParams))), responseCallback);
|
|
public virtual Task<TResponseData?> GetAllAsync<TResponseData>(int messageTag, object[]? contextParams) //where TResponseData : class
|
|
=> SendMessageToServerAsync<TResponseData>(messageTag, contextParams == null || contextParams.Length == 0 ? null : new SignalPostJsonDataMessage<IdMessage>(new IdMessage(contextParams)), AcDomain.NextUniqueInt32);
|
|
|
|
public virtual Task<TPostData?> PostDataAsync<TPostData>(int messageTag, TPostData postData) where TPostData : class
|
|
=> SendMessageToServerAsync<TPostData>(messageTag, new SignalPostJsonDataMessage<TPostData>(postData), AcDomain.NextUniqueInt32);
|
|
public virtual Task<TResponseData?> PostDataAsync<TPostData, TResponseData>(int messageTag, TPostData postData) //where TPostData : class where TResponseData : class
|
|
=> SendMessageToServerAsync<TResponseData>(messageTag, new SignalPostJsonDataMessage<TPostData>(postData), AcDomain.NextUniqueInt32);
|
|
|
|
public virtual Task PostDataAsync<TPostData>(int messageTag, TPostData postData, Func<ISignalResponseMessage<TPostData?>, Task> responseCallback) //where TPostData : class
|
|
=> SendMessageToServerAsync(messageTag, new SignalPostJsonDataMessage<TPostData>(postData), responseCallback);
|
|
public virtual Task PostDataAsync<TPostData, TResponseData>(int messageTag, TPostData postData, Func<ISignalResponseMessage<TResponseData?>, Task> responseCallback) //where TPostData : class where TResponseData : class
|
|
=> SendMessageToServerAsync(messageTag, new SignalPostJsonDataMessage<TPostData>(postData), responseCallback);
|
|
|
|
|
|
public Task GetAllIntoAsync<T>(List<T> intoList, int messageTag, object[]? contextParams = null) where T : IEntityGuid
|
|
{
|
|
return GetAllAsync<List<T>>(messageTag, response =>
|
|
{
|
|
if (response.Status != SignalResponseStatus.Success || response.ResponseData == null)
|
|
{
|
|
Logger.Error($"GetAllIntoAsync<{typeof(T).Name}>(); status: {response.Status}; dataCount: {response.ResponseData?.Count}; {ConstHelper.NameByValue(TagsName, messageTag)};");
|
|
return Task.CompletedTask;
|
|
}
|
|
|
|
intoList.Clear();
|
|
intoList.AddRange(response.ResponseData);
|
|
|
|
return Task.CompletedTask;
|
|
}, contextParams);
|
|
}
|
|
|
|
#endregion CRUD
|
|
|
|
public virtual Task<TResponse?> SendMessageToServerAsync<TResponse>(int messageTag) //where TResponse : class
|
|
=> SendMessageToServerAsync<TResponse>(messageTag, null, AcDomain.NextUniqueInt32);
|
|
|
|
public virtual Task<TResponse?> SendMessageToServerAsync<TResponse>(int messageTag, ISignalRMessage? message) //where TResponse : class
|
|
=> SendMessageToServerAsync<TResponse>(messageTag, message, AcDomain.NextUniqueInt32);
|
|
|
|
protected virtual async Task<TResponse?> SendMessageToServerAsync<TResponse>(int messageTag, ISignalRMessage? message, int requestId) //where TResponse : class
|
|
{
|
|
Logger.DebugConditional($"Client SendMessageToServerAsync<TResult>; {nameof(requestId)}: {requestId}; {ConstHelper.NameByValue(TagsName, messageTag)}");
|
|
|
|
_responseByRequestId[requestId] = new SignalRRequestModel();
|
|
await SendMessageToServerAsync(messageTag, message, requestId);
|
|
|
|
try
|
|
{
|
|
if (await TaskHelper.WaitToAsync(() => _responseByRequestId[requestId].ResponseByRequestId != null, Timeout, 25, 50) &&
|
|
_responseByRequestId.TryRemove(requestId, out var obj) && obj.ResponseByRequestId is ISignalResponseMessage<string> responseMessage)
|
|
{
|
|
if (responseMessage.Status == SignalResponseStatus.Error || responseMessage.ResponseData == null)
|
|
{
|
|
var errorText = $"Client SendMessageToServerAsync<TResponseData> response error; await; tag: {messageTag}; Status: {responseMessage.Status}; requestId: {requestId};";
|
|
|
|
Logger.Error(errorText);
|
|
|
|
//TODO: Ideiglenes, majd a ResponseMessage-et kell visszaadni a Status miatt! - J.
|
|
throw new Exception(errorText);
|
|
//return default;
|
|
}
|
|
|
|
return responseMessage.ResponseData.JsonTo<TResponse>();
|
|
}
|
|
}
|
|
catch (Exception ex)
|
|
{
|
|
Logger.Error($"SendMessageToServerAsync; requestId: {requestId}; {ex.Message}; {ConstHelper.NameByValue(TagsName, messageTag)}", ex);
|
|
}
|
|
|
|
_responseByRequestId.TryRemove(requestId, out _);
|
|
return default;
|
|
}
|
|
|
|
public virtual Task SendMessageToServerAsync<TResponseData>(int messageTag, Func<ISignalResponseMessage<TResponseData?>, Task> responseCallback)
|
|
=> SendMessageToServerAsync(messageTag, null, responseCallback);
|
|
|
|
public virtual Task SendMessageToServerAsync<TResponseData>(int messageTag, ISignalRMessage? message, Func<ISignalResponseMessage<TResponseData?>, Task> responseCallback)
|
|
{
|
|
if (messageTag == 0)
|
|
Logger.Error($"SendMessageToServerAsync; messageTag == 0");
|
|
|
|
var requestId = AcDomain.NextUniqueInt32;
|
|
|
|
_responseByRequestId[requestId] = new SignalRRequestModel(new Action<ISignalResponseMessage<string>>(responseMessage =>
|
|
{
|
|
TResponseData? responseData = default;
|
|
|
|
if (responseMessage.Status == SignalResponseStatus.Success)
|
|
{
|
|
responseData = string.IsNullOrEmpty(responseMessage.ResponseData) ? default : responseMessage.ResponseData.JsonTo<TResponseData?>();
|
|
}
|
|
else Logger.Error($"Client SendMessageToServerAsync<TResponseData> response error; callback; Status: {responseMessage.Status}; requestId: {requestId}; {ConstHelper.NameByValue(TagsName, messageTag)}");
|
|
|
|
responseCallback(new SignalResponseMessage<TResponseData?>(messageTag, responseMessage.Status, responseData));
|
|
}));
|
|
|
|
return SendMessageToServerAsync(messageTag, message, requestId);
|
|
}
|
|
|
|
public virtual Task OnReceiveMessage(int messageTag, byte[] message, int? requestId)
|
|
{
|
|
var logText = $"Client OnReceiveMessage; {nameof(requestId)}: {requestId}; {ConstHelper.NameByValue(TagsName, messageTag)}";
|
|
|
|
if (message.Length == 0) Logger.Warning($"message.Length == 0! {logText}");
|
|
|
|
try
|
|
{
|
|
if (requestId.HasValue && _responseByRequestId.ContainsKey(requestId.Value))
|
|
{
|
|
var reqId = requestId.Value;
|
|
|
|
_responseByRequestId[reqId].ResponseDateTime = DateTime.UtcNow;
|
|
Logger.Info($"[{_responseByRequestId[reqId].ResponseDateTime.Subtract(_responseByRequestId[reqId].RequestDateTime).TotalMilliseconds:N0}ms][{(message.Length/1024)}kb]{logText}");
|
|
|
|
var responseMessage = message.MessagePackTo<SignalResponseJsonMessage>(ContractlessStandardResolver.Options);
|
|
|
|
switch (_responseByRequestId[reqId].ResponseByRequestId)
|
|
{
|
|
case null:
|
|
_responseByRequestId[reqId].ResponseByRequestId = responseMessage;
|
|
return Task.CompletedTask;
|
|
|
|
case Action<ISignalResponseMessage<string>> messagePackCallback:
|
|
_responseByRequestId.TryRemove(reqId, out _);
|
|
|
|
messagePackCallback.Invoke(responseMessage);
|
|
return Task.CompletedTask;
|
|
|
|
//case Action<string> jsonCallback:
|
|
// _responseByRequestId.TryRemove(reqId, out _);
|
|
|
|
// jsonCallback.Invoke(responseMessage);
|
|
// return Task.CompletedTask;
|
|
|
|
default:
|
|
Logger.Error($"Client OnReceiveMessage switch; unknown message type: {_responseByRequestId[reqId].ResponseByRequestId?.GetType().Name}; {ConstHelper.NameByValue(TagsName, messageTag)}");
|
|
break;
|
|
}
|
|
|
|
_responseByRequestId.TryRemove(reqId, out _);
|
|
}
|
|
else Logger.Info(logText);
|
|
|
|
OnMessageReceived(messageTag, message, requestId);
|
|
}
|
|
catch (Exception ex)
|
|
{
|
|
if (requestId.HasValue)
|
|
_responseByRequestId.TryRemove(requestId.Value, out _);
|
|
|
|
Logger.Error($"Client OnReceiveMessage; requestId: {requestId}; {ex.Message}; {ConstHelper.NameByValue(TagsName, messageTag)}", ex);
|
|
throw;
|
|
}
|
|
|
|
return Task.CompletedTask;
|
|
}
|
|
//public virtual Task OnRequestMessage(int messageTag, int requestId)
|
|
//{
|
|
// Logger.DebugConditional($"Client OnRequestMessage; {nameof(messageTag)}: {messageTag}; {nameof(requestId)}: {requestId};");
|
|
|
|
// try
|
|
// {
|
|
// OnMessageRequested(messageTag, requestId);
|
|
// }
|
|
// catch(Exception ex)
|
|
// {
|
|
// Logger.Error($"Client OnReceiveMessage; {nameof(messageTag)}: {messageTag}; {nameof(requestId)}: {requestId}; {ex.Message}", ex);
|
|
// throw;
|
|
// }
|
|
|
|
// return Task.CompletedTask;
|
|
|
|
//}
|
|
}
|
|
}
|