tom/hermes-common-library
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Fixed several socket issues. Added backoff for reconnection.

Browse Source
This commit is contained in:
Tom 2024-08-10 19:31:08 +00:00
parent aa9e3dbcd7
commit ab90d47b89
8 changed files with 207 additions and 106 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

38
Abstract/HandlerTypeManager.cs
View File

@ -2,41 +2,55 @@
namespace CommonSocketLibrary.Abstract
{
public abstract class HandlerTypeManager<Client, Handler>
public interface ICodedOperation
{
int OperationCode { get; }
}
public interface IMessageTypeManager
{
Type? GetMessageTypeByCode(int code);
}
public abstract class MessageTypeManager<Handler> : IMessageTypeManager where Handler : ICodedOperation
{
private readonly IDictionary<int, Type> _types;
public IDictionary<int, Type> HandlerTypes { get => _types; }
protected readonly ILogger _logger;
public HandlerTypeManager(ILogger logger, HandlerManager<Client, Handler> handlers)
public MessageTypeManager(IEnumerable<Handler> handlers, ILogger logger)
{
_types = new Dictionary<int, Type>();
_logger = logger;
GenerateHandlerTypes(handlers.Handlers);
GenerateHandlerTypes(handlers);
}
public Type? GetMessageTypeByCode(int code)
{
_types.TryGetValue(code, out Type? type);
return type;
}
private void GenerateHandlerTypes(IDictionary<int, Handler> handlers)
private void GenerateHandlerTypes(IEnumerable<Handler> handlers)
{
foreach (var entry in handlers)
foreach (var handler in handlers)
{
if (entry.Value == null)
if (handler == null)
{
_logger.Error($"Failed to link websocket handler #{entry.Key} due to null value.");
_logger.Error($"Failed to link websocket handler due to null value.");
continue;
}
var type = entry.Value.GetType();
var type = handler.GetType();
var target = FetchMessageType(type);
if (target == null)
{
_logger.Error($"Failed to link websocket handler #{entry.Key} due to no match for {target}.");
_logger.Error($"Failed to link websocket handler #{handler.OperationCode} due to no match for {target}.");
continue;
}
_types.Add(entry.Key, target);
_logger.Debug($"Linked websocket handler #{entry.Key} to type {target.AssemblyQualifiedName}.");
_types.Add(handler.OperationCode, target);
_logger.Debug($"Linked websocket handler #{handler.OperationCode} to type {target.AssemblyQualifiedName}.");
}
}

135
Abstract/SocketClient.cs
View File

@ -1,51 +1,56 @@
using CommonSocketLibrary.Backoff;
using Serilog;
using System.Collections;
using System.Net.WebSockets;
using System.Text;
using System.Text.Json;
namespace CommonSocketLibrary.Abstract
{
public abstract class SocketClient<Message> : IDisposable
public abstract class SocketClient<Message> : IDisposable where Message : class
{
private ClientWebSocket? _socket;
private CancellationTokenSource? _cts;
protected ClientWebSocket? _socket;
protected CancellationTokenSource? _cts;
private readonly int ReceiveBufferSize = 8192;
protected readonly ILogger _logger;
protected readonly JsonSerializerOptions _options;
private bool _disposed;
public bool Connected { get; set; }
public int ReceiveBufferSize { get; } = 8192;
public event EventHandler<EventArgs> OnConnected;
public event EventHandler<SocketDisconnectionEventArgs> OnDisconnected;
public SocketClient(ILogger logger, JsonSerializerOptions options)
{
_logger = logger;
_options = options;
Connected = false;
_disposed = false;
}
public async Task ConnectAsync(string url)
protected async Task ConnectAsync(string url)
{
if (_socket != null)
{
if (_socket.State == WebSocketState.Open) return;
else _socket.Dispose();
else if (!_disposed) _socket.Dispose();
}
_socket = new ClientWebSocket();
_socket.Options.RemoteCertificateValidationCallback = (o, c, ch, er) => true;
_socket.Options.UseDefaultCredentials = false;
_disposed = false;
if (_cts != null) _cts.Dispose();
_cts = new CancellationTokenSource();
await _socket.ConnectAsync(new Uri(url), _cts.Token);
await Task.Factory.StartNew(ReceiveLoop, _cts.Token, TaskCreationOptions.LongRunning, TaskScheduler.Default);
await OnConnection();
OnConnected?.Invoke(this, EventArgs.Empty);
}
public async Task DisconnectAsync()
public async Task DisconnectAsync(SocketDisconnectionEventArgs args)
{
if (_socket == null || _cts == null) return;
// TODO: requests cleanup code, sub-protocol dependent.
if (_disposed || _socket == null || _cts == null)
return;
if (_socket.State == WebSocketState.Open)
{
_cts.CancelAfter(TimeSpan.FromMilliseconds(500));
@ -53,14 +58,20 @@ namespace CommonSocketLibrary.Abstract
await _socket.CloseAsync(WebSocketCloseStatus.NormalClosure, "", CancellationToken.None);
}
Connected = false;
OnDisconnected?.Invoke(this, args);
_socket.Dispose();
_socket = null;
_cts.Dispose();
_cts = null;
}
public void Dispose() => DisconnectAsync().Wait();
public void Dispose()
{
if (_disposed)
return;
_disposed = true;
}
private async Task ReceiveLoop()
{
@ -87,78 +98,84 @@ namespace CommonSocketLibrary.Abstract
await ResponseReceived(outputStream);
}
}
catch (TaskCanceledException) { }
catch (WebSocketException wse)
{
string data = string.Join(string.Empty, wse.Data.Cast<DictionaryEntry>().Select(e => e.Key + "=" + e.Value));
_logger.Error($"Websocket connection problem while receiving data [state: {_socket.State}][code: {wse.ErrorCode}][data: {data}]");
}
catch (TaskCanceledException)
{
_logger.Error($"Socket's receive loop got canceled forcefully [state: {_socket.State}]");
}
finally
{
outputStream?.Dispose();
if (_socket.State.ToString().Contains("Close") || _socket.State == WebSocketState.Aborted)
await DisconnectAsync(new SocketDisconnectionEventArgs(_socket.CloseStatus.ToString()!, _socket.CloseStatusDescription ?? string.Empty));
}
}
public async Task SendRaw(string content)
protected async Task Reconnect(IBackoff backoff, Action reconnect)
{
if (!Connected) return;
var bytes = new byte[1024 * 4];
var array = new ArraySegment<byte>(bytes);
var total = Encoding.UTF8.GetBytes(content).Length;
var current = 0;
while (current < total)
{
var size = Encoding.UTF8.GetBytes(content.Substring(current), array);
await _socket.SendAsync(array, WebSocketMessageType.Text, true, _cts.Token);
current += size;
}
await OnMessageSend(-1, content);
}
public async Task Send<T>(int opcode, T data)
while (true)
{
try
{
var message = GenerateMessage(opcode, data);
var content = JsonSerializer.Serialize(message, _options);
var bytes = Encoding.UTF8.GetBytes(content);
var array = new ArraySegment<byte>(bytes);
var total = bytes.Length;
var current = 0;
while (current < total)
{
var size = Encoding.UTF8.GetBytes(content.Substring(current), array);
await _socket.SendAsync(array, WebSocketMessageType.Text, current + size >= total, _cts.Token);
current += size;
TimeSpan delay = backoff.GetNextDelay();
await Task.Delay(delay);
reconnect.Invoke();
backoff.Reset();
break;
}
await OnMessageSend(opcode, content);
}
catch (Exception e)
catch (Exception)
{
Connected = false;
_logger.Error(e, "Failed to send a message: " + opcode);
_logger.Error("Unable to reconnect to server.");
}
}
}
private async Task ResponseReceived(Stream stream)
{
Message? data = null;
try
{
var data = await JsonSerializer.DeserializeAsync<Message>(stream);
await OnResponseReceived(data);
data = await JsonSerializer.DeserializeAsync<Message>(stream, _options);
}
catch (Exception ex)
{
_logger.Error(ex, "Failed to read or execute a websocket message.");
_logger.Error(ex, "Failed to read a websocket message.");
}
finally
{
stream.Dispose();
}
if (data == null)
{
_logger.Error("Failed to read a websocket message.");
return;
}
try
{
await OnResponseReceived(data);
}
catch (Exception ex)
{
_logger.Error(ex, "Failed to execute a websocket message.");
}
}
protected abstract Message GenerateMessage<T>(int opcode, T data);
protected abstract Task OnResponseReceived(Message? content);
protected abstract Task OnMessageSend(int opcode, string? content);
protected abstract Task OnConnection();
}
public class SocketDisconnectionEventArgs : EventArgs
{
public string Status { get; }
public string Reason { get; }
public SocketDisconnectionEventArgs(string status, string reason)
{
Status = status;
Reason = reason;
}
}
}

32
Backoff/ExponentialBackoff.cs Normal file
View File

@ -0,0 +1,32 @@
namespace CommonSocketLibrary.Backoff
{
public class ExponentialBackoff : IBackoff
{
private int _initial;
private int _current;
private int _maximum;
public ExponentialBackoff(int initial, int maximum)
{
if (maximum < initial)
throw new InvalidOperationException("Initial backoff cannot be larger than maximum backoff.");
_initial = initial;
_maximum = maximum;
Reset();
}
public TimeSpan GetNextDelay()
{
_current = Math.Min(_current * 2, _maximum);
return TimeSpan.FromMilliseconds(_current);
}
public void Reset()
{
_current = _initial / 2;
}
}
}

8
Backoff/IBackoff.cs Normal file
View File

@ -0,0 +1,8 @@
namespace CommonSocketLibrary.Backoff
{
public interface IBackoff
{
TimeSpan GetNextDelay();
void Reset();
}
}

3
Common/IWebSocketHandler.cs
View File

@ -2,9 +2,8 @@ using CommonSocketLibrary.Abstract;
namespace CommonSocketLibrary.Common
{
public interface IWebSocketHandler
public interface IWebSocketHandler : ICodedOperation
{
int OperationCode { get; }
Task Execute<Data>(SocketClient<WebSocketMessage> sender, Data data);
}
}

78
Common/WebSocketClient.cs
View File

@ -1,4 +1,6 @@
using System.Text.Json;
using System.Net.WebSockets;
using System.Text;
using System.Text.Json;
using CommonSocketLibrary.Abstract;
using Serilog;
@ -6,21 +8,22 @@ namespace CommonSocketLibrary.Common
{
public class WebSocketClient : SocketClient<WebSocketMessage>
{
private readonly HandlerManager<WebSocketClient, IWebSocketHandler> _handlerManager;
private readonly HandlerTypeManager<WebSocketClient, IWebSocketHandler> _handlerTypeManager;
protected IDictionary<int, IWebSocketHandler> _handlers;
private readonly MessageTypeManager<IWebSocketHandler> _messageTypeManager;
public WebSocketClient(
ILogger logger,
HandlerManager<WebSocketClient, IWebSocketHandler> handlerManager,
HandlerTypeManager<WebSocketClient, IWebSocketHandler> typeManager,
JsonSerializerOptions serializerOptions
IEnumerable<IWebSocketHandler> handlers,
MessageTypeManager<IWebSocketHandler> typeManager,
JsonSerializerOptions serializerOptions,
ILogger logger
) : base(logger, serializerOptions)
{
_handlerManager = handlerManager;
_handlerTypeManager = typeManager;
_handlers = handlers.ToDictionary(h => h.OperationCode, h => h);
_messageTypeManager = typeManager;
}
protected override WebSocketMessage GenerateMessage<T>(int opcode, T data)
protected WebSocketMessage GenerateMessage<T>(int opcode, T data)
{
return new WebSocketMessage()
{
@ -29,28 +32,61 @@ namespace CommonSocketLibrary.Common
};
}
protected override async Task OnResponseReceived(WebSocketMessage? data)
protected override async Task OnResponseReceived(WebSocketMessage? message)
{
if (data == null)
if (message == null)
return;
string content = data.Data?.ToString() ?? string.Empty;
_logger.Verbose("RX #" + data.OpCode + ": " + content);
string content = message.Data?.ToString() ?? string.Empty;
_logger.Verbose("RX #" + message.OpCode + ": " + content);
if (!_handlerTypeManager.HandlerTypes.TryGetValue(data.OpCode, out Type? type) || type == null)
var type = _messageTypeManager.GetMessageTypeByCode(message.OpCode);
if (type == null)
{
return;
var obj = JsonSerializer.Deserialize(content, type, _options);
await _handlerManager.Execute(this, data.OpCode, obj);
}
protected override async Task OnMessageSend(int opcode, string? content)
var data = JsonSerializer.Deserialize(content, type, _options);
if (!_handlers.TryGetValue(message.OpCode, out IWebSocketHandler? handler) || handler == null)
{
return;
}
await handler.Execute(this, data);
}
public async Task Send<T>(int opcode, T data)
{
if (_socket == null)
return;
try
{
var message = GenerateMessage(opcode, data);
var content = JsonSerializer.Serialize(message, _options);
var bytes = Encoding.UTF8.GetBytes(content);
var array = new ArraySegment<byte>(bytes);
var total = bytes.Length;
var current = 0;
while (current < total)
{
var size = Encoding.UTF8.GetBytes(content.Substring(current), array);
await _socket!.SendAsync(array, WebSocketMessageType.Text, current + size >= total, _cts!.Token);
current += size;
}
_logger.Verbose("TX #" + opcode + ": " + content);
}
protected override async Task OnConnection()
catch (Exception e)
{
if (_socket.State.ToString().Contains("Close") || _socket.State == WebSocketState.Aborted)
{
await DisconnectAsync(new SocketDisconnectionEventArgs(_socket.CloseStatus.ToString()!, _socket.CloseStatusDescription ?? string.Empty));
_logger.Warning($"Socket state on closing = {_socket.State} | {_socket.CloseStatus?.ToString()} | {_socket.CloseStatusDescription}");
}
_logger.Error(e, $"Failed to send a websocket message [op code: {opcode}]");
}
}
}
}

4
Common/WebSocketHandlerTypeManager.cs
View File

@ -4,9 +4,9 @@ using Serilog;
namespace CommonSocketLibrary.Socket.Manager
{
public abstract class WebSocketHandlerTypeManager : HandlerTypeManager<WebSocketClient, IWebSocketHandler>
public abstract class WebSocketMessageTypeManager : MessageTypeManager<IWebSocketHandler>
{
public WebSocketHandlerTypeManager(ILogger logger, HandlerManager<WebSocketClient, IWebSocketHandler> handlers) : base(logger, handlers)
public WebSocketMessageTypeManager(IEnumerable<IWebSocketHandler> handlers, ILogger logger) : base(handlers, logger)
{
}

5
CommonSocketLibrary.csproj
View File

@ -13,13 +13,8 @@
<ItemGroup>
<PackageReference Include="MathParser.org-mXparser" Version="6.0.0" />
<PackageReference Include="Serilog" Version="4.0.0" />
<PackageReference Include="Serilog.Extensions.Logging" Version="8.0.1-dev-10391" />
<PackageReference Include="Serilog.Formatting.Compact" Version="3.0.0" />
<PackageReference Include="Serilog.Settings.Configuration" Version="8.0.1" />
<PackageReference Include="Serilog.Sinks.Async" Version="2.0.0" />
<PackageReference Include="Serilog.Sinks.Console" Version="6.0.0" />
<PackageReference Include="Serilog.Sinks.Debug" Version="3.0.0" />
<PackageReference Include="Serilog.Sinks.Trace" Version="4.0.0" />
</ItemGroup>
</Project>