Replace NetMQ with custom TCP logic

.gitignore

@@ -192,7 +192,6 @@ ClientBin/
 *.dbmdl
 *.dbproj.schemaview
 *.jfm
-*.pfx
 *.publishsettings
 orleans.codegen.cs
 

.run/Agent 1.run.xml

@@ -5,7 +5,7 @@
     <option name="WORKING_DIRECTORY" value="$PROJECT_DIR$/.workdir/Agent1" />
     <option name="PASS_PARENT_ENVS" value="1" />
     <envs>
-      <env name="AGENT_KEY" value="JXBQQYG5T267RQS75MXWBTCJZY5CKTCCGQY22MCZPHSQQSJYCHH2NG2TCNXQY6TBSXM9NQDRS2CMX" />
+      <env name="AGENT_KEY" value="K5ZRZYYJ9GWM2FS6XH5N5QQ7WZRPNDGHYMN5QP7RP6PPY27KRPMSYGCN" />
       <env name="AGENT_NAME" value="Agent 1" />
       <env name="ALLOWED_RCON_PORTS" value="25575,27000,27001" />
       <env name="ALLOWED_SERVER_PORTS" value="25565,26000,26001" />
@@ -15,14 +15,19 @@
       <env name="MAX_MEMORY" value="12G" />
     </envs>
     <option name="USE_EXTERNAL_CONSOLE" value="0" />
+    <option name="ENV_FILE_PATHS" value="" />
+    <option name="REDIRECT_INPUT_PATH" value="" />
+    <option name="PTY_MODE" value="Auto" />
     <option name="USE_MONO" value="0" />
     <option name="RUNTIME_ARGUMENTS" value="" />
+    <option name="AUTO_ATTACH_CHILDREN" value="0" />
+    <option name="MIXED_MODE_DEBUG" value="0" />
     <option name="PROJECT_PATH" value="$PROJECT_DIR$/Agent/Phantom.Agent/Phantom.Agent.csproj" />
     <option name="PROJECT_EXE_PATH_TRACKING" value="1" />
     <option name="PROJECT_ARGUMENTS_TRACKING" value="1" />
     <option name="PROJECT_WORKING_DIRECTORY_TRACKING" value="0" />
     <option name="PROJECT_KIND" value="DotNetCore" />
-    <option name="PROJECT_TFM" value="net8.0" />
+    <option name="PROJECT_TFM" value="net9.0" />
     <method v="2">
       <option name="Build" />
     </method>

.run/Agent 2.run.xml

@@ -5,7 +5,7 @@
     <option name="WORKING_DIRECTORY" value="$PROJECT_DIR$/.workdir/Agent2" />
     <option name="PASS_PARENT_ENVS" value="1" />
     <envs>
-      <env name="AGENT_KEY" value="JXBQQYG5T267RQS75MXWBTCJZY5CKTCCGQY22MCZPHSQQSJYCHH2NG2TCNXQY6TBSXM9NQDRS2CMX" />
+      <env name="AGENT_KEY" value="K5ZRZYYJ9GWM2FS6XH5N5QQ7WZRPNDGHYMN5QP7RP6PPY27KRPMSYGCN" />
       <env name="AGENT_NAME" value="Agent 2" />
       <env name="ALLOWED_RCON_PORTS" value="27002-27006" />
       <env name="ALLOWED_SERVER_PORTS" value="26002-26006" />
@@ -15,14 +15,19 @@
       <env name="MAX_MEMORY" value="10G" />
     </envs>
     <option name="USE_EXTERNAL_CONSOLE" value="0" />
+    <option name="ENV_FILE_PATHS" value="" />
+    <option name="REDIRECT_INPUT_PATH" value="" />
+    <option name="PTY_MODE" value="Auto" />
     <option name="USE_MONO" value="0" />
     <option name="RUNTIME_ARGUMENTS" value="" />
+    <option name="AUTO_ATTACH_CHILDREN" value="0" />
+    <option name="MIXED_MODE_DEBUG" value="0" />
     <option name="PROJECT_PATH" value="$PROJECT_DIR$/Agent/Phantom.Agent/Phantom.Agent.csproj" />
     <option name="PROJECT_EXE_PATH_TRACKING" value="1" />
     <option name="PROJECT_ARGUMENTS_TRACKING" value="1" />
     <option name="PROJECT_WORKING_DIRECTORY_TRACKING" value="0" />
     <option name="PROJECT_KIND" value="DotNetCore" />
-    <option name="PROJECT_TFM" value="net8.0" />
+    <option name="PROJECT_TFM" value="net9.0" />
     <method v="2">
       <option name="Build" />
     </method>

.run/Agent 3.run.xml

@@ -5,7 +5,7 @@
     <option name="WORKING_DIRECTORY" value="$PROJECT_DIR$/.workdir/Agent3" />
     <option name="PASS_PARENT_ENVS" value="1" />
     <envs>
-      <env name="AGENT_KEY" value="JXBQQYG5T267RQS75MXWBTCJZY5CKTCCGQY22MCZPHSQQSJYCHH2NG2TCNXQY6TBSXM9NQDRS2CMX" />
+      <env name="AGENT_KEY" value="K5ZRZYYJ9GWM2FS6XH5N5QQ7WZRPNDGHYMN5QP7RP6PPY27KRPMSYGCN" />
       <env name="AGENT_NAME" value="Agent 3" />
       <env name="ALLOWED_RCON_PORTS" value="27007" />
       <env name="ALLOWED_SERVER_PORTS" value="26007" />
@@ -15,14 +15,19 @@
       <env name="MAX_MEMORY" value="2560M" />
     </envs>
     <option name="USE_EXTERNAL_CONSOLE" value="0" />
+    <option name="ENV_FILE_PATHS" value="" />
+    <option name="REDIRECT_INPUT_PATH" value="" />
+    <option name="PTY_MODE" value="Auto" />
     <option name="USE_MONO" value="0" />
     <option name="RUNTIME_ARGUMENTS" value="" />
+    <option name="AUTO_ATTACH_CHILDREN" value="0" />
+    <option name="MIXED_MODE_DEBUG" value="0" />
     <option name="PROJECT_PATH" value="$PROJECT_DIR$/Agent/Phantom.Agent/Phantom.Agent.csproj" />
     <option name="PROJECT_EXE_PATH_TRACKING" value="1" />
     <option name="PROJECT_ARGUMENTS_TRACKING" value="1" />
     <option name="PROJECT_WORKING_DIRECTORY_TRACKING" value="0" />
     <option name="PROJECT_KIND" value="DotNetCore" />
-    <option name="PROJECT_TFM" value="net8.0" />
+    <option name="PROJECT_TFM" value="net9.0" />
     <method v="2">
       <option name="Build" />
     </method>

.run/Controller.run.xml

@@ -14,14 +14,19 @@
       <env name="WEB_RPC_SERVER_HOST" value="localhost" />
     </envs>
     <option name="USE_EXTERNAL_CONSOLE" value="0" />
+    <option name="ENV_FILE_PATHS" value="" />
+    <option name="REDIRECT_INPUT_PATH" value="" />
+    <option name="PTY_MODE" value="Auto" />
     <option name="USE_MONO" value="0" />
     <option name="RUNTIME_ARGUMENTS" value="" />
+    <option name="AUTO_ATTACH_CHILDREN" value="0" />
+    <option name="MIXED_MODE_DEBUG" value="0" />
     <option name="PROJECT_PATH" value="$PROJECT_DIR$/Controller/Phantom.Controller/Phantom.Controller.csproj" />
     <option name="PROJECT_EXE_PATH_TRACKING" value="1" />
     <option name="PROJECT_ARGUMENTS_TRACKING" value="1" />
     <option name="PROJECT_WORKING_DIRECTORY_TRACKING" value="0" />
     <option name="PROJECT_KIND" value="DotNetCore" />
-    <option name="PROJECT_TFM" value="net8.0" />
+    <option name="PROJECT_TFM" value="net9.0" />
     <method v="2">
       <option name="Build" />
     </method>

.run/Web.run.xml

@@ -7,18 +7,23 @@
     <envs>
       <env name="ASPNETCORE_ENVIRONMENT" value="Development" />
       <env name="CONTROLLER_HOST" value="localhost" />
-      <env name="WEB_KEY" value="BMNHM9RRPMCBBY29D9XHS6KBKZSRY7F5XFN27YZX96XXWJC2NM2D6YRHM9PZN9JGQGCSJ6FMB2GGZ" />
+      <env name="WEB_KEY" value="T5Y722D2GZBXT2H27QS95P2YQRFB2GCTKHSWT5CZFDTFKW52TCM9GDRW" />
       <env name="WEB_SERVER_HOST" value="localhost" />
     </envs>
     <option name="USE_EXTERNAL_CONSOLE" value="0" />
+    <option name="ENV_FILE_PATHS" value="" />
+    <option name="REDIRECT_INPUT_PATH" value="" />
+    <option name="PTY_MODE" value="Auto" />
     <option name="USE_MONO" value="0" />
     <option name="RUNTIME_ARGUMENTS" value="" />
+    <option name="AUTO_ATTACH_CHILDREN" value="0" />
+    <option name="MIXED_MODE_DEBUG" value="0" />
     <option name="PROJECT_PATH" value="$PROJECT_DIR$/Web/Phantom.Web/Phantom.Web.csproj" />
     <option name="PROJECT_EXE_PATH_TRACKING" value="1" />
     <option name="PROJECT_ARGUMENTS_TRACKING" value="1" />
     <option name="PROJECT_WORKING_DIRECTORY_TRACKING" value="0" />
     <option name="PROJECT_KIND" value="DotNetCore" />
-    <option name="PROJECT_TFM" value="net8.0" />
+    <option name="PROJECT_TFM" value="net9.0" />
     <method v="2">
       <option name="Build" />
     </method>

.workdir/Controller/secrets/agent.auth

@@ -0,0 +1 @@
+满<EFBFBD>H<EFBFBD>c<EFBFBD>og<EFBFBD>

.workdir/Controller/secrets/agent.secret

@@ -1 +0,0 @@
-<EFBFBD>Z<EFBFBD>t<>MPI<49>GMZ<4D><5A><EFBFBD><EFBFBD>kN<6B>VF1X<><58>p

.workdir/Controller/secrets/web.auth

@@ -0,0 +1 @@
+<07>U<EFBFBD>/<2F><04><><EFBFBD>q

.workdir/Controller/secrets/web.key

@@ -1,2 +0,0 @@
-<EFBFBD><EFBFBD>h?Ο<05>Bx
-<02>

.workdir/Controller/secrets/web.secret

@@ -1 +0,0 @@
-T<EFBFBD>./g<11><>N<EFBFBD><4E>t<EFBFBD>$<24>!<21>(<28><>#<23>~<7E><>}<14><:

Agent/Phantom.Agent.Rpc/ControllerConnection.cs

@@ -1,21 +1,15 @@
 using Phantom.Common.Messages.Agent;
-using Phantom.Utils.Logging;
 using Phantom.Utils.Rpc.Runtime;
-using Serilog;
 
 namespace Phantom.Agent.Rpc;
 
-public sealed class ControllerConnection {
-	private static readonly ILogger Logger = PhantomLogger.Create(nameof(ControllerConnection));
-	
-	private readonly RpcConnectionToServer<IMessageToController> connection;
-	
-	public ControllerConnection(RpcConnectionToServer<IMessageToController> connection) {
-		this.connection = connection;
-		Logger.Information("Connection ready.");
+public sealed class ControllerConnection(RpcSendChannel<IMessageToController> sendChannel) {
+	public ValueTask Send<TMessage>(TMessage message) where TMessage : IMessageToController {
+		return sendChannel.SendMessage(message, CancellationToken.None /* TODO */);
 	}
 	
-	public Task Send<TMessage>(TMessage message) where TMessage : IMessageToController {
-		return connection.Send(message);
+	// TODO handle properly
+	public bool TrySend<TMessage>(TMessage message) where TMessage : IMessageToController {
+		return sendChannel.TrySendMessage(message);
 	}
 }

Agent/Phantom.Agent.Rpc/KeepAliveLoop.cs

@@ -1,7 +1,6 @@
 using Phantom.Common.Messages.Agent;
 using Phantom.Common.Messages.Agent.ToController;
 using Phantom.Utils.Logging;
-using Phantom.Utils.Rpc.Runtime;
 using Serilog;
 
 namespace Phantom.Agent.Rpc;

Agent/Phantom.Agent.Rpc/RpcClientAgentHandshake.cs

@@ -0,0 +1,7 @@
+using Phantom.Common.Data;
+
+namespace Phantom.Agent.Rpc;
+
+sealed class RpcClientAgentHandshake(AuthToken authToken) {
+	
+}

Agent/Phantom.Agent.Rpc/RpcClientRuntime.cs

@@ -1,37 +0,0 @@
-using NetMQ;
-using NetMQ.Sockets;
-using Phantom.Common.Messages.Agent;
-using Phantom.Common.Messages.Agent.BiDirectional;
-using Phantom.Common.Messages.Agent.ToController;
-using Phantom.Utils.Actor;
-using Phantom.Utils.Rpc.Runtime;
-using Phantom.Utils.Rpc.Sockets;
-using Serilog;
-
-namespace Phantom.Agent.Rpc;
-
-public sealed class RpcClientRuntime : RpcClientRuntime<IMessageToAgent, IMessageToController, ReplyMessage> {
-	public static Task Launch(RpcClientSocket<IMessageToAgent, IMessageToController, ReplyMessage> socket, ActorRef<IMessageToAgent> handlerActorRef, SemaphoreSlim disconnectSemaphore, CancellationToken receiveCancellationToken) {
-		return new RpcClientRuntime(socket, handlerActorRef, disconnectSemaphore, receiveCancellationToken).Launch();
-	}
-	
-	private RpcClientRuntime(RpcClientSocket<IMessageToAgent, IMessageToController, ReplyMessage> socket, ActorRef<IMessageToAgent> handlerActor, SemaphoreSlim disconnectSemaphore, CancellationToken receiveCancellationToken) : base(socket, handlerActor, disconnectSemaphore, receiveCancellationToken) {}
-	
-	protected override async Task RunWithConnection(ClientSocket socket, RpcConnectionToServer<IMessageToController> connection) {
-		var keepAliveLoop = new KeepAliveLoop(connection);
-		try {
-			await base.RunWithConnection(socket, connection);
-		} finally {
-			keepAliveLoop.Cancel();
-		}
-	}
-	
-	protected override async Task SendDisconnectMessage(ClientSocket socket, ILogger logger) {
-		var unregisterMessageBytes = AgentMessageRegistries.ToController.Write(new UnregisterAgentMessage()).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.");
-		}
-	}
-}

Agent/Phantom.Agent.Services/Instances/InstanceActor.cs

@@ -58,7 +58,7 @@ sealed class InstanceActor : ReceiveActor<InstanceActor.ICommand> {
 	
 	private void ReportCurrentStatus() {
 		agentState.UpdateInstance(new Instance(instanceGuid, currentStatus));
-		instanceServices.ControllerConnection.Send(new ReportInstanceStatusMessage(instanceGuid, currentStatus));
+		instanceServices.ControllerConnection.TrySend(new ReportInstanceStatusMessage(instanceGuid, currentStatus));
 	}
 	
 	private void TransitionState(InstanceRunningState? newState) {

Agent/Phantom.Agent.Services/Instances/InstanceContext.cs

@@ -7,6 +7,6 @@ namespace Phantom.Agent.Services.Instances;
 
 sealed record InstanceContext(Guid InstanceGuid, string ShortName, ILogger Logger, InstanceServices Services, ActorRef<InstanceActor.ICommand> Actor, CancellationToken ActorCancellationToken) {
 	public void ReportEvent(IInstanceEvent instanceEvent) {
-		Services.ControllerConnection.Send(new ReportInstanceEventMessage(Guid.NewGuid(), DateTime.UtcNow, InstanceGuid, instanceEvent));
+		Services.ControllerConnection.TrySend(new ReportInstanceEventMessage(Guid.NewGuid(), DateTime.UtcNow, InstanceGuid, instanceEvent));
 	}
 }

Agent/Phantom.Agent.Services/Instances/InstanceTicketManager.cs

@@ -91,7 +91,7 @@ sealed class InstanceTicketManager {
 	
 	public void RefreshAgentStatus() {
 		lock (this) {
-			controllerConnection.Send(new ReportAgentStatusMessage(activeTicketGuids.Count, usedMemory));
+			controllerConnection.TrySend(new ReportAgentStatusMessage(activeTicketGuids.Count, usedMemory));
 		}
 	}
 	

Agent/Phantom.Agent.Services/Instances/State/InstanceLogSender.cs

@@ -63,7 +63,7 @@ sealed class InstanceLogSender : CancellableBackgroundTask {
 	
 	private void SendOutputToServer(ImmutableArray<string> lines) {
 		if (!lines.IsEmpty) {
-			controllerConnection.Send(new InstanceOutputMessage(instanceGuid, lines));
+			controllerConnection.TrySend(new InstanceOutputMessage(instanceGuid, lines));
 		}
 	}
 	

Agent/Phantom.Agent.Services/Instances/State/InstancePlayerCountTracker.cs

@@ -38,7 +38,7 @@ sealed class InstancePlayerCountTracker : CancellableBackgroundTask {
 			}
 			
 			onlinePlayerCountChanged?.Invoke(this, value?.Online);
-			controllerConnection.Send(new ReportInstancePlayerCountsMessage(instanceGuid, value));
+			controllerConnection.TrySend(new ReportInstancePlayerCountsMessage(instanceGuid, value));
 		}
 	}
 	

Agent/Phantom.Agent.Services/Rpc/ControllerMessageHandlerActor.cs

@@ -3,7 +3,6 @@ using Phantom.Common.Data;
 using Phantom.Common.Data.Instance;
 using Phantom.Common.Data.Replies;
 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.Actor;
@@ -16,18 +15,18 @@ namespace Phantom.Agent.Services.Rpc;
 public sealed class ControllerMessageHandlerActor : ReceiveActor<IMessageToAgent> {
 	private static ILogger Logger { get; } = PhantomLogger.Create<ControllerMessageHandlerActor>();
 	
-	public readonly record struct Init(RpcConnectionToServer<IMessageToController> Connection, AgentServices Agent, CancellationTokenSource ShutdownTokenSource);
+	public readonly record struct Init(RpcSendChannel<IMessageToController> SendChannel, AgentServices Agent, CancellationTokenSource ShutdownTokenSource);
 	
 	public static Props<IMessageToAgent> Factory(Init init) {
 		return Props<IMessageToAgent>.Create(() => new ControllerMessageHandlerActor(init), new ActorConfiguration { SupervisorStrategy = SupervisorStrategies.Resume });
 	}
 	
-	private readonly RpcConnectionToServer<IMessageToController> connection;
+	private readonly RpcSendChannel<IMessageToController> sendChannel;
 	private readonly AgentServices agent;
 	private readonly CancellationTokenSource shutdownTokenSource;
 	
 	private ControllerMessageHandlerActor(Init init) {
-		this.connection = init.Connection;
+		this.sendChannel = init.SendChannel;
 		this.agent = init.Agent;
 		this.shutdownTokenSource = init.ShutdownTokenSource;
 		
@@ -37,7 +36,6 @@ public sealed class ControllerMessageHandlerActor : ReceiveActor<IMessageToAgent
 		ReceiveAndReplyLater<LaunchInstanceMessage, Result<LaunchInstanceResult, InstanceActionFailure>>(HandleLaunchInstance);
 		ReceiveAndReplyLater<StopInstanceMessage, Result<StopInstanceResult, InstanceActionFailure>>(HandleStopInstance);
 		ReceiveAndReplyLater<SendCommandToInstanceMessage, Result<SendCommandToInstanceResult, InstanceActionFailure>>(HandleSendCommandToInstance);
-		Receive<ReplyMessage>(HandleReply);
 	}
 	
 	private async Task HandleRegisterAgentSuccess(RegisterAgentSuccessMessage message) {
@@ -56,9 +54,7 @@ public sealed class ControllerMessageHandlerActor : ReceiveActor<IMessageToAgent
 			}
 		}
 		
-		connection.SetIsReady();
-		
-		await connection.Send(new AdvertiseJavaRuntimesMessage(agent.JavaRuntimeRepository.All));
+		await sendChannel.SendMessage(new AdvertiseJavaRuntimesMessage(agent.JavaRuntimeRepository.All), CancellationToken.None);
 		agent.InstanceTicketManager.RefreshAgentStatus();
 	}
 	
@@ -94,8 +90,4 @@ public sealed class ControllerMessageHandlerActor : ReceiveActor<IMessageToAgent
 	private async Task<Result<SendCommandToInstanceResult, InstanceActionFailure>> HandleSendCommandToInstance(SendCommandToInstanceMessage message) {
 		return await agent.InstanceManager.Request(new InstanceManagerActor.SendCommandToInstanceCommand(message.InstanceGuid, message.Command));
 	}
-	
-	private void HandleReply(ReplyMessage message) {
-		connection.Receive(message);
-	}
 }

Agent/Phantom.Agent/AgentKey.cs

@@ -1,4 +1,4 @@
-using NetMQ;
+using System.Text;
 using Phantom.Common.Data;
 using Phantom.Utils.Cryptography;
 using Phantom.Utils.IO;
@@ -10,7 +10,7 @@ namespace Phantom.Agent;
 static class AgentKey {
 	private static ILogger Logger { get; } = PhantomLogger.Create(nameof(AgentKey));
 	
-	public static Task<(NetMQCertificate, AuthToken)?> Load(string? agentKeyToken, string? agentKeyFilePath) {
+	public static Task<ConnectionKey?> Load(string? agentKeyToken, string? agentKeyFilePath) {
 		if (agentKeyFilePath != null) {
 			return LoadFromFile(agentKeyFilePath);
 		}
@@ -22,18 +22,19 @@ static class AgentKey {
 		}
 	}
 	
-	private static async Task<(NetMQCertificate, AuthToken)?> LoadFromFile(string agentKeyFilePath) {
+	private static async Task<ConnectionKey?> LoadFromFile(string agentKeyFilePath) {
 		if (!File.Exists(agentKeyFilePath)) {
 			Logger.Fatal("Missing agent key file: {AgentKeyFilePath}", agentKeyFilePath);
 			return null;
 		}
 		
 		try {
-			Files.RequireMaximumFileSize(agentKeyFilePath, 64);
-			return LoadFromBytes(await File.ReadAllBytesAsync(agentKeyFilePath));
+			Files.RequireMaximumFileSize(agentKeyFilePath, maximumBytes: 64);
+			string[] lines = await File.ReadAllLinesAsync(agentKeyFilePath, Encoding.UTF8);
+			return LoadFromToken(lines[0]);
 		} catch (IOException e) {
 			Logger.Fatal("Error loading agent key from file: {AgentKeyFilePath}", agentKeyFilePath);
-			Logger.Fatal(e.Message);
+			Logger.Fatal("{}", e.Message);
 			return null;
 		} catch (Exception) {
 			Logger.Fatal("File does not contain a valid agent key: {AgentKeyFilePath}", agentKeyFilePath);
@@ -41,7 +42,7 @@ static class AgentKey {
 		}
 	}
 	
-	private static (NetMQCertificate, AuthToken)? LoadFromToken(string agentKey) {
+	private static ConnectionKey? LoadFromToken(string agentKey) {
 		try {
 			return LoadFromBytes(TokenGenerator.DecodeBytes(agentKey));
 		} catch (Exception) {
@@ -50,11 +51,9 @@ static class AgentKey {
 		}
 	}
 	
-	private static (NetMQCertificate, AuthToken)? LoadFromBytes(byte[] agentKey) {
-		var (publicKey, agentToken) = ConnectionCommonKey.FromBytes(agentKey);
-		var controllerCertificate = NetMQCertificate.FromPublicKey(publicKey);
-		
+	private static ConnectionKey? LoadFromBytes(byte[] agentKey) {
+		var connectionKey = ConnectionKey.FromBytes(agentKey);
 		Logger.Information("Loaded agent key.");
-		return (controllerCertificate, agentToken);
+		return connectionKey;
 	}
 }

Agent/Phantom.Agent/Program.cs

@@ -1,5 +1,4 @@
 using System.Reflection;
-using NetMQ;
 using Phantom.Agent;
 using Phantom.Agent.Rpc;
 using Phantom.Agent.Services;
@@ -9,9 +8,9 @@ using Phantom.Common.Messages.Agent;
 using Phantom.Common.Messages.Agent.ToController;
 using Phantom.Utils.Actor;
 using Phantom.Utils.Logging;
-using Phantom.Utils.Rpc;
-using Phantom.Utils.Rpc.Sockets;
+using Phantom.Utils.Rpc.Runtime;
 using Phantom.Utils.Runtime;
+using Phantom.Utils.Threading;
 
 const int ProtocolVersion = 1;
 
@@ -48,33 +47,44 @@ try {
 		return 1;
 	}
 	
-	var (controllerCertificate, agentToken) = agentKey.Value;
+	var (certificateThumbprint, authToken) = agentKey.Value;
 	var agentInfo = new AgentInfo(agentGuid.Value, agentName, ProtocolVersion, fullVersion, maxInstances, maxMemory, allowedServerPorts, allowedRconPorts);
 	
 	PhantomLogger.Root.InformationHeading("Launching Phantom Panel agent...");
 	
-	var rpcConfiguration = new RpcConfiguration("Agent", controllerHost, controllerPort, controllerCertificate);
-	var rpcSocket = RpcClientSocket.Connect(rpcConfiguration, AgentMessageRegistries.Definitions, new RegisterAgentMessage(agentToken, agentInfo));
+	using var rpcClient = await RpcClient<IMessageToController, IMessageToAgent>.Connect("Controller", controllerHost, controllerPort, "phantom-controller", certificateThumbprint, null, AgentMessageRegistries.Definitions, shutdownCancellationToken);
+	if (rpcClient == null) {
+		return 1;
+	}
 	
-	var agentServices = new AgentServices(agentInfo, folders, new AgentServiceConfiguration(maxConcurrentBackupCompressionTasks), new ControllerConnection(rpcSocket.Connection));
-	await agentServices.Initialize();
-	
-	var rpcMessageHandlerInit = new ControllerMessageHandlerActor.Init(rpcSocket.Connection, agentServices, shutdownCancellationTokenSource);
-	var rpcMessageHandlerActor = agentServices.ActorSystem.ActorOf(ControllerMessageHandlerActor.Factory(rpcMessageHandlerInit), "ControllerMessageHandler");
-	
-	var rpcDisconnectSemaphore = new SemaphoreSlim(0, 1);
-	var rpcTask = RpcClientRuntime.Launch(rpcSocket, rpcMessageHandlerActor, rpcDisconnectSemaphore, shutdownCancellationToken);
+	Task? rpcClientListener = null;
 	try {
-		await rpcTask.WaitAsync(shutdownCancellationToken);
-	} finally {
-		shutdownCancellationTokenSource.Cancel();
+		// var rpcConfiguration = new RpcConfiguration("Agent", 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(rpcClient.SendChannel));
+		await agentServices.Initialize();
+		
+		var rpcMessageHandlerInit = new ControllerMessageHandlerActor.Init(rpcClient.SendChannel, agentServices, shutdownCancellationTokenSource);
+		var rpcMessageHandlerActor = agentServices.ActorSystem.ActorOf(ControllerMessageHandlerActor.Factory(rpcMessageHandlerInit), "ControllerMessageHandler");
+		
+		rpcClientListener = rpcClient.Listen(rpcMessageHandlerActor);
+		
+		await shutdownCancellationToken.WaitHandle.WaitOneAsync();
 		await agentServices.Shutdown();
-		
-		rpcDisconnectSemaphore.Release();
-		await rpcTask;
-		rpcDisconnectSemaphore.Dispose();
-		
-		NetMQConfig.Cleanup();
+	} finally {
+		try {
+			await rpcClient.SendChannel.SendMessage(new UnregisterAgentMessage(), CancellationToken.None);
+			// TODO wait for acknowledgment
+		} catch (Exception e) {
+			PhantomLogger.Root.Warning(e, "Could not unregister agent after shutdown.");
+		} finally {
+			await rpcClient.Shutdown();
+			
+			if (rpcClientListener != null) {
+				await rpcClientListener;
+			}
+		}
 	}
 	
 	return 0;

Common/Phantom.Common.Data/AuthToken.cs

@@ -1,4 +1,5 @@
-using System.Diagnostics.CodeAnalysis;
+using System.Collections.Immutable;
+using System.Diagnostics.CodeAnalysis;
 using System.Security.Cryptography;
 using MemoryPack;
 
@@ -7,31 +8,33 @@ namespace Phantom.Common.Data;
 [MemoryPackable(GenerateType.VersionTolerant)]
 [SuppressMessage("ReSharper", "MemberCanBePrivate.Global")]
 public sealed partial class AuthToken {
-	internal const int Length = 12;
+	public const int Length = 12;
 	
 	[MemoryPackOrder(0)]
 	[MemoryPackInclude]
-	private readonly byte[] bytes;
+	public readonly ImmutableArray<byte> Bytes;
 	
-	internal AuthToken(byte[]? bytes) {
-		ArgumentNullException.ThrowIfNull(bytes);
-		
+	public AuthToken(ImmutableArray<byte> bytes) {
 		if (bytes.Length != Length) {
 			throw new ArgumentOutOfRangeException(nameof(bytes), "Invalid token length: " + bytes.Length + ". Token length must be exactly " + Length + " bytes.");
 		}
 		
-		this.bytes = bytes;
+		this.Bytes = bytes;
 	}
 	
 	public bool FixedTimeEquals(AuthToken providedAuthToken) {
-		return CryptographicOperations.FixedTimeEquals(bytes, providedAuthToken.bytes);
+		return FixedTimeEquals(providedAuthToken.Bytes.AsSpan());
+	}
+	
+	public bool FixedTimeEquals(ReadOnlySpan<byte> other) {
+		return CryptographicOperations.FixedTimeEquals(Bytes.AsSpan(), other);
 	}
 	
 	internal void WriteTo(Span<byte> span) {
-		bytes.CopyTo(span);
+		Bytes.CopyTo(span);
 	}
 	
 	public static AuthToken Generate() {
-		return new AuthToken(RandomNumberGenerator.GetBytes(Length));
+		return new AuthToken([..RandomNumberGenerator.GetBytes(Length)]);
 	}
 }

Common/Phantom.Common.Data/ConnectionCommonKey.cs

@@ -1,18 +0,0 @@
-namespace Phantom.Common.Data;
-
-public readonly record struct ConnectionCommonKey(byte[] CertificatePublicKey, AuthToken AuthToken) {
-	private const byte TokenLength = AuthToken.Length;
-	
-	public byte[] ToBytes() {
-		Span<byte> result = stackalloc byte[TokenLength + CertificatePublicKey.Length];
-		AuthToken.WriteTo(result[..TokenLength]);
-		CertificatePublicKey.CopyTo(result[TokenLength..]);
-		return result.ToArray();
-	}
-	
-	public static ConnectionCommonKey FromBytes(byte[] agentKey) {
-		var authToken = new AuthToken(agentKey[..TokenLength]);
-		var certificatePublicKey = agentKey[TokenLength..];
-		return new ConnectionCommonKey(certificatePublicKey, authToken);
-	}
-}

Common/Phantom.Common.Data/ConnectionKey.cs

@@ -0,0 +1,20 @@
+using Phantom.Utils.Rpc.Runtime.Tls;
+
+namespace Phantom.Common.Data;
+
+public readonly record struct ConnectionKey(RpcCertificateThumbprint CertificateThumbprint, AuthToken AuthToken) {
+	private const byte TokenLength = AuthToken.Length;
+
+	public byte[] ToBytes() {
+		Span<byte> result = stackalloc byte[TokenLength + CertificateThumbprint.Bytes.Length];
+		AuthToken.WriteTo(result[..TokenLength]);
+		CertificateThumbprint.Bytes.CopyTo(result[TokenLength..]);
+		return result.ToArray();
+	}
+
+	public static ConnectionKey FromBytes(ReadOnlySpan<byte> data) {
+		var authToken = new AuthToken([..data[..TokenLength]]);
+		var certificateThumbprint = RpcCertificateThumbprint.From(data[TokenLength..]);
+		return new ConnectionKey(certificateThumbprint, authToken);
+	}
+}

Common/Phantom.Common.Data/Phantom.Common.Data.csproj

@@ -11,6 +11,7 @@
   
   <ItemGroup>
     <ProjectReference Include="..\..\Utils\Phantom.Utils\Phantom.Utils.csproj" />
+    <ProjectReference Include="..\..\Utils\Phantom.Utils.Rpc\Phantom.Utils.Rpc.csproj" />
   </ItemGroup>
 
 </Project>

Common/Phantom.Common.Messages.Agent/AgentMessageRegistries.cs

@@ -1,6 +1,5 @@
 using Phantom.Common.Data;
 using Phantom.Common.Data.Replies;
-using Phantom.Common.Messages.Agent.BiDirectional;
 using Phantom.Common.Messages.Agent.ToAgent;
 using Phantom.Common.Messages.Agent.ToController;
 using Phantom.Utils.Logging;
@@ -12,7 +11,7 @@ public static class AgentMessageRegistries {
 	public static MessageRegistry<IMessageToAgent> ToAgent { get; } = new (PhantomLogger.Create("MessageRegistry", nameof(ToAgent)));
 	public static MessageRegistry<IMessageToController> ToController { get; } = new (PhantomLogger.Create("MessageRegistry", nameof(ToController)));
 	
-	public static IMessageDefinitions<IMessageToAgent, IMessageToController, ReplyMessage> Definitions { get; } = new MessageDefinitions();
+	public static IMessageDefinitions<IMessageToController, IMessageToAgent> Definitions { get; } = new MessageDefinitions();
 	
 	static AgentMessageRegistries() {
 		ToAgent.Add<RegisterAgentSuccessMessage>(0);
@@ -21,7 +20,6 @@ public static class AgentMessageRegistries {
 		ToAgent.Add<LaunchInstanceMessage, Result<LaunchInstanceResult, InstanceActionFailure>>(3);
 		ToAgent.Add<StopInstanceMessage, Result<StopInstanceResult, InstanceActionFailure>>(4);
 		ToAgent.Add<SendCommandToInstanceMessage, Result<SendCommandToInstanceResult, InstanceActionFailure>>(5);
-		ToAgent.Add<ReplyMessage>(127);
 		
 		ToController.Add<RegisterAgentMessage>(0);
 		ToController.Add<UnregisterAgentMessage>(1);
@@ -32,15 +30,10 @@ public static class AgentMessageRegistries {
 		ToController.Add<ReportAgentStatusMessage>(6);
 		ToController.Add<ReportInstanceEventMessage>(7);
 		ToController.Add<ReportInstancePlayerCountsMessage>(8);
-		ToController.Add<ReplyMessage>(127);
 	}
 	
-	private sealed class MessageDefinitions : IMessageDefinitions<IMessageToAgent, IMessageToController, ReplyMessage> {
+	private sealed class MessageDefinitions : IMessageDefinitions<IMessageToController, IMessageToAgent> {
 		public MessageRegistry<IMessageToAgent> ToClient => ToAgent;
 		public MessageRegistry<IMessageToController> ToServer => ToController;
-		
-		public ReplyMessage CreateReplyMessage(uint sequenceId, byte[] serializedReply) {
-			return new ReplyMessage(sequenceId, serializedReply);
-		}
 	}
 }

Common/Phantom.Common.Messages.Agent/BiDirectional/ReplyMessage.cs

@@ -1,10 +0,0 @@
-using MemoryPack;
-using Phantom.Utils.Rpc.Message;
-
-namespace Phantom.Common.Messages.Agent.BiDirectional;
-
-[MemoryPackable(GenerateType.VersionTolerant)]
-public sealed partial record ReplyMessage(
-	[property: MemoryPackOrder(0)] uint SequenceId,
-	[property: MemoryPackOrder(1)] byte[] SerializedReply
-) : IMessageToController, IMessageToAgent, IReply;

Common/Phantom.Common.Messages.Agent/Handshake/IAgentHandshakeResult.cs

@@ -0,0 +1,10 @@
+using MemoryPack;
+
+namespace Phantom.Common.Messages.Agent.Handshake;
+
+[MemoryPackable]
+[MemoryPackUnion(tag: 0, typeof(InvalidAuthToken))]
+public partial interface IAgentHandshakeResult;
+
+[MemoryPackable(GenerateType.VersionTolerant)]
+public sealed partial record InvalidAuthToken : IAgentHandshakeResult;

Common/Phantom.Common.Messages.Web/BiDirectional/ReplyMessage.cs

@@ -1,10 +0,0 @@
-using MemoryPack;
-using Phantom.Utils.Rpc.Message;
-
-namespace Phantom.Common.Messages.Web.BiDirectional;
-
-[MemoryPackable(GenerateType.VersionTolerant)]
-public sealed partial record ReplyMessage(
-	[property: MemoryPackOrder(0)] uint SequenceId,
-	[property: MemoryPackOrder(1)] byte[] SerializedReply
-) : IMessageToController, IMessageToWeb, IReply;

Common/Phantom.Common.Messages.Web/WebMessageRegistries.cs

@@ -7,7 +7,6 @@ using Phantom.Common.Data.Web.AuditLog;
 using Phantom.Common.Data.Web.EventLog;
 using Phantom.Common.Data.Web.Instance;
 using Phantom.Common.Data.Web.Users;
-using Phantom.Common.Messages.Web.BiDirectional;
 using Phantom.Common.Messages.Web.ToController;
 using Phantom.Common.Messages.Web.ToWeb;
 using Phantom.Utils.Logging;
@@ -19,7 +18,7 @@ public static class WebMessageRegistries {
 	public static MessageRegistry<IMessageToController> ToController { get; } = new (PhantomLogger.Create("MessageRegistry", nameof(ToController)));
 	public static MessageRegistry<IMessageToWeb> ToWeb { get; } = new (PhantomLogger.Create("MessageRegistry", nameof(ToWeb)));
 	
-	public static IMessageDefinitions<IMessageToWeb, IMessageToController, ReplyMessage> Definitions { get; } = new MessageDefinitions();
+	public static IMessageDefinitions<IMessageToController, IMessageToWeb> Definitions { get; } = new MessageDefinitions();
 	
 	static WebMessageRegistries() {
 		ToController.Add<RegisterWebMessage>(0);
@@ -42,22 +41,16 @@ public static class WebMessageRegistries {
 		ToController.Add<GetAgentJavaRuntimesMessage, ImmutableDictionary<Guid, ImmutableArray<TaggedJavaRuntime>>>(17);
 		ToController.Add<GetAuditLogMessage, Result<ImmutableArray<AuditLogItem>, UserActionFailure>>(18);
 		ToController.Add<GetEventLogMessage, Result<ImmutableArray<EventLogItem>, UserActionFailure>>(19);
-		ToController.Add<ReplyMessage>(127);
 		
 		ToWeb.Add<RegisterWebResultMessage>(0);
 		ToWeb.Add<RefreshAgentsMessage>(1);
 		ToWeb.Add<RefreshInstancesMessage>(2);
 		ToWeb.Add<InstanceOutputMessage>(3);
 		ToWeb.Add<RefreshUserSessionMessage>(4);
-		ToWeb.Add<ReplyMessage>(127);
 	}
 	
-	private sealed class MessageDefinitions : IMessageDefinitions<IMessageToWeb, IMessageToController, ReplyMessage> {
+	private sealed class MessageDefinitions : IMessageDefinitions<IMessageToController, IMessageToWeb> {
 		public MessageRegistry<IMessageToWeb> ToClient => ToWeb;
 		public MessageRegistry<IMessageToController> ToServer => ToController;
-		
-		public ReplyMessage CreateReplyMessage(uint sequenceId, byte[] serializedReply) {
-			return new ReplyMessage(sequenceId, serializedReply);
-		}
 	}
 }

Controller/Phantom.Controller.Services/Agents/AgentActor.cs

@@ -21,7 +21,6 @@ using Phantom.Utils.Actor.Mailbox;
 using Phantom.Utils.Actor.Tasks;
 using Phantom.Utils.Collections;
 using Phantom.Utils.Logging;
-using Phantom.Utils.Rpc.Runtime;
 using Serilog;
 
 namespace Phantom.Controller.Services.Agents;

Controller/Phantom.Controller.Services/Agents/AgentConnection.cs

@@ -1,7 +1,6 @@
 using Phantom.Common.Messages.Agent;
 using Phantom.Utils.Actor;
 using Phantom.Utils.Logging;
-using Phantom.Utils.Rpc.Runtime;
 using Serilog;
 
 namespace Phantom.Controller.Services.Agents;

Controller/Phantom.Controller.Services/Agents/AgentManager.cs

@@ -13,7 +13,6 @@ using Phantom.Controller.Minecraft;
 using Phantom.Controller.Services.Users.Sessions;
 using Phantom.Utils.Actor;
 using Phantom.Utils.Logging;
-using Phantom.Utils.Rpc.Runtime;
 using Serilog;
 
 namespace Phantom.Controller.Services.Agents;

Controller/Phantom.Controller.Services/ControllerServices.cs

@@ -13,7 +13,7 @@ using Phantom.Controller.Services.Rpc;
 using Phantom.Controller.Services.Users;
 using Phantom.Controller.Services.Users.Sessions;
 using Phantom.Utils.Actor;
-using Phantom.Utils.Rpc.Runtime;
+using Phantom.Utils.Rpc.Runtime2;
 using IMessageFromAgentToController = Phantom.Common.Messages.Agent.IMessageToController;
 using IMessageFromWebToController = Phantom.Common.Messages.Web.IMessageToController;
 

Controller/Phantom.Controller.Services/Rpc/AgentMessageHandlerActor.cs

@@ -1,13 +1,11 @@
 using Phantom.Common.Data.Replies;
 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.Controller.Services.Agents;
 using Phantom.Controller.Services.Events;
 using Phantom.Controller.Services.Instances;
 using Phantom.Utils.Actor;
-using Phantom.Utils.Rpc.Runtime;
 
 namespace Phantom.Controller.Services.Rpc;
 

Controller/Phantom.Controller.Services/Rpc/AgentRegistrationHandler.cs

@@ -4,7 +4,7 @@ using Phantom.Controller.Services.Agents;
 using Phantom.Controller.Services.Events;
 using Phantom.Controller.Services.Instances;
 using Phantom.Utils.Actor;
-using Phantom.Utils.Rpc.Runtime;
+using Phantom.Utils.Rpc.Runtime2;
 
 namespace Phantom.Controller.Services.Rpc;
 

Controller/Phantom.Controller.Services/Rpc/RpcServerAgentHandshake.cs

@@ -0,0 +1,35 @@
+using Phantom.Common.Data;
+using Phantom.Common.Data.Agent;
+using Phantom.Common.Messages.Agent.Handshake;
+using Phantom.Utils.Logging;
+using Phantom.Utils.Rpc.Runtime;
+using Phantom.Utils.Rpc.Runtime.Utils;
+using Serilog;
+
+namespace Phantom.Controller.Services.Rpc;
+
+public sealed class RpcServerAgentHandshake(AuthToken authToken) : RpcServerHandshake {
+	private static readonly ILogger Logger = PhantomLogger.Create<RpcServerAgentHandshake>();
+	
+	protected override async Task<bool> AcceptClient(string remoteAddress, Stream stream, CancellationToken cancellationToken) {
+		Memory<byte> buffer = new Memory<byte>(new byte[AuthToken.Length]);
+		await stream.ReadExactlyAsync(buffer, cancellationToken: cancellationToken);
+		
+		if (!authToken.FixedTimeEquals(buffer.Span)) {
+			Logger.Warning("Rejected client {}, invalid authorization token.", remoteAddress);
+			await Respond(remoteAddress, stream, new InvalidAuthToken(), cancellationToken);
+			return false;
+		}
+		
+		AgentInfo agentInfo = await Serialization.Deserialize<AgentInfo>(stream, cancellationToken);
+		return true;
+	}
+	
+	private async ValueTask Respond(string remoteAddress, Stream stream, IAgentHandshakeResult result, CancellationToken cancellationToken) {
+		try {
+			await Serialization.Serialize(result, stream, cancellationToken);
+		} catch (Exception e) {
+			Logger.Error(e, "Could not send handshake result to client {}.", remoteAddress);
+		}
+	}
+}

Controller/Phantom.Controller.Services/Rpc/WebMessageDataUpdateSenderActor.cs

@@ -5,7 +5,6 @@ using Phantom.Common.Messages.Web;
 using Phantom.Common.Messages.Web.ToWeb;
 using Phantom.Controller.Services.Instances;
 using Phantom.Utils.Actor;
-using Phantom.Utils.Rpc.Runtime;
 
 namespace Phantom.Controller.Services.Rpc;
 

Controller/Phantom.Controller.Services/Rpc/WebMessageHandlerActor.cs

@@ -7,7 +7,6 @@ using Phantom.Common.Data.Web.AuditLog;
 using Phantom.Common.Data.Web.EventLog;
 using Phantom.Common.Data.Web.Instance;
 using Phantom.Common.Data.Web.Users;
-using Phantom.Common.Messages.Agent.BiDirectional;
 using Phantom.Common.Messages.Web;
 using Phantom.Common.Messages.Web.ToController;
 using Phantom.Controller.Minecraft;
@@ -17,7 +16,6 @@ using Phantom.Controller.Services.Instances;
 using Phantom.Controller.Services.Users;
 using Phantom.Controller.Services.Users.Sessions;
 using Phantom.Utils.Actor;
-using Phantom.Utils.Rpc.Runtime;
 
 namespace Phantom.Controller.Services.Rpc;
 

Controller/Phantom.Controller.Services/Rpc/WebRegistrationHandler.cs

@@ -10,7 +10,7 @@ using Phantom.Controller.Services.Users;
 using Phantom.Controller.Services.Users.Sessions;
 using Phantom.Utils.Actor;
 using Phantom.Utils.Logging;
-using Phantom.Utils.Rpc.Runtime;
+using Phantom.Utils.Rpc.Runtime2;
 using Serilog;
 
 namespace Phantom.Controller.Services.Rpc;

Controller/Phantom.Controller/ConnectionKeyData.cs

@@ -1,6 +1,6 @@
-using NetMQ;
-using Phantom.Common.Data;
+using Phantom.Common.Data;
+using Phantom.Utils.Rpc.Runtime;
 
 namespace Phantom.Controller;
 
-readonly record struct ConnectionKeyData(NetMQCertificate Certificate, AuthToken AuthToken);
+readonly record struct ConnectionKeyData(RpcServerCertificate Certificate, AuthToken AuthToken);

Controller/Phantom.Controller/ConnectionKeyFiles.cs

@@ -1,39 +1,37 @@
-using NetMQ;
-using Phantom.Common.Data;
+using Phantom.Common.Data;
 using Phantom.Utils.Cryptography;
 using Phantom.Utils.IO;
 using Phantom.Utils.Logging;
+using Phantom.Utils.Monads;
+using Phantom.Utils.Rpc.Runtime;
+using Phantom.Utils.Rpc.Runtime.Tls;
 using Serilog;
 
 namespace Phantom.Controller;
 
 abstract class ConnectionKeyFiles {
-	private const string CommonKeyFileExtension = ".key";
-	private const string SecretKeyFileExtension = ".secret";
-	
 	private readonly ILogger logger;
-	private readonly string commonKeyFileName;
-	private readonly string secretKeyFileName;
+	private readonly string certificateFileName;
+	private readonly string authTokenFileName;
 	
 	private ConnectionKeyFiles(ILogger logger, string name) {
 		this.logger = logger;
-		this.commonKeyFileName = name + CommonKeyFileExtension;
-		this.secretKeyFileName = name + SecretKeyFileExtension;
+		this.certificateFileName = name + ".pfx";
+		this.authTokenFileName = name + ".auth";
 	}
 	
 	public async Task<ConnectionKeyData?> CreateOrLoad(string folderPath) {
-		string commonKeyFilePath = Path.Combine(folderPath, commonKeyFileName);
-		string secretKeyFilePath = Path.Combine(folderPath, secretKeyFileName);
+		string certificateFilePath = Path.Combine(folderPath, certificateFileName);
+		string authTokenFilePath = Path.Combine(folderPath, authTokenFileName);
 		
-		bool commonKeyFileExists = File.Exists(commonKeyFilePath);
-		bool secretKeyFileExists = File.Exists(secretKeyFilePath);
+		bool certificateFileExists = File.Exists(certificateFilePath);
+		bool authTokenFileExists = File.Exists(authTokenFilePath);
 		
-		if (commonKeyFileExists && secretKeyFileExists) {
+		if (certificateFileExists && authTokenFileExists) {
 			try {
-				return await ReadKeyFiles(commonKeyFilePath, secretKeyFilePath);
+				return await ReadKeyFiles(certificateFilePath, authTokenFilePath);
 			} catch (IOException e) {
-				logger.Fatal("Error reading connection key files.");
-				logger.Fatal(e.Message);
+				logger.Fatal(e, "Error reading connection key files.");
 				return null;
 			} catch (Exception) {
 				logger.Fatal("Connection key files contain invalid data.");
@@ -41,72 +39,75 @@ abstract class ConnectionKeyFiles {
 			}
 		}
 		
-		if (commonKeyFileExists || secretKeyFileExists) {
-			string existingKeyFilePath = commonKeyFileExists ? commonKeyFilePath : secretKeyFilePath;
-			string missingKeyFileName = commonKeyFileExists ? secretKeyFileName : commonKeyFileName;
-			logger.Fatal("The connection key file {ExistingKeyFilePath} exists but {MissingKeyFileName} does not. Please delete it to regenerate both files.", existingKeyFilePath, missingKeyFileName);
+		if (certificateFileExists || authTokenFileExists) {
+			string existingKeyFilePath = certificateFileExists ? certificateFilePath : authTokenFilePath;
+			string missingKeyFileName = certificateFileExists ? authTokenFileName : certificateFileName;
+			logger.Fatal("Connection key file {ExistingKeyFilePath} exists but {MissingKeyFileName} does not. Please delete it to regenerate both files.", existingKeyFilePath, missingKeyFileName);
 			return null;
 		}
 		
 		logger.Information("Creating connection key files in: {FolderPath}", folderPath);
 		
 		try {
-			return await GenerateKeyFiles(commonKeyFilePath, secretKeyFilePath);
+			return await GenerateKeyFiles(certificateFilePath, authTokenFilePath);
 		} catch (Exception e) {
-			logger.Fatal("Error creating connection key files.");
-			logger.Fatal(e.Message);
+			logger.Fatal(e, "Error creating connection key files.");
 			return null;
 		}
 	}
 	
-	private async Task<ConnectionKeyData?> ReadKeyFiles(string commonKeyFilePath, string secretKeyFilePath) {
-		byte[] commonKeyBytes = await ReadKeyFile(commonKeyFilePath);
-		byte[] secretKeyBytes = await ReadKeyFile(secretKeyFilePath);
+	private async Task<ConnectionKeyData?> ReadKeyFiles(string certificateFilePath, string authTokenFilePath) {
+		RpcServerCertificate certificate = null!;
 		
-		var (publicKey, authToken) = ConnectionCommonKey.FromBytes(commonKeyBytes);
-		var certificate = new NetMQCertificate(secretKeyBytes, publicKey);
+		switch (RpcServerCertificate.Load(certificateFilePath)) {
+			case Left<RpcServerCertificate, DisallowedAlgorithmError>(var rpcServerCertificate):
+				certificate = rpcServerCertificate;
+				break;
+			
+			case Right<RpcServerCertificate, DisallowedAlgorithmError>(var error):
+				logger.Fatal("Certificate {CertificateFilePath} was expected to use {ExpectedAlgorithmName}, instead it uses {ActualAlgorithmName}.", certificateFilePath, error.ExpectedAlgorithmName, error.ActualAlgorithmName);
+				return null;
+		}
 		
+		var authToken = new AuthToken([..await ReadKeyFile(authTokenFilePath)]);
 		logger.Information("Loaded connection key files.");
-		LogCommonKey(commonKeyFilePath, TokenGenerator.EncodeBytes(commonKeyBytes));
+		
+		var connectionKey = new ConnectionKey(certificate.Thumbprint, authToken);
+		LogCommonKey(TokenGenerator.EncodeBytes(connectionKey.ToBytes()));
 		
 		return new ConnectionKeyData(certificate, authToken);
 	}
 	
 	private static Task<byte[]> ReadKeyFile(string filePath) {
-		Files.RequireMaximumFileSize(filePath, 64);
+		Files.RequireMaximumFileSize(filePath, maximumBytes: 64);
 		return File.ReadAllBytesAsync(filePath);
 	}
 	
-	private async Task<ConnectionKeyData> GenerateKeyFiles(string commonKeyFilePath, string secretKeyFilePath) {
-		var certificate = new NetMQCertificate();
+	private async Task<ConnectionKeyData> GenerateKeyFiles(string certificateFilePath, string authTokenFilePath) {
+		var certificateBytes = RpcServerCertificate.CreateAndExport("phantom-controller");
 		var authToken = AuthToken.Generate();
-		var commonKey = new ConnectionCommonKey(certificate.PublicKey, authToken).ToBytes();
-		
-		await Files.WriteBytesAsync(secretKeyFilePath, certificate.SecretKey, FileMode.Create, Chmod.URW_GR);
-		await Files.WriteBytesAsync(commonKeyFilePath, commonKey, FileMode.Create, Chmod.URW_GR);
 		
+		await Files.WriteBytesAsync(certificateFilePath, certificateBytes, FileMode.Create, Chmod.URW_GR);
+		await Files.WriteBytesAsync(authTokenFilePath, authToken.Bytes.ToArray(), FileMode.Create, Chmod.URW_GR);
 		logger.Information("Created new connection key files.");
-		LogCommonKey(commonKeyFilePath, TokenGenerator.EncodeBytes(commonKey));
+		
+		var certificate = RpcServerCertificate.Load(certificateFilePath).RequireLeft;
+		var connectionKey = new ConnectionKey(certificate.Thumbprint, authToken);
+		LogCommonKey(TokenGenerator.EncodeBytes(connectionKey.ToBytes()));
 		
 		return new ConnectionKeyData(certificate, authToken);
 	}
 	
-	protected abstract void LogCommonKey(string commonKeyFilePath, string commonKeyEncoded);
+	protected abstract void LogCommonKey(string commonKeyEncoded);
 	
-	internal sealed class Agent : ConnectionKeyFiles {
-		public Agent() : base(PhantomLogger.Create<ConnectionKeyFiles, Agent>(), "agent") {}
-		
-		protected override void LogCommonKey(string commonKeyFilePath, string commonKeyEncoded) {
-			logger.Information("Agent key file: {AgentKeyFilePath}", commonKeyFilePath);
+	internal sealed class Agent() : ConnectionKeyFiles(PhantomLogger.Create<ConnectionKeyFiles, Agent>(), "agent") {
+		protected override void LogCommonKey(string commonKeyEncoded) {
 			logger.Information("Agent key: {AgentKey}", commonKeyEncoded);
 		}
 	}
 	
-	internal sealed class Web : ConnectionKeyFiles {
-		public Web() : base(PhantomLogger.Create<ConnectionKeyFiles, Web>(), "web") {}
-		
-		protected override void LogCommonKey(string commonKeyFilePath, string commonKeyEncoded) {
-			logger.Information("Web key file: {WebKeyFilePath}", commonKeyFilePath);
+	internal sealed class Web() : ConnectionKeyFiles(PhantomLogger.Create<ConnectionKeyFiles, Web>(), "web") {
+		protected override void LogCommonKey(string commonKeyEncoded) {
 			logger.Information("Web key: {WebKey}", commonKeyEncoded);
 		}
 	}

Controller/Phantom.Controller/Program.cs

@@ -1,15 +1,13 @@
 using System.Reflection;
-using NetMQ;
-using Phantom.Common.Messages.Agent;
-using Phantom.Common.Messages.Web;
 using Phantom.Controller;
 using Phantom.Controller.Database.Postgres;
 using Phantom.Controller.Services;
+using Phantom.Controller.Services.Rpc;
 using Phantom.Utils.IO;
 using Phantom.Utils.Logging;
-using Phantom.Utils.Rpc;
 using Phantom.Utils.Rpc.Runtime;
 using Phantom.Utils.Runtime;
+using Phantom.Utils.Tasks;
 
 var shutdownCancellationTokenSource = new CancellationTokenSource();
 var shutdownCancellationToken = shutdownCancellationTokenSource.Token;
@@ -37,7 +35,7 @@ try {
 	PhantomLogger.Root.InformationHeading("Initializing Phantom Panel controller...");
 	PhantomLogger.Root.Information("Controller version: {Version}", fullVersion);
 	
-	var (agentRpcServerHost, agentRpcServerPort, webRpcServerHost, webRpcServerPort, sqlConnectionString) = Variables.LoadOrStop();
+	var (agentRpcServerHost, webRpcServerHost, sqlConnectionString) = Variables.LoadOrStop();
 	
 	string secretsPath = Path.GetFullPath("./secrets");
 	CreateFolderOrStop(secretsPath, Chmod.URWX_GRX);
@@ -59,20 +57,26 @@ try {
 	using var controllerServices = new ControllerServices(dbContextFactory, agentKeyData.AuthToken, webKeyData.AuthToken, shutdownCancellationToken);
 	await controllerServices.Initialize();
 	
-	static RpcConfiguration ConfigureRpc(string serviceName, string host, ushort port, ConnectionKeyData connectionKey) {
-		return new RpcConfiguration(serviceName, host, port, connectionKey.Certificate);
-	}
+	LinkedTasks<bool> rpcServerTasks = new LinkedTasks<bool>([
+		new RpcServer("Agent", agentRpcServerHost, agentKeyData.Certificate, new RpcServerAgentHandshake(agentKeyData.AuthToken)).Run(shutdownCancellationToken),
+		// new RpcServer("Web", webRpcServerHost, webKeyData.Certificate).Run(shutdownCancellationToken),
+	]);
 	
-	try {
-		await Task.WhenAll(
-			RpcServerRuntime.Launch(ConfigureRpc("Agent", agentRpcServerHost, agentRpcServerPort, agentKeyData), AgentMessageRegistries.Definitions, controllerServices.AgentRegistrationHandler, controllerServices.ActorSystem, shutdownCancellationToken),
-			RpcServerRuntime.Launch(ConfigureRpc("Web", webRpcServerHost, webRpcServerPort, webKeyData), WebMessageRegistries.Definitions, controllerServices.WebRegistrationHandler, controllerServices.ActorSystem, shutdownCancellationToken)
-		);
-	} finally {
-		NetMQConfig.Cleanup();
+	// If either RPC server crashes, stop the whole process.
+	await rpcServerTasks.CancelTokenWhenAnyCompletes(shutdownCancellationTokenSource);
+	
+	foreach (Task<bool> rpcServerTask in await rpcServerTasks.WaitForAll()) {
+		if (rpcServerTask.IsFaulted || rpcServerTask is { IsCompletedSuccessfully: true, Result: false }) {
+			return 1;
+		}
 	}
 	
 	return 0;
+	
+	// await Task.WhenAll(
+	// 	RpcServerRuntime.Launch(ConfigureRpc("Agent", agentRpcServerHost, agentRpcServerPort, agentKeyData), AgentMessageRegistries.Definitions, controllerServices.AgentRegistrationHandler, controllerServices.ActorSystem, shutdownCancellationToken),
+	// 	RpcServerRuntime.Launch(ConfigureRpc("Web", webRpcServerHost, webRpcServerPort, webKeyData), WebMessageRegistries.Definitions, controllerServices.WebRegistrationHandler, controllerServices.ActorSystem, shutdownCancellationToken)
+	// );
 } catch (OperationCanceledException) {
 	return 0;
 } catch (StopProcedureException) {

Controller/Phantom.Controller/Variables.cs

@@ -1,14 +1,13 @@
-using Npgsql;
+using System.Net;
+using Npgsql;
 using Phantom.Utils.Logging;
 using Phantom.Utils.Runtime;
 
 namespace Phantom.Controller;
 
 sealed record Variables(
-	string AgentRpcServerHost,
-	ushort AgentRpcServerPort,
-	string WebRpcServerHost,
-	ushort WebRpcServerPort,
+	EndPoint AgentRpcServerHost,
+	EndPoint WebRpcServerHost,
 	string SqlConnectionString
 ) {
 	private static Variables LoadOrThrow() {
@@ -20,11 +19,19 @@ sealed record Variables(
 			Database = EnvironmentVariables.GetString("PG_DATABASE").Require
 		};
 		
+		EndPoint agentRpcServerHost = new IPEndPoint(
+			EnvironmentVariables.GetIpAddress("AGENT_RPC_SERVER_HOST").WithDefault(IPAddress.Any),
+			EnvironmentVariables.GetPortNumber("AGENT_RPC_SERVER_PORT").WithDefault(9401)
+		);
+		
+		EndPoint webRpcServerHost = new IPEndPoint(
+			EnvironmentVariables.GetIpAddress("WEB_RPC_SERVER_HOST").WithDefault(IPAddress.Any),
+			EnvironmentVariables.GetPortNumber("WEB_RPC_SERVER_PORT").WithDefault(9401)
+		);
+		
 		return new Variables(
-			EnvironmentVariables.GetString("AGENT_RPC_SERVER_HOST").WithDefault("0.0.0.0"),
-			EnvironmentVariables.GetPortNumber("AGENT_RPC_SERVER_PORT").WithDefault(9401),
-			EnvironmentVariables.GetString("WEB_RPC_SERVER_HOST").WithDefault("0.0.0.0"),
-			EnvironmentVariables.GetPortNumber("WEB_RPC_SERVER_PORT").WithDefault(9402),
+			agentRpcServerHost,
+			webRpcServerHost,
 			connectionStringBuilder.ToString()
 		);
 	}

README.md

@@ -28,10 +28,13 @@ This project is **work-in-progress**, and currently has no official releases. Fe
 
 For a quick start, I recommend using [Docker](https://www.docker.com/) or another containerization platform. The `Dockerfile` in the root of the repository can build three target images: `phantom-web`, `phantom-controller`, and `phantom-agent`.
 
-All images put the built application into the `/app` folder. The Agent image also installs Java 8, 16, 17, and 18.
+All images put the built application into the `/app` folder. The Agent image also installs Java 8, 16, 17, and 21.
 
 Files are stored relative to the working directory. In the provided images, the working directory is set to `/data`.
 
+> [!IMPORTANT]
+> The 3 services communicate with each other using TLS. Due to inconsistent TLS support and implementation quirks between operating systems, Phantom Panel is intended to run only on Linux with up-to-date OpenSSL libraries. Support for other operating systems only exists for the purposes of local development, and components running on different operating systems may not be able to communicate with each other.
+
 ## Controller
 
 The Controller comprises 3 key areas:

Utils/Phantom.Utils.Rpc/Frame/IFrame.cs

@@ -0,0 +1,46 @@
+using Phantom.Utils.Rpc.Frame.Types;
+
+namespace Phantom.Utils.Rpc.Frame;
+
+interface IFrame {
+	private const byte TypeMessageId = 1;
+	private const byte TypeReplyId = 2;
+	private const byte TypeErrorId = 3;
+	
+	static readonly ReadOnlyMemory<byte> TypeMessage = new ([TypeMessageId]);
+	static readonly ReadOnlyMemory<byte> TypeReply = new ([TypeReplyId]);
+	static readonly ReadOnlyMemory<byte> TypeError = new ([TypeErrorId]);
+	
+	internal static async Task ReadFrom(Stream stream, IFrameReader reader, CancellationToken cancellationToken) {
+		byte[] oneByteBuffer = new byte[1];
+		
+		while (!cancellationToken.IsCancellationRequested) {
+			await stream.ReadExactlyAsync(oneByteBuffer, cancellationToken);
+			
+			switch (oneByteBuffer[0]) {
+				case TypeMessageId:
+					var messageFrame = await MessageFrame.Read(stream, cancellationToken);
+					await reader.OnMessageFrame(messageFrame, stream, cancellationToken);
+					break;
+				
+				case TypeReplyId:
+					var replyFrame = await ReplyFrame.Read(stream, cancellationToken);
+					reader.OnReplyFrame(replyFrame);
+					break;
+				
+				case TypeErrorId:
+					var errorFrame = await ErrorFrame.Read(stream, cancellationToken);
+					reader.OnErrorFrame(errorFrame);
+					break;
+				
+				default:
+					reader.OnUnknownFrameStart(oneByteBuffer[0]);
+					break;
+			}
+		}
+	}
+	
+	ReadOnlyMemory<byte> Type { get; }
+	
+	Task Write(Stream stream, CancellationToken cancellationToken);
+}

Utils/Phantom.Utils.Rpc/Frame/IFrameReader.cs

@@ -0,0 +1,10 @@
+using Phantom.Utils.Rpc.Frame.Types;
+
+namespace Phantom.Utils.Rpc.Frame;
+
+interface IFrameReader {
+	Task OnMessageFrame(MessageFrame frame, Stream stream, CancellationToken cancellationToken);
+	void OnReplyFrame(ReplyFrame frame);
+	void OnErrorFrame(ErrorFrame frame);
+	void OnUnknownFrameStart(byte id);
+}

Utils/Phantom.Utils.Rpc/Frame/Types/ErrorFrame.cs

@@ -0,0 +1,19 @@
+using Phantom.Utils.Rpc.Runtime;
+using Phantom.Utils.Rpc.Runtime.Utils;
+
+namespace Phantom.Utils.Rpc.Frame.Types;
+
+sealed record ErrorFrame(uint ReplyingToMessageId, RpcError Error) : IFrame {
+	public ReadOnlyMemory<byte> Type => IFrame.TypeError;
+	
+	public async Task Write(Stream stream, CancellationToken cancellationToken) {
+		await Serialization.WriteUnsignedInt(ReplyingToMessageId, stream, cancellationToken);
+		await Serialization.WriteByte((byte) Error, stream, cancellationToken);
+	}
+	
+	public static async Task<ErrorFrame> Read(Stream stream, CancellationToken cancellationToken) {
+		var replyingToMessageId = await Serialization.ReadUnsignedInt(stream, cancellationToken);
+		var messageError = (RpcError) await Serialization.ReadByte(stream, cancellationToken);
+		return new ErrorFrame(replyingToMessageId, messageError);
+	}
+}

Utils/Phantom.Utils.Rpc/Frame/Types/MessageFrame.cs

@@ -0,0 +1,40 @@
+using Phantom.Utils.Rpc.Runtime;
+using Phantom.Utils.Rpc.Runtime.Utils;
+
+namespace Phantom.Utils.Rpc.Frame.Types;
+
+sealed record MessageFrame(uint MessageId, ushort RegistryCode, ReadOnlyMemory<byte> SerializedMessage) : IFrame {
+	public const int MaxMessageBytes = 1024 * 1024 * 8;
+	
+	public ReadOnlyMemory<byte> Type => IFrame.TypeMessage;
+	
+	public async Task Write(Stream stream, CancellationToken cancellationToken) {
+		int messageLength = SerializedMessage.Length;
+		CheckMessageLength(messageLength);
+		
+		await Serialization.WriteUnsignedInt(MessageId, stream, cancellationToken);
+		await Serialization.WriteUnsignedShort(RegistryCode, stream, cancellationToken);
+		await Serialization.WriteSignedInt(messageLength, stream, cancellationToken);
+		await stream.WriteAsync(SerializedMessage, cancellationToken);
+	}
+	
+	public static async Task<MessageFrame> Read(Stream stream, CancellationToken cancellationToken) {
+		var messageId = await Serialization.ReadUnsignedInt(stream, cancellationToken);
+		var registryCode = await Serialization.ReadUnsignedShort(stream, cancellationToken);
+		var essageLength = await Serialization.ReadSignedInt(stream, cancellationToken);
+		CheckMessageLength(essageLength);
+		var serializedMessage = await Serialization.ReadBytes(essageLength, stream, cancellationToken);
+		
+		return new MessageFrame(messageId, registryCode, serializedMessage);
+	}
+	
+	private static void CheckMessageLength(int messageLength) {
+		if (messageLength < 0) {
+			throw new RpcErrorException("Message length is negative", RpcError.InvalidData);
+		}
+		
+		if (messageLength > MaxMessageBytes) {
+			throw new RpcErrorException("Message is too large: " + messageLength + " > " + MaxMessageBytes + " bytes", RpcError.MessageTooLarge);
+		}
+	}
+}

Utils/Phantom.Utils.Rpc/Frame/Types/ReplyFrame.cs

@@ -0,0 +1,38 @@
+using Phantom.Utils.Rpc.Runtime;
+using Phantom.Utils.Rpc.Runtime.Utils;
+
+namespace Phantom.Utils.Rpc.Frame.Types;
+
+sealed record ReplyFrame(uint ReplyingToMessageId, ReadOnlyMemory<byte> SerializedReply) : IFrame {
+	public const int MaxReplyBytes = 1024 * 1024 * 32;
+	
+	public ReadOnlyMemory<byte> Type => IFrame.TypeReply;
+	
+	public async Task Write(Stream stream, CancellationToken cancellationToken) {
+		int replyLength = SerializedReply.Length;
+		CheckReplyLength(replyLength);
+		
+		await Serialization.WriteUnsignedInt(ReplyingToMessageId, stream, cancellationToken);
+		await Serialization.WriteSignedInt(replyLength, stream, cancellationToken);
+		await stream.WriteAsync(SerializedReply, cancellationToken);
+	}
+	
+	public static async Task<ReplyFrame> Read(Stream stream, CancellationToken cancellationToken) {
+		var replyingToMessageId = await Serialization.ReadUnsignedInt(stream, cancellationToken);
+		var replyLength = await Serialization.ReadSignedInt(stream, cancellationToken);
+		CheckReplyLength(replyLength);
+		var reply = await Serialization.ReadBytes(replyLength, stream, cancellationToken);
+		
+		return new ReplyFrame(replyingToMessageId, reply);
+	}
+	
+	private static void CheckReplyLength(int replyLength) {
+		if (replyLength < 0) {
+			throw new RpcErrorException("Reply length is negative", RpcError.InvalidData);
+		}
+		
+		if (replyLength > MaxReplyBytes) {
+			throw new RpcErrorException("Reply is too large: " + replyLength + " > " + MaxReplyBytes + " bytes", RpcError.MessageTooLarge);
+		}
+	}
+}

Utils/Phantom.Utils.Rpc/Message/IMessageDefinitions.cs

@@ -1,6 +1,6 @@
 namespace Phantom.Utils.Rpc.Message;
 
-public interface IMessageDefinitions<TClientMessage, TServerMessage, TReplyMessage> : IReplyMessageFactory<TReplyMessage> where TReplyMessage : TClientMessage, TServerMessage {
-	MessageRegistry<TClientMessage> ToClient { get; }
-	MessageRegistry<TServerMessage> ToServer { get; }
+public interface IMessageDefinitions<TClientToServerMessage, TServerToClientMessage> {
+	MessageRegistry<TServerToClientMessage> ToClient { get; }
+	MessageRegistry<TClientToServerMessage> ToServer { get; }
 }

Utils/Phantom.Utils.Rpc/Message/IReply.cs

@@ -1,6 +0,0 @@
-namespace Phantom.Utils.Rpc.Message;
-
-public interface IReply {
-	uint SequenceId { get; }
-	byte[] SerializedReply { get; }
-}

Utils/Phantom.Utils.Rpc/Message/IReplyMessageFactory.cs

@@ -1,5 +0,0 @@
-namespace Phantom.Utils.Rpc.Message;
-
-public interface IReplyMessageFactory<TReplyMessage> {
-	TReplyMessage CreateReplyMessage(uint sequenceId, byte[] serializedReply);
-}

Utils/Phantom.Utils.Rpc/Message/IReplySender.cs

@@ -1,5 +0,0 @@
-namespace Phantom.Utils.Rpc.Message;
-
-interface IReplySender {
-	Task SendReply(uint sequenceId, byte[] serializedReply);
-}

Utils/Phantom.Utils.Rpc/Message/MessageHandler.cs

@@ -1,35 +1,11 @@
 using Phantom.Utils.Actor;
-using Phantom.Utils.Logging;
-using Serilog;
+using Phantom.Utils.Rpc.Runtime;
 
 namespace Phantom.Utils.Rpc.Message;
 
-sealed class MessageHandler<TMessageBase> {
-	private readonly ILogger logger;
-	private readonly ActorRef<TMessageBase> handlerActor;
-	private readonly IReplySender replySender;
+interface MessageHandler<TMessageBase> {
+	ActorRef<TMessageBase> Actor { get; }
 	
-	public MessageHandler(string loggerName, ActorRef<TMessageBase> handlerActor, IReplySender replySender) {
-		this.logger = PhantomLogger.Create("MessageHandler", loggerName);
-		this.handlerActor = handlerActor;
-		this.replySender = replySender;
-	}
-	
-	public void Tell(TMessageBase message) {
-		handlerActor.Tell(message);
-	}
-	
-	public Task TellAndReply<TMessage, TReply>(TMessage message, uint sequenceId) where TMessage : ICanReply<TReply> {
-		return handlerActor.Request(message).ContinueWith(task => {
-			if (task.IsCompletedSuccessfully) {
-				return replySender.SendReply(sequenceId, MessageSerializer.Serialize(task.Result));
-			}
-			
-			if (task.IsFaulted) {
-				logger.Error(task.Exception, "Failed to handle message {Type}.", message.GetType().Name);
-			}
-			
-			return task;
-		}, TaskScheduler.Default);
-	}
+	ValueTask OnReply<TMessage, TReply>(uint messageId, TReply reply, CancellationToken cancellationToken) where TMessage : TMessageBase, ICanReply<TReply>;
+	ValueTask OnError(uint messageId, RpcError error, CancellationToken cancellationToken);
 }

Utils/Phantom.Utils.Rpc/Message/MessageRegistry.cs

@@ -1,22 +1,15 @@
-using System.Buffers;
-using System.Diagnostics.CodeAnalysis;
-using Phantom.Utils.Actor;
+using Phantom.Utils.Actor;
+using Phantom.Utils.Rpc.Frame.Types;
+using Phantom.Utils.Rpc.Runtime;
+using Phantom.Utils.Rpc.Runtime.Utils;
 using Serilog;
-using Serilog.Events;
 
 namespace Phantom.Utils.Rpc.Message;
 
-public sealed class MessageRegistry<TMessageBase> {
-	private const int DefaultBufferSize = 512;
-	
-	private readonly ILogger logger;
+public sealed class MessageRegistry<TMessageBase>(ILogger logger) {
 	private readonly Dictionary<Type, ushort> typeToCodeMapping = new ();
 	private readonly Dictionary<ushort, Type> codeToTypeMapping = new ();
-	private readonly Dictionary<ushort, Action<ReadOnlyMemory<byte>, ushort, MessageHandler<TMessageBase>>> codeToHandlerMapping = new ();
-	
-	public MessageRegistry(ILogger logger) {
-		this.logger = logger;
-	}
+	private readonly Dictionary<ushort, Func<uint, ReadOnlyMemory<byte>, MessageHandler<TMessageBase>, CancellationToken, Task>> codeToHandlerMapping = new ();
 	
 	public void Add<TMessage>(ushort code) where TMessage : TMessageBase {
 		if (HasReplyType(typeof(TMessage))) {
@@ -44,140 +37,64 @@ public sealed class MessageRegistry<TMessageBase> {
 		return messageType.GetInterfaces().Any(type => type.FullName is {} name && name.StartsWith(replyInterfaceName, StringComparison.Ordinal));
 	}
 	
-	internal bool TryGetType(ReadOnlyMemory<byte> data, [NotNullWhen(true)] out Type? type) {
-		try {
-			var code = MessageSerializer.ReadCode(ref data);
-			return codeToTypeMapping.TryGetValue(code, out type);
-		} catch (Exception) {
-			type = null;
-			return false;
-		}
-	}
-	
-	public ReadOnlySpan<byte> Write<TMessage>(TMessage message) where TMessage : TMessageBase {
-		if (!GetMessageCode<TMessage>(out var code)) {
-			return default;
-		}
-		
-		var buffer = new ArrayBufferWriter<byte>(DefaultBufferSize);
-		
-		try {
-			MessageSerializer.WriteCode(buffer, code);
-			MessageSerializer.Serialize(buffer, message);
-			
-			CheckWrittenBufferLength<TMessage>(buffer);
-			return buffer.WrittenSpan;
-		} catch (Exception e) {
-			LogWriteFailure<TMessage>(e);
-			return default;
-		}
-	}
-	
-	public ReadOnlySpan<byte> Write<TMessage, TReply>(uint sequenceId, TMessage message) where TMessage : TMessageBase, ICanReply<TReply> {
-		if (!GetMessageCode<TMessage>(out var code)) {
-			return default;
-		}
-		
-		var buffer = new ArrayBufferWriter<byte>(DefaultBufferSize);
-		
-		try {
-			MessageSerializer.WriteCode(buffer, code);
-			MessageSerializer.WriteSequenceId(buffer, sequenceId);
-			MessageSerializer.Serialize(buffer, message);
-			
-			CheckWrittenBufferLength<TMessage>(buffer);
-			return buffer.WrittenSpan;
-		} catch (Exception e) {
-			LogWriteFailure<TMessage>(e);
-			return default;
-		}
-	}
-	
-	private bool GetMessageCode<TMessage>(out ushort code) where TMessage : TMessageBase {
-		if (typeToCodeMapping.TryGetValue(typeof(TMessage), out code)) {
-			return true;
+	internal MessageFrame CreateFrame<TMessage>(uint messageId, TMessage message) where TMessage : TMessageBase {
+		if (typeToCodeMapping.TryGetValue(typeof(TMessage), out ushort code)) {
+			return new MessageFrame(messageId, code, Serialization.Serialize(message));
 		}
 		else {
-			logger.Error("Unknown message type {Type}.", typeof(TMessage));
-			return false;
+			throw new ArgumentException("Unknown message type: " + typeof(TMessage));
 		}
 	}
 	
-	private void CheckWrittenBufferLength<TMessage>(ArrayBufferWriter<byte> buffer) where TMessage : TMessageBase {
-		if (buffer.WrittenCount > DefaultBufferSize && logger.IsEnabled(LogEventLevel.Verbose)) {
-			logger.Verbose("Serializing {Type} exceeded default buffer size: {WrittenSize} B > {DefaultBufferSize} B", typeof(TMessage).Name, buffer.WrittenCount, DefaultBufferSize);
-		}
-	}
-	
-	private void LogWriteFailure<TMessage>(Exception e) where TMessage : TMessageBase {
-		logger.Error(e, "Failed to serialize message {Type}.", typeof(TMessage).Name);
-	}
-	
-	internal bool Read<TMessage>(ReadOnlyMemory<byte> data, out TMessage message) where TMessage : TMessageBase {
-		if (ReadTypeCode(ref data, out ushort code) && codeToTypeMapping.TryGetValue(code, out var expectedType) && expectedType == typeof(TMessage) && ReadMessage(data, out message)) {
-			return true;
+	internal async Task Handle(MessageFrame frame, MessageHandler<TMessageBase> handler, CancellationToken cancellationToken) {
+		uint messageId = frame.MessageId;
+		
+		if (codeToHandlerMapping.TryGetValue(frame.RegistryCode, out var action)) {
+			await action(messageId, frame.SerializedMessage, handler, cancellationToken);
 		}
 		else {
-			message = default!;
-			return false;
+			logger.Error("Unknown message code {Code} for message {MessageId}.", frame.RegistryCode, messageId);
+			await handler.OnError(messageId, RpcError.UnknownMessageRegistryCode, cancellationToken);
 		}
 	}
 	
-	internal void Handle(ReadOnlyMemory<byte> data, MessageHandler<TMessageBase> handler) {
-		if (!ReadTypeCode(ref data, out var code)) {
+	private async Task DeserializationHandler<TMessage>(uint messageId, ReadOnlyMemory<byte> serializedMessage, MessageHandler<TMessageBase> handler, CancellationToken cancellationToken) where TMessage : TMessageBase {
+		TMessage message;
+		try {
+			message = Serialization.Deserialize<TMessage>(serializedMessage);
+		} catch (Exception e) {
+			logger.Error(e, "Could not deserialize message {MessageId} ({MessageType}).", messageId, typeof(TMessage).Name);
+			await handler.OnError(messageId, RpcError.MessageDeserializationError, cancellationToken);
 			return;
 		}
 		
-		if (!codeToHandlerMapping.TryGetValue(code, out var handle)) {
-			logger.Error("Unknown message code {Code}.", code);
+		handler.Actor.Tell(message);
+	}
+	
+	private async Task DeserializationHandler<TMessage, TReply>(uint messageId, ReadOnlyMemory<byte> serializedMessage, MessageHandler<TMessageBase> handler, CancellationToken cancellationToken) where TMessage : TMessageBase, ICanReply<TReply> {
+		TMessage message;
+		try {
+			message = Serialization.Deserialize<TMessage>(serializedMessage);
+		} catch (Exception e) {
+			logger.Error(e, "Could not deserialize message {MessageId} ({MessageType}).", messageId, typeof(TMessage).Name);
+			await handler.OnError(messageId, RpcError.MessageDeserializationError, cancellationToken);
 			return;
 		}
 		
-		handle(data, code, handler);
-	}
-	
-	private bool ReadTypeCode(ref ReadOnlyMemory<byte> data, out ushort code) {
+		TReply reply;
 		try {
-			code = MessageSerializer.ReadCode(ref data);
-			return true;
+			reply = await handler.Actor.Request(message, cancellationToken);
 		} catch (Exception e) {
-			code = default;
-			logger.Error(e, "Failed to deserialize message code.");
-			return false;
+			logger.Error(e, "Could not handle message {MessageId} ({MessageType}).", messageId, typeof(TMessage).Name);
+			await handler.OnError(messageId, RpcError.MessageHandlingError, cancellationToken);
+			return;
 		}
-	}
-	
-	private bool ReadSequenceId<TMessage, TReply>(ref ReadOnlyMemory<byte> data, out uint sequenceId) where TMessage : TMessageBase, ICanReply<TReply> {
+		
 		try {
-			sequenceId = MessageSerializer.ReadSequenceId(ref data);
-			return true;
+			await handler.OnReply<TMessage, TReply>(messageId, reply, cancellationToken);
 		} catch (Exception e) {
-			sequenceId = default;
-			logger.Error(e, "Failed to deserialize sequence ID of message {Type}.", typeof(TMessage).Name);
-			return false;
-		}
-	}
-	
-	private bool ReadMessage<TMessage>(ReadOnlyMemory<byte> data, out TMessage message) where TMessage : TMessageBase {
-		try {
-			message = MessageSerializer.Deserialize<TMessage>(data);
-			return true;
-		} catch (Exception e) {
-			message = default!;
-			logger.Error(e, "Failed to deserialize message {Type}.", typeof(TMessage).Name);
-			return false;
-		}
-	}
-	
-	private void DeserializationHandler<TMessage>(ReadOnlyMemory<byte> data, ushort code, MessageHandler<TMessageBase> handler) where TMessage : TMessageBase {
-		if (ReadMessage<TMessage>(data, out var message)) {
-			handler.Tell(message);
-		}
-	}
-	
-	private void DeserializationHandler<TMessage, TReply>(ReadOnlyMemory<byte> data, ushort code, MessageHandler<TMessageBase> handler) where TMessage : TMessageBase, ICanReply<TReply> {
-		if (ReadSequenceId<TMessage, TReply>(ref data, out var sequenceId) && ReadMessage<TMessage>(data, out var message)) {
-			handler.TellAndReply<TMessage, TReply>(message, sequenceId);
+			logger.Error(e, "Could not reply to message {MessageId} ({MessageType}).", messageId, typeof(TMessage).Name);
+			await handler.OnError(messageId, RpcError.MessageHandlingError, cancellationToken);
 		}
 	}
 }

Utils/Phantom.Utils.Rpc/Message/MessageReplyTracker.cs

@@ -1,5 +1,7 @@
 using System.Collections.Concurrent;
 using Phantom.Utils.Logging;
+using Phantom.Utils.Rpc.Runtime;
+using Phantom.Utils.Rpc.Runtime.Utils;
 using Phantom.Utils.Tasks;
 using Serilog;
 
@@ -7,55 +9,57 @@ namespace Phantom.Utils.Rpc.Message;
 
 sealed class MessageReplyTracker {
 	private readonly ILogger logger;
-	private readonly ConcurrentDictionary<uint, TaskCompletionSource<byte[]>> replyTasks = new (4, 16);
-	
-	private uint lastSequenceId;
+	private readonly ConcurrentDictionary<uint, TaskCompletionSource<ReadOnlyMemory<byte>>> replyTasks = new (concurrencyLevel: 4, capacity: 16);
 	
 	internal MessageReplyTracker(string loggerName) {
 		this.logger = PhantomLogger.Create<MessageReplyTracker>(loggerName);
 	}
 	
-	public uint RegisterReply() {
-		var sequenceId = Interlocked.Increment(ref lastSequenceId);
-		replyTasks[sequenceId] = AsyncTasks.CreateCompletionSource<byte[]>();
-		return sequenceId;
+	public void RegisterReply(uint messageId) {
+		replyTasks[messageId] = AsyncTasks.CreateCompletionSource<ReadOnlyMemory<byte>>();
 	}
 	
-	public async Task<TReply> WaitForReply<TReply>(uint sequenceId, TimeSpan waitForReplyTime, CancellationToken cancellationToken) {
-		if (!replyTasks.TryGetValue(sequenceId, out var completionSource)) {
-			logger.Warning("No reply callback for id {SequenceId}.", sequenceId);
-			throw new ArgumentException("No reply callback for id: " + sequenceId, nameof(sequenceId));
+	public async Task<TReply> WaitForReply<TReply>(uint messageId, TimeSpan waitForReplyTime, CancellationToken cancellationToken) {
+		if (!replyTasks.TryGetValue(messageId, out var completionSource)) {
+			logger.Warning("No reply callback for id {MessageId}.", messageId);
+			throw new ArgumentException("No reply callback for id: " + messageId, nameof(messageId));
 		}
 		
 		try {
-			byte[] replyBytes = await completionSource.Task.WaitAsync(waitForReplyTime, cancellationToken);
-			return MessageSerializer.Deserialize<TReply>(replyBytes);
+			ReadOnlyMemory<byte> serializedReply = await completionSource.Task.WaitAsync(waitForReplyTime, cancellationToken);
+			return Serialization.Deserialize<TReply>(serializedReply);
 		} catch (TimeoutException) {
-			logger.Debug("Timed out waiting for reply with id {SequenceId}.", sequenceId);
+			logger.Debug("Timed out waiting for reply with id {MessageId}.", messageId);
 			throw;
 		} catch (OperationCanceledException) {
-			logger.Debug("Cancelled waiting for reply with id {SequenceId}.", sequenceId);
+			logger.Debug("Cancelled waiting for reply with id {MessageId}.", messageId);
 			throw;
 		} catch (Exception e) {
-			logger.Warning(e, "Error processing reply with id {SequenceId}.", sequenceId);
+			logger.Warning(e, "Error processing reply with id {MessageId}.", messageId);
 			throw;
 		} finally {
-			ForgetReply(sequenceId);
+			ForgetReply(messageId);
 		}
 	}
 	
-	public void ForgetReply(uint sequenceId) {
-		if (replyTasks.TryRemove(sequenceId, out var task)) {
+	public void ForgetReply(uint messageId) {
+		if (replyTasks.TryRemove(messageId, out var task)) {
 			task.SetCanceled();
 		}
 	}
 	
-	public void ReceiveReply(uint sequenceId, byte[] serializedReply) {
-		if (replyTasks.TryRemove(sequenceId, out var task)) {
+	public void FailReply(uint messageId, RpcErrorException e) {
+		if (replyTasks.TryRemove(messageId, out var task)) {
+			task.SetException(e);
+		}
+	}
+	
+	public void ReceiveReply(uint messageId, ReadOnlyMemory<byte> serializedReply) {
+		if (replyTasks.TryRemove(messageId, out var task)) {
 			task.SetResult(serializedReply);
 		}
 		else {
-			logger.Warning("Received a reply with id {SequenceId} but no registered callback.", sequenceId);
+			logger.Warning("Received a reply with id {MessageId} but no registered callback.", messageId);
 		}
 	}
 }

Utils/Phantom.Utils.Rpc/Message/MessageSerializer.cs

@@ -1,45 +0,0 @@
-using System.Buffers;
-using System.Buffers.Binary;
-using MemoryPack;
-
-namespace Phantom.Utils.Rpc.Message;
-
-static class MessageSerializer {
-	private static readonly MemoryPackSerializerOptions SerializerOptions = MemoryPackSerializerOptions.Utf8;
-	
-	public static byte[] Serialize<T>(T message) {
-		return MemoryPackSerializer.Serialize(message, SerializerOptions);
-	}
-	
-	public static void Serialize<T>(IBufferWriter<byte> destination, T message) {
-		MemoryPackSerializer.Serialize(typeof(T), destination, message, SerializerOptions);
-	}
-	
-	public static T Deserialize<T>(ReadOnlyMemory<byte> memory) {
-		return MemoryPackSerializer.Deserialize<T>(memory.Span) ?? throw new NullReferenceException();
-	}
-	
-	public static void WriteCode(IBufferWriter<byte> destination, ushort value) {
-		Span<byte> buffer = stackalloc byte[2];
-		BinaryPrimitives.WriteUInt16LittleEndian(buffer, value);
-		destination.Write(buffer);
-	}
-	
-	public static ushort ReadCode(ref ReadOnlyMemory<byte> memory) {
-		ushort value = BinaryPrimitives.ReadUInt16LittleEndian(memory.Span);
-		memory = memory[2..];
-		return value;
-	}
-	
-	public static void WriteSequenceId(IBufferWriter<byte> destination, uint sequenceId) {
-		Span<byte> buffer = stackalloc byte[4];
-		BinaryPrimitives.WriteUInt32LittleEndian(buffer, sequenceId);
-		destination.Write(buffer);
-	}
-	
-	public static uint ReadSequenceId(ref ReadOnlyMemory<byte> memory) {
-		uint value = BinaryPrimitives.ReadUInt32LittleEndian(memory.Span);
-		memory = memory[4..];
-		return value;
-	}
-}

Utils/Phantom.Utils.Rpc/Message/ReplySender.cs

@@ -1,17 +0,0 @@
-using Phantom.Utils.Rpc.Runtime;
-
-namespace Phantom.Utils.Rpc.Message;
-
-sealed class ReplySender<TMessageBase, TReplyMessage> : IReplySender where TReplyMessage : TMessageBase {
-	private readonly RpcConnection<TMessageBase> connection;
-	private readonly IReplyMessageFactory<TReplyMessage> replyMessageFactory;
-	
-	public ReplySender(RpcConnection<TMessageBase> connection, IReplyMessageFactory<TReplyMessage> replyMessageFactory) {
-		this.connection = connection;
-		this.replyMessageFactory = replyMessageFactory;
-	}
-	
-	public Task SendReply(uint sequenceId, byte[] serializedReply) {
-		return connection.Send(replyMessageFactory.CreateReplyMessage(sequenceId, serializedReply));
-	}
-}

Utils/Phantom.Utils.Rpc/RpcConfiguration.cs

@@ -1,8 +0,0 @@
-using NetMQ;
-
-namespace Phantom.Utils.Rpc;
-
-public sealed record RpcConfiguration(string ServiceName, string Host, ushort Port, NetMQCertificate ServerCertificate) {
-	internal string LoggerName => "Rpc:" + ServiceName;
-	internal string TcpUrl => "tcp://" + Host + ":" + Port;
-}

Utils/Phantom.Utils.Rpc/RpcExtensions.cs

@@ -1,32 +0,0 @@
-using NetMQ;
-using NetMQ.Sockets;
-
-namespace Phantom.Utils.Rpc;
-
-static class RpcExtensions {
-	public static ReadOnlyMemory<byte> Receive(this ClientSocket socket, CancellationToken cancellationToken) {
-		var msg = new Msg();
-		msg.InitEmpty();
-		
-		try {
-			socket.Receive(ref msg, cancellationToken);
-			return msg.SliceAsMemory();
-		} finally {
-			// Only releases references, so the returned ReadOnlyMemory is safe.
-			msg.Close();
-		}
-	}
-	
-	public static (uint, ReadOnlyMemory<byte>) Receive(this ServerSocket socket, CancellationToken cancellationToken) {
-		var msg = new Msg();
-		msg.InitEmpty();
-		
-		try {
-			socket.Receive(ref msg, cancellationToken);
-			return (msg.RoutingId, msg.SliceAsMemory());
-		} finally {
-			// Only releases references, so the returned ReadOnlyMemory is safe.
-			msg.Close();
-		}
-	}
-}

Utils/Phantom.Utils.Rpc/Runtime/IRpcConnectionProvider.cs

@@ -0,0 +1,5 @@
+namespace Phantom.Utils.Rpc.Runtime;
+
+interface IRpcConnectionProvider {
+	Task<Stream> GetStream();
+}

Utils/Phantom.Utils.Rpc/Runtime/RpcClient.cs

@@ -0,0 +1,98 @@
+using Phantom.Utils.Actor;
+using Phantom.Utils.Logging;
+using Phantom.Utils.Rpc.Frame;
+using Phantom.Utils.Rpc.Frame.Types;
+using Phantom.Utils.Rpc.Message;
+using Phantom.Utils.Rpc.Runtime.Tls;
+using Serilog;
+
+namespace Phantom.Utils.Rpc.Runtime;
+
+public sealed class RpcClient<TClientToServerMessage, TServerToClientMessage> : IDisposable {
+	public static async Task<RpcClient<TClientToServerMessage, TServerToClientMessage>?> Connect(string loggerName, string host, ushort port, string distinguishedName, RpcCertificateThumbprint certificateThumbprint, RpcClientHandshake handshake, IMessageDefinitions<TClientToServerMessage, TServerToClientMessage> messageDefinitions, CancellationToken cancellationToken) {
+		RpcClientConnector connector = new RpcClientConnector(loggerName, host, port, distinguishedName, certificateThumbprint, handshake);
+		RpcClientConnector.Connection? connection = await connector.EstablishNewConnection(cancellationToken);
+		return connection == null ? null : new RpcClient<TClientToServerMessage, TServerToClientMessage>(loggerName, messageDefinitions, connector, connection);
+	}
+	
+	private readonly ILogger logger;
+	private readonly MessageRegistry<TServerToClientMessage> serverToClientMessageRegistry;
+	private readonly RpcClientConnection connection;
+	
+	public RpcSendChannel<TClientToServerMessage> SendChannel { get; }
+	
+	private RpcClient(string loggerName, IMessageDefinitions<TClientToServerMessage, TServerToClientMessage> messageDefinitions, RpcClientConnector connector, RpcClientConnector.Connection connection) {
+		this.logger = PhantomLogger.Create<RpcClient<TClientToServerMessage, TServerToClientMessage>>(loggerName);
+		this.serverToClientMessageRegistry = messageDefinitions.ToClient;
+		
+		this.connection = new RpcClientConnection(loggerName, connector, connection);
+		this.SendChannel = new RpcSendChannel<TClientToServerMessage>(loggerName, this.connection, messageDefinitions.ToServer, sendQueueCapacity: 500);
+	}
+	
+	public async Task Listen(ActorRef<TServerToClientMessage> actor) {
+		try {
+			await connection.ReadConnection(stream => Receive(stream, new MessageHandlerImpl(SendChannel, actor)));
+		} catch (OperationCanceledException) {
+			// Ignore.
+		}
+	}
+	
+	private async Task Receive(Stream stream, MessageHandlerImpl handler) {
+		await IFrame.ReadFrom(stream, new FrameReader(this, handler), CancellationToken.None);
+	}
+	
+	private sealed class FrameReader(RpcClient<TClientToServerMessage, TServerToClientMessage> client, MessageHandlerImpl handler) : IFrameReader {
+		public Task OnMessageFrame(MessageFrame frame, Stream stream, CancellationToken cancellationToken) {
+			return client.serverToClientMessageRegistry.Handle(frame, handler, cancellationToken);
+		}
+		
+		public void OnReplyFrame(ReplyFrame frame) {
+			client.SendChannel.ReceiveReply(frame);
+		}
+		
+		public void OnErrorFrame(ErrorFrame frame) {
+			client.SendChannel.ReceiveError(frame.ReplyingToMessageId, frame.Error);
+		}
+		
+		public void OnUnknownFrameStart(byte id) {
+			client.logger.Error("Received unknown frame ID: {Id}", id);
+		}
+	}
+	
+	private sealed class MessageHandlerImpl(RpcSendChannel<TClientToServerMessage> sendChannel, ActorRef<TServerToClientMessage> actor) : MessageHandler<TServerToClientMessage> {
+		public ActorRef<TServerToClientMessage> Actor => actor;
+		
+		public ValueTask OnReply<TMessage, TReply>(uint messageId, TReply reply, CancellationToken cancellationToken) where TMessage : TServerToClientMessage, ICanReply<TReply> {
+			return sendChannel.SendReply(messageId, reply, cancellationToken);
+		}
+		
+		public ValueTask OnError(uint messageId, RpcError error, CancellationToken cancellationToken) {
+			return sendChannel.SendError(messageId, error, cancellationToken);
+		}
+	}
+	
+	public async Task Shutdown() {
+		logger.Information("Shutting down client...");
+		
+		try {
+			await SendChannel.Close();
+		} catch (Exception e) {
+			logger.Error(e, "Caught exception while closing send channel.");
+		}
+		
+		try {
+			connection.Close();
+		} catch (Exception e) {
+			logger.Error(e, "Caught exception while closing connection.");
+		}
+		
+		// TODO disconnection handshake?
+		
+		logger.Information("Client shut down.");
+	}
+	
+	public void Dispose() {
+		connection.Dispose();
+		SendChannel.Dispose();
+	}
+}

Utils/Phantom.Utils.Rpc/Runtime/RpcClientConnection.cs

@@ -0,0 +1,85 @@
+using Phantom.Utils.Logging;
+using Serilog;
+
+namespace Phantom.Utils.Rpc.Runtime;
+
+sealed class RpcClientConnection(string loggerName, RpcClientConnector connector, RpcClientConnector.Connection initialConnection) : IRpcConnectionProvider, IDisposable {
+	private readonly ILogger logger = PhantomLogger.Create<RpcClientConnection>(loggerName);
+	
+	private readonly SemaphoreSlim semaphore = new (1);
+	private RpcClientConnector.Connection currentConnection = initialConnection;
+	
+	private readonly CancellationTokenSource newConnectionCancellationTokenSource = new ();
+	
+	public async Task<Stream> GetStream() {
+		return (await GetConnection()).Stream;
+	}
+	
+	private async Task<RpcClientConnector.Connection> GetConnection() {
+		CancellationToken cancellationToken = newConnectionCancellationTokenSource.Token;
+		
+		await semaphore.WaitAsync(cancellationToken);
+		try {
+			if (!currentConnection.Socket.Connected) {
+				currentConnection = await connector.EstablishNewConnectionWithRetry(cancellationToken);
+			}
+			
+			return currentConnection;
+		} finally {
+			semaphore.Release();
+		}
+	}
+	
+	public async Task ReadConnection(Func<Stream, Task> reader) {
+		RpcClientConnector.Connection? connection = null;
+		
+		try {
+			while (true) {
+				connection?.Dispose();
+				connection = null;
+				
+				try {
+					connection = await GetConnection();
+				} catch (OperationCanceledException) {
+					throw;
+				} catch (Exception e) {
+					logger.Warning(e, "Could not obtain a new connection.");
+					continue;
+				}
+				
+				try {
+					await reader(connection.Stream);
+				} catch (OperationCanceledException) {
+					throw;
+				} catch (EndOfStreamException) {
+					logger.Warning("Socket was closed.");
+				} catch (Exception e) {
+					logger.Error(e, "Closing socket due to an exception while reading it.");
+					
+					try {
+						await connection.Shutdown();
+					} catch (Exception e2) {
+						logger.Error(e2, "Caught exception closing the socket.");
+					}
+				}
+			}
+		} finally {
+			if (connection != null) {
+				try {
+					await connection.Disconnect(); // TODO what happens if already disconnected?
+				} finally {
+					connection.Dispose();
+				}
+			}
+		}
+	}
+	
+	public void Close() {
+		newConnectionCancellationTokenSource.Cancel();
+	}
+	
+	public void Dispose() {
+		semaphore.Dispose();
+		newConnectionCancellationTokenSource.Dispose();
+	}
+}

Utils/Phantom.Utils.Rpc/Runtime/RpcClientConnectionClosedEventArgs.cs

@@ -1,9 +0,0 @@
-namespace Phantom.Utils.Rpc.Runtime;
-
-sealed class RpcClientConnectionClosedEventArgs : EventArgs {
-	internal uint RoutingId { get; }
-	
-	internal RpcClientConnectionClosedEventArgs(uint routingId) {
-		RoutingId = routingId;
-	}
-}

Utils/Phantom.Utils.Rpc/Runtime/RpcClientConnector.cs

@@ -0,0 +1,173 @@
+using System.Net.Security;
+using System.Net.Sockets;
+using System.Security.Authentication;
+using System.Security.Cryptography.X509Certificates;
+using Phantom.Utils.Collections;
+using Phantom.Utils.Logging;
+using Phantom.Utils.Rpc.Runtime.Tls;
+using Serilog;
+
+namespace Phantom.Utils.Rpc.Runtime;
+
+internal sealed class RpcClientConnector {
+	private static readonly TimeSpan InitialRetryDelay = TimeSpan.FromMilliseconds(100);
+	private static readonly TimeSpan MaximumRetryDelay = TimeSpan.FromSeconds(30);
+	private static readonly TimeSpan DisconnectTimeout = TimeSpan.FromSeconds(10);
+	
+	private readonly ILogger logger;
+	private readonly string host;
+	private readonly ushort port;
+	private readonly RpcCertificateThumbprint certificateThumbprint;
+	private readonly RpcClientHandshake handshake;
+	private readonly SslClientAuthenticationOptions sslOptions;
+	
+	private bool loggedCertificateValidationError = false;
+	
+	public RpcClientConnector(string loggerName, string host, ushort port, string distinguishedName, RpcCertificateThumbprint certificateThumbprint, RpcClientHandshake handshake) {
+		this.logger = PhantomLogger.Create<RpcClientConnector>(loggerName);
+		this.host = host;
+		this.port = port;
+		this.certificateThumbprint = certificateThumbprint;
+		this.handshake = handshake;
+		
+		this.sslOptions = new SslClientAuthenticationOptions {
+			AllowRenegotiation = false,
+			AllowTlsResume = true,
+			CertificateRevocationCheckMode = X509RevocationMode.NoCheck,
+			EnabledSslProtocols = TlsSupport.SupportedProtocols,
+			EncryptionPolicy = EncryptionPolicy.RequireEncryption,
+			RemoteCertificateValidationCallback = ValidateServerCertificate,
+			TargetHost = distinguishedName,
+		};
+	}
+	
+	internal async Task<Connection> EstablishNewConnectionWithRetry(CancellationToken cancellationToken) {
+		TimeSpan nextAttemptDelay = InitialRetryDelay;
+		
+		while (true) {
+			Connection? newConnection;
+			try {
+				newConnection = await EstablishNewConnection(cancellationToken);
+			} catch (Exception e) {
+				logger.Error(e, "Caught unhandled exception while connecting.");
+				newConnection = null;
+			}
+			
+			if (newConnection != null) {
+				return newConnection;
+			}
+			
+			logger.Warning("Failed to connect to server, trying again in {}.", nextAttemptDelay.TotalSeconds.ToString("F1"));
+			
+			await Task.Delay(nextAttemptDelay, cancellationToken);
+			nextAttemptDelay = Comparables.Min(nextAttemptDelay.Multiply(1.5), MaximumRetryDelay);
+		}
+	}
+	
+	internal async Task<Connection?> EstablishNewConnection(CancellationToken cancellationToken) {
+		logger.Information("Connecting to {Host}:{Port}...", host, port);
+		
+		Socket clientSocket = new Socket(SocketType.Stream, ProtocolType.Tcp);
+		try {
+			await clientSocket.ConnectAsync(host, port, cancellationToken);
+		} catch (Exception e) {
+			logger.Error(e, "Could not connect.");
+			throw;
+		}
+		
+		SslStream? stream = null;
+		bool handledException = false;
+		try {
+			stream = new SslStream(new NetworkStream(clientSocket, ownsSocket: false), leaveInnerStreamOpen: false);
+			
+			try {
+				loggedCertificateValidationError = false;
+				await stream.AuthenticateAsClientAsync(sslOptions, cancellationToken);
+			} catch (AuthenticationException e) {
+				if (!loggedCertificateValidationError) {
+					logger.Error(e, "Could not establish a secure connection.");
+				}
+				
+				handledException = true;
+				throw;
+			}
+			
+			logger.Information("Established a secure connection.");
+			
+			try {
+				await handshake.AcceptServer(stream, cancellationToken);
+			} catch (EndOfStreamException) {
+				logger.Warning("Could not perform application handshake, connection lost.");
+				handledException = true;
+				throw;
+			} catch (Exception e) {
+				logger.Warning(e, "Could not perform application handshake.");
+				handledException = true;
+				throw;
+			}
+			
+			return new Connection(clientSocket, stream);
+		} catch (Exception e) {
+			if (!handledException) {
+				logger.Error(e, "Caught unhandled exception.");
+			}
+			
+			try {
+				await DisconnectSocket(clientSocket, stream);
+			} finally {
+				clientSocket.Close();
+			}
+			
+			return null;
+		}
+	}
+	
+	private bool ValidateServerCertificate(object sender, X509Certificate? certificate, X509Chain? chain, SslPolicyErrors sslPolicyErrors) {
+		if (certificate == null || sslPolicyErrors.HasFlag(SslPolicyErrors.RemoteCertificateNotAvailable)) {
+			logger.Error("Could not establish a secure connection, server did not provide a certificate.");
+		}
+		else if (sslPolicyErrors.HasFlag(SslPolicyErrors.RemoteCertificateNameMismatch)) {
+			logger.Error("Could not establish a secure connection, server certificate has the wrong name: {Name}", certificate.Subject);
+		}
+		else if (!certificateThumbprint.Check(certificate)) {
+			logger.Error("Could not establish a secure connection, server certificate does not match.");
+		}
+		else if (TlsSupport.CheckAlgorithm((X509Certificate2) certificate) is {} error) {
+			logger.Error("Could not establish a secure connection, server certificate rejected because it uses {ActualAlgorithmName} instead of {ExpectedAlgorithmName}.", error.ActualAlgorithmName, error.ExpectedAlgorithmName);
+		}
+		else if ((sslPolicyErrors & ~SslPolicyErrors.RemoteCertificateChainErrors) != SslPolicyErrors.None) {
+			logger.Error("Could not establish a secure connection, server certificate validation failed.");
+		}
+		else {
+			return true;
+		}
+		
+		loggedCertificateValidationError = true;
+		return false;
+	}
+	
+	private static async Task DisconnectSocket(Socket socket, Stream? stream) {
+		if (stream != null) {
+			await stream.DisposeAsync();
+		}
+		
+		using CancellationTokenSource timeoutTokenSource = new CancellationTokenSource(DisconnectTimeout);
+		await socket.DisconnectAsync(reuseSocket: false, timeoutTokenSource.Token);
+	}
+	
+	internal sealed record Connection(Socket Socket, Stream Stream) : IDisposable {
+		public async Task Disconnect() {
+			await DisconnectSocket(Socket, Stream);
+		}
+		
+		public async ValueTask Shutdown() {
+			await Stream.DisposeAsync();
+			Socket.Shutdown(SocketShutdown.Both);
+		}
+		
+		public void Dispose() {
+			Stream.Dispose();
+			Socket.Close();
+		}
+	}
+}

Utils/Phantom.Utils.Rpc/Runtime/RpcClientHandshake.cs

@@ -0,0 +1,5 @@
+namespace Phantom.Utils.Rpc.Runtime;
+
+public abstract class RpcClientHandshake {
+	protected internal abstract Task<bool> AcceptServer(Stream stream, CancellationToken cancellationToken);
+}

Utils/Phantom.Utils.Rpc/Runtime/RpcClientRuntime.cs

@@ -1,72 +0,0 @@
-using NetMQ.Sockets;
-using Phantom.Utils.Actor;
-using Phantom.Utils.Rpc.Message;
-using Phantom.Utils.Rpc.Sockets;
-using Serilog;
-using Serilog.Events;
-
-namespace Phantom.Utils.Rpc.Runtime;
-
-public abstract class RpcClientRuntime<TClientMessage, TServerMessage, TReplyMessage> : RpcRuntime<ClientSocket> where TReplyMessage : TClientMessage, TServerMessage {
-	private readonly RpcConnectionToServer<TServerMessage> connection;
-	private readonly IMessageDefinitions<TClientMessage, TServerMessage, TReplyMessage> messageDefinitions;
-	private readonly ActorRef<TClientMessage> handlerActor;
-	
-	private readonly SemaphoreSlim disconnectSemaphore;
-	private readonly CancellationToken receiveCancellationToken;
-	
-	protected RpcClientRuntime(RpcClientSocket<TClientMessage, TServerMessage, TReplyMessage> socket, ActorRef<TClientMessage> handlerActor, SemaphoreSlim disconnectSemaphore, CancellationToken receiveCancellationToken) : base(socket) {
-		this.connection = socket.Connection;
-		this.messageDefinitions = socket.MessageDefinitions;
-		this.handlerActor = handlerActor;
-		this.disconnectSemaphore = disconnectSemaphore;
-		this.receiveCancellationToken = receiveCancellationToken;
-	}
-	
-	private protected sealed override Task Run(ClientSocket socket) {
-		return RunWithConnection(socket, connection);
-	}
-	
-	protected virtual async Task RunWithConnection(ClientSocket socket, RpcConnectionToServer<TServerMessage> connection) {
-		var replySender = new ReplySender<TServerMessage, TReplyMessage>(connection, messageDefinitions);
-		var messageHandler = new MessageHandler<TClientMessage>(LoggerName, handlerActor, replySender);
-		
-		try {
-			while (!receiveCancellationToken.IsCancellationRequested) {
-				var data = socket.Receive(receiveCancellationToken);
-				
-				LogMessageType(RuntimeLogger, data);
-				
-				if (data.Length > 0) {
-					messageDefinitions.ToClient.Handle(data, messageHandler);
-				}
-			}
-		} catch (OperationCanceledException) {
-			// Ignore.
-		} finally {
-			await handlerActor.Stop();
-			RuntimeLogger.Debug("ZeroMQ client stopped receiving messages.");
-			
-			await disconnectSemaphore.WaitAsync(CancellationToken.None);
-		}
-	}
-	
-	private protected sealed override async Task Disconnect(ClientSocket socket) {
-		await SendDisconnectMessage(socket, RuntimeLogger);
-	}
-	
-	protected abstract Task SendDisconnectMessage(ClientSocket socket, ILogger logger);
-	
-	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);
-		}
-	}
-}

Utils/Phantom.Utils.Rpc/Runtime/RpcConnection.cs

@@ -1,40 +0,0 @@
-using Phantom.Utils.Actor;
-using Phantom.Utils.Rpc.Message;
-
-namespace Phantom.Utils.Rpc.Runtime;
-
-public abstract class RpcConnection<TMessageBase> {
-	private readonly MessageRegistry<TMessageBase> messageRegistry;
-	private readonly MessageReplyTracker replyTracker;
-	
-	internal RpcConnection(MessageRegistry<TMessageBase> messageRegistry, MessageReplyTracker replyTracker) {
-		this.messageRegistry = messageRegistry;
-		this.replyTracker = replyTracker;
-	}
-	
-	private protected abstract ValueTask Send(byte[] bytes);
-	
-	public async Task Send<TMessage>(TMessage message) where TMessage : TMessageBase {
-		var bytes = messageRegistry.Write(message).ToArray();
-		if (bytes.Length > 0) {
-			await Send(bytes);
-		}
-	}
-	
-	public async Task<TReply> Send<TMessage, TReply>(TMessage message, TimeSpan waitForReplyTime, CancellationToken waitForReplyCancellationToken) where TMessage : TMessageBase, ICanReply<TReply> {
-		var sequenceId = replyTracker.RegisterReply();
-		
-		var bytes = messageRegistry.Write<TMessage, TReply>(sequenceId, message).ToArray();
-		if (bytes.Length == 0) {
-			replyTracker.ForgetReply(sequenceId);
-			throw new ArgumentException("Could not write message.", nameof(message));
-		}
-		
-		await Send(bytes);
-		return await replyTracker.WaitForReply<TReply>(sequenceId, waitForReplyTime, waitForReplyCancellationToken);
-	}
-	
-	public void Receive(IReply message) {
-		replyTracker.ReceiveReply(message.SequenceId, message.SerializedReply);
-	}
-}

Utils/Phantom.Utils.Rpc/Runtime/RpcConnectionToClient.cs

@@ -1,41 +0,0 @@
-using NetMQ;
-using NetMQ.Sockets;
-using Phantom.Utils.Rpc.Message;
-
-namespace Phantom.Utils.Rpc.Runtime;
-
-public sealed class RpcConnectionToClient<TMessageBase> : RpcConnection<TMessageBase> {
-	private readonly ServerSocket socket;
-	private readonly uint routingId;
-	
-	internal event EventHandler<RpcClientConnectionClosedEventArgs>? Closed;
-	private bool isClosed;
-	
-	internal RpcConnectionToClient(ServerSocket socket, uint routingId, MessageRegistry<TMessageBase> messageRegistry, MessageReplyTracker replyTracker) : base(messageRegistry, replyTracker) {
-		this.socket = socket;
-		this.routingId = routingId;
-	}
-	
-	public bool IsSame(RpcConnectionToClient<TMessageBase> other) {
-		return this.routingId == other.routingId && this.socket == other.socket;
-	}
-	
-	public void Close() {
-		bool hasClosed = false;
-		
-		lock (this) {
-			if (!isClosed) {
-				isClosed = true;
-				hasClosed = true;
-			}
-		}
-		
-		if (hasClosed) {
-			Closed?.Invoke(this, new RpcClientConnectionClosedEventArgs(routingId));
-		}
-	}
-	
-	private protected override ValueTask Send(byte[] bytes) {
-		return socket.SendAsync(routingId, bytes);
-	}
-}

Utils/Phantom.Utils.Rpc/Runtime/RpcConnectionToServer.cs

@@ -1,25 +0,0 @@
-using NetMQ;
-using NetMQ.Sockets;
-using Phantom.Utils.Rpc.Message;
-using Phantom.Utils.Tasks;
-
-namespace Phantom.Utils.Rpc.Runtime;
-
-public sealed class RpcConnectionToServer<TMessageBase> : RpcConnection<TMessageBase> {
-	private readonly ClientSocket socket;
-	private readonly TaskCompletionSource isReady = AsyncTasks.CreateCompletionSource();
-	
-	public Task IsReady => isReady.Task;
-	
-	internal RpcConnectionToServer(ClientSocket socket, MessageRegistry<TMessageBase> messageRegistry, MessageReplyTracker replyTracker) : base(messageRegistry, replyTracker) {
-		this.socket = socket;
-	}
-	
-	public void SetIsReady() {
-		isReady.TrySetResult();
-	}
-	
-	private protected override ValueTask Send(byte[] bytes) {
-		return socket.SendAsync(bytes);
-	}
-}

Utils/Phantom.Utils.Rpc/Runtime/RpcError.cs

@@ -0,0 +1,9 @@
+namespace Phantom.Utils.Rpc.Runtime;
+
+public enum RpcError : byte {
+	InvalidData = 0,
+	UnknownMessageRegistryCode = 1,
+	MessageDeserializationError = 2,
+	MessageHandlingError = 3,
+	MessageTooLarge = 4,
+}

Utils/Phantom.Utils.Rpc/Runtime/RpcErrorException.cs

@@ -0,0 +1,5 @@
+namespace Phantom.Utils.Rpc.Runtime;
+
+public sealed class RpcErrorException(string message, RpcError error) : Exception(message) {
+	public RpcError Error => error;
+}

Utils/Phantom.Utils.Rpc/Runtime/RpcHandshakeResult.cs

@@ -0,0 +1,6 @@
+namespace Phantom.Utils.Rpc.Runtime;
+
+public enum RpcHandshakeResult : byte {
+	UnknownError = 0,
+	InvalidFormat = 1,
+}

Utils/Phantom.Utils.Rpc/Runtime/RpcReceiveChannel.cs

@@ -0,0 +1,5 @@
+namespace Phantom.Utils.Rpc.Runtime;
+
+sealed class RpcReceiveChannel {
+	
+}

Utils/Phantom.Utils.Rpc/Runtime/RpcReceiverActor.cs

@@ -1,75 +0,0 @@
-using Akka.Actor;
-using Akka.Event;
-using Phantom.Utils.Actor;
-using Phantom.Utils.Rpc.Message;
-
-namespace Phantom.Utils.Rpc.Runtime;
-
-sealed class RpcReceiverActor<TClientMessage, TServerMessage, TRegistrationMessage, TReplyMessage> : ReceiveActor<RpcReceiverActor<TClientMessage, TServerMessage, TRegistrationMessage, TReplyMessage>.ReceiveMessageCommand>, IWithStash where TRegistrationMessage : TServerMessage where TReplyMessage : TClientMessage, TServerMessage {
-	public readonly record struct Init(
-		string LoggerName,
-		IMessageDefinitions<TClientMessage, TServerMessage, TReplyMessage> MessageDefinitions,
-		IRegistrationHandler<TClientMessage, TServerMessage, TRegistrationMessage> RegistrationHandler,
-		RpcConnectionToClient<TClientMessage> Connection
-	);
-	
-	public static Props<ReceiveMessageCommand> Factory(Init init) {
-		return Props<ReceiveMessageCommand>.Create(() => new RpcReceiverActor<TClientMessage, TServerMessage, TRegistrationMessage, TReplyMessage>(init), new ActorConfiguration {
-			SupervisorStrategy = SupervisorStrategies.Resume,
-			StashCapacity = 100
-		});
-	}
-	
-	public IStash Stash { get; set; } = null!;
-	
-	private readonly string loggerName;
-	private readonly IMessageDefinitions<TClientMessage, TServerMessage, TReplyMessage> messageDefinitions;
-	private readonly IRegistrationHandler<TClientMessage, TServerMessage, TRegistrationMessage> registrationHandler;
-	private readonly RpcConnectionToClient<TClientMessage> connection;
-	
-	private RpcReceiverActor(Init init) {
-		this.loggerName = init.LoggerName;
-		this.messageDefinitions = init.MessageDefinitions;
-		this.registrationHandler = init.RegistrationHandler;
-		this.connection = init.Connection;
-		
-		ReceiveAsync<ReceiveMessageCommand>(ReceiveMessageUnauthorized);
-	}
-	
-	public sealed record ReceiveMessageCommand(Type MessageType, ReadOnlyMemory<byte> Data);
-	
-	private async Task ReceiveMessageUnauthorized(ReceiveMessageCommand command) {
-		if (command.MessageType == typeof(TRegistrationMessage)) {
-			await HandleRegistrationMessage(command);
-		}
-		else if (Stash.IsFull) {
-			Context.GetLogger().Warning("Stash is full, dropping message: {MessageType}", command.MessageType);
-		}
-		else {
-			Stash.Stash();
-		}
-	}
-	
-	private async Task HandleRegistrationMessage(ReceiveMessageCommand command) {
-		if (!messageDefinitions.ToServer.Read(command.Data, out TRegistrationMessage message)) {
-			return;
-		}
-		
-		var props = await registrationHandler.TryRegister(connection, message);
-		if (props == null) {
-			return;
-		}
-		
-		var handlerActor = Context.ActorOf(props, "Handler");
-		var replySender = new ReplySender<TClientMessage, TReplyMessage>(connection, messageDefinitions);
-		BecomeAuthorized(new MessageHandler<TServerMessage>(loggerName, handlerActor, replySender));
-	}
-	
-	private void BecomeAuthorized(MessageHandler<TServerMessage> handler) {
-		Stash.UnstashAll();
-		
-		Become(() => {
-			Receive<ReceiveMessageCommand>(command => messageDefinitions.ToServer.Handle(command.Data, handler));
-		});
-	}
-}

Utils/Phantom.Utils.Rpc/Runtime/RpcRuntime.cs

@@ -1,50 +0,0 @@
-using System.Diagnostics.CodeAnalysis;
-using NetMQ;
-using Phantom.Utils.Logging;
-using Phantom.Utils.Rpc.Message;
-using Phantom.Utils.Rpc.Sockets;
-using Serilog;
-
-namespace Phantom.Utils.Rpc.Runtime;
-
-public abstract class RpcRuntime<TSocket> where TSocket : ThreadSafeSocket {
-	private readonly TSocket socket;
-	
-	private protected string LoggerName { get; }
-	private protected ILogger RuntimeLogger { get; }
-	private protected MessageReplyTracker ReplyTracker { get; }
-	
-	protected RpcRuntime(RpcSocket<TSocket> socket) {
-		this.socket = socket.Socket;
-		
-		this.LoggerName = socket.Config.LoggerName;
-		this.RuntimeLogger = PhantomLogger.Create(LoggerName);
-		this.ReplyTracker = socket.ReplyTracker;
-	}
-	
-	protected async Task Launch() {
-		[SuppressMessage("ReSharper", "AccessToDisposedClosure")]
-		async Task RunTask() {
-			try {
-				await Run(socket);
-			} catch (Exception e) {
-				RuntimeLogger.Error(e, "Caught exception in RPC thread.");
-			}
-		}
-		
-		try {
-			await Task.Factory.StartNew(RunTask, CancellationToken.None, TaskCreationOptions.LongRunning, TaskScheduler.Default).Unwrap();
-		} catch (OperationCanceledException) {
-			// Ignore.
-		} finally {
-			await Disconnect(socket);
-			
-			socket.Dispose();
-			RuntimeLogger.Information("ZeroMQ runtime stopped.");
-		}
-	}
-	
-	private protected abstract Task Run(TSocket socket);
-	
-	private protected abstract Task Disconnect(TSocket socket);
-}

Utils/Phantom.Utils.Rpc/Runtime/RpcSendChannel.cs

@@ -0,0 +1,128 @@
+using System.Threading.Channels;
+using Phantom.Utils.Actor;
+using Phantom.Utils.Rpc.Frame;
+using Phantom.Utils.Rpc.Frame.Types;
+using Phantom.Utils.Rpc.Message;
+using Phantom.Utils.Rpc.Runtime.Utils;
+
+namespace Phantom.Utils.Rpc.Runtime;
+
+public sealed class RpcSendChannel<TMessageBase> : IDisposable {
+	private readonly IRpcConnectionProvider connectionProvider;
+	private readonly MessageRegistry<TMessageBase> messageRegistry;
+	private readonly MessageReplyTracker messageReplyTracker;
+	
+	private readonly Channel<IFrame> sendQueue;
+	private readonly Task sendQueueTask;
+	
+	private readonly CancellationTokenSource cancellationTokenSource = new ();
+	
+	private uint nextMessageId;
+	
+	internal RpcSendChannel(string loggerName, IRpcConnectionProvider connectionProvider, MessageRegistry<TMessageBase> messageRegistry, int sendQueueCapacity) {
+		this.connectionProvider = connectionProvider;
+		this.messageRegistry = messageRegistry;
+		this.messageReplyTracker = new MessageReplyTracker(loggerName);
+		
+		this.sendQueue = Channel.CreateBounded<IFrame>(new BoundedChannelOptions(sendQueueCapacity) {
+			AllowSynchronousContinuations = false,
+			FullMode = BoundedChannelFullMode.Wait,
+			SingleReader = true,
+			SingleWriter = false,
+		});
+		
+		this.sendQueueTask = ProcessSendQueue();
+	}
+	
+	public bool TrySendMessage<TMessage>(TMessage message) where TMessage : TMessageBase {
+		return sendQueue.Writer.TryWrite(NextMessageFrame(message));
+	}
+	
+	public async ValueTask SendMessage<TMessage>(TMessage message, CancellationToken cancellationToken) where TMessage : TMessageBase {
+		await SendFrame(NextMessageFrame(message), cancellationToken);
+	}
+	
+	public async Task<TReply> SendMessage<TMessage, TReply>(TMessage message, TimeSpan waitForReplyTime, CancellationToken cancellationToken) where TMessage : TMessageBase, ICanReply<TReply> {
+		MessageFrame frame = NextMessageFrame(message);
+		uint messageId = frame.MessageId;
+		
+		messageReplyTracker.RegisterReply(messageId);
+		try {
+			await SendFrame(frame, cancellationToken);
+		} catch (Exception) {
+			messageReplyTracker.ForgetReply(messageId);
+			throw;
+		}
+		
+		return await messageReplyTracker.WaitForReply<TReply>(messageId, waitForReplyTime, cancellationToken);
+	}
+	
+	internal async ValueTask SendReply<TReply>(uint replyingToMessageId, TReply reply, CancellationToken cancellationToken) {
+		await SendFrame(new ReplyFrame(replyingToMessageId, Serialization.Serialize(reply)), cancellationToken);
+	}
+	
+	internal async ValueTask SendError(uint replyingToMessageId, RpcError error, CancellationToken cancellationToken) {
+		await SendFrame(new ErrorFrame(replyingToMessageId, error), cancellationToken);
+	}
+	
+	private async ValueTask SendFrame(IFrame frame, CancellationToken cancellationToken) {
+		if (!sendQueue.Writer.TryWrite(frame)) {
+			await sendQueue.Writer.WriteAsync(frame, cancellationToken);
+		}
+	}
+	
+	private MessageFrame NextMessageFrame<T>(T message) where T : TMessageBase {
+		uint messageId = Interlocked.Increment(ref nextMessageId);
+		return messageRegistry.CreateFrame(messageId, message);
+	}
+	
+	private async Task ProcessSendQueue() {
+		CancellationToken cancellationToken = cancellationTokenSource.Token;
+		
+		// TODO figure out cancellation
+		await foreach (IFrame frame in sendQueue.Reader.ReadAllAsync(cancellationToken)) {
+			while (!cancellationToken.IsCancellationRequested) {
+				Stream stream;
+				try {
+					stream = await connectionProvider.GetStream();
+				} catch (OperationCanceledException) {
+					throw;
+				} catch (Exception) {
+					continue;
+				}
+				
+				await stream.WriteAsync(frame.Type, cancellationToken);
+				await frame.Write(stream, cancellationToken);
+				await stream.FlushAsync(cancellationToken);
+			}
+		}
+	}
+	
+	internal void ReceiveReply(ReplyFrame frame) {
+		messageReplyTracker.ReceiveReply(frame.ReplyingToMessageId, frame.SerializedReply);
+	}
+	
+	internal void ReceiveError(uint messageId, RpcError error) {
+		messageReplyTracker.FailReply(messageId, error switch {
+			RpcError.UnknownMessageRegistryCode  => new RpcErrorException("Unknown message registry code", error),
+			RpcError.MessageDeserializationError => new RpcErrorException("Message deserialization error", error),
+			RpcError.MessageHandlingError        => new RpcErrorException("Message handling error", error),
+			_                                    => new RpcErrorException("Unknown error", error),
+		});
+	}
+	
+	internal async Task Close() {
+		sendQueue.Writer.TryComplete();
+		
+		try {
+			await sendQueueTask;
+		} catch (Exception) {
+			// Ignore.
+		}
+	}
+	
+	public void Dispose() {
+		sendQueueTask.Dispose();
+		cancellationTokenSource.Dispose();
+	}
+}

Utils/Phantom.Utils.Rpc/Runtime/RpcServer.cs

@@ -0,0 +1,133 @@
+using System.Net;
+using System.Net.Security;
+using System.Net.Sockets;
+using System.Security.Cryptography.X509Certificates;
+using Phantom.Utils.Logging;
+using Phantom.Utils.Rpc.Runtime.Tls;
+using Serilog;
+
+namespace Phantom.Utils.Rpc.Runtime;
+
+public sealed class RpcServer(string loggerName, EndPoint endPoint, RpcServerCertificate certificate, RpcServerHandshake handshake) {
+	private readonly ILogger logger = PhantomLogger.Create<RpcServer>(loggerName);
+	private readonly List<Client> clients = [];
+	
+	public async Task<bool> Run(CancellationToken shutdownToken) {
+		SslServerAuthenticationOptions sslOptions = new () {
+			AllowRenegotiation = false,
+			AllowTlsResume = true,
+			CertificateRevocationCheckMode = X509RevocationMode.NoCheck,
+			ClientCertificateRequired = false,
+			EnabledSslProtocols = TlsSupport.SupportedProtocols,
+			EncryptionPolicy = EncryptionPolicy.RequireEncryption,
+			ServerCertificate = certificate.Certificate,
+		};
+		
+		try {
+			using var serverSocket = new Socket(endPoint.AddressFamily, SocketType.Stream, ProtocolType.Tcp);
+			
+			try {
+				serverSocket.Bind(endPoint);
+				serverSocket.Listen(5);
+			} catch (Exception e) {
+				logger.Error(e, "Could not bind to {EndPoint}.", endPoint);
+				return false;
+			}
+			
+			try {
+				logger.Information("Server listening on {EndPoint}.", endPoint);
+				
+				while (!shutdownToken.IsCancellationRequested) {
+					Socket clientSocket = await serverSocket.AcceptAsync(shutdownToken);
+					clients.Add(new Client(logger, clientSocket, sslOptions, handshake, shutdownToken));
+					clients.RemoveAll(static client => client.Task.IsCompleted);
+				}
+			} finally {
+				await Stop(serverSocket);
+			}
+		} catch (Exception e) {
+			logger.Error(e, "Server crashed with uncaught exception.");
+			return false;
+		}
+		
+		return true;
+	}
+	
+	private async Task Stop(Socket serverSocket) {
+		try {
+			serverSocket.Close();
+		} catch (Exception e) {
+			logger.Error(e, "Server socket failed to close.");
+			return;
+		}
+		
+		logger.Information("Server socket closed, waiting for client sockets to close.");
+		
+		try {
+			await Task.WhenAll(clients.Select(static client => client.Task));
+		} catch (Exception) {
+			// Ignore exceptions when shutting down.
+		}
+		
+		logger.Information("Server stopped.");
+	}
+	
+	private sealed class Client {
+		private static readonly TimeSpan DisconnectTimeout = TimeSpan.FromSeconds(10);
+		
+		public Task Task { get; }
+		
+		private readonly ILogger logger;
+		private readonly Socket socket;
+		private readonly SslServerAuthenticationOptions sslOptions;
+		private readonly RpcServerHandshake handshake;
+		private readonly CancellationToken shutdownToken;
+		
+		public Client(ILogger logger, Socket socket, SslServerAuthenticationOptions sslOptions, RpcServerHandshake handshake, CancellationToken shutdownToken) {
+			this.logger = logger;
+			this.socket = socket;
+			this.sslOptions = sslOptions;
+			this.handshake = handshake;
+			this.shutdownToken = shutdownToken;
+			this.Task = Run();
+		}
+		
+		private async Task Run() {
+			try {
+				await using var stream = new SslStream(new NetworkStream(socket, ownsSocket: false), leaveInnerStreamOpen: false);
+				
+				try {
+					await stream.AuthenticateAsServerAsync(sslOptions, shutdownToken);
+				} catch (Exception e) {
+					logger.Error(e, "Could not establish a secure connection.");
+					return;
+				}
+				
+				try {
+					await handshake.AcceptClient(socket.RemoteEndPoint?.ToString() ?? "<unknown address>", stream, shutdownToken);
+				} catch (EndOfStreamException) {
+					logger.Warning("Could not perform application handshake, connection lost.");
+					return;
+				} catch (Exception e) {
+					logger.Warning(e, "Could not perform application handshake.");
+					return;
+				}
+				
+				byte[] buffer = new byte[1024];
+				int readBytes;
+				while ((readBytes = await stream.ReadAsync(buffer, shutdownToken)) > 0) {}
+			} finally {
+				try {
+					using var timeoutTokenSource = new CancellationTokenSource(DisconnectTimeout);
+					await socket.DisconnectAsync(reuseSocket: false, timeoutTokenSource.Token);
+				} catch (OperationCanceledException) {
+					logger.Error("Could not disconnect client socket, disconnection timed out.");
+				} catch (Exception e) {
+					logger.Error(e, "Could not disconnect client socket.");
+				} finally {
+					socket.Close();
+				}
+			}
+		}
+	}
+}

Utils/Phantom.Utils.Rpc/Runtime/RpcServerCertificate.cs

@@ -0,0 +1,28 @@
+using System.Security.Cryptography.X509Certificates;
+using Phantom.Utils.Monads;
+using Phantom.Utils.Rpc.Runtime.Tls;
+
+namespace Phantom.Utils.Rpc.Runtime;
+
+public sealed class RpcServerCertificate {
+	public static byte[] CreateAndExport(string commonName) {
+		var distinguishedNameBuilder = new X500DistinguishedNameBuilder();
+		distinguishedNameBuilder.AddCommonName(commonName);
+		var distinguishedName = distinguishedNameBuilder.Build();
+		
+		using var certificate = TlsSupport.CreateSelfSignedCertificate(distinguishedName);
+		return certificate.Export(X509ContentType.Pkcs12);
+	}
+	
+	public static Either<RpcServerCertificate, DisallowedAlgorithmError> Load(string path) {
+		return TlsSupport.LoadPkcs12FromFile(path).MapLeft(static certificate => new RpcServerCertificate(certificate));
+	}
+	
+	internal X509Certificate2 Certificate { get; }
+	
+	public RpcCertificateThumbprint Thumbprint => RpcCertificateThumbprint.From(Certificate);
+	
+	private RpcServerCertificate(X509Certificate2 certificate) {
+		this.Certificate = certificate;
+	}
+}

Utils/Phantom.Utils.Rpc/Runtime/RpcServerHandshake.cs

@@ -0,0 +1,5 @@
+namespace Phantom.Utils.Rpc.Runtime;
+
+public abstract class RpcServerHandshake {
+	protected internal abstract Task<bool> AcceptClient(string remoteAddress, Stream stream, CancellationToken cancellationToken);
+}

Utils/Phantom.Utils.Rpc/Runtime/Tls/DisallowedAlgorithmError.cs

@@ -0,0 +1,3 @@
+namespace Phantom.Utils.Rpc.Runtime.Tls;
+
+public sealed record DisallowedAlgorithmError(string ExpectedAlgorithmName, string ActualAlgorithmName);

Utils/Phantom.Utils.Rpc/Runtime/Tls/RpcCertificateThumbprint.cs

@@ -0,0 +1,31 @@
+using System.Collections.Immutable;
+using System.Security.Cryptography;
+using System.Security.Cryptography.X509Certificates;
+
+namespace Phantom.Utils.Rpc.Runtime.Tls;
+
+public sealed record RpcCertificateThumbprint {
+	private const int Length = 20;
+	
+	public static RpcCertificateThumbprint From(ReadOnlySpan<byte> bytes) {
+		return new RpcCertificateThumbprint([..bytes]);
+	}
+	
+	internal static RpcCertificateThumbprint From(X509Certificate certificate) {
+		return new RpcCertificateThumbprint([..certificate.GetCertHash()]);
+	}
+	
+	public ImmutableArray<byte> Bytes { get; init; }
+	
+	private RpcCertificateThumbprint(ImmutableArray<byte> bytes) {
+		if (bytes.Length != Length) {
+			throw new ArgumentOutOfRangeException(nameof(bytes), "Invalid thumbprint length: " + bytes.Length + ". Thumbprint length must be exactly " + Length + " bytes.");
+		}
+		
+		this.Bytes = bytes;
+	}
+	
+	internal bool Check(X509Certificate certificate) {
+		return CryptographicOperations.FixedTimeEquals(Bytes.AsSpan(), certificate.GetCertHash());
+	}
+}

Utils/Phantom.Utils.Rpc/Runtime/Tls/TlsSupport.cs

@@ -0,0 +1,56 @@
+using System.Security.Authentication;
+using System.Security.Cryptography;
+using System.Security.Cryptography.X509Certificates;
+using Phantom.Utils.Monads;
+
+namespace Phantom.Utils.Rpc.Runtime.Tls;
+
+/// <summary>
+/// <para>
+/// .NET uses the operating system's native TLS implementation, which is much worse on Windows than it is on Linux.
+/// </para>
+/// <para>
+/// On Linux, the client and server will use TLS 1.3 with ECDSA. On other operating systems, the requirements are reduced for the purposes of local development.
+/// The client and server are not designed to be able to communicate if they run on different operating systems.
+/// </para>
+/// </summary>
+static class TlsSupport {
+	public static SslProtocols SupportedProtocols => OperatingSystem.IsLinux() ? SslProtocols.Tls13 : SslProtocols.None /* OS default */;
+	
+	public static X509Certificate2 CreateSelfSignedCertificate(X500DistinguishedName distinguishedName) {
+		if (OperatingSystem.IsLinux()) {
+			using var keys = ECDsa.Create();
+			return CreateSelfSignedCertificate(new CertificateRequest(distinguishedName, keys, HashAlgorithmName.SHA512));
+		}
+		else {
+			using var keys = RSA.Create(keySizeInBits: 4096);
+			return CreateSelfSignedCertificate(new CertificateRequest(distinguishedName, keys, HashAlgorithmName.SHA384, RSASignaturePadding.Pkcs1));
+		}
+	}
+	
+	private static X509Certificate2 CreateSelfSignedCertificate(CertificateRequest request) {
+		return request.CreateSelfSigned(DateTimeOffset.UtcNow, DateTimeOffset.MaxValue);
+	}
+	
+	public static Either<X509Certificate2, DisallowedAlgorithmError> LoadPkcs12FromFile(string path) {
+		X509KeyStorageFlags storageFlags = OperatingSystem.IsLinux() ? X509KeyStorageFlags.EphemeralKeySet : X509KeyStorageFlags.DefaultKeySet;
+		X509Certificate2 certificate = X509CertificateLoader.LoadPkcs12FromFile(path, password: null, storageFlags);
+		
+		if (CheckAlgorithm(certificate) is {} unsupportedCertificateAlgorithm) {
+			return Either.Right(unsupportedCertificateAlgorithm);
+		}
+		else {
+			return Either.Left(certificate);
+		}
+	}
+	
+	public static DisallowedAlgorithmError? CheckAlgorithm(X509Certificate2 certificate) {
+		if (OperatingSystem.IsLinux() && certificate.GetECDsaPublicKey() == null) {
+			string actualAlgorithm = certificate.GetKeyAlgorithm();
+			Oid actualAlgorithmOid = Oid.FromOidValue(actualAlgorithm, OidGroup.PublicKeyAlgorithm);
+			return new DisallowedAlgorithmError("ECC", actualAlgorithmOid.FriendlyName ?? actualAlgorithm);
+		}
+		
+		return null;
+	}
+}

Utils/Phantom.Utils.Rpc/Runtime/Utils/RentedMemory.cs

@@ -0,0 +1,22 @@
+using System.Buffers;
+
+namespace Phantom.Utils.Rpc.Runtime.Utils;
+
+readonly record struct RentedMemory<T>(T[] Array, int Length) : IDisposable {
+	public Span<T> AsSpan => Array.AsSpan(..Length);
+	public Memory<T> AsMemory => Array.AsMemory(..Length);
+	
+	public void Dispose() {
+		ArrayPool<T>.Shared.Return(Array);
+	}
+	
+	public static RentedMemory<T> Rent(int bytes) {
+		T[] buffer = ArrayPool<T>.Shared.Rent(bytes);
+		try {
+			return new RentedMemory<T>(buffer, bytes);
+		} catch (Exception) {
+			ArrayPool<T>.Shared.Return(buffer);
+			throw;
+		}
+	}
+}

Utils/Phantom.Utils.Rpc/Runtime/Utils/Serialization.cs

@@ -0,0 +1,69 @@
+using System.Buffers;
+using System.Buffers.Binary;
+using MemoryPack;
+
+namespace Phantom.Utils.Rpc.Runtime.Utils;
+
+public static class Serialization {
+	private static readonly MemoryPackSerializerOptions SerializerOptions = MemoryPackSerializerOptions.Utf8;
+	
+	private static async ValueTask WritePrimitive<T>(T value, int size, Action<Span<byte>, T> writer, Stream stream, CancellationToken cancellationToken) {
+		using var buffer = RentedMemory<byte>.Rent(size);
+		writer(buffer.AsSpan, value);
+		await stream.WriteAsync(buffer.AsMemory, cancellationToken);
+	}
+	
+	private static async ValueTask<T> ReadPrimitive<T>(Func<ReadOnlySpan<byte>, T> reader, int size, Stream stream, CancellationToken cancellationToken) {
+		using var buffer = RentedMemory<byte>.Rent(size);
+		await stream.ReadExactlyAsync(buffer.AsMemory, cancellationToken);
+		return reader(buffer.AsSpan);
+	}
+	
+	public static ValueTask WriteByte(byte value, Stream stream, CancellationToken cancellationToken) {
+		return WritePrimitive(value, sizeof(byte), static (span, value) => span[0] = value, stream, cancellationToken);
+	}
+	
+	public static ValueTask<byte> ReadByte(Stream stream, CancellationToken cancellationToken) {
+		return ReadPrimitive(static span => span[0], sizeof(byte), stream, cancellationToken);
+	}
+	
+	public static ValueTask WriteUnsignedShort(ushort value, Stream stream, CancellationToken cancellationToken) {
+		return WritePrimitive(value, sizeof(ushort), BinaryPrimitives.WriteUInt16LittleEndian, stream, cancellationToken);
+	}
+	
+	public static ValueTask<ushort> ReadUnsignedShort(Stream stream, CancellationToken cancellationToken) {
+		return ReadPrimitive(BinaryPrimitives.ReadUInt16LittleEndian, sizeof(ushort), stream, cancellationToken);
+	}
+	
+	public static ValueTask WriteSignedInt(int value, Stream stream, CancellationToken cancellationToken) {
+		return WritePrimitive(value, sizeof(int), BinaryPrimitives.WriteInt32LittleEndian, stream, cancellationToken);
+	}
+	
+	public static ValueTask<int> ReadSignedInt(Stream stream, CancellationToken cancellationToken) {
+		return ReadPrimitive(BinaryPrimitives.ReadInt32LittleEndian, sizeof(int), stream, cancellationToken);
+	}
+	
+	public static ValueTask WriteUnsignedInt(uint value, Stream stream, CancellationToken cancellationToken) {
+		return WritePrimitive(value, sizeof(uint), BinaryPrimitives.WriteUInt32LittleEndian, stream, cancellationToken);
+	}
+	
+	public static ValueTask<uint> ReadUnsignedInt(Stream stream, CancellationToken cancellationToken) {
+		return ReadPrimitive(BinaryPrimitives.ReadUInt32LittleEndian, sizeof(uint), stream, cancellationToken);
+	}
+	
+	public static async ValueTask<ReadOnlyMemory<byte>> ReadBytes(int length, Stream stream, CancellationToken cancellationToken) {
+		Memory<byte> buffer = new byte[length];
+		await stream.ReadExactlyAsync(buffer, cancellationToken);
+		return buffer;
+	}
+	
+	public static ReadOnlyMemory<byte> Serialize<T>(T value) {
+		var buffer = new ArrayBufferWriter<byte>();
+		MemoryPackSerializer.Serialize(buffer, value, SerializerOptions);
+		return buffer.WrittenMemory;
+	}
+	
+	public static T Deserialize<T>(ReadOnlyMemory<byte> buffer) {
+		return MemoryPackSerializer.Deserialize<T>(buffer.Span, SerializerOptions)!;
+	}
+}

Utils/Phantom.Utils.Rpc/Runtime2/IRegistrationHandler.cs

@@ -1,6 +1,6 @@
 using Phantom.Utils.Actor;
 
-namespace Phantom.Utils.Rpc.Runtime;
+namespace Phantom.Utils.Rpc.Runtime2;
 
 public interface IRegistrationHandler<TClientMessage, TServerMessage, TRegistrationMessage> where TRegistrationMessage : TServerMessage {
 	Task<Props<TServerMessage>?> TryRegister(RpcConnectionToClient<TClientMessage> connection, TRegistrationMessage message);

Utils/Phantom.Utils.Rpc/Runtime2/RpcServerRuntime.cs

@@ -3,12 +3,10 @@ using Akka.Actor;
 using NetMQ.Sockets;
 using Phantom.Utils.Actor;
 using Phantom.Utils.Logging;
-using Phantom.Utils.Rpc.Message;
-using Phantom.Utils.Rpc.Sockets;
 using Serilog;
 using Serilog.Events;
 
-namespace Phantom.Utils.Rpc.Runtime;
+namespace Phantom.Utils.Rpc.Runtime2;
 
 public static class RpcServerRuntime {
 	public static Task Launch<TClientMessage, TServerMessage, TRegistrationMessage, TReplyMessage>(

Utils/Phantom.Utils.Rpc/Sockets/RpcClientSocket.cs

@@ -1,42 +0,0 @@
-using NetMQ;
-using NetMQ.Sockets;
-using Phantom.Utils.Logging;
-using Phantom.Utils.Rpc.Message;
-using Phantom.Utils.Rpc.Runtime;
-
-namespace Phantom.Utils.Rpc.Sockets;
-
-public static class RpcClientSocket {
-	public static RpcClientSocket<TClientMessage, TServerMessage, TReplyMessage> Connect<TClientMessage, TServerMessage, TReplyMessage, THelloMessage>(RpcConfiguration config, IMessageDefinitions<TClientMessage, TServerMessage, TReplyMessage> messageDefinitions, THelloMessage helloMessage) where THelloMessage : TServerMessage where TReplyMessage : TClientMessage, TServerMessage {
-		return RpcClientSocket<TClientMessage, TServerMessage, TReplyMessage>.Connect(config, messageDefinitions, helloMessage);
-	}
-}
-
-public sealed class RpcClientSocket<TClientMessage, TServerMessage, TReplyMessage> : RpcSocket<ClientSocket> where TReplyMessage : TClientMessage, TServerMessage {
-	internal static RpcClientSocket<TClientMessage, TServerMessage, TReplyMessage> Connect<THelloMessage>(RpcConfiguration config, IMessageDefinitions<TClientMessage, TServerMessage, TReplyMessage> messageDefinitions, THelloMessage helloMessage) where THelloMessage : TServerMessage {
-		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 = PhantomLogger.Create(config.LoggerName);
-		
-		logger.Information("Starting ZeroMQ client and connecting to {Url}...", url);
-		socket.Connect(url);
-		logger.Information("ZeroMQ client ready.");
-		
-		return new RpcClientSocket<TClientMessage, TServerMessage, TReplyMessage>(socket, config, messageDefinitions);
-	}
-	
-	public RpcConnectionToServer<TServerMessage> Connection { get; }
-	internal IMessageDefinitions<TClientMessage, TServerMessage, TReplyMessage> MessageDefinitions { get; }
-	
-	private RpcClientSocket(ClientSocket socket, RpcConfiguration config, IMessageDefinitions<TClientMessage, TServerMessage, TReplyMessage> messageDefinitions) : base(socket, config) {
-		MessageDefinitions = messageDefinitions;
-		Connection = new RpcConnectionToServer<TServerMessage>(socket, messageDefinitions.ToServer, ReplyTracker);
-	}
-}

Utils/Phantom.Utils.Rpc/Sockets/RpcServerSocket.cs

@@ -1,26 +0,0 @@
-using NetMQ.Sockets;
-using Phantom.Utils.Logging;
-
-namespace Phantom.Utils.Rpc.Sockets;
-
-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 = PhantomLogger.Create(config.LoggerName);
-		
-		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) {}
-}

Utils/Phantom.Utils.Rpc/Sockets/RpcSocket.cs

@@ -1,25 +0,0 @@
-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.LoggerName);
-	}
-}

Utils/Phantom.Utils/Collections/Comparables.cs

@@ -0,0 +1,11 @@
+namespace Phantom.Utils.Collections;
+
+public static class Comparables {
+	public static T Min<T>(T first, T second) where T : IComparable<T> {
+		return first.CompareTo(second) < 0 ? first : second;
+	}
+	
+	public static T Max<T>(T first, T second) where T : IComparable<T> {
+		return first.CompareTo(second) < 0 ? second : first;
+	}
+}

Utils/Phantom.Utils/Cryptography/TokenGenerator.cs

@@ -33,7 +33,7 @@ public static class TokenGenerator {
 		return Encoding.ASCII.GetBytes(token);
 	}
 	
-	public static string EncodeBytes(byte[] bytes) {
+	public static string EncodeBytes(ReadOnlySpan<byte> bytes) {
 		return Base24.Encode(bytes);
 	}
 	

Utils/Phantom.Utils/Monads/Either.cs

@@ -0,0 +1,22 @@
+namespace Phantom.Utils.Monads;
+
+public abstract record Either<TLeft, TRight> {
+	public abstract TLeft RequireLeft { get; }
+	public abstract TRight RequireRight { get; }
+	
+	public abstract Either<TNewLeft, TRight> MapLeft<TNewLeft>(Func<TLeft, TNewLeft> func);
+	public abstract Either<TLeft, TNewRight> MapRight<TNewRight>(Func<TRight, TNewRight> func);
+	
+	public static implicit operator Either<TLeft, TRight>(Left<TLeft> value) => new Left<TLeft, TRight>(value.Value);
+	public static implicit operator Either<TLeft, TRight>(Right<TRight> value) => new Right<TLeft, TRight>(value.Value);
+}
+
+public static class Either {
+	public static Left<TValue> Left<TValue>(TValue value) {
+		return new Left<TValue>(value);
+	}
+	
+	public static Right<TValue> Right<TValue>(TValue value) {
+		return new Right<TValue>(value);
+	}
+}

Utils/Phantom.Utils/Monads/Left.cs

@@ -0,0 +1,16 @@
+namespace Phantom.Utils.Monads;
+
+public sealed record Left<TLeft, TRight>(TLeft Value) : Either<TLeft, TRight> {
+	public override TLeft RequireLeft => Value;
+	public override TRight RequireRight => throw new InvalidOperationException("Either<" + typeof(TLeft).Name + ", " + typeof(TRight).Name + "> has a left value, but right value was requested.");
+	
+	public override Either<TNewLeft, TRight> MapLeft<TNewLeft>(Func<TLeft, TNewLeft> func) {
+		return new Left<TNewLeft, TRight>(func(Value));
+	}
+	
+	public override Either<TLeft, TNewRight> MapRight<TNewRight>(Func<TRight, TNewRight> func) {
+		return new Left<TLeft, TNewRight>(Value);
+	}
+}
+
+public sealed record Left<TValue>(TValue Value);