chylex/Minecraft-Phantom-Panel
1
0
Fork 0
mirror of https://github.com/chylex/Minecraft-Phantom-Panel.git synced 2025-04-30 15:34:06 +02:00
Code

Improve handling of missing registration message

Browse Source
This commit is contained in:
chylex 2023-12-17 18:05:55 +01:00
parent 5bc474d455
commit 2471dc04f1
Signed by: chylex
GPG Key ID: 4DE42C8F19A80548
1 changed files with 52 additions and 41 deletions

93
Utils/Phantom.Utils.Rpc/Runtime/RpcServerRuntime.cs
View File

@ -36,88 +36,90 @@ internal sealed class RpcServerRuntime<TClientListener, TServerListener, TReplyM
var clients = new ConcurrentDictionary<ulong, Client>(); var clients = new ConcurrentDictionary<ulong, Client>();
void OnConnectionClosed(object? sender, RpcClientConnectionClosedEventArgs e) { void OnConnectionClosed(object? sender, RpcClientConnectionClosedEventArgs e) {
if (!clients.Remove(e.RoutingId, out var client)) { if (clients.Remove(e.RoutingId, out var client)) {
return; client.Connection.Closed -= OnConnectionClosed;
} }
RuntimeLogger.Debug("Closing connection to {RoutingId}.", e.RoutingId);
client.Connection.Closed -= OnConnectionClosed;
taskManager.Run("Closing connection to " + e.RoutingId, async () => {
await client.StopReceiving();
await client.StopProcessing();
await client.Connection.StopSending();
RuntimeLogger.Debug("Closed connection to {RoutingId}.", e.RoutingId);
});
} }
while (!cancellationToken.IsCancellationRequested) { while (!cancellationToken.IsCancellationRequested) {
var (routingId, data) = socket.Receive(cancellationToken); var (routingId, data) = socket.Receive(cancellationToken);
if (data.Length == 0) { if (data.Length == 0) {
LogMessageType(routingId, data, messageType: null); LogUnknownMessage(routingId, data);
continue; continue;
} }
Type? messageType = messageDefinitions.ToServer.TryGetType(data, out var type) ? type : null; Type? messageType = messageDefinitions.ToServer.TryGetType(data, out var type) ? type : null;
if (messageType == null) {
LogUnknownMessage(routingId, data);
continue;
}
if (!clients.TryGetValue(routingId, out var client)) { if (!clients.TryGetValue(routingId, out var client)) {
if (!messageDefinitions.IsRegistrationMessage(messageType)) {
RuntimeLogger.Warning("Received {MessageType} ({Bytes} B) from unregistered client {RoutingId}.", messageType.Name, data.Length, routingId);
continue;
}
var clientLoggerName = LoggerName + ":" + routingId; var clientLoggerName = LoggerName + ":" + routingId;
var processingQueue = new RpcQueue(taskManager, "Process messages from " + routingId); var processingQueue = new RpcQueue(taskManager, "Process messages from " + routingId);
var connection = new RpcConnectionToClient<TClientListener>(clientLoggerName, socket, routingId, messageDefinitions.ToClient, ReplyTracker); var connection = new RpcConnectionToClient<TClientListener>(clientLoggerName, socket, routingId, messageDefinitions.ToClient, ReplyTracker);
connection.Closed += OnConnectionClosed; connection.Closed += OnConnectionClosed;
client = new Client(clientLoggerName, connection, processingQueue, messageDefinitions, listenerFactory(connection)); client = new Client(clientLoggerName, connection, processingQueue, messageDefinitions, listenerFactory(connection), taskManager);
clients[routingId] = client; clients[routingId] = client;
client.EnqueueRegistrationMessage(messageType, data);
}
else {
client.Enqueue(messageType, data);
} }
LogMessageType(routingId, data, messageType);
client.Enqueue(messageType, data);
} }
foreach (var client in clients.Values) { foreach (var client in clients.Values) {
client.Connection.Close(); client.Connection.Close();
} }
return Task.CompletedTask; return taskManager.Stop();
}
private void LogUnknownMessage(uint routingId, ReadOnlyMemory<byte> data) {
RuntimeLogger.Warning("Received unknown message ({Bytes} B) from {RoutingId}.", data.Length, routingId);
} }
private protected override Task Disconnect(ServerSocket socket) { private protected override Task Disconnect(ServerSocket socket) {
return Task.CompletedTask; return Task.CompletedTask;
} }
private void LogMessageType(uint routingId, ReadOnlyMemory<byte> data, Type? messageType) {
if (!RuntimeLogger.IsEnabled(LogEventLevel.Verbose)) {
return;
}
if (data.Length > 0 && messageType != null) {
RuntimeLogger.Verbose("Received {MessageType} ({Bytes} B) from {RoutingId}.", messageType.Name, data.Length, routingId);
}
else {
RuntimeLogger.Verbose("Received {Bytes} B message from {RoutingId}.", data.Length, routingId);
}
}
private sealed class Client : MessageHandler<TServerListener> { private sealed class Client : MessageHandler<TServerListener> {
public RpcConnectionToClient<TClientListener> Connection { get; } public RpcConnectionToClient<TClientListener> Connection { get; }
private readonly RpcQueue processingQueue; private readonly RpcQueue processingQueue;
private readonly IMessageDefinitions<TClientListener, TServerListener, TReplyMessage> messageDefinitions; private readonly IMessageDefinitions<TClientListener, TServerListener, TReplyMessage> messageDefinitions;
private readonly TaskManager taskManager;
public Client(string loggerName, RpcConnectionToClient<TClientListener> connection, RpcQueue processingQueue, IMessageDefinitions<TClientListener, TServerListener, TReplyMessage> messageDefinitions, TServerListener listener) : base(loggerName, listener) {
public Client(string loggerName, RpcConnectionToClient<TClientListener> connection, RpcQueue processingQueue, IMessageDefinitions<TClientListener, TServerListener, TReplyMessage> messageDefinitions, TServerListener listener, TaskManager taskManager) : base(loggerName, listener) {
this.Connection = connection; this.Connection = connection;
this.Connection.Closed += OnConnectionClosed;
this.processingQueue = processingQueue; this.processingQueue = processingQueue;
this.messageDefinitions = messageDefinitions; this.messageDefinitions = messageDefinitions;
this.taskManager = taskManager;
} }
internal void Enqueue(Type? messageType, ReadOnlyMemory<byte> data) { internal void EnqueueRegistrationMessage(Type messageType, ReadOnlyMemory<byte> data) {
if (!Connection.GetAuthorization().IsCompleted && messageType != null && messageDefinitions.IsRegistrationMessage(messageType)) { LogMessageType(messageType, data);
processingQueue.Enqueue(() => Handle(data)); processingQueue.Enqueue(() => Handle(data));
} }
else {
processingQueue.Enqueue(() => WaitForAuthorizationAndHandle(data)); internal void Enqueue(Type messageType, ReadOnlyMemory<byte> data) {
LogMessageType(messageType, data);
processingQueue.Enqueue(() => WaitForAuthorizationAndHandle(data));
}
private void LogMessageType(Type messageType, ReadOnlyMemory<byte> data) {
if (Logger.IsEnabled(LogEventLevel.Verbose)) {
Logger.Verbose("Received {MessageType} ({Bytes} B).", messageType.Name, data.Length);
} }
} }
@ -138,8 +140,17 @@ internal sealed class RpcServerRuntime<TClientListener, TServerListener, TReplyM
return Connection.Send(messageDefinitions.CreateReplyMessage(sequenceId, serializedReply)); return Connection.Send(messageDefinitions.CreateReplyMessage(sequenceId, serializedReply));
} }
internal Task StopProcessing() { private void OnConnectionClosed(object? sender, RpcClientConnectionClosedEventArgs e) {
return processingQueue.Stop(); Connection.Closed -= OnConnectionClosed;
Logger.Debug("Closing connection...");
taskManager.Run("Closing connection to " + e.RoutingId, async () => {
await StopReceiving();
await processingQueue.Stop();
await Connection.StopSending();
Logger.Debug("Connection closed.");
});
} }
} }
} }