Replace NetMQ with custom TCP logic

Agent/Phantom.Agent.Services/AgentServices.cs

@@ -52,6 +52,8 @@ public sealed class AgentServices {
 	public  async Task<bool> Register(ControllerConnection connection, CancellationToken cancellationToken) {
 		Logger.Information("Registering with the controller...");
 		
+		// TODO NEED TO SEND WHEN SERVER RESTARTS!!!
+		
 		ImmutableArray<ConfigureInstanceMessage> configureInstanceMessages;
 		try {
 			configureInstanceMessages = await connection.Send<RegisterAgentMessage, ImmutableArray<ConfigureInstanceMessage>>(new RegisterAgentMessage(AgentInfo), TimeSpan.FromMinutes(1), cancellationToken);

Agent/Phantom.Agent.Services/Instances/InstanceManagerActor.cs

@@ -56,11 +56,6 @@ sealed class InstanceManagerActor : ReceiveActor<InstanceManagerActor.ICommand>
 		ReceiveAsync<ShutdownCommand>(Shutdown);
 	}
 	
-	private string GetInstanceLoggerName(Guid guid) {
-		var prefix = guid.ToString();
-		return prefix[..prefix.IndexOf('-')] + "/" + Interlocked.Increment(ref instanceLoggerSequenceId);
-	}
-	
 	private sealed record InstanceInfo(ActorRef<InstanceActor.ICommand> Actor, InstanceConfiguration Configuration, IServerLauncher Launcher);
 	
 	public interface ICommand {}
@@ -118,7 +113,8 @@ sealed class InstanceManagerActor : ReceiveActor<InstanceManagerActor.ICommand>
 			}
 		}
 		else {
-			var instanceInit = new InstanceActor.Init(agentState, instanceGuid, GetInstanceLoggerName(instanceGuid), instanceServices, instanceTicketManager, shutdownCancellationToken);
+			var instanceLoggerName = PhantomLogger.ShortenGuid(instanceGuid) + "/" + Interlocked.Increment(ref instanceLoggerSequenceId);;
+			var instanceInit = new InstanceActor.Init(agentState, instanceGuid, instanceLoggerName, instanceServices, instanceTicketManager, shutdownCancellationToken);
 			instances[instanceGuid] = instance = new InstanceInfo(Context.ActorOf(InstanceActor.Factory(instanceInit), "Instance-" + instanceGuid), configuration, launcher);
 			
 			Logger.Information("Created instance \"{Name}\" (GUID {Guid}).", configuration.InstanceName, instanceGuid);

Agent/Phantom.Agent/Program.cs

@@ -7,6 +7,7 @@ using Phantom.Common.Messages.Agent;
 using Phantom.Common.Messages.Agent.ToController;
 using Phantom.Utils.Actor;
 using Phantom.Utils.Logging;
+using Phantom.Utils.Rpc.Message;
 using Phantom.Utils.Rpc.Runtime.Client;
 using Phantom.Utils.Runtime;
 using Phantom.Utils.Threading;
@@ -74,7 +75,7 @@ try {
 		var rpcMessageHandlerInit = new ControllerMessageHandlerActor.Init(agentServices);
 		var rpcMessageHandlerActor = agentServices.ActorSystem.ActorOf(ControllerMessageHandlerActor.Factory(rpcMessageHandlerInit), "ControllerMessageHandler");
 		
-		rpcClientListener = rpcClient.Listen(rpcMessageHandlerActor);
+		rpcClientListener = rpcClient.Listen(new IMessageReceiver<IMessageToAgent>.Actor(rpcMessageHandlerActor));
 		
 		if (await agentServices.Register(controllerConnection, shutdownCancellationToken)) {
 			PhantomLogger.Root.Information("Phantom Panel agent is ready.");

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

@@ -22,7 +22,6 @@ public static class AgentMessageRegistries {
 		
 		ToController.Add<RegisterAgentMessage, ImmutableArray<ConfigureInstanceMessage>>(0);
 		ToController.Add<UnregisterAgentMessage>(1);
-		ToController.Add<AgentIsAliveMessage>(2);
 		ToController.Add<AdvertiseJavaRuntimesMessage>(3);
 		ToController.Add<ReportInstanceStatusMessage>(4);
 		ToController.Add<InstanceOutputMessage>(5);

Common/Phantom.Common.Messages.Agent/ToController/AgentIsAliveMessage.cs

@@ -1,6 +0,0 @@
-using MemoryPack;
-
-namespace Phantom.Common.Messages.Agent.ToController;
-
-[MemoryPackable(GenerateType.VersionTolerant)]
-public sealed partial record AgentIsAliveMessage : IMessageToController;

Common/Phantom.Common.Messages.Web/ToWeb/RegisterWebResultMessage.cs

@@ -1,8 +0,0 @@
-using MemoryPack;
-
-namespace Phantom.Common.Messages.Web.ToWeb;
-
-[MemoryPackable(GenerateType.VersionTolerant)]
-public sealed partial record RegisterWebResultMessage(
-	[property: MemoryPackOrder(0)] bool Success
-) : IMessageToWeb;

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

@@ -21,6 +21,7 @@ using Phantom.Utils.Actor.Mailbox;
 using Phantom.Utils.Actor.Tasks;
 using Phantom.Utils.Collections;
 using Phantom.Utils.Logging;
+using Phantom.Utils.Rpc.Runtime.Server;
 using Serilog;
 
 namespace Phantom.Controller.Services.Agents;
@@ -167,13 +168,13 @@ sealed class AgentActor : ReceiveActor<AgentActor.ICommand> {
 		return configurationMessages.ToImmutable();
 	}
 	
-	public interface ICommand {}
+	public interface ICommand;
 	
 	private sealed record InitializeCommand : ICommand;
 	
-	public sealed record RegisterCommand(AgentConfiguration Configuration, RpcConnectionToClient<IMessageToAgent> Connection) : ICommand, ICanReply<ImmutableArray<ConfigureInstanceMessage>>;
+	public sealed record RegisterCommand(AgentConfiguration Configuration, RpcServerToClientConnection<IMessageToController, IMessageToAgent> Connection) : ICommand, ICanReply<ImmutableArray<ConfigureInstanceMessage>>;
 	
-	public sealed record UnregisterCommand(RpcConnectionToClient<IMessageToAgent> Connection) : ICommand;
+	public sealed record UnregisterCommand(RpcServerToClientConnection<IMessageToController, IMessageToAgent> Connection) : ICommand;
 	
 	private sealed record RefreshConnectionStatusCommand : ICommand;
 	

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

@@ -1,35 +1,30 @@
 using Phantom.Common.Messages.Agent;
 using Phantom.Utils.Actor;
 using Phantom.Utils.Logging;
+using Phantom.Utils.Rpc.Runtime.Server;
 using Serilog;
 
 namespace Phantom.Controller.Services.Agents;
 
-sealed class AgentConnection {
+sealed class AgentConnection(Guid agentGuid, string agentName) {
 	private static readonly ILogger Logger = PhantomLogger.Create<AgentConnection>();
 	
-	private readonly Guid agentGuid;
-	private string agentName;
+	private string agentName = agentName;
+	private RpcServerToClientConnection<IMessageToController, IMessageToAgent>? connection;
 	
-	private RpcConnectionToClient<IMessageToAgent>? connection;
-	
-	public AgentConnection(Guid agentGuid, string agentName) {
-		this.agentName = agentName;
-		this.agentGuid = agentGuid;
-	}
-	
-	public void UpdateConnection(RpcConnectionToClient<IMessageToAgent> newConnection, string newAgentName) {
+	public void UpdateConnection(RpcServerToClientConnection<IMessageToController, IMessageToAgent> newConnection, string newAgentName) {
 		lock (this) {
-			connection?.Close();
+			connection?.CloseSession();
 			connection = newConnection;
 			agentName = newAgentName;
 		}
 	}
 	
-	public bool CloseIfSame(RpcConnectionToClient<IMessageToAgent> expected) {
+	public bool CloseIfSame(RpcServerToClientConnection<IMessageToController, IMessageToAgent> expected) {
 		lock (this) {
-			if (connection != null && connection.IsSame(expected)) {
-				connection.Close();
+			if (connection != null && ReferenceEquals(connection, expected)) {
+				connection.CloseSession();
+				connection = null;
 				return true;
 			}
 		}
@@ -44,7 +39,7 @@ sealed class AgentConnection {
 				return Task.CompletedTask;
 			}
 			
-			return connection.Send(message);
+			return connection.SendChannel.SendMessage(message).AsTask();
 		}
 	}
 	
@@ -52,10 +47,10 @@ sealed class AgentConnection {
 		lock (this) {
 			if (connection == null) {
 				LogAgentOffline();
-				return Task.FromResult<TReply?>(default);
+				return Task.FromResult<TReply?>(null);
 			}
 			
-			return connection.Send<TMessage, TReply>(message, waitForReplyTime, waitForReplyCancellationToken)!;
+			return connection.SendChannel.SendMessage<TMessage, TReply>(message, waitForReplyTime, waitForReplyCancellationToken)!;
 		}
 	}
 	

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

@@ -13,7 +13,7 @@ using Phantom.Controller.Minecraft;
 using Phantom.Controller.Services.Users.Sessions;
 using Phantom.Utils.Actor;
 using Phantom.Utils.Logging;
-using Phantom.Utils.Rpc;
+using Phantom.Utils.Rpc.Runtime.Server;
 using Serilog;
 
 namespace Phantom.Controller.Services.Agents;
@@ -22,19 +22,18 @@ sealed class AgentManager {
 	private static readonly ILogger Logger = PhantomLogger.Create<AgentManager>();
 	
 	private readonly IActorRefFactory actorSystem;
-	private readonly AuthToken authToken;
 	private readonly ControllerState controllerState;
 	private readonly MinecraftVersions minecraftVersions;
 	private readonly UserLoginManager userLoginManager;
 	private readonly IDbContextProvider dbProvider;
 	private readonly CancellationToken cancellationToken;
 	
-	private readonly ConcurrentDictionary<Guid, ActorRef<AgentActor.ICommand>> agentsByGuid = new ();
+	private readonly ConcurrentDictionary<Guid, ActorRef<AgentActor.ICommand>> agentsByAgentGuid = new ();
+	private readonly ConcurrentDictionary<Guid, Guid> agentGuidsBySessionGuid = new ();
 	private readonly Func<Guid, AgentConfiguration, ActorRef<AgentActor.ICommand>> addAgentActorFactory;
 	
-	public AgentManager(IActorRefFactory actorSystem, AuthToken authToken, ControllerState controllerState, MinecraftVersions minecraftVersions, UserLoginManager userLoginManager, IDbContextProvider dbProvider, CancellationToken cancellationToken) {
+	public AgentManager(IActorRefFactory actorSystem, ControllerState controllerState, MinecraftVersions minecraftVersions, UserLoginManager userLoginManager, IDbContextProvider dbProvider, CancellationToken cancellationToken) {
 		this.actorSystem = actorSystem;
-		this.authToken = authToken;
 		this.controllerState = controllerState;
 		this.minecraftVersions = minecraftVersions;
 		this.userLoginManager = userLoginManager;
@@ -57,28 +56,21 @@ sealed class AgentManager {
 			var agentGuid = entity.AgentGuid;
 			var agentConfiguration = new AgentConfiguration(entity.Name, entity.ProtocolVersion, entity.BuildVersion, entity.MaxInstances, entity.MaxMemory);
 			
-			if (agentsByGuid.TryAdd(agentGuid, CreateAgentActor(agentGuid, agentConfiguration))) {
+			if (agentsByAgentGuid.TryAdd(agentGuid, CreateAgentActor(agentGuid, agentConfiguration))) {
 				Logger.Information("Loaded agent \"{AgentName}\" (GUID {AgentGuid}) from database.", agentConfiguration.AgentName, agentGuid);
 			}
 		}
 	}
 	
-	public async Task<bool> RegisterAgent(AuthToken authToken, AgentInfo agentInfo, RpcConnectionToClient<IMessageToAgent> connection) {
-		if (!this.authToken.FixedTimeEquals(authToken)) {
-			await connection.Send(new RegisterAgentFailureMessage(RegisterAgentError.InvalidToken));
-			return false;
-		}
-		
+	public async Task<ImmutableArray<ConfigureInstanceMessage>> RegisterAgent(AgentInfo agentInfo, RpcServerToClientConnection<IMessageToController, IMessageToAgent> connection) {
 		var agentProperties = AgentConfiguration.From(agentInfo);
-		var agentActor = agentsByGuid.GetOrAdd(agentInfo.AgentGuid, addAgentActorFactory, agentProperties);
-		var configureInstanceMessages = await agentActor.Request(new AgentActor.RegisterCommand(agentProperties, connection), cancellationToken);
-		await connection.Send(new RegisterAgentSuccessMessage(configureInstanceMessages));
-		
-		return true;
+		var agentActor = agentsByAgentGuid.GetOrAdd(agentInfo.AgentGuid, addAgentActorFactory, agentProperties);
+		agentGuidsBySessionGuid[connection.SessionId] = agentInfo.AgentGuid;
+		return await agentActor.Request(new AgentActor.RegisterCommand(agentProperties, connection), cancellationToken);
 	}
 	
 	public bool TellAgent(Guid agentGuid, AgentActor.ICommand command) {
-		if (agentsByGuid.TryGetValue(agentGuid, out var agent)) {
+		if (agentsByAgentGuid.TryGetValue(agentGuid, out var agent)) {
 			agent.Tell(command);
 			return true;
 		}
@@ -94,7 +86,7 @@ sealed class AgentManager {
 			return (UserInstanceActionFailure) UserActionFailure.NotAuthorized;
 		}
 		
-		if (!agentsByGuid.TryGetValue(agentGuid, out var agent)) {
+		if (!agentsByAgentGuid.TryGetValue(agentGuid, out var agent)) {
 			return (UserInstanceActionFailure) InstanceActionFailure.AgentDoesNotExist;
 		}
 		
@@ -102,4 +94,12 @@ sealed class AgentManager {
 		var result = await agent.Request(command, cancellationToken);
 		return result.MapError(static error => (UserInstanceActionFailure) error);
 	}
+	
+	public bool TryGetAgentGuidBySessionGuid(Guid sessionGuid, out Guid agentGuid) {
+		return agentGuidsBySessionGuid.TryGetValue(sessionGuid, out agentGuid);
+	}
+	
+	public void OnSessionClosed(Guid sessionId, Guid agentGuid) {
+		agentGuidsBySessionGuid.TryRemove(new KeyValuePair<Guid, Guid>(sessionId, agentGuid));
+	}
 }

Controller/Phantom.Controller.Services/ControllerServices.cs

@@ -1,8 +1,4 @@
 using Akka.Actor;
-using Phantom.Common.Messages.Agent;
-using Phantom.Common.Messages.Agent.ToController;
-using Phantom.Common.Messages.Web;
-using Phantom.Common.Messages.Web.ToController;
 using Phantom.Controller.Database;
 using Phantom.Controller.Minecraft;
 using Phantom.Controller.Services.Agents;
@@ -13,8 +9,8 @@ using Phantom.Controller.Services.Users;
 using Phantom.Controller.Services.Users.Sessions;
 using Phantom.Utils.Actor;
 using Phantom.Utils.Rpc;
-using IMessageFromAgentToController = Phantom.Common.Messages.Agent.IMessageToController;
-using IMessageFromWebToController = Phantom.Common.Messages.Web.IMessageToController;
+using IRpcAgentRegistrar = Phantom.Utils.Rpc.Runtime.Server.IRpcServerClientRegistrar<Phantom.Common.Messages.Agent.IMessageToController, Phantom.Common.Messages.Agent.IMessageToAgent>;
+using IRpcWebRegistrar = Phantom.Utils.Rpc.Runtime.Server.IRpcServerClientRegistrar<Phantom.Common.Messages.Web.IMessageToController, Phantom.Common.Messages.Web.IMessageToWeb>;
 
 namespace Phantom.Controller.Services;
 
@@ -37,13 +33,13 @@ public sealed class ControllerServices : IDisposable {
 	private AuditLogManager AuditLogManager { get; }
 	private EventLogManager EventLogManager { get; }
 	
-	public IRegistrationHandler<IMessageToAgent, IMessageFromAgentToController, RegisterAgentMessage> AgentRegistrationHandler { get; }
-	public IRegistrationHandler<IMessageToWeb, IMessageFromWebToController, RegisterWebMessage> WebRegistrationHandler { get; }
+	public IRpcAgentRegistrar AgentRegistrar { get; }
+	public IRpcWebRegistrar WebRegistrar { get; }
 	
 	private readonly IDbContextProvider dbProvider;
 	private readonly CancellationToken cancellationToken;
 	
-	public ControllerServices(IDbContextProvider dbProvider, AuthToken agentAuthToken, AuthToken webAuthToken, CancellationToken shutdownCancellationToken) {
+	public ControllerServices(IDbContextProvider dbProvider, AuthToken agentAuthToken, CancellationToken shutdownCancellationToken) {
 		this.dbProvider = dbProvider;
 		this.cancellationToken = shutdownCancellationToken;
 		
@@ -59,14 +55,14 @@ public sealed class ControllerServices : IDisposable {
 		this.UserLoginManager = new UserLoginManager(AuthenticatedUserCache, UserManager, dbProvider);
 		this.PermissionManager = new PermissionManager(dbProvider);
 		
-		this.AgentManager = new AgentManager(ActorSystem, agentAuthToken, ControllerState, MinecraftVersions, UserLoginManager, dbProvider, cancellationToken);
+		this.AgentManager = new AgentManager(ActorSystem, ControllerState, MinecraftVersions, UserLoginManager, dbProvider, cancellationToken);
 		this.InstanceLogManager = new InstanceLogManager();
 		
 		this.AuditLogManager = new AuditLogManager(dbProvider);
 		this.EventLogManager = new EventLogManager(ControllerState, ActorSystem, dbProvider, shutdownCancellationToken);
 		
-		this.AgentRegistrationHandler = new AgentRegistrationHandler(AgentManager, InstanceLogManager, EventLogManager);
-		this.WebRegistrationHandler = new WebRegistrationHandler(webAuthToken, ControllerState, InstanceLogManager, UserManager, RoleManager, UserRoleManager, UserLoginManager, AuditLogManager, AgentManager, MinecraftVersions, EventLogManager);
+		this.AgentRegistrar = new AgentClientRegistrar(ActorSystem, AgentManager, InstanceLogManager, EventLogManager);
+		this.WebRegistrar = new WebClientRegistrar(ActorSystem, ControllerState, InstanceLogManager, UserManager, RoleManager, UserRoleManager, UserLoginManager, AuditLogManager, AgentManager, MinecraftVersions, EventLogManager);
 	}
 	
 	public async Task Initialize() {

Controller/Phantom.Controller.Services/Rpc/AgentClientRegistrar.cs

@@ -0,0 +1,31 @@
+using Akka.Actor;
+using Phantom.Common.Messages.Agent;
+using Phantom.Controller.Services.Agents;
+using Phantom.Controller.Services.Events;
+using Phantom.Controller.Services.Instances;
+using Phantom.Utils.Actor;
+using Phantom.Utils.Rpc.Message;
+using Phantom.Utils.Rpc.Runtime.Server;
+
+namespace Phantom.Controller.Services.Rpc;
+
+sealed class AgentClientRegistrar(
+	IActorRefFactory actorSystem,
+	AgentManager agentManager,
+	InstanceLogManager instanceLogManager,
+	EventLogManager eventLogManager
+) : IRpcServerClientRegistrar<IMessageToController, IMessageToAgent> {
+	public IMessageReceiver<IMessageToController> Register(RpcServerToClientConnection<IMessageToController, IMessageToAgent> connection) {
+		var name = "AgentClient-" + connection.SessionId;
+		var init = new AgentMessageHandlerActor.Init(connection, agentManager, instanceLogManager, eventLogManager);
+		return new Receiver(connection.SessionId, agentManager, actorSystem.ActorOf(AgentMessageHandlerActor.Factory(init), name));
+	}
+	
+	private sealed class Receiver(Guid sessionId, AgentManager agentManager, ActorRef<IMessageToController> actor) : IMessageReceiver<IMessageToController>.Actor(actor) {
+		public override void OnPing() {
+			if (agentManager.TryGetAgentGuidBySessionGuid(sessionId, out var agentGuid)) {
+				agentManager.TellAgent(agentGuid, new AgentActor.NotifyIsAliveCommand());
+			}
+		}
+	}
+}

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

@@ -1,4 +1,4 @@
-using Phantom.Common.Data.Replies;
+using System.Collections.Immutable;
 using Phantom.Common.Messages.Agent;
 using Phantom.Common.Messages.Agent.ToAgent;
 using Phantom.Common.Messages.Agent.ToController;
@@ -6,34 +6,32 @@ using Phantom.Controller.Services.Agents;
 using Phantom.Controller.Services.Events;
 using Phantom.Controller.Services.Instances;
 using Phantom.Utils.Actor;
+using Phantom.Utils.Rpc.Runtime.Server;
 
 namespace Phantom.Controller.Services.Rpc;
 
 sealed class AgentMessageHandlerActor : ReceiveActor<IMessageToController> {
-	public readonly record struct Init(Guid AgentGuid, RpcConnectionToClient<IMessageToAgent> Connection, AgentRegistrationHandler AgentRegistrationHandler, AgentManager AgentManager, InstanceLogManager InstanceLogManager, EventLogManager EventLogManager);
+	public readonly record struct Init(RpcServerToClientConnection<IMessageToController, IMessageToAgent> Connection, AgentManager AgentManager, InstanceLogManager InstanceLogManager, EventLogManager EventLogManager);
 	
 	public static Props<IMessageToController> Factory(Init init) {
 		return Props<IMessageToController>.Create(() => new AgentMessageHandlerActor(init), new ActorConfiguration { SupervisorStrategy = SupervisorStrategies.Resume });
 	}
 	
-	private readonly Guid agentGuid;
-	private readonly RpcConnectionToClient<IMessageToAgent> connection;
-	private readonly AgentRegistrationHandler agentRegistrationHandler;
+	private readonly RpcServerToClientConnection<IMessageToController, IMessageToAgent> connection;
 	private readonly AgentManager agentManager;
 	private readonly InstanceLogManager instanceLogManager;
 	private readonly EventLogManager eventLogManager;
 	
+	private Guid? registeredAgentGuid;
+	
 	private AgentMessageHandlerActor(Init init) {
-		this.agentGuid = init.AgentGuid;
 		this.connection = init.Connection;
-		this.agentRegistrationHandler = init.AgentRegistrationHandler;
 		this.agentManager = init.AgentManager;
 		this.instanceLogManager = init.InstanceLogManager;
 		this.eventLogManager = init.EventLogManager;
 		
-		ReceiveAsync<RegisterAgentMessage>(HandleRegisterAgent);
+		ReceiveAsyncAndReply<RegisterAgentMessage, ImmutableArray<ConfigureInstanceMessage>>(HandleRegisterAgent);
 		Receive<UnregisterAgentMessage>(HandleUnregisterAgent);
-		Receive<AgentIsAliveMessage>(HandleAgentIsAlive);
 		Receive<AdvertiseJavaRuntimesMessage>(HandleAdvertiseJavaRuntimes);
 		Receive<ReportAgentStatusMessage>(HandleReportAgentStatus);
 		Receive<ReportInstanceStatusMessage>(HandleReportInstanceStatus);
@@ -42,42 +40,40 @@ sealed class AgentMessageHandlerActor : ReceiveActor<IMessageToController> {
 		Receive<InstanceOutputMessage>(HandleInstanceOutput);
 	}
 	
-	private async Task HandleRegisterAgent(RegisterAgentMessage message) {
-		if (agentGuid != message.AgentInfo.AgentGuid) {
-			await connection.Send(new RegisterAgentFailureMessage(RegisterAgentError.ConnectionAlreadyHasAnAgent));
-		}
-		else {
-			await agentRegistrationHandler.TryRegisterImpl(connection, message);
+	private Guid RequireAgentGuid() {
+		return registeredAgentGuid ?? throw new InvalidOperationException("Agent has not registered yet.");
 	}
+	
+	private Task<ImmutableArray<ConfigureInstanceMessage>> HandleRegisterAgent(RegisterAgentMessage message) {
+		registeredAgentGuid = message.AgentInfo.AgentGuid;
+		return agentManager.RegisterAgent(message.AgentInfo, connection);
 	}
 	
 	private void HandleUnregisterAgent(UnregisterAgentMessage message) {
+		Guid agentGuid = RequireAgentGuid();
 		agentManager.TellAgent(agentGuid, new AgentActor.UnregisterCommand(connection));
-		connection.Close();
-	}
-	
-	private void HandleAgentIsAlive(AgentIsAliveMessage message) {
-		agentManager.TellAgent(agentGuid, new AgentActor.NotifyIsAliveCommand());
+		agentManager.OnSessionClosed(connection.SessionId, agentGuid);
+		connection.CloseSession();
 	}
 	
 	private void HandleAdvertiseJavaRuntimes(AdvertiseJavaRuntimesMessage message) {
-		agentManager.TellAgent(agentGuid, new AgentActor.UpdateJavaRuntimesCommand(message.Runtimes));
+		agentManager.TellAgent(RequireAgentGuid(), new AgentActor.UpdateJavaRuntimesCommand(message.Runtimes));
 	}
 	
 	private void HandleReportAgentStatus(ReportAgentStatusMessage message) {
-		agentManager.TellAgent(agentGuid, new AgentActor.UpdateStatsCommand(message.RunningInstanceCount, message.RunningInstanceMemory));
+		agentManager.TellAgent(RequireAgentGuid(), new AgentActor.UpdateStatsCommand(message.RunningInstanceCount, message.RunningInstanceMemory));
 	}
 	
 	private void HandleReportInstanceStatus(ReportInstanceStatusMessage message) {
-		agentManager.TellAgent(agentGuid, new AgentActor.UpdateInstanceStatusCommand(message.InstanceGuid, message.InstanceStatus));
+		agentManager.TellAgent(RequireAgentGuid(), new AgentActor.UpdateInstanceStatusCommand(message.InstanceGuid, message.InstanceStatus));
 	}
 	
 	private void HandleReportInstancePlayerCounts(ReportInstancePlayerCountsMessage message) {
-		agentManager.TellAgent(agentGuid, new AgentActor.UpdateInstancePlayerCountsCommand(message.InstanceGuid, message.PlayerCounts));
+		agentManager.TellAgent(RequireAgentGuid(), new AgentActor.UpdateInstancePlayerCountsCommand(message.InstanceGuid, message.PlayerCounts));
 	}
 	
 	private void HandleReportInstanceEvent(ReportInstanceEventMessage message) {
-		message.Event.Accept(eventLogManager.CreateInstanceEventVisitor(message.EventGuid, message.UtcTime, agentGuid, message.InstanceGuid));
+		message.Event.Accept(eventLogManager.CreateInstanceEventVisitor(message.EventGuid, message.UtcTime, RequireAgentGuid(), message.InstanceGuid));
 	}
 	
 	private void HandleInstanceOutput(InstanceOutputMessage message) {

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

@@ -1,34 +0,0 @@
-using Phantom.Common.Messages.Agent;
-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.Runtime2;
-
-namespace Phantom.Controller.Services.Rpc;
-
-sealed class AgentRegistrationHandler : IRegistrationHandler<IMessageToAgent, IMessageToController, RegisterAgentMessage> {
-	private readonly AgentManager agentManager;
-	private readonly InstanceLogManager instanceLogManager;
-	private readonly EventLogManager eventLogManager;
-	
-	public AgentRegistrationHandler(AgentManager agentManager, InstanceLogManager instanceLogManager, EventLogManager eventLogManager) {
-		this.agentManager = agentManager;
-		this.instanceLogManager = instanceLogManager;
-		this.eventLogManager = eventLogManager;
-	}
-	
-	async Task<Props<IMessageToController>?> IRegistrationHandler<IMessageToAgent, IMessageToController, RegisterAgentMessage>.TryRegister(RpcConnectionToClient<IMessageToAgent> connection, RegisterAgentMessage message) {
-		return await TryRegisterImpl(connection, message) ? CreateMessageHandlerActorProps(message.AgentInfo.AgentGuid, connection) : null;
-	}
-	
-	public Task<bool> TryRegisterImpl(RpcConnectionToClient<IMessageToAgent> connection, RegisterAgentMessage message) {
-		return agentManager.RegisterAgent(message.AuthToken, message.AgentInfo, connection);
-	}
-	
-	private Props<IMessageToController> CreateMessageHandlerActorProps(Guid agentGuid, RpcConnectionToClient<IMessageToAgent> connection) {
-		var init = new AgentMessageHandlerActor.Init(agentGuid, connection, this, agentManager, instanceLogManager, eventLogManager);
-		return AgentMessageHandlerActor.Factory(init);
-	}
-}

Controller/Phantom.Controller.Services/Rpc/WebClientRegistrar.cs

@@ -0,0 +1,33 @@
+using Akka.Actor;
+using Phantom.Common.Messages.Web;
+using Phantom.Controller.Minecraft;
+using Phantom.Controller.Services.Agents;
+using Phantom.Controller.Services.Events;
+using Phantom.Controller.Services.Instances;
+using Phantom.Controller.Services.Users;
+using Phantom.Controller.Services.Users.Sessions;
+using Phantom.Utils.Actor;
+using Phantom.Utils.Rpc.Message;
+using Phantom.Utils.Rpc.Runtime.Server;
+
+namespace Phantom.Controller.Services.Rpc;
+
+sealed class WebClientRegistrar(
+	IActorRefFactory actorSystem,
+	ControllerState controllerState,
+	InstanceLogManager instanceLogManager,
+	UserManager userManager,
+	RoleManager roleManager,
+	UserRoleManager userRoleManager,
+	UserLoginManager userLoginManager,
+	AuditLogManager auditLogManager,
+	AgentManager agentManager,
+	MinecraftVersions minecraftVersions,
+	EventLogManager eventLogManager
+) : IRpcServerClientRegistrar<IMessageToController, IMessageToWeb> {
+	public IMessageReceiver<IMessageToController> Register(RpcServerToClientConnection<IMessageToController, IMessageToWeb> connection) {
+		var name = "WebClient-" + connection.SessionId;
+		var init = new WebMessageHandlerActor.Init(connection, this, controllerState, instanceLogManager, userManager, roleManager, userRoleManager, userLoginManager, auditLogManager, agentManager, minecraftVersions, eventLogManager);
+		return new IMessageReceiver<IMessageToController>.Actor(actorSystem.ActorOf(WebMessageHandlerActor.Factory(init), name));
+	}
+}

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

@@ -5,17 +5,18 @@ 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;
 
 sealed class WebMessageDataUpdateSenderActor : ReceiveActor<WebMessageDataUpdateSenderActor.ICommand> {
-	public readonly record struct Init(RpcConnectionToClient<IMessageToWeb> Connection, ControllerState ControllerState, InstanceLogManager InstanceLogManager);
+	public readonly record struct Init(RpcSendChannel<IMessageToWeb> Connection, ControllerState ControllerState, InstanceLogManager InstanceLogManager);
 	
 	public static Props<ICommand> Factory(Init init) {
 		return Props<ICommand>.Create(() => new WebMessageDataUpdateSenderActor(init), new ActorConfiguration { SupervisorStrategy = SupervisorStrategies.Resume });
 	}
 	
-	private readonly RpcConnectionToClient<IMessageToWeb> connection;
+	private readonly RpcSendChannel<IMessageToWeb> connection;
 	private readonly ControllerState controllerState;
 	private readonly InstanceLogManager instanceLogManager;
 	private readonly ActorRef<ICommand> selfCached;
@@ -69,18 +70,18 @@ sealed class WebMessageDataUpdateSenderActor : ReceiveActor<WebMessageDataUpdate
 	private sealed record RefreshUserSessionCommand(Guid UserGuid) : ICommand;
 	
 	private Task RefreshAgents(RefreshAgentsCommand command) {
-		return connection.Send(new RefreshAgentsMessage(command.Agents.Values.ToImmutableArray()));
+		return connection.SendMessage(new RefreshAgentsMessage(command.Agents.Values.ToImmutableArray())).AsTask();
 	}
 	
 	private Task RefreshInstances(RefreshInstancesCommand command) {
-		return connection.Send(new RefreshInstancesMessage(command.Instances.Values.ToImmutableArray()));
+		return connection.SendMessage(new RefreshInstancesMessage(command.Instances.Values.ToImmutableArray())).AsTask();
 	}
 	
 	private Task ReceiveInstanceLogs(ReceiveInstanceLogsCommand command) {
-		return connection.Send(new InstanceOutputMessage(command.InstanceGuid, command.Lines));
+		return connection.SendMessage(new InstanceOutputMessage(command.InstanceGuid, command.Lines)).AsTask();
 	}
 	
 	private Task RefreshUserSession(RefreshUserSessionCommand command) {
-		return connection.Send(new RefreshUserSessionMessage(command.UserGuid));
+		return connection.SendMessage(new RefreshUserSessionMessage(command.UserGuid)).AsTask();
 	}
 }

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

@@ -16,13 +16,14 @@ 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.Server;
 
 namespace Phantom.Controller.Services.Rpc;
 
 sealed class WebMessageHandlerActor : ReceiveActor<IMessageToController> {
 	public readonly record struct Init(
-		RpcConnectionToClient<IMessageToWeb> Connection,
-		WebRegistrationHandler WebRegistrationHandler,
+		RpcServerToClientConnection<IMessageToController, IMessageToWeb> Connection,
+		WebClientRegistrar WebClientRegistrar,
 		ControllerState ControllerState,
 		InstanceLogManager InstanceLogManager,
 		UserManager UserManager,
@@ -39,8 +40,7 @@ sealed class WebMessageHandlerActor : ReceiveActor<IMessageToController> {
 		return Props<IMessageToController>.Create(() => new WebMessageHandlerActor(init), new ActorConfiguration { SupervisorStrategy = SupervisorStrategies.Resume });
 	}
 	
-	private readonly RpcConnectionToClient<IMessageToWeb> connection;
-	private readonly WebRegistrationHandler webRegistrationHandler;
+	private readonly RpcServerToClientConnection<IMessageToController, IMessageToWeb> connection;
 	private readonly ControllerState controllerState;
 	private readonly UserManager userManager;
 	private readonly RoleManager roleManager;
@@ -53,7 +53,6 @@ sealed class WebMessageHandlerActor : ReceiveActor<IMessageToController> {
 	
 	private WebMessageHandlerActor(Init init) {
 		this.connection = init.Connection;
-		this.webRegistrationHandler = init.WebRegistrationHandler;
 		this.controllerState = init.ControllerState;
 		this.userManager = init.UserManager;
 		this.roleManager = init.RoleManager;
@@ -64,11 +63,10 @@ sealed class WebMessageHandlerActor : ReceiveActor<IMessageToController> {
 		this.minecraftVersions = init.MinecraftVersions;
 		this.eventLogManager = init.EventLogManager;
 		
-		var senderActorInit = new WebMessageDataUpdateSenderActor.Init(connection, controllerState, init.InstanceLogManager);
+		var senderActorInit = new WebMessageDataUpdateSenderActor.Init(connection.SendChannel, controllerState, init.InstanceLogManager);
 		Context.ActorOf(WebMessageDataUpdateSenderActor.Factory(senderActorInit), "DataUpdateSender");
 		
-		ReceiveAsync<RegisterWebMessage>(HandleRegisterWeb);
-		Receive<UnregisterWebMessage>(HandleUnregisterWeb);
+		ReceiveAsync<UnregisterWebMessage>(HandleUnregisterWeb);
 		ReceiveAndReplyLater<LogInMessage, Optional<LogInSuccess>>(HandleLogIn);
 		Receive<LogOutMessage>(HandleLogOut);
 		ReceiveAndReply<GetAuthenticatedUser, Optional<AuthenticatedUserInfo>>(GetAuthenticatedUser);
@@ -89,12 +87,8 @@ sealed class WebMessageHandlerActor : ReceiveActor<IMessageToController> {
 		ReceiveAndReplyLater<GetEventLogMessage, Result<ImmutableArray<EventLogItem>, UserActionFailure>>(HandleGetEventLog);
 	}
 	
-	private async Task HandleRegisterWeb(RegisterWebMessage message) {
-		await webRegistrationHandler.TryRegisterImpl(connection, message);
-	}
-	
-	private void HandleUnregisterWeb(UnregisterWebMessage message) {
-		connection.Close();
+	private Task HandleUnregisterWeb(UnregisterWebMessage message) {
+		return connection.CloseSession();
 	}
 	
 	private Task<Optional<LogInSuccess>> HandleLogIn(LogInMessage message) {

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

@@ -1,67 +0,0 @@
-using Phantom.Common.Messages.Web;
-using Phantom.Common.Messages.Web.ToController;
-using Phantom.Common.Messages.Web.ToWeb;
-using Phantom.Controller.Minecraft;
-using Phantom.Controller.Services.Agents;
-using Phantom.Controller.Services.Events;
-using Phantom.Controller.Services.Instances;
-using Phantom.Controller.Services.Users;
-using Phantom.Controller.Services.Users.Sessions;
-using Phantom.Utils.Actor;
-using Phantom.Utils.Logging;
-using Phantom.Utils.Rpc;
-using Serilog;
-
-namespace Phantom.Controller.Services.Rpc;
-
-sealed class WebRegistrationHandler : IRegistrationHandler<IMessageToWeb, IMessageToController, RegisterWebMessage> {
-	private static readonly ILogger Logger = PhantomLogger.Create<WebRegistrationHandler>();
-	
-	private readonly AuthToken webAuthToken;
-	private readonly ControllerState controllerState;
-	private readonly InstanceLogManager instanceLogManager;
-	private readonly UserManager userManager;
-	private readonly RoleManager roleManager;
-	private readonly UserRoleManager userRoleManager;
-	private readonly UserLoginManager userLoginManager;
-	private readonly AuditLogManager auditLogManager;
-	private readonly AgentManager agentManager;
-	private readonly MinecraftVersions minecraftVersions;
-	private readonly EventLogManager eventLogManager;
-	
-	public WebRegistrationHandler(AuthToken webAuthToken, ControllerState controllerState, InstanceLogManager instanceLogManager, UserManager userManager, RoleManager roleManager, UserRoleManager userRoleManager, UserLoginManager userLoginManager, AuditLogManager auditLogManager, AgentManager agentManager, MinecraftVersions minecraftVersions, EventLogManager eventLogManager) {
-		this.webAuthToken = webAuthToken;
-		this.controllerState = controllerState;
-		this.userManager = userManager;
-		this.roleManager = roleManager;
-		this.userRoleManager = userRoleManager;
-		this.userLoginManager = userLoginManager;
-		this.auditLogManager = auditLogManager;
-		this.agentManager = agentManager;
-		this.minecraftVersions = minecraftVersions;
-		this.eventLogManager = eventLogManager;
-		this.instanceLogManager = instanceLogManager;
-	}
-	
-	async Task<Props<IMessageToController>?> IRegistrationHandler<IMessageToWeb, IMessageToController, RegisterWebMessage>.TryRegister(RpcConnectionToClient<IMessageToWeb> connection, RegisterWebMessage message) {
-		return await TryRegisterImpl(connection, message) ? CreateMessageHandlerActorProps(connection) : null;
-	}
-	
-	public async Task<bool> TryRegisterImpl(RpcConnectionToClient<IMessageToWeb> connection, RegisterWebMessage message) {
-		if (webAuthToken.FixedTimeEquals(message.AuthToken)) {
-			Logger.Information("Web authorized successfully.");
-			await connection.Send(new RegisterWebResultMessage(true));
-			return true;
-		}
-		else {
-			Logger.Warning("Web failed to authorize, invalid token.");
-			await connection.Send(new RegisterWebResultMessage(false));
-			return false;
-		}
-	}
-	
-	private Props<IMessageToController> CreateMessageHandlerActorProps(RpcConnectionToClient<IMessageToWeb> connection) {
-		var init = new WebMessageHandlerActor.Init(connection, this, controllerState, instanceLogManager, userManager, roleManager, userRoleManager, userLoginManager, auditLogManager, agentManager, minecraftVersions, eventLogManager);
-		return WebMessageHandlerActor.Factory(init);
-	}
-}

Controller/Phantom.Controller/Program.cs

@@ -1,4 +1,6 @@
 using System.Reflection;
+using Phantom.Common.Messages.Agent;
+using Phantom.Common.Messages.Web;
 using Phantom.Controller;
 using Phantom.Controller.Database.Postgres;
 using Phantom.Controller.Services;
@@ -7,6 +9,8 @@ using Phantom.Utils.Logging;
 using Phantom.Utils.Rpc.Runtime.Server;
 using Phantom.Utils.Runtime;
 using Phantom.Utils.Tasks;
+using RpcAgentServer = Phantom.Utils.Rpc.Runtime.Server.RpcServer<Phantom.Common.Messages.Agent.IMessageToController, Phantom.Common.Messages.Agent.IMessageToAgent>;
+using RpcWebServer = Phantom.Utils.Rpc.Runtime.Server.RpcServer<Phantom.Common.Messages.Web.IMessageToController, Phantom.Common.Messages.Web.IMessageToWeb>;
 
 var shutdownCancellationTokenSource = new CancellationTokenSource();
 var shutdownCancellationToken = shutdownCancellationTokenSource.Token;
@@ -53,12 +57,28 @@ try {
 	
 	var dbContextFactory = new ApplicationDbContextFactory(sqlConnectionString);
 	
-	using var controllerServices = new ControllerServices(dbContextFactory, agentKeyData.AuthToken, webKeyData.AuthToken, shutdownCancellationToken);
+	using var controllerServices = new ControllerServices(dbContextFactory, agentKeyData.AuthToken, shutdownCancellationToken);
 	await controllerServices.Initialize();
 	
+	var agentConnectionParameters = new RpcServerConnectionParameters(
+		EndPoint: agentRpcServerHost,
+		Certificate: agentKeyData.Certificate,
+		AuthToken: agentKeyData.AuthToken,
+		SendQueueCapacity: 100,
+		PingInterval: TimeSpan.FromSeconds(10)
+	);
+	
+	var webConnectionParameters = new RpcServerConnectionParameters(
+		EndPoint: webRpcServerHost,
+		Certificate: webKeyData.Certificate,
+		AuthToken: webKeyData.AuthToken,
+		SendQueueCapacity: 500,
+		PingInterval: TimeSpan.FromMinutes(1)
+	);
+	
 	LinkedTasks<bool> rpcServerTasks = new LinkedTasks<bool>([
-		new RpcServer("Agent", agentRpcServerHost, agentKeyData.AuthToken, agentKeyData.Certificate).Run(shutdownCancellationToken),
-		new RpcServer("Web", webRpcServerHost, agentKeyData.AuthToken, webKeyData.Certificate).Run(shutdownCancellationToken),
+		new RpcAgentServer("Agent", agentConnectionParameters, AgentMessageRegistries.Definitions, controllerServices.AgentRegistrar).Run(shutdownCancellationToken),
+		new RpcWebServer("Web", webConnectionParameters, WebMessageRegistries.Definitions, controllerServices.WebRegistrar).Run(shutdownCancellationToken),
 	]);
 	
 	// If either RPC server crashes, stop the whole process.
@@ -71,11 +91,6 @@ try {
 	}
 	
 	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

@@ -26,7 +26,7 @@ sealed record Variables(
 		
 		EndPoint webRpcServerHost = new IPEndPoint(
 			EnvironmentVariables.GetIpAddress("WEB_RPC_SERVER_HOST").WithDefault(IPAddress.Any),
-			EnvironmentVariables.GetPortNumber("WEB_RPC_SERVER_PORT").WithDefault(9401)
+			EnvironmentVariables.GetPortNumber("WEB_RPC_SERVER_PORT").WithDefault(9402)
 		);
 		
 		return new Variables(

Packages.props

@@ -23,7 +23,6 @@
   <ItemGroup>
     <PackageReference Update="BCrypt.Net-Next.StrongName" Version="4.0.3" />
     <PackageReference Update="MemoryPack"                 Version="1.10.0" />
-    <PackageReference Update="NetMQ"                      Version="4.0.1.13" />
   </ItemGroup>
   
   <ItemGroup>

Utils/Phantom.Utils.Logging/PhantomLogger.cs

@@ -31,29 +31,48 @@ public static class PhantomLogger {
 	}
 	
 	public static ILogger Create<T>() {
-		return Create(typeof(T).Name);
+		return Create(TypeName<T>());
 	}
 	
 	public static ILogger Create<T>(string name) {
-		return Create(ConcatNames(typeof(T).Name, name));
+		return Create(ConcatNames(TypeName<T>(), name));
 	}
 	
 	public static ILogger Create<T>(string name1, string name2) {
-		return Create(ConcatNames(typeof(T).Name, ConcatNames(name1, name2)));
+		return Create(ConcatNames(TypeName<T>(), name1, name2));
 	}
 	
 	public static ILogger Create<T1, T2>() {
-		return Create(ConcatNames(typeof(T1).Name, typeof(T2).Name));
+		return Create(ConcatNames(TypeName<T1>(), TypeName<T2>()));
 	}
 	
 	public static ILogger Create<T1, T2>(string name) {
-		return Create(ConcatNames(typeof(T1).Name, ConcatNames(typeof(T2).Name, name)));
+		return Create(ConcatNames(TypeName<T1>(), TypeName<T2>(), name));
 	}
 	
-	private static string ConcatNames(string name1, string name2) {
+	public static ILogger Create<T1, T2>(string name1, string name2) {
+		return Create(ConcatNames(TypeName<T1>(), TypeName<T2>(), ConcatNames(name1, name2)));
+	}
+	
+	private static string TypeName<T>() {
+		string typeName = typeof(T).Name;
+		int genericsStartIndex = typeName.IndexOf('`');
+		return genericsStartIndex > 0 ? typeName[..genericsStartIndex] : typeName;
+	}
+	
+	public static string ConcatNames(string name1, string name2) {
 		return name1 + ":" + name2;
 	}
 	
+	public static string ConcatNames(string name1, string name2, string name3) {
+		return ConcatNames(name1, ConcatNames(name2, name3));
+	}
+	
+	public static string ShortenGuid(Guid guid) {
+		var prefix = guid.ToString();
+		return prefix[..prefix.IndexOf('-')];
+	}
+	
 	public static void Dispose() {
 		Root.Dispose();
 		Base.Dispose();

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

@@ -4,11 +4,13 @@ namespace Phantom.Utils.Rpc.Frame;
 
 interface IFrame {
 	private const byte TypePingId = 0;
-	private const byte TypeMessageId = 1;
-	private const byte TypeReplyId = 2;
-	private const byte TypeErrorId = 3;
+	private const byte TypePongId = 1;
+	private const byte TypeMessageId = 2;
+	private const byte TypeReplyId = 3;
+	private const byte TypeErrorId = 4;
 	
 	static readonly ReadOnlyMemory<byte> TypePing = new ([TypePingId]);
+	static readonly ReadOnlyMemory<byte> TypePong = new ([TypePongId]);
 	static readonly ReadOnlyMemory<byte> TypeMessage = new ([TypeMessageId]);
 	static readonly ReadOnlyMemory<byte> TypeReply = new ([TypeReplyId]);
 	static readonly ReadOnlyMemory<byte> TypeError = new ([TypeErrorId]);
@@ -21,11 +23,18 @@ interface IFrame {
 			
 			switch (oneByteBuffer[0]) {
 				case TypePingId:
+					var pingTime = await PingFrame.Read(stream, cancellationToken);
+					await reader.OnPing(pingTime, cancellationToken);
+					break;
+				
+				case TypePongId:
+					var pongFrame = await PongFrame.Read(stream, cancellationToken);
+					reader.OnPongFrame(pongFrame);
 					break;
 				
 				case TypeMessageId:
 					var messageFrame = await MessageFrame.Read(stream, cancellationToken);
-					await reader.OnMessageFrame(messageFrame, stream, cancellationToken);
+					await reader.OnMessageFrame(messageFrame, cancellationToken);
 					break;
 				
 				case TypeReplyId:
@@ -45,7 +54,7 @@ interface IFrame {
 		}
 	}
 	
-	ReadOnlyMemory<byte> Type { get; }
+	ReadOnlyMemory<byte> FrameType { get; }
 	
 	Task Write(Stream stream, CancellationToken cancellationToken = default);
 }

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

@@ -3,7 +3,9 @@
 namespace Phantom.Utils.Rpc.Frame;
 
 interface IFrameReader {
-	Task OnMessageFrame(MessageFrame frame, Stream stream, CancellationToken cancellationToken);
+	ValueTask OnPing(DateTimeOffset pingTime, CancellationToken cancellationToken);
+	void OnPongFrame(PongFrame frame);
+	Task OnMessageFrame(MessageFrame frame, CancellationToken cancellationToken);
 	void OnReplyFrame(ReplyFrame frame);
 	void OnErrorFrame(ErrorFrame frame);
 	void OnUnknownFrameStart(byte id);

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

@@ -3,7 +3,7 @@
 namespace Phantom.Utils.Rpc.Frame.Types;
 
 sealed record ErrorFrame(uint ReplyingToMessageId, RpcError Error) : IFrame {
-	public ReadOnlyMemory<byte> Type => IFrame.TypeError;
+	public ReadOnlyMemory<byte> FrameType => IFrame.TypeError;
 	
 	public async Task Write(Stream stream, CancellationToken cancellationToken) {
 		await Serialization.WriteUnsignedInt(ReplyingToMessageId, stream, cancellationToken);

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

@@ -5,7 +5,7 @@ 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 ReadOnlyMemory<byte> FrameType => IFrame.TypeMessage;
 	
 	public async Task Write(Stream stream, CancellationToken cancellationToken) {
 		int messageLength = SerializedMessage.Length;

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

@@ -1,11 +1,15 @@
 namespace Phantom.Utils.Rpc.Frame.Types;
 
 sealed record PingFrame : IFrame {
-	public static PingFrame Instance { get; } = new PingFrame();
+	public static PingFrame Instance { get; } = new ();
 	
-	public ReadOnlyMemory<byte> Type => IFrame.TypePing;
+	public ReadOnlyMemory<byte> FrameType => IFrame.TypePing;
 	
-	public Task Write(Stream stream, CancellationToken cancellationToken) {
-		return Task.CompletedTask;
+	public async Task Write(Stream stream, CancellationToken cancellationToken) {
+		await Serialization.WriteSignedLong(DateTimeOffset.UtcNow.ToUnixTimeMilliseconds(), stream, cancellationToken);
+	}
+	
+	public static async Task<DateTimeOffset> Read(Stream stream, CancellationToken cancellationToken) {
+		return DateTimeOffset.FromUnixTimeMilliseconds(await Serialization.ReadSignedLong(stream, cancellationToken));
 	}
 }

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

@@ -0,0 +1,13 @@
+namespace Phantom.Utils.Rpc.Frame.Types;
+
+sealed record PongFrame(DateTimeOffset PingTime) : IFrame {
+	public ReadOnlyMemory<byte> FrameType => IFrame.TypePong;
+	
+	public async Task Write(Stream stream, CancellationToken cancellationToken) {
+		await Serialization.WriteSignedLong(PingTime.ToUnixTimeMilliseconds(), stream, cancellationToken);
+	}
+	
+	public static async Task<PongFrame> Read(Stream stream, CancellationToken cancellationToken) {
+		return new PongFrame(DateTimeOffset.FromUnixTimeMilliseconds(await Serialization.ReadSignedLong(stream, cancellationToken)));
+	}
+}

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

@@ -5,7 +5,7 @@ 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 ReadOnlyMemory<byte> FrameType => IFrame.TypeReply;
 	
 	public async Task Write(Stream stream, CancellationToken cancellationToken) {
 		int replyLength = SerializedReply.Length;

Utils/Phantom.Utils.Rpc/IRpcListener.cs

@@ -0,0 +1,5 @@
+namespace Phantom.Utils.Rpc;
+
+public interface IRpcListener {
+	
+}

Utils/Phantom.Utils.Rpc/IRpcReplySender.cs

@@ -0,0 +1,8 @@
+using Phantom.Utils.Rpc.Runtime;
+
+namespace Phantom.Utils.Rpc;
+
+interface IRpcReplySender {
+	ValueTask SendReply<TReply>(uint replyingToMessageId, TReply reply, CancellationToken cancellationToken);
+	ValueTask SendError(uint replyingToMessageId, RpcError error, CancellationToken cancellationToken);
+}

Utils/Phantom.Utils.Rpc/Message/IMessageReceiver.cs

@@ -0,0 +1,21 @@
+using Phantom.Utils.Actor;
+
+namespace Phantom.Utils.Rpc.Message;
+
+public interface IMessageReceiver<TMessageBase> {
+	void OnPing();
+	void OnMessage(TMessageBase message);
+	Task<TReply> OnMessage<TMessage, TReply>(TMessage message, CancellationToken cancellationToken = default) where TMessage : TMessageBase, ICanReply<TReply>;
+	
+	class Actor(ActorRef<TMessageBase> actor) : IMessageReceiver<TMessageBase> {
+		public virtual void OnPing() {}
+		
+		public void OnMessage(TMessageBase message) {
+			actor.Tell(message);
+		}
+		
+		public Task<TReply> OnMessage<TMessage, TReply>(TMessage message, CancellationToken cancellationToken = default) where TMessage : TMessageBase, ICanReply<TReply> {
+			return actor.Request(message, cancellationToken);
+		}
+	}
+}

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

@@ -1,11 +0,0 @@
-using Phantom.Utils.Actor;
-using Phantom.Utils.Rpc.Runtime;
-
-namespace Phantom.Utils.Rpc.Message;
-
-interface MessageHandler<TMessageBase> {
-	ActorRef<TMessageBase> Actor { get; }
-	
-	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,4 +1,5 @@
-using Phantom.Utils.Actor;
+using System.Diagnostics.CodeAnalysis;
+using Phantom.Utils.Actor;
 using Phantom.Utils.Rpc.Frame.Types;
 using Phantom.Utils.Rpc.Runtime;
 using Serilog;
@@ -8,7 +9,7 @@ namespace Phantom.Utils.Rpc.Message;
 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, Func<uint, ReadOnlyMemory<byte>, MessageHandler<TMessageBase>, CancellationToken, Task>> codeToHandlerMapping = new ();
+	private readonly Dictionary<ushort, Func<uint, ReadOnlyMemory<byte>, RpcMessageHandler<TMessageBase>, CancellationToken, Task>> codeToHandlerMapping = new ();
 	
 	public void Add<TMessage>(ushort code) where TMessage : TMessageBase {
 		if (HasReplyType(typeof(TMessage))) {
@@ -36,6 +37,10 @@ public sealed class MessageRegistry<TMessageBase>(ILogger logger) {
 		return messageType.GetInterfaces().Any(type => type.FullName is {} name && name.StartsWith(replyInterfaceName, StringComparison.Ordinal));
 	}
 	
+	internal bool TryGetType(MessageFrame frame, [NotNullWhen(true)] out Type? type) {
+		return codeToTypeMapping.TryGetValue(frame.RegistryCode, out type);
+	}
+	
 	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));
@@ -45,7 +50,7 @@ public sealed class MessageRegistry<TMessageBase>(ILogger logger) {
 		}
 	}
 	
-	internal async Task Handle(MessageFrame frame, MessageHandler<TMessageBase> handler, CancellationToken cancellationToken) {
+	internal async Task Handle(MessageFrame frame, RpcMessageHandler<TMessageBase> handler, CancellationToken cancellationToken) {
 		uint messageId = frame.MessageId;
 		
 		if (codeToHandlerMapping.TryGetValue(frame.RegistryCode, out var action)) {
@@ -53,47 +58,47 @@ public sealed class MessageRegistry<TMessageBase>(ILogger logger) {
 		}
 		else {
 			logger.Error("Unknown message code {Code} for message {MessageId}.", frame.RegistryCode, messageId);
-			await handler.OnError(messageId, RpcError.UnknownMessageRegistryCode, cancellationToken);
+			await handler.SendError(messageId, RpcError.UnknownMessageRegistryCode, cancellationToken);
 		}
 	}
 	
-	private async Task DeserializationHandler<TMessage>(uint messageId, ReadOnlyMemory<byte> serializedMessage, MessageHandler<TMessageBase> handler, CancellationToken cancellationToken) where TMessage : TMessageBase {
+	private async Task DeserializationHandler<TMessage>(uint messageId, ReadOnlyMemory<byte> serializedMessage, RpcMessageHandler<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);
+			await handler.SendError(messageId, RpcError.MessageDeserializationError, cancellationToken);
 			return;
 		}
 		
-		handler.Actor.Tell(message);
+		handler.Receiver.OnMessage(message);
 	}
 	
-	private async Task DeserializationHandler<TMessage, TReply>(uint messageId, ReadOnlyMemory<byte> serializedMessage, MessageHandler<TMessageBase> handler, CancellationToken cancellationToken) where TMessage : TMessageBase, ICanReply<TReply> {
+	private async Task DeserializationHandler<TMessage, TReply>(uint messageId, ReadOnlyMemory<byte> serializedMessage, RpcMessageHandler<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);
+			await handler.SendError(messageId, RpcError.MessageDeserializationError, cancellationToken);
 			return;
 		}
 		
 		TReply reply;
 		try {
-			reply = await handler.Actor.Request(message, cancellationToken);
+			reply = await handler.Receiver.OnMessage<TMessage, TReply>(message, cancellationToken);
 		} catch (Exception e) {
 			logger.Error(e, "Could not handle message {MessageId} ({MessageType}).", messageId, typeof(TMessage).Name);
-			await handler.OnError(messageId, RpcError.MessageHandlingError, cancellationToken);
+			await handler.SendError(messageId, RpcError.MessageHandlingError, cancellationToken);
 			return;
 		}
 		
 		try {
-			await handler.OnReply<TMessage, TReply>(messageId, reply, cancellationToken);
+			await handler.SendReply(messageId, reply, cancellationToken);
 		} catch (Exception e) {
 			logger.Error(e, "Could not reply to message {MessageId} ({MessageType}).", messageId, typeof(TMessage).Name);
-			await handler.OnError(messageId, RpcError.MessageHandlingError, cancellationToken);
+			await handler.SendError(messageId, RpcError.MessageHandlingError, cancellationToken);
 		}
 	}
 }

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

@@ -8,7 +8,7 @@ namespace Phantom.Utils.Rpc.Message;
 
 sealed class MessageReplyTracker {
 	private readonly ILogger logger;
-	private readonly ConcurrentDictionary<uint, TaskCompletionSource<ReadOnlyMemory<byte>>> replyTasks = new (concurrencyLevel: 4, capacity: 16);
+	private readonly ConcurrentDictionary<uint, TaskCompletionSource<ReadOnlyMemory<byte>>> replyTasks = new (concurrencyLevel: 2, capacity: 16);
 	
 	internal MessageReplyTracker(string loggerName) {
 		this.logger = PhantomLogger.Create<MessageReplyTracker>(loggerName);

Utils/Phantom.Utils.Rpc/Phantom.Utils.Rpc.csproj

@@ -7,7 +7,6 @@
   
   <ItemGroup>
     <PackageReference Include="MemoryPack" />
-    <PackageReference Include="NetMQ" />
     <PackageReference Include="Serilog" />
   </ItemGroup>
   

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

@@ -1,7 +1,4 @@
-using Phantom.Utils.Actor;
-using Phantom.Utils.Logging;
-using Phantom.Utils.Rpc.Frame;
-using Phantom.Utils.Rpc.Frame.Types;
+using Phantom.Utils.Logging;
 using Phantom.Utils.Rpc.Message;
 using Serilog;
 
@@ -9,67 +6,37 @@ namespace Phantom.Utils.Rpc.Runtime.Client;
 
 public sealed class RpcClient<TClientToServerMessage, TServerToClientMessage> : IDisposable {
 	public static async Task<RpcClient<TClientToServerMessage, TServerToClientMessage>?> Connect(string loggerName, RpcClientConnectionParameters connectionParameters, IMessageDefinitions<TClientToServerMessage, TServerToClientMessage> messageDefinitions, CancellationToken cancellationToken) {
-		RpcClientConnector connector = new RpcClientConnector(loggerName, connectionParameters);
-		RpcClientConnector.Connection? connection = await connector.EstablishNewConnection(cancellationToken);
+		RpcClientToServerConnector connector = new RpcClientToServerConnector(loggerName, connectionParameters);
+		RpcClientToServerConnector.Connection? connection = await connector.EstablishNewConnection(cancellationToken);
 		return connection == null ? null : new RpcClient<TClientToServerMessage, TServerToClientMessage>(loggerName, connectionParameters, messageDefinitions, connector, connection);
 	}
 	
+	private readonly string loggerName;
 	private readonly ILogger logger;
 	private readonly MessageRegistry<TServerToClientMessage> serverToClientMessageRegistry;
-	private readonly RpcClientConnection connection;
+	private readonly RpcClientToServerConnection connection;
 	
 	public RpcSendChannel<TClientToServerMessage> SendChannel { get; }
 	
-	private RpcClient(string loggerName, RpcClientConnectionParameters connectionParameters, IMessageDefinitions<TClientToServerMessage, TServerToClientMessage> messageDefinitions, RpcClientConnector connector, RpcClientConnector.Connection connection) {
+	private RpcClient(string loggerName, RpcClientConnectionParameters connectionParameters, IMessageDefinitions<TClientToServerMessage, TServerToClientMessage> messageDefinitions, RpcClientToServerConnector connector, RpcClientToServerConnector.Connection connection) {
+		this.loggerName = loggerName;
 		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, connectionParameters, this.connection, messageDefinitions.ToServer);
+		this.connection = new RpcClientToServerConnection(loggerName, connector, connection);
+		this.SendChannel = new RpcSendChannel<TClientToServerMessage>(loggerName, connectionParameters.Common, this.connection, messageDefinitions.ToServer);
 	}
 	
-	public async Task Listen(ActorRef<TServerToClientMessage> actor) {
+	public async Task Listen(IMessageReceiver<TServerToClientMessage> receiver) {
+		var messageHandler = new RpcMessageHandler<TServerToClientMessage>(receiver, SendChannel);
+		var frameReader = new RpcFrameReader<TClientToServerMessage, TServerToClientMessage>(loggerName, serverToClientMessageRegistry, messageHandler, SendChannel);
 		try {
-			await connection.ReadConnection(stream => Receive(stream, new MessageHandlerImpl(SendChannel, actor)));
+			await connection.ReadConnection(frameReader, CancellationToken.None);
 		} 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...");
 		

Utils/Phantom.Utils.Rpc/Runtime/Client/RpcClientConnectionParameters.cs

@@ -10,4 +10,6 @@ public readonly record struct RpcClientConnectionParameters(
 	AuthToken AuthToken,
 	ushort SendQueueCapacity,
 	TimeSpan PingInterval
-);
+) {
+	internal RpcCommonConnectionParameters Common => new (SendQueueCapacity, PingInterval);
+}

Utils/Phantom.Utils.Rpc/Runtime/Client/RpcClientToServerConnection.cs

@@ -1,13 +1,14 @@
 using Phantom.Utils.Logging;
+using Phantom.Utils.Rpc.Frame;
 using Serilog;
 
 namespace Phantom.Utils.Rpc.Runtime.Client;
 
-sealed class RpcClientConnection(string loggerName, RpcClientConnector connector, RpcClientConnector.Connection initialConnection) : IRpcConnectionProvider, IDisposable {
-	private readonly ILogger logger = PhantomLogger.Create<RpcClientConnection>(loggerName);
+sealed class RpcClientToServerConnection(string loggerName, RpcClientToServerConnector connector, RpcClientToServerConnector.Connection initialConnection) : IRpcConnectionProvider, IDisposable {
+	private readonly ILogger logger = PhantomLogger.Create<RpcClientToServerConnection>(loggerName);
 	
 	private readonly SemaphoreSlim semaphore = new (1);
-	private RpcClientConnector.Connection currentConnection = initialConnection;
+	private RpcClientToServerConnector.Connection currentConnection = initialConnection;
 	
 	private readonly CancellationTokenSource newConnectionCancellationTokenSource = new ();
 	
@@ -15,7 +16,7 @@ sealed class RpcClientConnection(string loggerName, RpcClientConnector connector
 		return (await GetConnection()).Stream;
 	}
 	
-	private async Task<RpcClientConnector.Connection> GetConnection() {
+	private async Task<RpcClientToServerConnector.Connection> GetConnection() {
 		CancellationToken cancellationToken = newConnectionCancellationTokenSource.Token;
 		
 		await semaphore.WaitAsync(cancellationToken);
@@ -30,8 +31,8 @@ sealed class RpcClientConnection(string loggerName, RpcClientConnector connector
 		}
 	}
 	
-	public async Task ReadConnection(Func<Stream, Task> reader) {
-		RpcClientConnector.Connection? connection = null;
+	public async Task ReadConnection(IFrameReader frameReader, CancellationToken cancellationToken) {
+		RpcClientToServerConnector.Connection? connection = null;
 		
 		try {
 			while (true) {
@@ -48,7 +49,7 @@ sealed class RpcClientConnection(string loggerName, RpcClientConnector connector
 				}
 				
 				try {
-					await reader(connection.Stream);
+					await IFrame.ReadFrom(connection.Stream, frameReader, cancellationToken);
 				} catch (OperationCanceledException) {
 					throw;
 				} catch (EndOfStreamException) {

Utils/Phantom.Utils.Rpc/Runtime/Client/RpcClientToServerConnector.cs

@@ -9,7 +9,7 @@ using Serilog;
 
 namespace Phantom.Utils.Rpc.Runtime.Client;
 
-internal sealed class RpcClientConnector {
+internal sealed class RpcClientToServerConnector {
 	private static readonly TimeSpan InitialRetryDelay = TimeSpan.FromMilliseconds(100);
 	private static readonly TimeSpan MaximumRetryDelay = TimeSpan.FromSeconds(30);
 	private static readonly TimeSpan DisconnectTimeout = TimeSpan.FromSeconds(10);
@@ -21,8 +21,8 @@ internal sealed class RpcClientConnector {
 	
 	private bool loggedCertificateValidationError = false;
 	
-	public RpcClientConnector(string loggerName, RpcClientConnectionParameters parameters) {
-		this.logger = PhantomLogger.Create<RpcClientConnector>(loggerName);
+	public RpcClientToServerConnector(string loggerName, RpcClientConnectionParameters parameters) {
+		this.logger = PhantomLogger.Create<RpcClientToServerConnector>(loggerName);
 		this.sessionId = Guid.NewGuid();
 		this.parameters = parameters;
 		

Utils/Phantom.Utils.Rpc/Runtime/RpcCommonConnectionParameters.cs

@@ -0,0 +1,6 @@
+namespace Phantom.Utils.Rpc.Runtime;
+
+readonly record struct RpcCommonConnectionParameters(
+	ushort SendQueueCapacity,
+	TimeSpan PingInterval
+);

Utils/Phantom.Utils.Rpc/Runtime/RpcFrameReader.cs

@@ -0,0 +1,47 @@
+using Phantom.Utils.Logging;
+using Phantom.Utils.Rpc.Frame;
+using Phantom.Utils.Rpc.Frame.Types;
+using Phantom.Utils.Rpc.Message;
+using Serilog;
+
+namespace Phantom.Utils.Rpc.Runtime;
+
+sealed class RpcFrameReader<TSentMessage, TReceivedMessage>(
+	string loggerName,
+	MessageRegistry<TReceivedMessage> messageRegistry,
+	RpcMessageHandler<TReceivedMessage> messageHandler,
+	RpcSendChannel<TSentMessage> sendChannel
+) : IFrameReader {
+	private readonly ILogger logger = PhantomLogger.Create<RpcFrameReader<TSentMessage, TReceivedMessage>>(loggerName);
+	
+	public ValueTask OnPing(DateTimeOffset pingTime, CancellationToken cancellationToken) {
+		messageHandler.OnPing();
+		return sendChannel.SendPong(pingTime, cancellationToken);
+	}
+	
+	public void OnPongFrame(PongFrame frame) {
+		sendChannel.ReceivePong(frame);
+	}
+	
+	public Task OnMessageFrame(MessageFrame frame, CancellationToken cancellationToken) {
+		if (messageRegistry.TryGetType(frame, out var messageType)) {
+			logger.Verbose("Received message {MesageId} of type {MessageType} ({Bytes} B).", frame.MessageId, messageType.Name, frame.SerializedMessage.Length);
+		}
+		
+		return messageRegistry.Handle(frame, messageHandler, cancellationToken);
+	}
+	
+	public void OnReplyFrame(ReplyFrame frame) {
+		logger.Verbose("Received reply to message {MesageId} ({Bytes} B).", frame.ReplyingToMessageId, frame.SerializedReply.Length);
+		sendChannel.ReceiveReply(frame);
+	}
+	
+	public void OnErrorFrame(ErrorFrame frame) {
+		logger.Warning("Received error response to message {MesageId}: {Error}", frame.ReplyingToMessageId, frame.Error);
+		sendChannel.ReceiveError(frame.ReplyingToMessageId, frame.Error);
+	}
+	
+	public void OnUnknownFrameStart(byte id) {
+		logger.Error("Received unknown frame ID: {FrameId}", id);
+	}
+}

Utils/Phantom.Utils.Rpc/Runtime/RpcMessageHandler.cs

@@ -0,0 +1,19 @@
+using Phantom.Utils.Rpc.Message;
+
+namespace Phantom.Utils.Rpc.Runtime;
+
+sealed class RpcMessageHandler<TMessageBase>(IMessageReceiver<TMessageBase> receiver, IRpcReplySender replySender) {
+	public IMessageReceiver<TMessageBase> Receiver => receiver;
+	
+	public void OnPing() {
+		receiver.OnPing();
+	}
+	
+	public ValueTask SendReply<TReply>(uint messageId, TReply reply, CancellationToken cancellationToken) {
+		return replySender.SendReply(messageId, reply, cancellationToken);
+	}
+	
+	public ValueTask SendError(uint messageId, RpcError error, CancellationToken cancellationToken) {
+		return replySender.SendError(messageId, error, cancellationToken);
+	}
+}

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

@@ -5,12 +5,11 @@ 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.Client;
 using Serilog;
 
 namespace Phantom.Utils.Rpc.Runtime;
 
-public sealed class RpcSendChannel<TMessageBase> : IDisposable {
+public sealed class RpcSendChannel<TMessageBase> : IRpcReplySender, IDisposable {
 	private readonly ILogger logger;
 	private readonly IRpcConnectionProvider connectionProvider;
 	private readonly MessageRegistry<TMessageBase> messageRegistry;
@@ -23,8 +22,9 @@ public sealed class RpcSendChannel<TMessageBase> : IDisposable {
 	private readonly CancellationTokenSource pingCancellationTokenSource = new ();
 	
 	private uint nextMessageId;
+	private TaskCompletionSource<DateTimeOffset>? pongTask;
 	
-	internal RpcSendChannel(string loggerName, RpcClientConnectionParameters connectionParameters, IRpcConnectionProvider connectionProvider, MessageRegistry<TMessageBase> messageRegistry) {
+	internal RpcSendChannel(string loggerName, RpcCommonConnectionParameters connectionParameters, IRpcConnectionProvider connectionProvider, MessageRegistry<TMessageBase> messageRegistry) {
 		this.logger = PhantomLogger.Create<RpcSendChannel<TMessageBase>>(loggerName);
 		this.connectionProvider = connectionProvider;
 		this.messageRegistry = messageRegistry;
@@ -41,11 +41,15 @@ public sealed class RpcSendChannel<TMessageBase> : IDisposable {
 		this.pingTask = Ping(connectionParameters.PingInterval);
 	}
 	
+	internal async ValueTask SendPong(DateTimeOffset pingTime, CancellationToken cancellationToken) {
+		await SendFrame(new PongFrame(pingTime), cancellationToken);
+	}
+	
 	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 {
+	public async ValueTask SendMessage<TMessage>(TMessage message, CancellationToken cancellationToken = default) where TMessage : TMessageBase {
 		await SendFrame(NextMessageFrame(message), cancellationToken);
 	}
 	
@@ -64,11 +68,11 @@ public sealed class RpcSendChannel<TMessageBase> : IDisposable {
 		return await messageReplyTracker.WaitForReply<TReply>(messageId, waitForReplyTime, cancellationToken);
 	}
 	
-	internal async ValueTask SendReply<TReply>(uint replyingToMessageId, TReply reply, CancellationToken cancellationToken) {
+	async ValueTask IRpcReplySender.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) {
+	async ValueTask IRpcReplySender.SendError(uint replyingToMessageId, RpcError error, CancellationToken cancellationToken) {
 		await SendFrame(new ErrorFrame(replyingToMessageId, error), cancellationToken);
 	}
 	
@@ -86,18 +90,17 @@ public sealed class RpcSendChannel<TMessageBase> : IDisposable {
 	private async Task ProcessSendQueue() {
 		await foreach (IFrame frame in sendQueue.Reader.ReadAllAsync()) {
 			while (true) {
-				Stream stream;
 				try {
-					stream = await connectionProvider.GetStream();
+					Stream stream = await connectionProvider.GetStream();
+					await stream.WriteAsync(frame.FrameType);
+					await frame.Write(stream);
+					await stream.FlushAsync();
+					break;
 				} catch (OperationCanceledException) {
 					throw;
 				} catch (Exception) {
-					continue;
+					// Retry.
 				}
-				
-				await stream.WriteAsync(frame.Type);
-				await frame.Write(stream);
-				await stream.FlushAsync();
 			}
 		}
 	}
@@ -109,10 +112,23 @@ public sealed class RpcSendChannel<TMessageBase> : IDisposable {
 		while (true) {
 			await Task.Delay(interval, cancellationToken);
 			
+			pongTask = new TaskCompletionSource<DateTimeOffset>();
+			
 			if (!sendQueue.Writer.TryWrite(PingFrame.Instance)) {
 				logger.Warning("Skipped a ping due to a full queue.");
+				continue;
+			}
+			
+			DateTimeOffset pingTime = await pongTask.Task;
+			DateTimeOffset currentTime = DateTimeOffset.UtcNow;
+			
+			TimeSpan roundTripTime = currentTime - pingTime;
+			logger.Information("Received pong (rtt: {RoundTripTime} ms).", (long) roundTripTime.TotalMilliseconds);
 		}
 	}
+	
+	internal void ReceivePong(PongFrame frame) {
+		pongTask?.TrySetResult(frame.PingTime);
 	}
 	
 	internal void ReceiveReply(ReplyFrame frame) {

Utils/Phantom.Utils.Rpc/Runtime/Server/IRpcServerClientRegistrar.cs

@@ -0,0 +1,7 @@
+using Phantom.Utils.Rpc.Message;
+
+namespace Phantom.Utils.Rpc.Runtime.Server;
+
+public interface IRpcServerClientRegistrar<TClientToServerMessage, TServerToClientMessage> {
+	IMessageReceiver<TClientToServerMessage> Register(RpcServerToClientConnection<TClientToServerMessage, TServerToClientMessage> connection);
+}

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

@@ -4,17 +4,25 @@ using System.Net.Sockets;
 using System.Security.Cryptography.X509Certificates;
 using Phantom.Utils.Logging;
 using Phantom.Utils.Monads;
+using Phantom.Utils.Rpc.Message;
 using Phantom.Utils.Rpc.Runtime.Tls;
 using Serilog;
 
 namespace Phantom.Utils.Rpc.Runtime.Server;
 
-public sealed class RpcServer(string loggerName, EndPoint endPoint, AuthToken authToken, RpcServerCertificate certificate) {
-	private readonly ILogger logger = PhantomLogger.Create<RpcServer>(loggerName);
-	private readonly RpcServerClientManager clientManager = new ();
+public sealed class RpcServer<TClientToServerMessage, TServerToClientMessage>(
+	string loggerName,
+	RpcServerConnectionParameters connectionParameters,
+	IMessageDefinitions<TClientToServerMessage, TServerToClientMessage> messageDefinitions,
+	IRpcServerClientRegistrar<TClientToServerMessage, TServerToClientMessage> clientRegistrar
+) {
+	private readonly ILogger logger = PhantomLogger.Create<RpcServer<TClientToServerMessage, TServerToClientMessage>>(loggerName);
+	private readonly RpcServerClientSessions<TServerToClientMessage> clientSessions = new (loggerName, connectionParameters.Common, messageDefinitions.ToClient);
 	private readonly List<Client> clients = [];
 	
 	public async Task<bool> Run(CancellationToken shutdownToken) {
+		EndPoint endPoint = connectionParameters.EndPoint;
+		
 		SslServerAuthenticationOptions sslOptions = new () {
 			AllowRenegotiation = false,
 			AllowTlsResume = true,
@@ -22,7 +30,7 @@ public sealed class RpcServer(string loggerName, EndPoint endPoint, AuthToken au
 			ClientCertificateRequired = false,
 			EnabledSslProtocols = TlsSupport.SupportedProtocols,
 			EncryptionPolicy = EncryptionPolicy.RequireEncryption,
-			ServerCertificate = certificate.Certificate,
+			ServerCertificate = connectionParameters.Certificate.Certificate,
 		};
 		
 		try {
@@ -39,11 +47,13 @@ public sealed class RpcServer(string loggerName, EndPoint endPoint, AuthToken au
 			try {
 				logger.Information("Server listening on {EndPoint}.", endPoint);
 				
-				while (!shutdownToken.IsCancellationRequested) {
+				while (true) {
 					Socket clientSocket = await serverSocket.AcceptAsync(shutdownToken);
-					clients.Add(new Client(clientManager, clientSocket, sslOptions, authToken, shutdownToken));
+					clients.Add(new Client(loggerName, messageDefinitions, clientRegistrar, clientSessions, clientSocket, sslOptions, connectionParameters.AuthToken, shutdownToken));
 					clients.RemoveAll(static client => client.Task.IsCompleted);
 				}
+			} catch (OperationCanceledException) {
+				// Ignore.
 			} finally {
 				await Stop(serverSocket);
 			}
@@ -67,6 +77,7 @@ public sealed class RpcServer(string loggerName, EndPoint endPoint, AuthToken au
 		
 		try {
 			await Task.WhenAll(clients.Select(static client => client.Task));
+			await clientSessions.Shutdown();
 		} catch (Exception) {
 			// Ignore exceptions when shutting down.
 		}
@@ -75,26 +86,54 @@ public sealed class RpcServer(string loggerName, EndPoint endPoint, AuthToken au
 	}
 	
 	private sealed class Client {
-		private static readonly TimeSpan DisconnectTimeout = TimeSpan.FromSeconds(10);
+		private static TimeSpan DisconnectTimeout => TimeSpan.FromSeconds(10);
+		
+		private static string GetAddressDescriptor(Socket socket) {
+			EndPoint? endPoint = socket.RemoteEndPoint;
+			
+			return endPoint switch {
+				IPEndPoint ip => ip.Port.ToString(),
+				null          => "{unknown}",
+				_             => "{" + endPoint + "}",
+			};
+		}
 		
 		public Task Task { get; }
 		
 		private string Address => socket.RemoteEndPoint?.ToString() ?? "<unknown address>";
 		
 		private readonly ILogger logger;
-		private readonly RpcServerClientManager clientManager;
+		private readonly string serverLoggerName;
+		
+		private readonly IMessageDefinitions<TClientToServerMessage, TServerToClientMessage> messageDefinitions;
+		private readonly IRpcServerClientRegistrar<TClientToServerMessage, TServerToClientMessage> clientRegistrar;
+		private readonly RpcServerClientSessions<TServerToClientMessage> clientSessions;
 		private readonly Socket socket;
 		private readonly SslServerAuthenticationOptions sslOptions;
 		private readonly AuthToken authToken;
 		private readonly CancellationToken shutdownToken;
 		
-		public Client(RpcServerClientManager clientManager, Socket socket, SslServerAuthenticationOptions sslOptions, AuthToken authToken, CancellationToken shutdownToken) {
-			this.logger = PhantomLogger.Create<RpcServer, Client>(Address);
-			this.clientManager = clientManager;
+		public Client(
+			string serverLoggerName,
+			IMessageDefinitions<TClientToServerMessage, TServerToClientMessage> messageDefinitions,
+			IRpcServerClientRegistrar<TClientToServerMessage, TServerToClientMessage> clientRegistrar,
+			RpcServerClientSessions<TServerToClientMessage> clientSessions,
+			Socket socket,
+			SslServerAuthenticationOptions sslOptions,
+			AuthToken authToken,
+			CancellationToken shutdownToken
+		) {
+			this.logger = PhantomLogger.Create<RpcServer<TClientToServerMessage, TServerToClientMessage>, Client>(PhantomLogger.ConcatNames(serverLoggerName, GetAddressDescriptor(socket)));
+			this.serverLoggerName = serverLoggerName;
+			
+			this.messageDefinitions = messageDefinitions;
+			this.clientRegistrar = clientRegistrar;
+			this.clientSessions = clientSessions;
 			this.socket = socket;
 			this.sslOptions = sslOptions;
 			this.authToken = authToken;
 			this.shutdownToken = shutdownToken;
+			
 			this.Task = Run();
 		}
 		
@@ -103,21 +142,18 @@ public sealed class RpcServer(string loggerName, EndPoint endPoint, AuthToken au
 			try {
 				await using var stream = new SslStream(new NetworkStream(socket, ownsSocket: false), leaveInnerStreamOpen: false);
 				
-				Guid? sessionId;
+				Guid? sessionIdResult;
 				try {
-					sessionId = await InitializeConnection(stream);
+					sessionIdResult = await InitializeConnection(stream);
 				} catch (OperationCanceledException) {
 					logger.Warning("Cancelling incoming client due to shutdown.");
 					return;
 				}
 				
-				if (!sessionId.HasValue) {
-					return;
-				}
-				
+				if (sessionIdResult.HasValue) {
 					logger.Information("Client connected.");
-				
-				clientManager.SetConnection(sessionId.Value, stream);
+					await RunConnectedSession(sessionIdResult.Value, stream);
+				}
 			} finally {
 				logger.Information("Disconnecting client...");
 				try {
@@ -129,6 +165,7 @@ public sealed class RpcServer(string loggerName, EndPoint endPoint, AuthToken au
 					logger.Error(e, "Could not disconnect client socket.");
 				} finally {
 					socket.Close();
+					logger.Information("Client socket closed.");
 				}
 			}
 		}
@@ -180,5 +217,16 @@ public sealed class RpcServer(string loggerName, EndPoint endPoint, AuthToken au
 			var sessionId = await Serialization.ReadGuid(stream, cancellationToken);
 			return Either.Left(sessionId);
 		}
+		
+		private async Task RunConnectedSession(Guid sessionId, Stream stream) {
+			var loggerName = PhantomLogger.ConcatNames(serverLoggerName, clientSessions.NextLoggerName(sessionId));
+			var session = clientSessions.OnConnected(sessionId, stream);
+			try {
+				var connection = new RpcServerToClientConnection<TClientToServerMessage, TServerToClientMessage>(loggerName, clientSessions, sessionId, messageDefinitions.ToServer, stream, session);
+				await connection.Listen(clientRegistrar.Register(connection));
+			} finally {
+				clientSessions.OnDisconnected(sessionId);
+			}
+		}
 	}
 }

Utils/Phantom.Utils.Rpc/Runtime/Server/RpcServerClientConnection.cs

@@ -1,4 +0,0 @@
-namespace Phantom.Utils.Rpc.Runtime.Server;
-
-public sealed class RpcServerClientConnection {
-}

Utils/Phantom.Utils.Rpc/Runtime/Server/RpcServerClientManager.cs

@@ -1,13 +0,0 @@
-namespace Phantom.Utils.Rpc.Runtime.Server;
-
-sealed class RpcServerClientManager {
-	private readonly Dictionary<Guid, RpcServerClientConnection> connectionsBySessionId = new ();
-	
-	internal void SetConnection(Guid sessionId, Stream stream) {
-		connectionsBySessionId.AddOrUpdate(sessionId, id => {
-			return new RpcServerClientConnection(id, stream);
-		}, connection => {
-			return connection;
-		});
-	}
-}

Utils/Phantom.Utils.Rpc/Runtime/Server/RpcServerClientSession.cs

@@ -0,0 +1,48 @@
+using Phantom.Utils.Rpc.Message;
+
+namespace Phantom.Utils.Rpc.Runtime.Server;
+
+sealed class RpcServerClientSession<TServerToClientMessage> : IRpcConnectionProvider {
+	public RpcSendChannel<TServerToClientMessage> SendChannel { get; }
+	
+	private TaskCompletionSource<Stream> nextStream = new ();
+	
+	public RpcServerClientSession(string loggerName, RpcCommonConnectionParameters connectionParameters, MessageRegistry<TServerToClientMessage> messageRegistry) {
+		this.SendChannel = new RpcSendChannel<TServerToClientMessage>(loggerName, connectionParameters, this, messageRegistry);
+	}
+	
+	public void OnConnected(Stream stream) {
+		lock (this) {
+			if (!nextStream.Task.IsCanceled && !nextStream.TrySetResult(stream)) {
+				nextStream = new TaskCompletionSource<Stream>();
+				nextStream.SetResult(stream);
+			}
+		}
+	}
+	
+	public void OnDisconnected() {
+		lock (this) {
+			var task = nextStream.Task;
+			if (task is { IsCompleted: true, IsCanceled: false }) {
+				nextStream = new TaskCompletionSource<Stream>();
+			}
+		}
+	}
+	
+	Task<Stream> IRpcConnectionProvider.GetStream() {
+		lock (this) {
+			return nextStream.Task;
+		}
+	}
+	
+	public Task Close() {
+		lock (this) {
+			if (!nextStream.TrySetCanceled()) {
+				nextStream = new TaskCompletionSource<Stream>();
+				nextStream.SetCanceled();
+			}
+		}
+		
+		return SendChannel.Close();
+	}
+}

Utils/Phantom.Utils.Rpc/Runtime/Server/RpcServerClientSessions.cs

@@ -0,0 +1,50 @@
+using System.Collections.Concurrent;
+using Phantom.Utils.Logging;
+using Phantom.Utils.Rpc.Message;
+
+namespace Phantom.Utils.Rpc.Runtime.Server;
+
+sealed class RpcServerClientSessions<TServerToClientMessage>(string loggerName, RpcCommonConnectionParameters connectionParameters, MessageRegistry<TServerToClientMessage> messageRegistry) {
+	private readonly ConcurrentDictionary<Guid, RpcServerClientSession<TServerToClientMessage>> sessionsById = new ();
+	private readonly ConcurrentDictionary<Guid, uint> sessionLoggerSequenceIds = new ();
+	
+	public string NextLoggerName(Guid sessionId) {
+		string name = PhantomLogger.ShortenGuid(sessionId);
+		return name + "/" + sessionLoggerSequenceIds.AddOrUpdate(sessionId, static _ => 1, static (_, prev) => prev + 1);
+	}
+	
+	public RpcSendChannel<TServerToClientMessage> OnConnected(Guid sessionId, Stream stream) {
+		var session = sessionsById.GetOrAdd(sessionId, CreateSession);
+		session.OnConnected(stream);
+		return session.SendChannel;
+	}
+	
+	private RpcServerClientSession<TServerToClientMessage> CreateSession(Guid sessionId) {
+		return new RpcServerClientSession<TServerToClientMessage>(PhantomLogger.ConcatNames(loggerName, sessionId.ToString()), connectionParameters, messageRegistry);
+	}
+	
+	public void OnDisconnected(Guid sessionId) {
+		if (sessionsById.TryGetValue(sessionId, out var session)) {
+			session.OnDisconnected();
+		}
+	}
+	
+	public Task CloseSession(Guid sessionId) {
+		if (sessionsById.Remove(sessionId, out var session)) {
+			return session.Close();
+		}
+		else {
+			return Task.CompletedTask;
+		}
+	}
+	
+	public Task Shutdown() {
+		List<Task> tasks = [];
+		
+		foreach (Guid guid in sessionsById.Keys) {
+			tasks.Add(CloseSession(guid));
+		}
+		
+		return Task.WhenAll(tasks);
+	}
+}

Utils/Phantom.Utils.Rpc/Runtime/Server/RpcServerConnectionParameters.cs

@@ -0,0 +1,14 @@
+using System.Net;
+using Phantom.Utils.Rpc.Runtime.Tls;
+
+namespace Phantom.Utils.Rpc.Runtime.Server;
+
+public readonly record struct RpcServerConnectionParameters(
+	EndPoint EndPoint,
+	RpcServerCertificate Certificate,
+	AuthToken AuthToken,
+	ushort SendQueueCapacity,
+	TimeSpan PingInterval
+) {
+	internal RpcCommonConnectionParameters Common => new (SendQueueCapacity, PingInterval);
+}

Utils/Phantom.Utils.Rpc/Runtime/Server/RpcServerToClientConnection.cs

@@ -0,0 +1,38 @@
+using Phantom.Utils.Rpc.Frame;
+using Phantom.Utils.Rpc.Message;
+
+namespace Phantom.Utils.Rpc.Runtime.Server;
+
+public sealed class RpcServerToClientConnection<TClientToServerMessage, TServerToClientMessage> {
+	private readonly string loggerName;
+	private readonly RpcServerClientSessions<TServerToClientMessage> sessions;
+	private readonly MessageRegistry<TClientToServerMessage> messageRegistry;
+	private readonly Stream stream;
+	
+	public Guid SessionId { get; }
+	public RpcSendChannel<TServerToClientMessage> SendChannel { get; }
+	
+	internal RpcServerToClientConnection(string loggerName, RpcServerClientSessions<TServerToClientMessage> sessions, Guid sessionId, MessageRegistry<TClientToServerMessage> messageRegistry, Stream stream, RpcSendChannel<TServerToClientMessage> sendChannel) {
+		this.loggerName = loggerName;
+		this.sessions = sessions;
+		this.messageRegistry = messageRegistry;
+		this.stream = stream;
+		
+		this.SessionId = sessionId;
+		this.SendChannel = sendChannel;
+	}
+	
+	public Task CloseSession() {
+		return sessions.CloseSession(SessionId);
+	}
+	
+	internal async Task Listen(IMessageReceiver<TClientToServerMessage> receiver) {
+		var messageHandler = new RpcMessageHandler<TClientToServerMessage>(receiver, SendChannel);
+		var frameReader = new RpcFrameReader<TServerToClientMessage, TClientToServerMessage>(loggerName, messageRegistry, messageHandler, SendChannel);
+		try {
+			await IFrame.ReadFrom(stream, frameReader, CancellationToken.None);
+		} catch (OperationCanceledException) {
+			// Ignore.
+		}
+	}
+}

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

@@ -1,135 +0,0 @@
-// using System.Collections.Concurrent;
-// using Akka.Actor;
-// using NetMQ.Sockets;
-// using Phantom.Utils.Actor;
-// using Phantom.Utils.Logging;
-// using Serilog;
-// using Serilog.Events;
-//
-// namespace Phantom.Utils.Rpc.Runtime2;
-//
-// public static class RpcServerRuntime {
-// 	public static Task Launch<TClientMessage, TServerMessage, TRegistrationMessage, TReplyMessage>(
-// 		RpcConfiguration config,
-// 		IMessageDefinitions<TClientMessage, TServerMessage, TReplyMessage> messageDefinitions,
-// 		IRegistrationHandler<TClientMessage, TServerMessage, TRegistrationMessage> registrationHandler,
-// 		IActorRefFactory actorSystem,
-// 		CancellationToken cancellationToken
-// 	) where TRegistrationMessage : TServerMessage where TReplyMessage : TClientMessage, TServerMessage {
-// 		return RpcServerRuntime<TClientMessage, TServerMessage, TRegistrationMessage, TReplyMessage>.Launch(config, messageDefinitions, registrationHandler, actorSystem, cancellationToken);
-// 	}
-// }
-//
-// internal sealed class RpcServerRuntime<TClientMessage, TServerMessage, TRegistrationMessage, TReplyMessage> : RpcRuntime<ServerSocket> where TRegistrationMessage : TServerMessage where TReplyMessage : TClientMessage, TServerMessage {
-// 	internal static Task Launch(RpcConfiguration config, IMessageDefinitions<TClientMessage, TServerMessage, TReplyMessage> messageDefinitions, IRegistrationHandler<TClientMessage, TServerMessage, TRegistrationMessage> registrationHandler, IActorRefFactory actorSystem, CancellationToken cancellationToken) {
-// 		var socket = RpcServerSocket.Connect(config);
-// 		return new RpcServerRuntime<TClientMessage, TServerMessage, TRegistrationMessage, TReplyMessage>(socket, messageDefinitions, registrationHandler, actorSystem, cancellationToken).Launch();
-// 	}
-// 	
-// 	private readonly string serviceName;
-// 	private readonly IMessageDefinitions<TClientMessage, TServerMessage, TReplyMessage> messageDefinitions;
-// 	private readonly IRegistrationHandler<TClientMessage, TServerMessage, TRegistrationMessage> registrationHandler;
-// 	private readonly IActorRefFactory actorSystem;
-// 	private readonly CancellationToken cancellationToken;
-// 	
-// 	private RpcServerRuntime(RpcServerSocket socket, IMessageDefinitions<TClientMessage, TServerMessage, TReplyMessage> messageDefinitions, IRegistrationHandler<TClientMessage, TServerMessage, TRegistrationMessage> registrationHandler, IActorRefFactory actorSystem, CancellationToken cancellationToken) : base(socket) {
-// 		this.serviceName = socket.Config.ServiceName;
-// 		this.messageDefinitions = messageDefinitions;
-// 		this.registrationHandler = registrationHandler;
-// 		this.actorSystem = actorSystem;
-// 		this.cancellationToken = cancellationToken;
-// 	}
-// 	
-// 	private protected override Task Run(ServerSocket socket) {
-// 		var clients = new ConcurrentDictionary<ulong, Client>();
-// 		
-// 		void OnConnectionClosed(object? sender, RpcClientConnectionClosedEventArgs e) {
-// 			if (clients.Remove(e.RoutingId, out var client)) {
-// 				client.Connection.Closed -= OnConnectionClosed;
-// 			}
-// 		}
-// 		
-// 		while (!cancellationToken.IsCancellationRequested) {
-// 			var (routingId, data) = socket.Receive(cancellationToken);
-// 			
-// 			if (data.Length == 0) {
-// 				LogUnknownMessage(routingId, data);
-// 				continue;
-// 			}
-// 			
-// 			Type? messageType = messageDefinitions.ToServer.TryGetType(data, out var type) ? type : null;
-// 			if (messageType == null) {
-// 				LogUnknownMessage(routingId, data);
-// 				continue;
-// 			}
-// 			
-// 			if (!clients.TryGetValue(routingId, out var client)) {
-// 				if (messageType != typeof(TRegistrationMessage)) {
-// 					RuntimeLogger.Warning("Received {MessageType} ({Bytes} B) from unregistered client {RoutingId}.", messageType.Name, data.Length, routingId);
-// 					continue;
-// 				}
-// 				
-// 				var clientLoggerName = LoggerName + ":" + routingId;
-// 				var clientActorName = "Rpc-" + serviceName + "-" + routingId;
-// 				
-// 				// TODO add pings and tear down connection after too much inactivity
-// 				var connection = new RpcConnectionToClient<TClientMessage>(socket, routingId, messageDefinitions.ToClient, ReplyTracker);
-// 				connection.Closed += OnConnectionClosed;
-// 				
-// 				client = new Client(clientLoggerName, clientActorName, connection, actorSystem, messageDefinitions, registrationHandler);
-// 				clients[routingId] = client;
-// 			}
-// 			
-// 			client.Enqueue(messageType, data);
-// 		}
-// 		
-// 		foreach (var client in clients.Values) {
-// 			client.Connection.Close();
-// 		}
-// 		
-// 		return Task.CompletedTask;
-// 	}
-// 	
-// 	private void LogUnknownMessage(uint routingId, ReadOnlyMemory<byte> data) {
-// 		RuntimeLogger.Warning("Received unknown message ({Bytes} B) from {RoutingId}.", data.Length, routingId);
-// 	}
-// 	
-// 	private protected override Task Disconnect(ServerSocket socket) {
-// 		return Task.CompletedTask;
-// 	}
-// 	
-// 	private sealed class Client {
-// 		public RpcConnectionToClient<TClientMessage> Connection { get; }
-// 		
-// 		private readonly ILogger logger;
-// 		private readonly ActorRef<RpcReceiverActor<TClientMessage, TServerMessage, TRegistrationMessage, TReplyMessage>.ReceiveMessageCommand> receiverActor;
-// 		
-// 		public Client(string loggerName, string actorName, RpcConnectionToClient<TClientMessage> connection, IActorRefFactory actorSystem, IMessageDefinitions<TClientMessage, TServerMessage, TReplyMessage> messageDefinitions, IRegistrationHandler<TClientMessage, TServerMessage, TRegistrationMessage> registrationHandler) {
-// 			this.Connection = connection;
-// 			this.Connection.Closed += OnConnectionClosed;
-// 			
-// 			this.logger = PhantomLogger.Create(loggerName);
-// 			
-// 			var receiverActorInit = new RpcReceiverActor<TClientMessage, TServerMessage, TRegistrationMessage, TReplyMessage>.Init(loggerName, messageDefinitions, registrationHandler, Connection);
-// 			this.receiverActor = actorSystem.ActorOf(RpcReceiverActor<TClientMessage, TServerMessage, TRegistrationMessage, TReplyMessage>.Factory(receiverActorInit), actorName + "-Receiver");
-// 		}
-// 		
-// 		internal void Enqueue(Type messageType, ReadOnlyMemory<byte> data) {
-// 			LogMessageType(messageType, data);
-// 			receiverActor.Tell(new RpcReceiverActor<TClientMessage, TServerMessage, TRegistrationMessage, TReplyMessage>.ReceiveMessageCommand(messageType, data));
-// 		}
-// 		
-// 		private void LogMessageType(Type messageType, ReadOnlyMemory<byte> data) {
-// 			if (logger.IsEnabled(LogEventLevel.Verbose)) {
-// 				logger.Verbose("Received {MessageType} ({Bytes} B).", messageType.Name, data.Length);
-// 			}
-// 		}
-// 		
-// 		private void OnConnectionClosed(object? sender, RpcClientConnectionClosedEventArgs e) {
-// 			Connection.Closed -= OnConnectionClosed;
-// 			
-// 			logger.Debug("Closing connection...");
-// 			receiverActor.Stop();
-// 		}
-// 	}
-// }

Utils/Phantom.Utils.Rpc/Serialization.cs

@@ -52,6 +52,14 @@ static class Serialization {
 		return ReadValue(BinaryPrimitives.ReadUInt32LittleEndian, sizeof(uint), stream, cancellationToken);
 	}
 	
+	public static ValueTask WriteSignedLong(long value, Stream stream, CancellationToken cancellationToken) {
+		return WriteValue(value, sizeof(long), BinaryPrimitives.WriteInt64LittleEndian, stream, cancellationToken);
+	}
+	
+	public static ValueTask<long> ReadSignedLong(Stream stream, CancellationToken cancellationToken) {
+		return ReadValue(BinaryPrimitives.ReadInt64LittleEndian, sizeof(long), stream, cancellationToken);
+	}
+	
 	public static ValueTask WriteGuid(Guid guid, Stream stream, CancellationToken cancellationToken) {
 		static void Write(Span<byte> span, Guid guid) {
 			if (!guid.TryWriteBytes(span)) {

Web/Phantom.Web.Services/PhantomWebServices.cs

@@ -14,7 +14,7 @@ namespace Phantom.Web.Services;
 public static class PhantomWebServices {
 	public static void AddPhantomServices(this IServiceCollection services) {
 		services.AddSingleton<ControllerConnection>();
-		services.AddSingleton<ControllerMessageHandlerFactory>();
+		services.AddSingleton<ControllerMessageHandlerActorInitFactory>();
 		
 		services.AddSingleton<AgentManager>();
 		services.AddSingleton<InstanceManager>();

Web/Phantom.Web.Services/Rpc/ControllerConnection.cs

@@ -6,7 +6,7 @@ namespace Phantom.Web.Services.Rpc;
 
 public sealed class ControllerConnection(RpcSendChannel<IMessageToController> connection) {
 	public ValueTask Send<TMessage>(TMessage message) where TMessage : IMessageToController {
-		return connection.SendMessage(message, CancellationToken.None);
+		return connection.SendMessage(message);
 	}
 	
 	public Task<TReply> Send<TMessage, TReply>(TMessage message, TimeSpan waitForReplyTime, CancellationToken waitForReplyCancellationToken = default) where TMessage : IMessageToController, ICanReply<TReply> {

Web/Phantom.Web.Services/Rpc/ControllerMessageHandlerActor.cs

@@ -7,13 +7,12 @@ using Phantom.Web.Services.Instances;
 
 namespace Phantom.Web.Services.Rpc;
 
-sealed class ControllerMessageHandlerActor : ReceiveActor<IMessageToWeb> {
+public sealed class ControllerMessageHandlerActor : ReceiveActor<IMessageToWeb> {
 	public readonly record struct Init(
 		AgentManager AgentManager,
 		InstanceManager InstanceManager,
 		InstanceLogManager InstanceLogManager,
-		UserSessionRefreshManager UserSessionRefreshManager,
-		TaskCompletionSource<bool> RegisterSuccessWaiter
+		UserSessionRefreshManager UserSessionRefreshManager
 	);
 	
 	public static Props<IMessageToWeb> Factory(Init init) {
@@ -24,26 +23,19 @@ sealed class ControllerMessageHandlerActor : ReceiveActor<IMessageToWeb> {
 	private readonly InstanceManager instanceManager;
 	private readonly InstanceLogManager instanceLogManager;
 	private readonly UserSessionRefreshManager userSessionRefreshManager;
-	private readonly TaskCompletionSource<bool> registerSuccessWaiter;
 	
 	private ControllerMessageHandlerActor(Init init) {
 		this.agentManager = init.AgentManager;
 		this.instanceManager = init.InstanceManager;
 		this.instanceLogManager = init.InstanceLogManager;
 		this.userSessionRefreshManager = init.UserSessionRefreshManager;
-		this.registerSuccessWaiter = init.RegisterSuccessWaiter;
 		
-		Receive<RegisterWebResultMessage>(HandleRegisterWebResult);
 		Receive<RefreshAgentsMessage>(HandleRefreshAgents);
 		Receive<RefreshInstancesMessage>(HandleRefreshInstances);
 		Receive<InstanceOutputMessage>(HandleInstanceOutput);
 		Receive<RefreshUserSessionMessage>(HandleRefreshUserSession);
 	}
 	
-	private void HandleRegisterWebResult(RegisterWebResultMessage message) {
-		registerSuccessWaiter.TrySetResult(message.Success);
-	}
-	
 	private void HandleRefreshAgents(RefreshAgentsMessage message) {
 		agentManager.RefreshAgents(message.Agents);
 	}

Web/Phantom.Web.Services/Rpc/ControllerMessageHandlerActorInitFactory.cs

@@ -0,0 +1,16 @@
+using Phantom.Web.Services.Agents;
+using Phantom.Web.Services.Authentication;
+using Phantom.Web.Services.Instances;
+
+namespace Phantom.Web.Services.Rpc;
+
+public sealed class ControllerMessageHandlerActorInitFactory(
+	AgentManager agentManager,
+	InstanceManager instanceManager,
+	InstanceLogManager instanceLogManager,
+	UserSessionRefreshManager userSessionRefreshManager
+) {
+	public ControllerMessageHandlerActor.Init Create() {
+		return new ControllerMessageHandlerActor.Init(agentManager, instanceManager, instanceLogManager, userSessionRefreshManager);
+	}
+}

Web/Phantom.Web.Services/Rpc/ControllerMessageHandlerFactory.cs

@@ -1,28 +0,0 @@
-using Akka.Actor;
-using Phantom.Common.Messages.Web;
-using Phantom.Utils.Actor;
-using Phantom.Utils.Tasks;
-using Phantom.Web.Services.Agents;
-using Phantom.Web.Services.Authentication;
-using Phantom.Web.Services.Instances;
-
-namespace Phantom.Web.Services.Rpc;
-
-public sealed class ControllerMessageHandlerFactory(
-	AgentManager agentManager,
-	InstanceManager instanceManager,
-	InstanceLogManager instanceLogManager,
-	UserSessionRefreshManager userSessionRefreshManager
-) {
-	private readonly TaskCompletionSource<bool> registerSuccessWaiter = AsyncTasks.CreateCompletionSource<bool>();
-	
-	public Task<bool> RegisterSuccessWaiter => registerSuccessWaiter.Task;
-	
-	private int messageHandlerId = 0;
-	
-	public ActorRef<IMessageToWeb> Create(IActorRefFactory actorSystem) {
-		var init = new ControllerMessageHandlerActor.Init(agentManager, instanceManager, instanceLogManager, userSessionRefreshManager, registerSuccessWaiter);
-		var name = "ControllerMessageHandler-" + Interlocked.Increment(ref messageHandlerId);
-		return actorSystem.ActorOf(ControllerMessageHandlerActor.Factory(init), name);
-	}
-}

Web/Phantom.Web/Program.cs

@@ -5,6 +5,7 @@ using Phantom.Utils.Actor;
 using Phantom.Utils.Cryptography;
 using Phantom.Utils.IO;
 using Phantom.Utils.Logging;
+using Phantom.Utils.Rpc.Message;
 using Phantom.Utils.Rpc.Runtime.Client;
 using Phantom.Utils.Runtime;
 using Phantom.Utils.Threading;
@@ -71,11 +72,6 @@ try {
 	
 	using var actorSystem = ActorSystemFactory.Create("Web");
 	
-	ControllerMessageHandlerFactory messageHandlerFactory;
-	await using (var scope = webApplication.Services.CreateAsyncScope()) {
-		messageHandlerFactory = scope.ServiceProvider.GetRequiredService<ControllerMessageHandlerFactory>();
-	}
-	
 	Task? rpcClientListener = null;
 	try {
 		PhantomLogger.Root.InformationHeading("Launching Phantom Panel web...");
@@ -83,14 +79,22 @@ try {
 		PhantomLogger.Root.Information("For administrator setup, visit: {HttpUrl}{SetupPath}", webConfiguration.HttpUrl, webConfiguration.BasePath + "setup");
 		
 		await WebLauncher.Launch(webConfiguration, webApplication);
-		rpcClientListener = rpcClient.Listen(messageHandlerFactory.Create(actorSystem));
+		
+		ActorRef<IMessageToWeb> rpcMessageHandlerActor;
+		await using (var scope = webApplication.Services.CreateAsyncScope()) {
+			var rpcMessageHandlerInit = scope.ServiceProvider.GetRequiredService<ControllerMessageHandlerActorInitFactory>().Create();
+			rpcMessageHandlerActor = actorSystem.ActorOf(ControllerMessageHandlerActor.Factory(rpcMessageHandlerInit), "ControllerMessageHandler");
+		}
+		
+		rpcClientListener = rpcClient.Listen(new IMessageReceiver<IMessageToWeb>.Actor(rpcMessageHandlerActor));
 		
 		PhantomLogger.Root.Information("Phantom Panel web is ready.");
 		
 		await shutdownCancellationToken.WaitHandle.WaitOneAsync();
+		await webApplication.StopAsync();
 	} finally {
 		try {
-			await rpcClient.SendChannel.SendMessage(new UnregisterWebMessage(), CancellationToken.None);
+			await rpcClient.SendChannel.SendMessage(new UnregisterWebMessage());
 			// TODO wait for acknowledgment
 		} catch (Exception e) {
 			PhantomLogger.Root.Warning(e, "Could not unregister agent after shutdown.");

Web/Phantom.Web/WebLauncher.cs

@@ -62,7 +62,7 @@ static class WebLauncher {
 		application.MapFallbackToPage("/_Host");
 		
 		logger.Information("Starting Web server on port {Port}...", config.Port);
-		return application.RunAsync(config.CancellationToken);
+		return application.StartAsync(config.CancellationToken);
 	}
 	
 	private sealed class NullLifetime : IHostLifetime {