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

Extract classes for client-to-server RPC connection

Browse Source
This commit is contained in:
chylex 2023-10-21 20:45:07 +02:00
parent 55b853d227
commit 2a9bb9e6ac
Signed by: chylex
GPG Key ID: 4DE42C8F19A80548
31 changed files with 378 additions and 301 deletions

25
Agent/Phantom.Agent.Rpc/ControllerConnection.cs Normal file
View File

@ -0,0 +1,25 @@
using Phantom.Common.Logging;
using Phantom.Common.Messages.Agent;
using Phantom.Utils.Rpc;
using Serilog;
namespace Phantom.Agent.Rpc;
public sealed class ControllerConnection {
private static readonly ILogger Logger = PhantomLogger.Create(nameof(ControllerConnection));
private readonly RpcConnectionToServer<IMessageToControllerListener> connection;
public ControllerConnection(RpcConnectionToServer<IMessageToControllerListener> connection) {
this.connection = connection;
Logger.Information("Connection ready.");
}
public Task Send<TMessage>(TMessage message) where TMessage : IMessageToController {
return connection.Send(message);
}
public Task<TReply?> Send<TMessage, TReply>(TMessage message, TimeSpan waitForReplyTime, CancellationToken waitForReplyCancellationToken) where TMessage : IMessageToController<TReply> where TReply : class {
return connection.Send<TMessage, TReply>(message, waitForReplyTime, waitForReplyCancellationToken);
}
}

6
Agent/Phantom.Agent.Rpc/KeepAliveLoop.cs
View File

@ -1,5 +1,7 @@
using Phantom.Common.Logging;
using Phantom.Common.Messages.Agent;
using Phantom.Common.Messages.Agent.ToController;
using Phantom.Utils.Rpc;
using Serilog;
namespace Phantom.Agent.Rpc;
@ -9,10 +11,10 @@ sealed class KeepAliveLoop {
private static readonly TimeSpan KeepAliveInterval = TimeSpan.FromSeconds(10);
private readonly RpcServerConnection connection;
private readonly RpcConnectionToServer<IMessageToControllerListener> connection;
private readonly CancellationTokenSource cancellationTokenSource = new ();
public KeepAliveLoop(RpcServerConnection connection) {
public KeepAliveLoop(RpcConnectionToServer<IMessageToControllerListener> connection) {
this.connection = connection;
Task.Run(Run);
}

42
Agent/Phantom.Agent.Rpc/RpcClientRuntime.cs Normal file
View File

@ -0,0 +1,42 @@
using NetMQ;
using NetMQ.Sockets;
using Phantom.Common.Data.Agent;
using Phantom.Common.Messages.Agent;
using Phantom.Common.Messages.Agent.BiDirectional;
using Phantom.Common.Messages.Agent.ToController;
using Phantom.Utils.Rpc;
using Phantom.Utils.Rpc.Sockets;
using Phantom.Utils.Tasks;
using Serilog;
namespace Phantom.Agent.Rpc;
public sealed class RpcClientRuntime : RpcClientRuntime<IMessageToAgentListener, IMessageToControllerListener, ReplyMessage> {
public static Task Launch(RpcClientSocket<IMessageToAgentListener, IMessageToControllerListener, ReplyMessage> socket, AgentInfo agentInfo, IMessageToAgentListener messageListener, SemaphoreSlim disconnectSemaphore, CancellationToken receiveCancellationToken) {
return new RpcClientRuntime(socket, agentInfo.Guid, messageListener, disconnectSemaphore, receiveCancellationToken).Launch();
}
private readonly Guid agentGuid;
private RpcClientRuntime(RpcClientSocket<IMessageToAgentListener, IMessageToControllerListener, ReplyMessage> socket, Guid agentGuid, IMessageToAgentListener messageListener, SemaphoreSlim disconnectSemaphore, CancellationToken receiveCancellationToken) : base(socket, messageListener, disconnectSemaphore, receiveCancellationToken) {
this.agentGuid = agentGuid;
}
protected override void RunWithConnection(ClientSocket socket, RpcConnectionToServer<IMessageToControllerListener> connection, ILogger logger, TaskManager taskManager) {
var keepAliveLoop = new KeepAliveLoop(connection);
try {
base.RunWithConnection(socket, connection, logger, taskManager);
} finally {
keepAliveLoop.Cancel();
}
}
protected override async Task Disconnect(ClientSocket socket, ILogger logger) {
var unregisterMessageBytes = AgentMessageRegistries.ToController.Write(new UnregisterAgentMessage(agentGuid)).ToArray();
try {
await socket.SendAsync(unregisterMessageBytes).AsTask().WaitAsync(TimeSpan.FromSeconds(5), CancellationToken.None);
} catch (TimeoutException) {
logger.Error("Timed out communicating agent shutdown with the controller.");
}
}
}

107
Agent/Phantom.Agent.Rpc/RpcLauncher.cs
View File

@ -1,107 +0,0 @@
using NetMQ;
using NetMQ.Sockets;
using Phantom.Common.Data.Agent;
using Phantom.Common.Messages.Agent;
using Phantom.Common.Messages.Agent.BiDirectional;
using Phantom.Common.Messages.Agent.ToController;
using Phantom.Utils.Rpc;
using Phantom.Utils.Rpc.Message;
using Phantom.Utils.Tasks;
using Serilog;
using Serilog.Events;
namespace Phantom.Agent.Rpc;
public sealed class RpcLauncher : RpcRuntime<ClientSocket> {
public static Task Launch(RpcConfiguration config, AuthToken authToken, AgentInfo agentInfo, Func<RpcServerConnection, IMessageToAgentListener> listenerFactory, SemaphoreSlim disconnectSemaphore, CancellationToken receiveCancellationToken) {
var socket = new ClientSocket();
var options = socket.Options;
options.CurveServerCertificate = config.ServerCertificate;
options.CurveCertificate = new NetMQCertificate();
options.HelloMessage = AgentMessageRegistries.ToController.Write(new RegisterAgentMessage(authToken, agentInfo)).ToArray();
return new RpcLauncher(config, socket, agentInfo.Guid, listenerFactory, disconnectSemaphore, receiveCancellationToken).Launch();
}
private readonly RpcConfiguration config;
private readonly Guid agentGuid;
private readonly Func<RpcServerConnection, IMessageToAgentListener> messageListenerFactory;
private readonly SemaphoreSlim disconnectSemaphore;
private readonly CancellationToken receiveCancellationToken;
private RpcLauncher(RpcConfiguration config, ClientSocket socket, Guid agentGuid, Func<RpcServerConnection, IMessageToAgentListener> messageListenerFactory, SemaphoreSlim disconnectSemaphore, CancellationToken receiveCancellationToken) : base(config, socket) {
this.config = config;
this.agentGuid = agentGuid;
this.messageListenerFactory = messageListenerFactory;
this.disconnectSemaphore = disconnectSemaphore;
this.receiveCancellationToken = receiveCancellationToken;
}
protected override void Connect(ClientSocket socket) {
var logger = config.RuntimeLogger;
var url = config.TcpUrl;
logger.Information("Starting ZeroMQ client and connecting to {Url}...", url);
socket.Connect(url);
logger.Information("ZeroMQ client ready.");
}
protected override void Run(ClientSocket socket, MessageReplyTracker replyTracker, TaskManager taskManager) {
var connection = new RpcServerConnection(socket, replyTracker);
ServerMessaging.SetCurrentConnection(connection);
var logger = config.RuntimeLogger;
var handler = new MessageToAgentHandler(messageListenerFactory(connection), logger, taskManager, receiveCancellationToken);
var keepAliveLoop = new KeepAliveLoop(connection);
try {
while (!receiveCancellationToken.IsCancellationRequested) {
var data = socket.Receive(receiveCancellationToken);
LogMessageType(logger, data);
if (data.Length > 0) {
AgentMessageRegistries.ToAgent.Handle(data, handler);
}
}
} catch (OperationCanceledException) {
// Ignore.
} finally {
logger.Debug("ZeroMQ client stopped receiving messages.");
disconnectSemaphore.Wait(CancellationToken.None);
keepAliveLoop.Cancel();
}
}
private static void LogMessageType(ILogger logger, ReadOnlyMemory<byte> data) {
if (!logger.IsEnabled(LogEventLevel.Verbose)) {
return;
}
if (data.Length > 0 && AgentMessageRegistries.ToAgent.TryGetType(data, out var type)) {
logger.Verbose("Received {MessageType} ({Bytes} B) from controller.", type.Name, data.Length);
}
else {
logger.Verbose("Received {Bytes} B message from controller.", data.Length);
}
}
protected override async Task Disconnect() {
var unregisterTimeoutTask = Task.Delay(TimeSpan.FromSeconds(5), CancellationToken.None);
var finishedTask = await Task.WhenAny(ServerMessaging.Send(new UnregisterAgentMessage(agentGuid)), unregisterTimeoutTask);
if (finishedTask == unregisterTimeoutTask) {
config.RuntimeLogger.Error("Timed out communicating agent shutdown with the controller.");
}
}
private sealed class MessageToAgentHandler : MessageHandler<IMessageToAgentListener> {
public MessageToAgentHandler(IMessageToAgentListener listener, ILogger logger, TaskManager taskManager, CancellationToken cancellationToken) : base(listener, logger, taskManager, cancellationToken) {}
protected override Task SendReply(uint sequenceId, byte[] serializedReply) {
return ServerMessaging.Send(new ReplyMessage(sequenceId, serializedReply));
}
}
}

41
Agent/Phantom.Agent.Rpc/RpcServerConnection.cs
View File

@ -1,41 +0,0 @@
using NetMQ;
using NetMQ.Sockets;
using Phantom.Common.Messages.Agent;
using Phantom.Common.Messages.Agent.BiDirectional;
using Phantom.Utils.Rpc.Message;
namespace Phantom.Agent.Rpc;
public sealed class RpcServerConnection {
private readonly ClientSocket socket;
private readonly MessageReplyTracker replyTracker;
internal RpcServerConnection(ClientSocket socket, MessageReplyTracker replyTracker) {
this.socket = socket;
this.replyTracker = replyTracker;
}
internal async Task Send<TMessage>(TMessage message) where TMessage : IMessageToController {
var bytes = AgentMessageRegistries.ToController.Write(message).ToArray();
if (bytes.Length > 0) {
await socket.SendAsync(bytes);
}
}
internal async Task<TReply?> Send<TMessage, TReply>(TMessage message, TimeSpan waitForReplyTime, CancellationToken waitForReplyCancellationToken) where TMessage : IMessageToController<TReply> where TReply : class {
var sequenceId = replyTracker.RegisterReply();
var bytes = AgentMessageRegistries.ToController.Write<TMessage, TReply>(sequenceId, message).ToArray();
if (bytes.Length == 0) {
replyTracker.ForgetReply(sequenceId);
return null;
}
await socket.SendAsync(bytes);
return await replyTracker.WaitForReply<TReply>(sequenceId, waitForReplyTime, waitForReplyCancellationToken);
}
public void Receive(ReplyMessage message) {
replyTracker.ReceiveReply(message.SequenceId, message.SerializedReply);
}
}

34
Agent/Phantom.Agent.Rpc/ServerMessaging.cs
View File

@ -1,34 +0,0 @@
using Phantom.Common.Logging;
using Phantom.Common.Messages.Agent;
using Serilog;
namespace Phantom.Agent.Rpc;
public static class ServerMessaging {
private static readonly ILogger Logger = PhantomLogger.Create(nameof(ServerMessaging));
private static RpcServerConnection? CurrentConnection { get; set; }
private static RpcServerConnection CurrentConnectionOrThrow => CurrentConnection ?? throw new InvalidOperationException("Server connection not ready.");
private static readonly object SetCurrentConnectionLock = new ();
internal static void SetCurrentConnection(RpcServerConnection connection) {
lock (SetCurrentConnectionLock) {
if (CurrentConnection != null) {
throw new InvalidOperationException("Server connection can only be set once.");
}
CurrentConnection = connection;
}
Logger.Information("Server connection ready.");
}
public static Task Send<TMessage>(TMessage message) where TMessage : IMessageToController {
return CurrentConnectionOrThrow.Send(message);
}
public static Task<TReply?> Send<TMessage, TReply>(TMessage message, TimeSpan waitForReplyTime, CancellationToken waitForReplyCancellationToken) where TMessage : IMessageToController<TReply> where TReply : class {
return CurrentConnectionOrThrow.Send<TMessage, TReply>(message, waitForReplyTime, waitForReplyCancellationToken);
}
}

5
Agent/Phantom.Agent.Services/AgentServices.cs
View File

@ -1,4 +1,5 @@
using Phantom.Agent.Minecraft.Java;
using Phantom.Agent.Rpc;
using Phantom.Agent.Services.Backups;
using Phantom.Agent.Services.Instances;
using Phantom.Common.Data.Agent;
@ -18,12 +19,12 @@ public sealed class AgentServices {
internal JavaRuntimeRepository JavaRuntimeRepository { get; }
internal InstanceSessionManager InstanceSessionManager { get; }
public AgentServices(AgentInfo agentInfo, AgentFolders agentFolders, AgentServiceConfiguration serviceConfiguration) {
public AgentServices(AgentInfo agentInfo, AgentFolders agentFolders, AgentServiceConfiguration serviceConfiguration, ControllerConnection controllerConnection) {
this.AgentFolders = agentFolders;
this.TaskManager = new TaskManager(PhantomLogger.Create<TaskManager, AgentServices>());
this.BackupManager = new BackupManager(agentFolders, serviceConfiguration.MaxConcurrentCompressionTasks);
this.JavaRuntimeRepository = new JavaRuntimeRepository();
this.InstanceSessionManager = new InstanceSessionManager(agentInfo, agentFolders, JavaRuntimeRepository, TaskManager, BackupManager);
this.InstanceSessionManager = new InstanceSessionManager(controllerConnection, agentInfo, agentFolders, JavaRuntimeRepository, TaskManager, BackupManager);
}
public async Task Initialize() {

7
Agent/Phantom.Agent.Services/Instances/Instance.cs
View File

@ -1,5 +1,4 @@
using Phantom.Agent.Minecraft.Launcher;
using Phantom.Agent.Rpc;
using Phantom.Agent.Services.Instances.Procedures;
using Phantom.Agent.Services.Instances.States;
using Phantom.Common.Data.Instance;
@ -57,7 +56,7 @@ sealed class Instance : IAsyncDisposable {
public void ReportLastStatus() {
TryUpdateStatus("Report last status of instance " + shortName, async () => {
await ServerMessaging.Send(new ReportInstanceStatusMessage(Configuration.InstanceGuid, currentStatus));
await Services.ControllerConnection.Send(new ReportInstanceStatusMessage(Configuration.InstanceGuid, currentStatus));
});
}
@ -65,14 +64,14 @@ sealed class Instance : IAsyncDisposable {
TryUpdateStatus("Report status of instance " + shortName + " as " + status.GetType().Name, async () => {
if (status != currentStatus) {
currentStatus = status;
await ServerMessaging.Send(new ReportInstanceStatusMessage(Configuration.InstanceGuid, status));
await Services.ControllerConnection.Send(new ReportInstanceStatusMessage(Configuration.InstanceGuid, status));
}
});
}
private void ReportEvent(IInstanceEvent instanceEvent) {
var message = new ReportInstanceEventMessage(Guid.NewGuid(), DateTime.UtcNow, Configuration.InstanceGuid, instanceEvent);
Services.TaskManager.Run("Report event for instance " + shortName, async () => await ServerMessaging.Send(message));
Services.TaskManager.Run("Report event for instance " + shortName, async () => await Services.ControllerConnection.Send(message));
}
internal void TransitionState(IInstanceState newState) {

6
Agent/Phantom.Agent.Services/Instances/InstanceLogSender.cs
View File

@ -16,12 +16,14 @@ sealed class InstanceLogSender : CancellableBackgroundTask {
private static readonly TimeSpan SendDelay = TimeSpan.FromMilliseconds(200);
private readonly ControllerConnection controllerConnection;
private readonly Guid instanceGuid;
private readonly Channel<string> outputChannel;
private int droppedLinesSinceLastSend;
public InstanceLogSender(TaskManager taskManager, Guid instanceGuid, string loggerName) : base(PhantomLogger.Create<InstanceLogSender>(loggerName), taskManager, "Instance log sender for " + loggerName) {
public InstanceLogSender(ControllerConnection controllerConnection, TaskManager taskManager, Guid instanceGuid, string loggerName) : base(PhantomLogger.Create<InstanceLogSender>(loggerName), taskManager, "Instance log sender for " + loggerName) {
this.controllerConnection = controllerConnection;
this.instanceGuid = instanceGuid;
this.outputChannel = Channel.CreateBounded<string>(BufferOptions, OnLineDropped);
Start();
@ -61,7 +63,7 @@ sealed class InstanceLogSender : CancellableBackgroundTask {
private async Task SendOutputToServer(ImmutableArray<string> lines) {
if (!lines.IsEmpty) {
await ServerMessaging.Send(new InstanceOutputMessage(instanceGuid, lines));
await controllerConnection.Send(new InstanceOutputMessage(instanceGuid, lines));
}
}

3
Agent/Phantom.Agent.Services/Instances/InstanceServices.cs
View File

@ -1,7 +1,8 @@
using Phantom.Agent.Minecraft.Launcher;
using Phantom.Agent.Rpc;
using Phantom.Agent.Services.Backups;
using Phantom.Utils.Tasks;
namespace Phantom.Agent.Services.Instances;
sealed record InstanceServices(TaskManager TaskManager, PortManager PortManager, BackupManager BackupManager, LaunchServices LaunchServices);
sealed record InstanceServices(ControllerConnection ControllerConnection, TaskManager TaskManager, PortManager PortManager, BackupManager BackupManager, LaunchServices LaunchServices);

8
Agent/Phantom.Agent.Services/Instances/InstanceSessionManager.cs
View File

@ -24,6 +24,7 @@ namespace Phantom.Agent.Services.Instances;
sealed class InstanceSessionManager : IAsyncDisposable {
private static readonly ILogger Logger = PhantomLogger.Create<InstanceSessionManager>();
private readonly ControllerConnection controllerConnection;
private readonly AgentInfo agentInfo;
private readonly string basePath;
@ -36,7 +37,8 @@ sealed class InstanceSessionManager : IAsyncDisposable {
private uint instanceLoggerSequenceId = 0;
public InstanceSessionManager(AgentInfo agentInfo, AgentFolders agentFolders, JavaRuntimeRepository javaRuntimeRepository, TaskManager taskManager, BackupManager backupManager) {
public InstanceSessionManager(ControllerConnection controllerConnection, AgentInfo agentInfo, AgentFolders agentFolders, JavaRuntimeRepository javaRuntimeRepository, TaskManager taskManager, BackupManager backupManager) {
this.controllerConnection = controllerConnection;
this.agentInfo = agentInfo;
this.basePath = agentFolders.InstancesFolderPath;
this.shutdownCancellationToken = shutdownCancellationTokenSource.Token;
@ -45,7 +47,7 @@ sealed class InstanceSessionManager : IAsyncDisposable {
var launchServices = new LaunchServices(minecraftServerExecutables, javaRuntimeRepository);
var portManager = new PortManager(agentInfo.AllowedServerPorts, agentInfo.AllowedRconPorts);
this.instanceServices = new InstanceServices(taskManager, portManager, backupManager, launchServices);
this.instanceServices = new InstanceServices(controllerConnection, taskManager, portManager, backupManager, launchServices);
}
private async Task<InstanceActionResult<T>> AcquireSemaphoreAndRun<T>(Func<Task<InstanceActionResult<T>>> func) {
@ -146,7 +148,7 @@ sealed class InstanceSessionManager : IAsyncDisposable {
var runningInstances = GetRunningInstancesInternal();
var runningInstanceCount = runningInstances.Length;
var runningInstanceMemory = runningInstances.Aggregate(RamAllocationUnits.Zero, static (total, instance) => total + instance.Configuration.MemoryAllocation);
await ServerMessaging.Send(new ReportAgentStatusMessage(runningInstanceCount, runningInstanceMemory));
await controllerConnection.Send(new ReportAgentStatusMessage(runningInstanceCount, runningInstanceMemory));
} finally {
semaphore.Release();
}

2
Agent/Phantom.Agent.Services/Instances/States/InstanceRunningState.cs
View File

@ -27,7 +27,7 @@ sealed class InstanceRunningState : IInstanceState, IDisposable {
this.context = context;
this.Process = process;
this.logSender = new InstanceLogSender(context.Services.TaskManager, configuration.InstanceGuid, context.ShortName);
this.logSender = new InstanceLogSender(context.Services.ControllerConnection, context.Services.TaskManager, configuration.InstanceGuid, context.ShortName);
this.backupScheduler = new BackupScheduler(context.Services.TaskManager, context.Services.BackupManager, process, context, configuration.ServerPort);
this.backupScheduler.BackupCompleted += OnScheduledBackupCompleted;

10
Agent/Phantom.Agent.Services/Rpc/MessageListener.cs
View File

@ -1,11 +1,11 @@
using Phantom.Agent.Rpc;
using Phantom.Common.Data.Instance;
using Phantom.Common.Data.Instance;
using Phantom.Common.Data.Replies;
using Phantom.Common.Logging;
using Phantom.Common.Messages.Agent;
using Phantom.Common.Messages.Agent.BiDirectional;
using Phantom.Common.Messages.Agent.ToAgent;
using Phantom.Common.Messages.Agent.ToController;
using Phantom.Utils.Rpc;
using Phantom.Utils.Rpc.Message;
using Serilog;
@ -14,11 +14,11 @@ namespace Phantom.Agent.Services.Rpc;
public sealed class MessageListener : IMessageToAgentListener {
private static ILogger Logger { get; } = PhantomLogger.Create<MessageListener>();
private readonly RpcServerConnection connection;
private readonly RpcConnectionToServer<IMessageToControllerListener> connection;
private readonly AgentServices agent;
private readonly CancellationTokenSource shutdownTokenSource;
public MessageListener(RpcServerConnection connection, AgentServices agent, CancellationTokenSource shutdownTokenSource) {
public MessageListener(RpcConnectionToServer<IMessageToControllerListener> connection, AgentServices agent, CancellationTokenSource shutdownTokenSource) {
this.connection = connection;
this.agent = agent;
this.shutdownTokenSource = shutdownTokenSource;
@ -40,7 +40,7 @@ public sealed class MessageListener : IMessageToAgentListener {
}
}
await ServerMessaging.Send(new AdvertiseJavaRuntimesMessage(agent.JavaRuntimeRepository.All));
await connection.Send(new AdvertiseJavaRuntimesMessage(agent.JavaRuntimeRepository.All));
await agent.InstanceSessionManager.RefreshAgentStatus();
return NoReply.Instance;

21
Agent/Phantom.Agent/Program.cs
View File

@ -1,11 +1,15 @@
using System.Reflection;
using NetMQ;
using Phantom.Agent;
using Phantom.Agent.Rpc;
using Phantom.Agent.Services;
using Phantom.Agent.Services.Rpc;
using Phantom.Common.Data.Agent;
using Phantom.Common.Logging;
using Phantom.Common.Messages.Agent;
using Phantom.Common.Messages.Agent.ToController;
using Phantom.Utils.Rpc;
using Phantom.Utils.Rpc.Sockets;
using Phantom.Utils.Runtime;
using Phantom.Utils.Tasks;
@ -45,19 +49,18 @@ try {
var (controllerCertificate, agentToken) = agentKey.Value;
var agentInfo = new AgentInfo(agentGuid.Value, agentName, ProtocolVersion, fullVersion, maxInstances, maxMemory, allowedServerPorts, allowedRconPorts);
var agentServices = new AgentServices(agentInfo, folders, new AgentServiceConfiguration(maxConcurrentBackupCompressionTasks));
MessageListener MessageListenerFactory(RpcServerConnection connection) {
return new MessageListener(connection, agentServices, shutdownCancellationTokenSource);
}
PhantomLogger.Root.InformationHeading("Launching Phantom Panel agent...");
var rpcConfiguration = new RpcConfiguration(PhantomLogger.Create("Rpc"), PhantomLogger.Create<TaskManager>("Rpc"), controllerHost, controllerPort, controllerCertificate);
var rpcSocket = RpcClientSocket.Connect(rpcConfiguration, AgentMessageRegistries.Definitions, new RegisterAgentMessage(agentToken, agentInfo));
var agentServices = new AgentServices(agentInfo, folders, new AgentServiceConfiguration(maxConcurrentBackupCompressionTasks), new ControllerConnection(rpcSocket.Connection));
await agentServices.Initialize();
var rpcDisconnectSemaphore = new SemaphoreSlim(0, 1);
var rpcConfiguration = new RpcConfiguration(PhantomLogger.Create("Rpc"), PhantomLogger.Create<TaskManager>("Rpc"), controllerHost, controllerPort, controllerCertificate);
var rpcTask = RpcLauncher.Launch(rpcConfiguration, agentToken, agentInfo, MessageListenerFactory, rpcDisconnectSemaphore, shutdownCancellationToken);
var rpcMessageListener = new MessageListener(rpcSocket.Connection, agentServices, shutdownCancellationTokenSource);
var rpcTask = RpcClientRuntime.Launch(rpcSocket, agentInfo, rpcMessageListener, rpcDisconnectSemaphore, shutdownCancellationToken);
try {
await rpcTask.WaitAsync(shutdownCancellationToken);
} finally {
@ -67,6 +70,8 @@ try {
rpcDisconnectSemaphore.Release();
await rpcTask;
rpcDisconnectSemaphore.Dispose();
NetMQConfig.Cleanup();
}
return 0;

6
Common/Phantom.Common.Messages.Agent/AgentMessageRegistries.cs
View File

@ -34,14 +34,14 @@ public static class AgentMessageRegistries {
}
private sealed class MessageDefinitions : IMessageDefinitions<IMessageToAgentListener, IMessageToControllerListener, ReplyMessage> {
public MessageRegistry<IMessageToAgentListener> Outgoing => ToAgent;
public MessageRegistry<IMessageToControllerListener> Incoming => ToController;
public MessageRegistry<IMessageToAgentListener> ToClient => ToAgent;
public MessageRegistry<IMessageToControllerListener> ToServer => ToController;
public bool IsRegistrationMessage(Type messageType) {
return messageType == typeof(RegisterAgentMessage);
}
public ReplyMessage CreateReplyMessage( uint sequenceId, byte[] serializedReply) {
public ReplyMessage CreateReplyMessage(uint sequenceId, byte[] serializedReply) {
return new ReplyMessage(sequenceId, serializedReply);
}
}

6
Common/Phantom.Common.Messages.Web/WebMessageRegistries.cs
View File

@ -17,14 +17,14 @@ public static class WebMessageRegistries {
}
private sealed class MessageDefinitions : IMessageDefinitions<IMessageToWebListener, IMessageToControllerListener, ReplyMessage> {
public MessageRegistry<IMessageToWebListener> Outgoing => ToWeb;
public MessageRegistry<IMessageToControllerListener> Incoming => ToController;
public MessageRegistry<IMessageToWebListener> ToClient => ToWeb;
public MessageRegistry<IMessageToControllerListener> ToServer => ToController;
public bool IsRegistrationMessage(Type messageType) {
return false;
}
public ReplyMessage CreateReplyMessage( uint sequenceId, byte[] serializedReply) {
public ReplyMessage CreateReplyMessage(uint sequenceId, byte[] serializedReply) {
return new ReplyMessage(sequenceId, serializedReply);
}
}

6
Controller/Phantom.Controller.Rpc/RpcClientConnection.cs → Controller/Phantom.Controller.Rpc/RpcConnectionToClient.cs
View File

@ -4,7 +4,7 @@ using Phantom.Utils.Rpc.Message;
namespace Phantom.Controller.Rpc;
public sealed class RpcClientConnection<TListener> {
public sealed class RpcConnectionToClient<TListener> {
private readonly ServerSocket socket;
private readonly uint routingId;
@ -14,14 +14,14 @@ public sealed class RpcClientConnection<TListener> {
internal event EventHandler<RpcClientConnectionClosedEventArgs>? Closed;
private bool isClosed;
internal RpcClientConnection(ServerSocket socket, uint routingId, MessageRegistry<TListener> messageRegistry, MessageReplyTracker messageReplyTracker) {
internal RpcConnectionToClient(ServerSocket socket, uint routingId, MessageRegistry<TListener> messageRegistry, MessageReplyTracker messageReplyTracker) {
this.socket = socket;
this.routingId = routingId;
this.messageRegistry = messageRegistry;
this.messageReplyTracker = messageReplyTracker;
}
public bool IsSame(RpcClientConnection<TListener> other) {
public bool IsSame(RpcConnectionToClient<TListener> other) {
return this.routingId == other.routingId && this.socket == other.socket;
}

58
Controller/Phantom.Controller.Rpc/RpcRuntime.cs
View File

@ -1,6 +1,7 @@
using NetMQ.Sockets;
using Phantom.Utils.Rpc;
using Phantom.Utils.Rpc.Message;
using Phantom.Utils.Rpc.Sockets;
using Phantom.Utils.Tasks;
using Serilog;
using Serilog.Events;
@ -8,49 +9,28 @@ using Serilog.Events;
namespace Phantom.Controller.Rpc;
public static class RpcRuntime {
public static Task Launch<TOutgoingListener, TIncomingListener, TReplyMessage>(RpcConfiguration config, IMessageDefinitions<TOutgoingListener, TIncomingListener, TReplyMessage> messageDefinitions, Func<RpcClientConnection<TOutgoingListener>, TIncomingListener> listenerFactory, CancellationToken cancellationToken) where TReplyMessage : IMessage<TOutgoingListener, NoReply>, IMessage<TIncomingListener, NoReply> {
return RpcRuntime<TOutgoingListener, TIncomingListener, TReplyMessage>.Launch(config, messageDefinitions, listenerFactory, cancellationToken);
public static Task Launch<TClientListener, TServerListener, TReplyMessage>(RpcConfiguration config, IMessageDefinitions<TClientListener, TServerListener, TReplyMessage> messageDefinitions, Func<RpcConnectionToClient<TClientListener>, TServerListener> listenerFactory, CancellationToken cancellationToken) where TReplyMessage : IMessage<TClientListener, NoReply>, IMessage<TServerListener, NoReply> {
return RpcRuntime<TClientListener, TServerListener, TReplyMessage>.Launch(config, messageDefinitions, listenerFactory, cancellationToken);
}
}
internal sealed class RpcRuntime<TOutgoingListener, TIncomingListener, TReplyMessage> : RpcRuntime<ServerSocket> where TReplyMessage : IMessage<TOutgoingListener, NoReply>, IMessage<TIncomingListener, NoReply> {
internal static Task Launch(RpcConfiguration config, IMessageDefinitions<TOutgoingListener, TIncomingListener, TReplyMessage> messageDefinitions, Func<RpcClientConnection<TOutgoingListener>, TIncomingListener> listenerFactory, CancellationToken cancellationToken) {
return new RpcRuntime<TOutgoingListener, TIncomingListener, TReplyMessage>(config, messageDefinitions, listenerFactory, cancellationToken).Launch();
internal sealed class RpcRuntime<TClientListener, TServerListener, TReplyMessage> : RpcRuntime<ServerSocket> where TReplyMessage : IMessage<TClientListener, NoReply>, IMessage<TServerListener, NoReply> {
internal static Task Launch(RpcConfiguration config, IMessageDefinitions<TClientListener, TServerListener, TReplyMessage> messageDefinitions, Func<RpcConnectionToClient<TClientListener>, TServerListener> listenerFactory, CancellationToken cancellationToken) {
var socket = RpcServerSocket.Connect(config);
return new RpcRuntime<TClientListener, TServerListener, TReplyMessage>(socket, messageDefinitions, listenerFactory, cancellationToken).Launch();
}
private static ServerSocket CreateSocket(RpcConfiguration config) {
var socket = new ServerSocket();
var options = socket.Options;
options.CurveServer = true;
options.CurveCertificate = config.ServerCertificate;
return socket;
}
private readonly RpcConfiguration config;
private readonly IMessageDefinitions<TOutgoingListener, TIncomingListener, TReplyMessage> messageDefinitions;
private readonly Func<RpcClientConnection<TOutgoingListener>, TIncomingListener> listenerFactory;
private readonly IMessageDefinitions<TClientListener, TServerListener, TReplyMessage> messageDefinitions;
private readonly Func<RpcConnectionToClient<TClientListener>, TServerListener> listenerFactory;
private readonly CancellationToken cancellationToken;
private RpcRuntime(RpcConfiguration config, IMessageDefinitions<TOutgoingListener, TIncomingListener, TReplyMessage> messageDefinitions, Func<RpcClientConnection<TOutgoingListener>, TIncomingListener> listenerFactory, CancellationToken cancellationToken) : base(config, CreateSocket(config)) {
this.config = config;
private RpcRuntime(RpcServerSocket socket, IMessageDefinitions<TClientListener, TServerListener, TReplyMessage> messageDefinitions, Func<RpcConnectionToClient<TClientListener>, TServerListener> listenerFactory, CancellationToken cancellationToken) : base(socket) {
this.messageDefinitions = messageDefinitions;
this.listenerFactory = listenerFactory;
this.cancellationToken = cancellationToken;
}
protected override void Connect(ServerSocket socket) {
var logger = config.RuntimeLogger;
var url = config.TcpUrl;
logger.Information("Starting ZeroMQ server on {Url}...", url);
socket.Bind(url);
logger.Information("ZeroMQ server initialized, listening for connections on port {Port}.", config.Port);
}
protected override void Run(ServerSocket socket, MessageReplyTracker replyTracker, TaskManager taskManager) {
var logger = config.RuntimeLogger;
protected override void Run(ServerSocket socket, ILogger logger, MessageReplyTracker replyTracker, TaskManager taskManager) {
var clients = new Dictionary<ulong, Client>();
void OnConnectionClosed(object? sender, RpcClientConnectionClosedEventArgs e) {
@ -71,7 +51,7 @@ internal sealed class RpcRuntime<TOutgoingListener, TIncomingListener, TReplyMes
continue;
}
var connection = new RpcClientConnection<TOutgoingListener>(socket, routingId, messageDefinitions.Outgoing, replyTracker);
var connection = new RpcConnectionToClient<TClientListener>(socket, routingId, messageDefinitions.ToClient, replyTracker);
connection.Closed += OnConnectionClosed;
client = new Client(connection, messageDefinitions, listenerFactory(connection), logger, taskManager, cancellationToken);
@ -79,7 +59,7 @@ internal sealed class RpcRuntime<TOutgoingListener, TIncomingListener, TReplyMes
}
LogMessageType(logger, routingId, data);
messageDefinitions.Incoming.Handle(data, client);
messageDefinitions.ToServer.Handle(data, client);
}
foreach (var client in clients.Values) {
@ -92,7 +72,7 @@ internal sealed class RpcRuntime<TOutgoingListener, TIncomingListener, TReplyMes
return;
}
if (data.Length > 0 && messageDefinitions.Incoming.TryGetType(data, out var type)) {
if (data.Length > 0 && messageDefinitions.ToServer.TryGetType(data, out var type)) {
logger.Verbose("Received {MessageType} ({Bytes} B) from {RoutingId}.", type.Name, data.Length, routingId);
}
else {
@ -101,7 +81,7 @@ internal sealed class RpcRuntime<TOutgoingListener, TIncomingListener, TReplyMes
}
private bool CheckIsRegistrationMessage(ReadOnlyMemory<byte> data, ILogger logger, uint routingId) {
if (messageDefinitions.Incoming.TryGetType(data, out var type) && messageDefinitions.IsRegistrationMessage(type)) {
if (messageDefinitions.ToServer.TryGetType(data, out var type) && messageDefinitions.IsRegistrationMessage(type)) {
return true;
}
@ -109,12 +89,12 @@ internal sealed class RpcRuntime<TOutgoingListener, TIncomingListener, TReplyMes
return false;
}
private sealed class Client : MessageHandler<TIncomingListener> {
public RpcClientConnection<TOutgoingListener> Connection { get; }
private sealed class Client : MessageHandler<TServerListener> {
public RpcConnectionToClient<TClientListener> Connection { get; }
private readonly IMessageDefinitions<TOutgoingListener, TIncomingListener, TReplyMessage> messageDefinitions;
private readonly IMessageDefinitions<TClientListener, TServerListener, TReplyMessage> messageDefinitions;
public Client(RpcClientConnection<TOutgoingListener> connection, IMessageDefinitions<TOutgoingListener, TIncomingListener, TReplyMessage> messageDefinitions, TIncomingListener listener, ILogger logger, TaskManager taskManager, CancellationToken cancellationToken) : base(listener, logger, taskManager, cancellationToken) {
public Client(RpcConnectionToClient<TClientListener> connection, IMessageDefinitions<TClientListener, TServerListener, TReplyMessage> messageDefinitions, TServerListener listener, ILogger logger, TaskManager taskManager, CancellationToken cancellationToken) : base(listener, logger, taskManager, cancellationToken) {
this.Connection = connection;
this.messageDefinitions = messageDefinitions;
}

6
Controller/Phantom.Controller.Services/Agents/AgentConnection.cs
View File

@ -4,13 +4,13 @@ using Phantom.Controller.Rpc;
namespace Phantom.Controller.Services.Agents;
sealed class AgentConnection {
private readonly RpcClientConnection<IMessageToAgentListener> connection;
private readonly RpcConnectionToClient<IMessageToAgentListener> connection;
internal AgentConnection(RpcClientConnection<IMessageToAgentListener> connection) {
internal AgentConnection(RpcConnectionToClient<IMessageToAgentListener> connection) {
this.connection = connection;
}
public bool IsSame(RpcClientConnection<IMessageToAgentListener> connection) {
public bool IsSame(RpcConnectionToClient<IMessageToAgentListener> connection) {
return this.connection.IsSame(connection);
}

4
Controller/Phantom.Controller.Services/Agents/AgentManager.cs
View File

@ -52,7 +52,7 @@ public sealed class AgentManager {
return agents.ByGuid.ToImmutable();
}
internal async Task<bool> RegisterAgent(AuthToken authToken, AgentInfo agentInfo, InstanceManager instanceManager, RpcClientConnection<IMessageToAgentListener> connection) {
internal async Task<bool> RegisterAgent(AuthToken authToken, AgentInfo agentInfo, InstanceManager instanceManager, RpcConnectionToClient<IMessageToAgentListener> connection) {
if (!this.authToken.FixedTimeEquals(authToken)) {
await connection.Send(new RegisterAgentFailureMessage(RegisterAgentFailure.InvalidToken));
return false;
@ -88,7 +88,7 @@ public sealed class AgentManager {
return true;
}
internal bool UnregisterAgent(Guid agentGuid, RpcClientConnection<IMessageToAgentListener> connection) {
internal bool UnregisterAgent(Guid agentGuid, RpcConnectionToClient<IMessageToAgentListener> connection) {
if (agents.ByGuid.TryReplaceIf(agentGuid, static oldAgent => oldAgent.AsOffline(), oldAgent => oldAgent.Connection?.IsSame(connection) == true)) {
Logger.Information("Unregistered agent with GUID {Guid}.", agentGuid);
return true;

4
Controller/Phantom.Controller.Services/ControllerServices.cs
View File

@ -53,11 +53,11 @@ public sealed class ControllerServices {
this.cancellationToken = shutdownCancellationToken;
}
public AgentMessageListener CreateAgentMessageListener(RpcClientConnection<IMessageToAgentListener> connection) {
public AgentMessageListener CreateAgentMessageListener(RpcConnectionToClient<IMessageToAgentListener> connection) {
return new AgentMessageListener(connection, AgentManager, AgentJavaRuntimesManager, InstanceManager, InstanceLogManager, EventLog, cancellationToken);
}
public WebMessageListener CreateWebMessageListener(RpcClientConnection<IMessageToWebListener> connection) {
public WebMessageListener CreateWebMessageListener(RpcConnectionToClient<IMessageToWebListener> connection) {
return new WebMessageListener(connection);
}

4
Controller/Phantom.Controller.Services/Rpc/AgentMessageListener.cs
View File

@ -14,7 +14,7 @@ using Phantom.Utils.Tasks;
namespace Phantom.Controller.Services.Rpc;
public sealed class AgentMessageListener : IMessageToControllerListener {
private readonly RpcClientConnection<IMessageToAgentListener> connection;
private readonly RpcConnectionToClient<IMessageToAgentListener> connection;
private readonly AgentManager agentManager;
private readonly AgentJavaRuntimesManager agentJavaRuntimesManager;
private readonly InstanceManager instanceManager;
@ -24,7 +24,7 @@ public sealed class AgentMessageListener : IMessageToControllerListener {
private readonly TaskCompletionSource<Guid> agentGuidWaiter = AsyncTasks.CreateCompletionSource<Guid>();
internal AgentMessageListener(RpcClientConnection<IMessageToAgentListener> connection, AgentManager agentManager, AgentJavaRuntimesManager agentJavaRuntimesManager, InstanceManager instanceManager, InstanceLogManager instanceLogManager, EventLog eventLog, CancellationToken cancellationToken) {
internal AgentMessageListener(RpcConnectionToClient<IMessageToAgentListener> connection, AgentManager agentManager, AgentJavaRuntimesManager agentJavaRuntimesManager, InstanceManager instanceManager, InstanceLogManager instanceLogManager, EventLog eventLog, CancellationToken cancellationToken) {
this.connection = connection;
this.agentManager = agentManager;
this.agentJavaRuntimesManager = agentJavaRuntimesManager;

4
Controller/Phantom.Controller.Services/Rpc/WebMessageListener.cs
View File

@ -6,9 +6,9 @@ using Phantom.Utils.Rpc.Message;
namespace Phantom.Controller.Services.Rpc;
public sealed class WebMessageListener : IMessageToControllerListener {
private readonly RpcClientConnection<IMessageToWebListener> connection;
private readonly RpcConnectionToClient<IMessageToWebListener> connection;
internal WebMessageListener(RpcClientConnection<IMessageToWebListener> connection) {
internal WebMessageListener(RpcConnectionToClient<IMessageToWebListener> connection) {
this.connection = connection;
}

13
Controller/Phantom.Controller/Program.cs
View File

@ -1,4 +1,5 @@
using System.Reflection;
using NetMQ;
using Phantom.Common.Logging;
using Phantom.Common.Messages.Agent;
using Phantom.Common.Messages.Web;
@ -61,10 +62,14 @@ try {
return new RpcConfiguration(PhantomLogger.Create("Rpc", serviceName), PhantomLogger.Create<TaskManager>("Rpc", serviceName), host, port, connectionKey.Certificate);
}
await Task.WhenAll(
RpcRuntime.Launch(ConfigureRpc("Agent", agentRpcServerHost, agentRpcServerPort, agentKeyData), AgentMessageRegistries.Definitions, controllerServices.CreateAgentMessageListener, shutdownCancellationToken),
RpcRuntime.Launch(ConfigureRpc("Web", webRpcServerHost, webRpcServerPort, webKeyData), WebMessageRegistries.Definitions, controllerServices.CreateWebMessageListener, shutdownCancellationToken)
);
try {
await Task.WhenAll(
RpcRuntime.Launch(ConfigureRpc("Agent", agentRpcServerHost, agentRpcServerPort, agentKeyData), AgentMessageRegistries.Definitions, controllerServices.CreateAgentMessageListener, shutdownCancellationToken),
RpcRuntime.Launch(ConfigureRpc("Web", webRpcServerHost, webRpcServerPort, webKeyData), WebMessageRegistries.Definitions, controllerServices.CreateWebMessageListener, shutdownCancellationToken)
);
} finally {
NetMQConfig.Cleanup();
}
return 0;
} catch (OperationCanceledException) {

6
Utils/Phantom.Utils.Rpc/Message/IMessageDefinitions.cs
View File

@ -1,8 +1,8 @@
namespace Phantom.Utils.Rpc.Message;
public interface IMessageDefinitions<TOutgoingListener, TIncomingListener, TReplyMessage> where TReplyMessage : IMessage<TOutgoingListener, NoReply>, IMessage<TIncomingListener, NoReply> {
MessageRegistry<TOutgoingListener> Outgoing { get; }
MessageRegistry<TIncomingListener> Incoming { get; }
public interface IMessageDefinitions<TClientListener, TServerListener, TReplyMessage> where TReplyMessage : IMessage<TClientListener, NoReply>, IMessage<TServerListener, NoReply> {
MessageRegistry<TClientListener> ToClient { get; }
MessageRegistry<TServerListener> ToServer { get; }
bool IsRegistrationMessage(Type messageType);
TReplyMessage CreateReplyMessage(uint sequenceId, byte[] serializedReply);

77
Utils/Phantom.Utils.Rpc/RpcClientRuntime.cs Normal file
View File

@ -0,0 +1,77 @@
using NetMQ.Sockets;
using Phantom.Utils.Rpc.Message;
using Phantom.Utils.Rpc.Sockets;
using Phantom.Utils.Tasks;
using Serilog;
using Serilog.Events;
namespace Phantom.Utils.Rpc;
public abstract class RpcClientRuntime<TClientListener, TServerListener, TReplyMessage> : RpcRuntime<ClientSocket> where TReplyMessage : IMessage<TClientListener, NoReply>, IMessage<TServerListener, NoReply> {
private readonly RpcConnectionToServer<TServerListener> connection;
private readonly IMessageDefinitions<TClientListener, TServerListener, TReplyMessage> messageDefinitions;
private readonly TClientListener messageListener;
private readonly SemaphoreSlim disconnectSemaphore;
private readonly CancellationToken receiveCancellationToken;
protected RpcClientRuntime(RpcClientSocket<TClientListener, TServerListener, TReplyMessage> socket, TClientListener messageListener, SemaphoreSlim disconnectSemaphore, CancellationToken receiveCancellationToken) : base(socket) {
this.connection = socket.Connection;
this.messageDefinitions = socket.MessageDefinitions;
this.messageListener = messageListener;
this.disconnectSemaphore = disconnectSemaphore;
this.receiveCancellationToken = receiveCancellationToken;
}
protected sealed override void Run(ClientSocket socket, ILogger logger, MessageReplyTracker replyTracker, TaskManager taskManager) {
RunWithConnection(socket, connection, logger, taskManager);
}
protected virtual void RunWithConnection(ClientSocket socket, RpcConnectionToServer<TServerListener> connection, ILogger logger, TaskManager taskManager) {
var handler = new Handler(connection, messageDefinitions, messageListener, logger, taskManager, receiveCancellationToken);
try {
while (!receiveCancellationToken.IsCancellationRequested) {
var data = socket.Receive(receiveCancellationToken);
LogMessageType(logger, data);
if (data.Length > 0) {
messageDefinitions.ToClient.Handle(data, handler);
}
}
} catch (OperationCanceledException) {
// Ignore.
} finally {
logger.Debug("ZeroMQ client stopped receiving messages.");
disconnectSemaphore.Wait(CancellationToken.None);
}
}
private void LogMessageType(ILogger logger, ReadOnlyMemory<byte> data) {
if (!logger.IsEnabled(LogEventLevel.Verbose)) {
return;
}
if (data.Length > 0 && messageDefinitions.ToClient.TryGetType(data, out var type)) {
logger.Verbose("Received {MessageType} ({Bytes} B).", type.Name, data.Length);
}
else {
logger.Verbose("Received {Bytes} B message.", data.Length);
}
}
private sealed class Handler : MessageHandler<TClientListener> {
private readonly RpcConnectionToServer<TServerListener> connection;
private readonly IMessageDefinitions<TClientListener, TServerListener, TReplyMessage> messageDefinitions;
public Handler(RpcConnectionToServer<TServerListener> connection, IMessageDefinitions<TClientListener, TServerListener, TReplyMessage> messageDefinitions, TClientListener listener, ILogger logger, TaskManager taskManager, CancellationToken cancellationToken) : base(listener, logger, taskManager, cancellationToken) {
this.connection = connection;
this.messageDefinitions = messageDefinitions;
}
protected override Task SendReply(uint sequenceId, byte[] serializedReply) {
return connection.Send(messageDefinitions.CreateReplyMessage(sequenceId, serializedReply));
}
}
}

41
Utils/Phantom.Utils.Rpc/RpcConnectionToServer.cs Normal file
View File

@ -0,0 +1,41 @@
using NetMQ;
using NetMQ.Sockets;
using Phantom.Utils.Rpc.Message;
namespace Phantom.Utils.Rpc;
public sealed class RpcConnectionToServer<TListener> {
private readonly ClientSocket socket;
private readonly MessageRegistry<TListener> messageRegistry;
private readonly MessageReplyTracker replyTracker;
internal RpcConnectionToServer(ClientSocket socket, MessageRegistry<TListener> messageRegistry, MessageReplyTracker replyTracker) {
this.socket = socket;
this.messageRegistry = messageRegistry;
this.replyTracker = replyTracker;
}
public async Task Send<TMessage>(TMessage message) where TMessage : IMessage<TListener, NoReply> {
var bytes = messageRegistry.Write(message).ToArray();
if (bytes.Length > 0) {
await socket.SendAsync(bytes);
}
}
public async Task<TReply?> Send<TMessage, TReply>(TMessage message, TimeSpan waitForReplyTime, CancellationToken waitForReplyCancellationToken) where TMessage : IMessage<TListener, TReply> where TReply : class {
var sequenceId = replyTracker.RegisterReply();
var bytes = messageRegistry.Write<TMessage, TReply>(sequenceId, message).ToArray();
if (bytes.Length == 0) {
replyTracker.ForgetReply(sequenceId);
return null;
}
await socket.SendAsync(bytes);
return await replyTracker.WaitForReply<TReply>(sequenceId, waitForReplyTime, waitForReplyCancellationToken);
}
public void Receive(IReply message) {
replyTracker.ReceiveReply(message.SequenceId, message.SerializedReply);
}
}

37
Utils/Phantom.Utils.Rpc/RpcRuntime.cs
View File

@ -1,39 +1,28 @@
using NetMQ;
using Phantom.Utils.Rpc.Message;
using Phantom.Utils.Rpc.Sockets;
using Phantom.Utils.Tasks;
using Serilog;
namespace Phantom.Utils.Rpc;
static class RpcRuntime {
internal static void SetDefaultSocketOptions(ThreadSafeSocketOptions options) {
// TODO test behavior when either agent or server are offline for a very long time
options.DelayAttachOnConnect = true;
options.ReceiveHighWatermark = 10_000;
options.SendHighWatermark = 10_000;
}
}
public abstract class RpcRuntime<TSocket> where TSocket : ThreadSafeSocket {
private readonly TSocket socket;
private readonly ILogger runtimeLogger;
private readonly MessageReplyTracker replyTracker;
private readonly TaskManager taskManager;
protected RpcRuntime(RpcConfiguration configuration, TSocket socket) {
RpcRuntime.SetDefaultSocketOptions(socket.Options);
this.socket = socket;
this.runtimeLogger = configuration.RuntimeLogger;
this.replyTracker = new MessageReplyTracker(runtimeLogger);
this.taskManager = new TaskManager(configuration.TaskManagerLogger);
protected RpcRuntime(RpcSocket<TSocket> socket) {
this.socket = socket.Socket;
this.runtimeLogger = socket.Config.RuntimeLogger;
this.replyTracker = socket.ReplyTracker;
this.taskManager = new TaskManager(socket.Config.TaskManagerLogger);
}
protected async Task Launch() {
Connect(socket);
void RunTask() {
try {
Run(socket, replyTracker, taskManager);
Run(socket, runtimeLogger, replyTracker, taskManager);
} catch (Exception e) {
runtimeLogger.Error(e, "Caught exception in RPC thread.");
}
@ -42,21 +31,19 @@ public abstract class RpcRuntime<TSocket> where TSocket : ThreadSafeSocket {
try {
await Task.Factory.StartNew(RunTask, CancellationToken.None, TaskCreationOptions.LongRunning, TaskScheduler.Default);
} catch (OperationCanceledException) {
// ignore
// Ignore.
} finally {
await taskManager.Stop();
await Disconnect();
await Disconnect(socket, runtimeLogger);
socket.Dispose();
NetMQConfig.Cleanup();
runtimeLogger.Information("ZeroMQ client stopped.");
runtimeLogger.Information("ZeroMQ runtime stopped.");
}
}
protected abstract void Connect(TSocket socket);
protected abstract void Run(TSocket socket, MessageReplyTracker replyTracker, TaskManager taskManager);
protected abstract void Run(TSocket socket, ILogger logger, MessageReplyTracker replyTracker, TaskManager taskManager);
protected virtual Task Disconnect() {
protected virtual Task Disconnect(TSocket socket, ILogger logger) {
return Task.CompletedTask;
}
}

40
Utils/Phantom.Utils.Rpc/Sockets/RpcClientSocket.cs Normal file
View File

@ -0,0 +1,40 @@
using NetMQ;
using NetMQ.Sockets;
using Phantom.Utils.Rpc.Message;
namespace Phantom.Utils.Rpc.Sockets;
public static class RpcClientSocket {
public static RpcClientSocket<TClientListener, TServerListener, TReplyMessage> Connect<TClientListener, TServerListener, TReplyMessage, THelloMessage>(RpcConfiguration config, IMessageDefinitions<TClientListener, TServerListener, TReplyMessage> messageDefinitions, THelloMessage helloMessage) where THelloMessage : IMessage<TServerListener, NoReply> where TReplyMessage : IMessage<TClientListener, NoReply>, IMessage<TServerListener, NoReply> {
return RpcClientSocket<TClientListener, TServerListener, TReplyMessage>.Connect(config, messageDefinitions, helloMessage);
}
}
public sealed class RpcClientSocket<TClientListener, TServerListener, TReplyMessage> : RpcSocket<ClientSocket> where TReplyMessage : IMessage<TClientListener, NoReply>, IMessage<TServerListener, NoReply> {
internal static RpcClientSocket<TClientListener, TServerListener, TReplyMessage> Connect<THelloMessage>(RpcConfiguration config, IMessageDefinitions<TClientListener, TServerListener, TReplyMessage> messageDefinitions, THelloMessage helloMessage) where THelloMessage : IMessage<TServerListener, NoReply> {
var socket = new ClientSocket();
var options = socket.Options;
options.CurveServerCertificate = config.ServerCertificate;
options.CurveCertificate = new NetMQCertificate();
options.HelloMessage = messageDefinitions.ToServer.Write(helloMessage).ToArray();
RpcSocket.SetDefaultSocketOptions(options);
var url = config.TcpUrl;
var logger = config.RuntimeLogger;
logger.Information("Starting ZeroMQ client and connecting to {Url}...", url);
socket.Connect(url);
logger.Information("ZeroMQ client ready.");
return new RpcClientSocket<TClientListener, TServerListener, TReplyMessage>(socket, config, messageDefinitions);
}
public RpcConnectionToServer<TServerListener> Connection { get; }
internal IMessageDefinitions<TClientListener, TServerListener, TReplyMessage> MessageDefinitions { get; }
private RpcClientSocket(ClientSocket socket, RpcConfiguration config, IMessageDefinitions<TClientListener, TServerListener, TReplyMessage> messageDefinitions) : base(socket, config) {
MessageDefinitions = messageDefinitions;
Connection = new RpcConnectionToServer<TServerListener>(socket, messageDefinitions.ToServer, ReplyTracker);
}
}

25
Utils/Phantom.Utils.Rpc/Sockets/RpcServerSocket.cs Normal file
View File

@ -0,0 +1,25 @@
using NetMQ.Sockets;
namespace Phantom.Utils.Rpc.Sockets;
public sealed class RpcServerSocket : RpcSocket<ServerSocket> {
public static RpcServerSocket Connect(RpcConfiguration config) {
var socket = new ServerSocket();
var options = socket.Options;
options.CurveServer = true;
options.CurveCertificate = config.ServerCertificate;
RpcSocket.SetDefaultSocketOptions(options);
var url = config.TcpUrl;
var logger = config.RuntimeLogger;
logger.Information("Starting ZeroMQ server on {Url}...", url);
socket.Bind(url);
logger.Information("ZeroMQ server initialized, listening for connections on port {Port}.", config.Port);
return new RpcServerSocket(socket, config);
}
private RpcServerSocket(ServerSocket socket, RpcConfiguration config) : base(socket, config) {}
}

25
Utils/Phantom.Utils.Rpc/Sockets/RpcSocket.cs Normal file
View File

@ -0,0 +1,25 @@
using NetMQ;
using Phantom.Utils.Rpc.Message;
namespace Phantom.Utils.Rpc.Sockets;
static class RpcSocket {
internal static void SetDefaultSocketOptions(ThreadSafeSocketOptions options) {
// TODO test behavior when either agent or server are offline for a very long time
options.DelayAttachOnConnect = true;
options.ReceiveHighWatermark = 10_000;
options.SendHighWatermark = 10_000;
}
}
public abstract class RpcSocket<TSocket> where TSocket : ThreadSafeSocket {
internal TSocket Socket { get; }
internal RpcConfiguration Config { get; }
internal MessageReplyTracker ReplyTracker { get; }
protected RpcSocket(TSocket socket, RpcConfiguration config) {
Socket = socket;
Config = config;
ReplyTracker = new MessageReplyTracker(config.RuntimeLogger);
}
}