1
0
mirror of https://github.com/chylex/Minecraft-Phantom-Panel.git synced 2025-09-16 00:32:12 +02:00

20 Commits

Author SHA1 Message Date
cf609a02f8 Replace NetMQ with custom TCP logic 2025-09-15 13:56:16 +02:00
23f9e079c1 Replace NetMQ with custom TCP logic 2025-09-15 13:15:57 +02:00
8f8e0cf2e0 Replace NetMQ with custom TCP logic 2025-09-15 13:03:36 +02:00
a65f22f97d Replace NetMQ with custom TCP logic 2025-09-15 09:37:39 +02:00
9a4ac4604e Replace NetMQ with custom TCP logic 2025-09-12 16:16:24 +02:00
a8fe48ec77 Replace NetMQ with custom TCP logic 2025-09-12 01:15:20 +02:00
62d10f248d Replace NetMQ with custom TCP logic 2025-09-11 20:20:16 +02:00
3f8e46ae2d Replace NetMQ with custom TCP logic 2025-09-11 19:44:43 +02:00
584e8acfd0 Replace NetMQ with custom TCP logic 2025-09-11 10:07:57 +02:00
47e563f7ce Replace NetMQ with custom TCP logic 2025-09-10 20:15:25 +02:00
29c403f2d4 Replace NetMQ with custom TCP logic 2025-09-10 19:48:32 +02:00
2947fa3522 Replace NetMQ with custom TCP logic 2025-09-10 19:14:06 +02:00
4d1a79307f Replace NetMQ with custom TCP logic 2025-09-10 19:06:49 +02:00
cc4eb8aa9c Replace NetMQ with custom TCP logic 2025-09-10 12:23:26 +02:00
de1767c876 Replace NetMQ with custom TCP logic 2025-09-09 16:20:05 +02:00
39f2fa4b17 Replace NetMQ with custom TCP logic 2025-09-08 14:52:34 +02:00
19e0d6fd3d Replace NetMQ with custom TCP logic 2025-09-06 02:36:41 +02:00
591a6a62ab Reformat code 2025-08-21 20:31:21 +02:00
ae32537d8c Update to .NET 9 and C# 13 2025-08-08 22:04:41 +02:00
8149d31d51 Update Java versions in Dockerfile 2024-07-14 10:40:36 +02:00
330 changed files with 3775 additions and 3171 deletions

1
.gitignore vendored
View File

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

View File

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

View File

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

View File

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

View File

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

View File

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

View File

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

Binary file not shown.

View File

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

View File

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

View File

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

Binary file not shown.

View File

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

View File

@@ -105,7 +105,7 @@ public abstract class BaseLauncher : IServerLauncher {
private static async Task AcceptEula(InstanceProperties instanceProperties) {
var eulaFilePath = Path.Combine(instanceProperties.InstanceFolder, "eula.txt");
await File.WriteAllLinesAsync(eulaFilePath, new [] { "# EULA", "eula=true" }, Encoding.UTF8);
await File.WriteAllLinesAsync(eulaFilePath, new[] { "# EULA", "eula=true" }, Encoding.UTF8);
}
private static async Task UpdateServerProperties(InstanceProperties instanceProperties) {

View File

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

View File

@@ -1,44 +0,0 @@
using Phantom.Common.Messages.Agent;
using Phantom.Common.Messages.Agent.ToController;
using Phantom.Utils.Logging;
using Phantom.Utils.Rpc.Runtime;
using Serilog;
namespace Phantom.Agent.Rpc;
sealed class KeepAliveLoop {
private static readonly ILogger Logger = PhantomLogger.Create<KeepAliveLoop>();
private static readonly TimeSpan KeepAliveInterval = TimeSpan.FromSeconds(10);
private readonly RpcConnectionToServer<IMessageToController> connection;
private readonly CancellationTokenSource cancellationTokenSource = new ();
public KeepAliveLoop(RpcConnectionToServer<IMessageToController> connection) {
this.connection = connection;
Task.Run(Run);
}
private async Task Run() {
var cancellationToken = cancellationTokenSource.Token;
try {
await connection.IsReady.WaitAsync(cancellationToken);
Logger.Information("Started keep-alive loop.");
while (true) {
await Task.Delay(KeepAliveInterval, cancellationToken);
await connection.Send(new AgentIsAliveMessage()).WaitAsync(cancellationToken);
}
} catch (OperationCanceledException) {
// Ignore.
} finally {
cancellationTokenSource.Dispose();
Logger.Information("Stopped keep-alive loop.");
}
}
public void Cancel() {
cancellationTokenSource.Cancel();
}
}

View File

@@ -1,12 +0,0 @@
<Project Sdk="Microsoft.NET.Sdk">
<PropertyGroup>
<ImplicitUsings>enable</ImplicitUsings>
<Nullable>enable</Nullable>
</PropertyGroup>
<ItemGroup>
<ProjectReference Include="..\..\Common\Phantom.Common.Messages.Agent\Phantom.Common.Messages.Agent.csproj" />
</ItemGroup>
</Project>

View File

@@ -1,37 +0,0 @@
using NetMQ;
using NetMQ.Sockets;
using Phantom.Common.Messages.Agent;
using Phantom.Common.Messages.Agent.BiDirectional;
using Phantom.Common.Messages.Agent.ToController;
using Phantom.Utils.Actor;
using Phantom.Utils.Rpc.Runtime;
using Phantom.Utils.Rpc.Sockets;
using Serilog;
namespace Phantom.Agent.Rpc;
public sealed class RpcClientRuntime : RpcClientRuntime<IMessageToAgent, IMessageToController, ReplyMessage> {
public static Task Launch(RpcClientSocket<IMessageToAgent, IMessageToController, ReplyMessage> socket, ActorRef<IMessageToAgent> handlerActorRef, SemaphoreSlim disconnectSemaphore, CancellationToken receiveCancellationToken) {
return new RpcClientRuntime(socket, handlerActorRef, disconnectSemaphore, receiveCancellationToken).Launch();
}
private RpcClientRuntime(RpcClientSocket<IMessageToAgent, IMessageToController, ReplyMessage> socket, ActorRef<IMessageToAgent> handlerActor, SemaphoreSlim disconnectSemaphore, CancellationToken receiveCancellationToken) : base(socket, handlerActor, disconnectSemaphore, receiveCancellationToken) {}
protected override async Task RunWithConnection(ClientSocket socket, RpcConnectionToServer<IMessageToController> connection) {
var keepAliveLoop = new KeepAliveLoop(connection);
try {
await base.RunWithConnection(socket, connection);
} finally {
keepAliveLoop.Cancel();
}
}
protected override async Task SendDisconnectMessage(ClientSocket socket, ILogger logger) {
var unregisterMessageBytes = AgentMessageRegistries.ToController.Write(new UnregisterAgentMessage()).ToArray();
try {
await socket.SendAsync(unregisterMessageBytes).AsTask().WaitAsync(TimeSpan.FromSeconds(5), CancellationToken.None);
} catch (TimeoutException) {
logger.Error("Timed out communicating agent shutdown with the controller.");
}
}
}

View File

@@ -1,9 +1,13 @@
using Akka.Actor;
using System.Collections.Immutable;
using Akka.Actor;
using Phantom.Agent.Minecraft.Java;
using Phantom.Agent.Rpc;
using Phantom.Agent.Services.Backups;
using Phantom.Agent.Services.Instances;
using Phantom.Agent.Services.Rpc;
using Phantom.Common.Data.Agent;
using Phantom.Common.Data.Replies;
using Phantom.Common.Messages.Agent.ToAgent;
using Phantom.Common.Messages.Agent.ToController;
using Phantom.Utils.Actor;
using Phantom.Utils.Logging;
using Serilog;
@@ -15,6 +19,7 @@ public sealed class AgentServices {
public ActorSystem ActorSystem { get; }
private AgentInfo AgentInfo { get; }
private AgentFolders AgentFolders { get; }
private AgentState AgentState { get; }
private BackupManager BackupManager { get; }
@@ -26,6 +31,7 @@ public sealed class AgentServices {
public AgentServices(AgentInfo agentInfo, AgentFolders agentFolders, AgentServiceConfiguration serviceConfiguration, ControllerConnection controllerConnection) {
this.ActorSystem = ActorSystemFactory.Create("Agent");
this.AgentInfo = agentInfo;
this.AgentFolders = agentFolders;
this.AgentState = new AgentState();
this.BackupManager = new BackupManager(agentFolders, serviceConfiguration.MaxConcurrentCompressionTasks);
@@ -43,6 +49,41 @@ 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);
} catch (Exception e) {
Logger.Fatal(e, "Registration failed.");
return false;
}
foreach (var configureInstanceMessage in configureInstanceMessages) {
var configureInstanceCommand = new InstanceManagerActor.ConfigureInstanceCommand(
configureInstanceMessage.InstanceGuid,
configureInstanceMessage.Configuration,
configureInstanceMessage.LaunchProperties,
configureInstanceMessage.LaunchNow,
AlwaysReportStatus: true
);
var configureInstanceResult = await InstanceManager.Request(configureInstanceCommand, cancellationToken);
if (!configureInstanceResult.Is(ConfigureInstanceResult.Success)) {
Logger.Fatal("Unable to configure instance \"{Name}\" (GUID {Guid}), shutting down.", configureInstanceMessage.Configuration.InstanceName, configureInstanceMessage.InstanceGuid);
return false;
}
}
await connection.Send(new AdvertiseJavaRuntimesMessage(JavaRuntimeRepository.All), cancellationToken);
InstanceTicketManager.RefreshAgentStatus();
return true;
}
public async Task Shutdown() {
Logger.Information("Stopping services...");

View File

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

View File

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

View File

@@ -4,8 +4,8 @@ using Phantom.Agent.Minecraft.Launcher;
using Phantom.Agent.Minecraft.Launcher.Types;
using Phantom.Agent.Minecraft.Properties;
using Phantom.Agent.Minecraft.Server;
using Phantom.Agent.Rpc;
using Phantom.Agent.Services.Backups;
using Phantom.Agent.Services.Rpc;
using Phantom.Common.Data;
using Phantom.Common.Data.Instance;
using Phantom.Common.Data.Minecraft;

View File

@@ -1,6 +1,6 @@
using Phantom.Agent.Minecraft.Launcher;
using Phantom.Agent.Rpc;
using Phantom.Agent.Services.Backups;
using Phantom.Agent.Services.Rpc;
namespace Phantom.Agent.Services.Instances;

View File

@@ -1,11 +1,10 @@
using Phantom.Agent.Rpc;
using Phantom.Agent.Services.Rpc;
using Phantom.Common.Data;
using Phantom.Common.Data.Agent;
using Phantom.Common.Data.Instance;
using Phantom.Common.Data.Replies;
using Phantom.Common.Messages.Agent.ToController;
using Phantom.Utils.Logging;
using Phantom.Utils.Tasks;
using Serilog;
namespace Phantom.Agent.Services.Instances;
@@ -92,7 +91,7 @@ sealed class InstanceTicketManager {
public void RefreshAgentStatus() {
lock (this) {
controllerConnection.Send(new ReportAgentStatusMessage(activeTicketGuids.Count, usedMemory));
controllerConnection.TrySend(new ReportAgentStatusMessage(activeTicketGuids.Count, usedMemory));
}
}

View File

@@ -1,6 +1,6 @@
using System.Collections.Immutable;
using System.Threading.Channels;
using Phantom.Agent.Rpc;
using Phantom.Agent.Services.Rpc;
using Phantom.Common.Messages.Agent.ToController;
using Phantom.Utils.Logging;
using Phantom.Utils.Tasks;
@@ -63,7 +63,7 @@ sealed class InstanceLogSender : CancellableBackgroundTask {
private void SendOutputToServer(ImmutableArray<string> lines) {
if (!lines.IsEmpty) {
controllerConnection.Send(new InstanceOutputMessage(instanceGuid, lines));
controllerConnection.TrySend(new InstanceOutputMessage(instanceGuid, lines));
}
}

View File

@@ -1,6 +1,6 @@
using Phantom.Agent.Minecraft.Instance;
using Phantom.Agent.Minecraft.Server;
using Phantom.Agent.Rpc;
using Phantom.Agent.Services.Rpc;
using Phantom.Common.Data.Instance;
using Phantom.Common.Messages.Agent.ToController;
using Phantom.Utils.Logging;
@@ -38,7 +38,7 @@ sealed class InstancePlayerCountTracker : CancellableBackgroundTask {
}
onlinePlayerCountChanged?.Invoke(this, value?.Online);
controllerConnection.Send(new ReportInstancePlayerCountsMessage(instanceGuid, value));
controllerConnection.TrySend(new ReportInstancePlayerCountsMessage(instanceGuid, value));
}
}

View File

@@ -8,7 +8,6 @@
<ItemGroup>
<ProjectReference Include="..\..\Common\Phantom.Common.Messages.Agent\Phantom.Common.Messages.Agent.csproj" />
<ProjectReference Include="..\Phantom.Agent.Minecraft\Phantom.Agent.Minecraft.csproj" />
<ProjectReference Include="..\Phantom.Agent.Rpc\Phantom.Agent.Rpc.csproj" />
</ItemGroup>
</Project>

View File

@@ -0,0 +1,20 @@
using Phantom.Common.Messages.Agent;
using Phantom.Utils.Actor;
using Phantom.Utils.Rpc.Runtime;
namespace Phantom.Agent.Services.Rpc;
public sealed class ControllerConnection(RpcSendChannel<IMessageToController> sendChannel) {
public ValueTask Send<TMessage>(TMessage message, CancellationToken cancellationToken) where TMessage : IMessageToController {
return sendChannel.SendMessage(message, cancellationToken);
}
// TODO handle properly
public bool TrySend<TMessage>(TMessage message) where TMessage : IMessageToController {
return sendChannel.TrySendMessage(message);
}
public Task<TReply> Send<TMessage, TReply>(TMessage message, TimeSpan waitForReplyTime, CancellationToken cancellationToken) where TMessage : IMessageToController, ICanReply<TReply> {
return sendChannel.SendMessage<TMessage, TReply>(message, waitForReplyTime, cancellationToken);
}
}

View File

@@ -1,86 +1,32 @@
using Phantom.Agent.Services.Instances;
using Phantom.Common.Data;
using Phantom.Common.Data.Instance;
using Phantom.Common.Data.Replies;
using Phantom.Common.Messages.Agent;
using Phantom.Common.Messages.Agent.BiDirectional;
using Phantom.Common.Messages.Agent.ToAgent;
using Phantom.Common.Messages.Agent.ToController;
using Phantom.Utils.Actor;
using Phantom.Utils.Logging;
using Phantom.Utils.Rpc.Runtime;
using Serilog;
namespace Phantom.Agent.Services.Rpc;
public sealed class ControllerMessageHandlerActor : ReceiveActor<IMessageToAgent> {
private static ILogger Logger { get; } = PhantomLogger.Create<ControllerMessageHandlerActor>();
public readonly record struct Init(RpcConnectionToServer<IMessageToController> Connection, AgentServices Agent, CancellationTokenSource ShutdownTokenSource);
public readonly record struct Init(AgentServices Agent);
public static Props<IMessageToAgent> Factory(Init init) {
return Props<IMessageToAgent>.Create(() => new ControllerMessageHandlerActor(init), new ActorConfiguration { SupervisorStrategy = SupervisorStrategies.Resume });
}
private readonly RpcConnectionToServer<IMessageToController> connection;
private readonly AgentServices agent;
private readonly CancellationTokenSource shutdownTokenSource;
private ControllerMessageHandlerActor(Init init) {
this.connection = init.Connection;
this.agent = init.Agent;
this.shutdownTokenSource = init.ShutdownTokenSource;
ReceiveAsync<RegisterAgentSuccessMessage>(HandleRegisterAgentSuccess);
Receive<RegisterAgentFailureMessage>(HandleRegisterAgentFailure);
ReceiveAndReplyLater<ConfigureInstanceMessage, Result<ConfigureInstanceResult, InstanceActionFailure>>(HandleConfigureInstance);
ReceiveAndReplyLater<LaunchInstanceMessage, Result<LaunchInstanceResult, InstanceActionFailure>>(HandleLaunchInstance);
ReceiveAndReplyLater<StopInstanceMessage, Result<StopInstanceResult, InstanceActionFailure>>(HandleStopInstance);
ReceiveAndReplyLater<SendCommandToInstanceMessage, Result<SendCommandToInstanceResult, InstanceActionFailure>>(HandleSendCommandToInstance);
Receive<ReplyMessage>(HandleReply);
}
private async Task HandleRegisterAgentSuccess(RegisterAgentSuccessMessage message) {
Logger.Information("Agent authentication successful.");
void ShutdownAfterConfigurationFailed(Guid instanceGuid, InstanceConfiguration configuration) {
Logger.Fatal("Unable to configure instance \"{Name}\" (GUID {Guid}), shutting down.", configuration.InstanceName, instanceGuid);
shutdownTokenSource.Cancel();
}
foreach (var configureInstanceMessage in message.InitialInstanceConfigurations) {
var result = await HandleConfigureInstance(configureInstanceMessage, alwaysReportStatus: true);
if (!result.Is(ConfigureInstanceResult.Success)) {
ShutdownAfterConfigurationFailed(configureInstanceMessage.InstanceGuid, configureInstanceMessage.Configuration);
return;
}
}
connection.SetIsReady();
await connection.Send(new AdvertiseJavaRuntimesMessage(agent.JavaRuntimeRepository.All));
agent.InstanceTicketManager.RefreshAgentStatus();
}
private void HandleRegisterAgentFailure(RegisterAgentFailureMessage message) {
string errorMessage = message.FailureKind switch {
RegisterAgentFailure.ConnectionAlreadyHasAnAgent => "This connection already has an associated agent.",
RegisterAgentFailure.InvalidToken => "Invalid token.",
_ => "Unknown error " + (byte) message.FailureKind + "."
};
Logger.Fatal("Agent authentication failed: {Error}", errorMessage);
PhantomLogger.Dispose();
Environment.Exit(1);
}
private Task<Result<ConfigureInstanceResult, InstanceActionFailure>> HandleConfigureInstance(ConfigureInstanceMessage message, bool alwaysReportStatus) {
return agent.InstanceManager.Request(new InstanceManagerActor.ConfigureInstanceCommand(message.InstanceGuid, message.Configuration, message.LaunchProperties, message.LaunchNow, alwaysReportStatus));
}
private async Task<Result<ConfigureInstanceResult, InstanceActionFailure>> HandleConfigureInstance(ConfigureInstanceMessage message) {
return await HandleConfigureInstance(message, alwaysReportStatus: false);
return await agent.InstanceManager.Request(new InstanceManagerActor.ConfigureInstanceCommand(message.InstanceGuid, message.Configuration, message.LaunchProperties, message.LaunchNow, AlwaysReportStatus: false));
}
private async Task<Result<LaunchInstanceResult, InstanceActionFailure>> HandleLaunchInstance(LaunchInstanceMessage message) {
@@ -94,8 +40,4 @@ public sealed class ControllerMessageHandlerActor : ReceiveActor<IMessageToAgent
private async Task<Result<SendCommandToInstanceResult, InstanceActionFailure>> HandleSendCommandToInstance(SendCommandToInstanceMessage message) {
return await agent.InstanceManager.Request(new InstanceManagerActor.SendCommandToInstanceCommand(message.InstanceGuid, message.Command));
}
private void HandleReply(ReplyMessage message) {
connection.Receive(message);
}
}

View File

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

View File

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

View File

@@ -8,7 +8,7 @@ namespace Phantom.Common.Data;
public sealed partial class AllowedPorts {
[MemoryPackOrder(0)]
[MemoryPackInclude]
private readonly ImmutableArray<PortRange> allDefinitions;
private readonly ImmutableArray<PortRange> allDefinitions;
private AllowedPorts(ImmutableArray<PortRange> allDefinitions) {
// TODO normalize and deduplicate ranges

View File

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

View File

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

View File

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

View File

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

View File

@@ -83,7 +83,7 @@ public readonly partial record struct RamAllocationUnits(
int unitMultiplier = char.ToUpperInvariant(definition[^1]) switch {
'M' => 1,
'G' => 1024,
_ => throw new ArgumentOutOfRangeException(nameof(definition), "Must end with 'M' or 'G'.")
_ => throw new ArgumentOutOfRangeException(nameof(definition), "Must end with 'M' or 'G'.")
};
if (!int.TryParse(definition[..^1], out int size)) {

View File

@@ -1,6 +0,0 @@
namespace Phantom.Common.Data.Replies;
public enum RegisterAgentFailure : byte {
ConnectionAlreadyHasAnAgent,
InvalidToken
}

View File

@@ -1,5 +1,6 @@
using System.Diagnostics.CodeAnalysis;
using MemoryPack;
using Phantom.Utils.Monads;
using Phantom.Utils.Result;
namespace Phantom.Common.Data;
@@ -24,6 +25,9 @@ public sealed partial class Result<TValue, TError> {
[MemoryPackIgnore]
public TError Error => !hasValue ? error! : throw new InvalidOperationException("Attempted to get error from a success result.");
[MemoryPackIgnore]
public Either<TValue, TError> AsEither => hasValue ? Either.Left(value!) : Either.Right(error!);
private Result(bool hasValue, TValue? value, TError? error) {
this.hasValue = hasValue;
this.value = value;

View File

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

View File

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

View File

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

View File

@@ -1,9 +0,0 @@
using MemoryPack;
using Phantom.Common.Data.Replies;
namespace Phantom.Common.Messages.Agent.ToAgent;
[MemoryPackable(GenerateType.VersionTolerant)]
public sealed partial record RegisterAgentFailureMessage(
[property: MemoryPackOrder(0)] RegisterAgentFailure FailureKind
) : IMessageToAgent;

View File

@@ -1,9 +0,0 @@
using System.Collections.Immutable;
using MemoryPack;
namespace Phantom.Common.Messages.Agent.ToAgent;
[MemoryPackable(GenerateType.VersionTolerant)]
public sealed partial record RegisterAgentSuccessMessage(
[property: MemoryPackOrder(0)] ImmutableArray<ConfigureInstanceMessage> InitialInstanceConfigurations
) : IMessageToAgent;

View File

@@ -1,11 +1,12 @@
using MemoryPack;
using Phantom.Common.Data;
using System.Collections.Immutable;
using MemoryPack;
using Phantom.Common.Data.Agent;
using Phantom.Common.Messages.Agent.ToAgent;
using Phantom.Utils.Actor;
namespace Phantom.Common.Messages.Agent.ToController;
[MemoryPackable(GenerateType.VersionTolerant)]
public sealed partial record RegisterAgentMessage(
[property: MemoryPackOrder(0)] AuthToken AuthToken,
[property: MemoryPackOrder(1)] AgentInfo AgentInfo
) : IMessageToController;
[property: MemoryPackOrder(0)] AgentInfo AgentInfo
) : IMessageToController, ICanReply<ImmutableArray<ConfigureInstanceMessage>>;

View File

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

View File

@@ -1,9 +0,0 @@
using MemoryPack;
using Phantom.Common.Data;
namespace Phantom.Common.Messages.Web.ToController;
[MemoryPackable(GenerateType.VersionTolerant)]
public sealed partial record RegisterWebMessage(
[property: MemoryPackOrder(0)] AuthToken AuthToken
) : IMessageToController;

View File

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

View File

@@ -21,7 +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;
using Phantom.Utils.Rpc.Runtime.Server;
using Serilog;
namespace Phantom.Controller.Services.Agents;
@@ -168,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;

View File

@@ -1,36 +1,29 @@
using Phantom.Common.Messages.Agent;
using Phantom.Utils.Actor;
using Phantom.Utils.Logging;
using Phantom.Utils.Rpc.Runtime;
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();
connection.CloseSession();
return true;
}
}
@@ -45,7 +38,7 @@ sealed class AgentConnection {
return Task.CompletedTask;
}
return connection.Send(message);
return connection.SendChannel.SendMessage(message).AsTask();
}
}
@@ -53,10 +46,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)!;
}
}

View File

@@ -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.Runtime;
using Phantom.Utils.Rpc.Runtime.Server;
using Serilog;
namespace Phantom.Controller.Services.Agents;
@@ -22,7 +22,6 @@ 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;
@@ -32,9 +31,8 @@ sealed class AgentManager {
private readonly ConcurrentDictionary<Guid, ActorRef<AgentActor.ICommand>> agentsByGuid = 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;
@@ -63,18 +61,10 @@ sealed class AgentManager {
}
}
public async Task<bool> RegisterAgent(AuthToken authToken, AgentInfo agentInfo, RpcConnectionToClient<IMessageToAgent> connection) {
if (!this.authToken.FixedTimeEquals(authToken)) {
await connection.Send(new RegisterAgentFailureMessage(RegisterAgentFailure.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;
return await agentActor.Request(new AgentActor.RegisterCommand(agentProperties, connection), cancellationToken);
}
public bool TellAgent(Guid agentGuid, AgentActor.ICommand command) {

View File

@@ -1,9 +1,4 @@
using Akka.Actor;
using Phantom.Common.Data;
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,9 +8,9 @@ using Phantom.Controller.Services.Rpc;
using Phantom.Controller.Services.Users;
using Phantom.Controller.Services.Users.Sessions;
using Phantom.Utils.Actor;
using Phantom.Utils.Rpc.Runtime;
using IMessageFromAgentToController = Phantom.Common.Messages.Agent.IMessageToController;
using IMessageFromWebToController = Phantom.Common.Messages.Web.IMessageToController;
using Phantom.Utils.Rpc;
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;
@@ -38,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;
@@ -60,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() {

View File

@@ -0,0 +1,21 @@
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.Runtime.Server;
namespace Phantom.Controller.Services.Rpc;
sealed class AgentClientRegistrar(
IActorRefFactory actorSystem,
AgentManager agentManager,
InstanceLogManager instanceLogManager,
EventLogManager eventLogManager
) : IRpcServerClientRegistrar<IMessageToController, IMessageToAgent> {
public ActorRef<IMessageToController> Register(Guid sessionId, RpcServerToClientConnection<IMessageToController, IMessageToAgent> connection) {
var init = new AgentMessageHandlerActor.Init(connection, agentManager, instanceLogManager, eventLogManager);
return actorSystem.ActorOf(AgentMessageHandlerActor.Factory(init), "AgentClient-" + sessionId);
}
}

View File

@@ -1,39 +1,36 @@
using Phantom.Common.Data.Replies;
using System.Collections.Immutable;
using Phantom.Common.Messages.Agent;
using Phantom.Common.Messages.Agent.BiDirectional;
using Phantom.Common.Messages.Agent.ToAgent;
using Phantom.Common.Messages.Agent.ToController;
using Phantom.Controller.Services.Agents;
using Phantom.Controller.Services.Events;
using Phantom.Controller.Services.Instances;
using Phantom.Utils.Actor;
using Phantom.Utils.Rpc.Runtime;
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? agentGuid;
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);
@@ -42,52 +39,47 @@ sealed class AgentMessageHandlerActor : ReceiveActor<IMessageToController> {
Receive<ReportInstancePlayerCountsMessage>(HandleReportInstancePlayerCounts);
Receive<ReportInstanceEventMessage>(HandleReportInstanceEvent);
Receive<InstanceOutputMessage>(HandleInstanceOutput);
Receive<ReplyMessage>(HandleReply);
}
private async Task HandleRegisterAgent(RegisterAgentMessage message) {
if (agentGuid != message.AgentInfo.AgentGuid) {
await connection.Send(new RegisterAgentFailureMessage(RegisterAgentFailure.ConnectionAlreadyHasAnAgent));
}
else {
await agentRegistrationHandler.TryRegisterImpl(connection, message);
}
private Guid RequireAgentGuid() {
return agentGuid ?? throw new InvalidOperationException("Agent has not registered yet.");
}
private Task<ImmutableArray<ConfigureInstanceMessage>> HandleRegisterAgent(RegisterAgentMessage message) {
agentGuid = message.AgentInfo.AgentGuid;
return agentManager.RegisterAgent(message.AgentInfo, connection);
}
private void HandleUnregisterAgent(UnregisterAgentMessage message) {
agentManager.TellAgent(agentGuid, new AgentActor.UnregisterCommand(connection));
connection.Close();
agentManager.TellAgent(RequireAgentGuid(), new AgentActor.UnregisterCommand(connection));
connection.CloseSession();
}
private void HandleAgentIsAlive(AgentIsAliveMessage message) {
agentManager.TellAgent(agentGuid, new AgentActor.NotifyIsAliveCommand());
agentManager.TellAgent(RequireAgentGuid(), new AgentActor.NotifyIsAliveCommand());
}
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) {
instanceLogManager.ReceiveLines(message.InstanceGuid, message.Lines);
}
private void HandleReply(ReplyMessage message) {
connection.Receive(message);
}
}

View File

@@ -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.Runtime;
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);
}
}

View File

@@ -0,0 +1,31 @@
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.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 ActorRef<IMessageToController> Register(Guid sessionId, RpcServerToClientConnection<IMessageToController, IMessageToWeb> connection) {
var init = new WebMessageHandlerActor.Init(connection, this, controllerState, instanceLogManager, userManager, roleManager, userRoleManager, userLoginManager, auditLogManager, agentManager, minecraftVersions, eventLogManager);
return actorSystem.ActorOf(WebMessageHandlerActor.Factory(init), "WebClient-" + sessionId);
}
}

View File

@@ -10,13 +10,13 @@ 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;
@@ -70,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();
}
}

View File

@@ -7,7 +7,6 @@ using Phantom.Common.Data.Web.AuditLog;
using Phantom.Common.Data.Web.EventLog;
using Phantom.Common.Data.Web.Instance;
using Phantom.Common.Data.Web.Users;
using Phantom.Common.Messages.Agent.BiDirectional;
using Phantom.Common.Messages.Web;
using Phantom.Common.Messages.Web.ToController;
using Phantom.Controller.Minecraft;
@@ -17,14 +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;
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,
@@ -41,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;
@@ -55,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;
@@ -66,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,15 +85,10 @@ sealed class WebMessageHandlerActor : ReceiveActor<IMessageToController> {
ReceiveAndReply<GetAgentJavaRuntimesMessage, ImmutableDictionary<Guid, ImmutableArray<TaggedJavaRuntime>>>(HandleGetAgentJavaRuntimes);
ReceiveAndReplyLater<GetAuditLogMessage, Result<ImmutableArray<AuditLogItem>, UserActionFailure>>(HandleGetAuditLog);
ReceiveAndReplyLater<GetEventLogMessage, Result<ImmutableArray<EventLogItem>, UserActionFailure>>(HandleGetEventLog);
Receive<ReplyMessage>(HandleReply);
}
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) {
@@ -191,8 +182,4 @@ sealed class WebMessageHandlerActor : ReceiveActor<IMessageToController> {
private Task<Result<ImmutableArray<EventLogItem>, UserActionFailure>> HandleGetEventLog(GetEventLogMessage message) {
return eventLogManager.GetMostRecentItems(userLoginManager.GetLoggedInUser(message.AuthToken), message.Count);
}
private void HandleReply(ReplyMessage message) {
connection.Receive(message);
}
}

View File

@@ -1,68 +0,0 @@
using Phantom.Common.Data;
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.Runtime;
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);
}
}

View File

@@ -1,6 +1,7 @@
using System.Collections.Immutable;
using Phantom.Common.Data;
using Phantom.Common.Data.Web.Users;
using Phantom.Common.Data.Web.Users.CreateOrUpdateAdministratorUserResults;
using Phantom.Controller.Database;
using Phantom.Controller.Database.Entities;
using Phantom.Controller.Database.Repositories;
@@ -56,12 +57,12 @@ sealed class UserManager {
wasCreated = true;
}
else {
return new Common.Data.Web.Users.CreateOrUpdateAdministratorUserResults.CreationFailed(result.Error);
return new CreationFailed(result.Error);
}
}
else {
if (userRepository.SetUserPassword(user, password).TryGetError(out var error)) {
return new Common.Data.Web.Users.CreateOrUpdateAdministratorUserResults.UpdatingFailed(error);
return new UpdatingFailed(error);
}
auditLogWriter.AdministratorUserModified(user);
@@ -70,7 +71,7 @@ sealed class UserManager {
var role = await new RoleRepository(db).GetByGuid(Role.Administrator.Guid);
if (role == null) {
return new Common.Data.Web.Users.CreateOrUpdateAdministratorUserResults.AddingToRoleFailed();
return new AddingToRoleFailed();
}
await new UserRoleRepository(db).Add(user, role);
@@ -84,10 +85,10 @@ sealed class UserManager {
Logger.Information("Updated administrator user \"{Username}\" (GUID {Guid}).", username, user.UserGuid);
}
return new Common.Data.Web.Users.CreateOrUpdateAdministratorUserResults.Success(user.ToUserInfo());
return new Success(user.ToUserInfo());
} catch (Exception e) {
Logger.Error(e, "Could not create or update administrator user \"{Username}\".", username);
return new Common.Data.Web.Users.CreateOrUpdateAdministratorUserResults.UnknownError();
return new UnknownError();
}
}

View File

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

View File

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

View File

@@ -1,5 +1,4 @@
using System.Reflection;
using NetMQ;
using Phantom.Common.Messages.Agent;
using Phantom.Common.Messages.Web;
using Phantom.Controller;
@@ -7,9 +6,11 @@ using Phantom.Controller.Database.Postgres;
using Phantom.Controller.Services;
using Phantom.Utils.IO;
using Phantom.Utils.Logging;
using Phantom.Utils.Rpc;
using Phantom.Utils.Rpc.Runtime;
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;
@@ -37,7 +38,7 @@ try {
PhantomLogger.Root.InformationHeading("Initializing Phantom Panel controller...");
PhantomLogger.Root.Information("Controller version: {Version}", fullVersion);
var (agentRpcServerHost, agentRpcServerPort, webRpcServerHost, webRpcServerPort, sqlConnectionString) = Variables.LoadOrStop();
var (agentRpcServerHost, webRpcServerHost, sqlConnectionString) = Variables.LoadOrStop();
string secretsPath = Path.GetFullPath("./secrets");
CreateFolderOrStop(secretsPath, Chmod.URWX_GRX);
@@ -56,20 +57,37 @@ 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();
static RpcConfiguration ConfigureRpc(string serviceName, string host, ushort port, ConnectionKeyData connectionKey) {
return new RpcConfiguration(serviceName, host, port, connectionKey.Certificate);
}
var agentConnectionParameters = new RpcServerConnectionParameters(
EndPoint: agentRpcServerHost,
Certificate: agentKeyData.Certificate,
AuthToken: agentKeyData.AuthToken,
SendQueueCapacity: 100,
PingInterval: TimeSpan.FromSeconds(10)
);
try {
await Task.WhenAll(
RpcServerRuntime.Launch(ConfigureRpc("Agent", agentRpcServerHost, agentRpcServerPort, agentKeyData), AgentMessageRegistries.Definitions, controllerServices.AgentRegistrationHandler, controllerServices.ActorSystem, shutdownCancellationToken),
RpcServerRuntime.Launch(ConfigureRpc("Web", webRpcServerHost, webRpcServerPort, webKeyData), WebMessageRegistries.Definitions, controllerServices.WebRegistrationHandler, controllerServices.ActorSystem, shutdownCancellationToken)
);
} finally {
NetMQConfig.Cleanup();
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 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.
await rpcServerTasks.CancelTokenWhenAnyCompletes(shutdownCancellationTokenSource);
foreach (Task<bool> rpcServerTask in await rpcServerTasks.WaitForAll()) {
if (rpcServerTask.IsFaulted || rpcServerTask is { IsCompletedSuccessfully: true, Result: false }) {
return 1;
}
}
return 0;

View File

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

View File

@@ -1,8 +1,8 @@
<Project>
<PropertyGroup>
<TargetFramework>net8.0</TargetFramework>
<LangVersion>11</LangVersion>
<TargetFramework>net9.0</TargetFramework>
<LangVersion>13</LangVersion>
</PropertyGroup>
<PropertyGroup>

View File

@@ -1,7 +1,7 @@
# +---------------+
# | Prepare build |
# +---------------+
FROM --platform=$BUILDPLATFORM mcr.microsoft.com/dotnet/nightly/sdk:8.0 AS phantom-builder
FROM --platform=$BUILDPLATFORM mcr.microsoft.com/dotnet/nightly/sdk:9.0 AS phantom-builder
ARG TARGETARCH
ADD . /app
@@ -19,7 +19,7 @@ RUN find .artifacts/publish/*/* -maxdepth 0 -execdir mv '{}' 'release' \;
# +---------------------+
# | Phantom Agent image |
# +---------------------+
FROM mcr.microsoft.com/dotnet/nightly/runtime:8.0 AS phantom-agent
FROM mcr.microsoft.com/dotnet/nightly/runtime:9.0 AS phantom-agent
RUN mkdir /data && chmod 777 /data
WORKDIR /data
@@ -27,7 +27,7 @@ WORKDIR /data
COPY --from=eclipse-temurin:8-jre /opt/java/openjdk /opt/java/8
COPY --from=eclipse-temurin:16-jdk /opt/java/openjdk /opt/java/16
COPY --from=eclipse-temurin:17-jre /opt/java/openjdk /opt/java/17
COPY --from=eclipse-temurin:20-jre /opt/java/openjdk /opt/java/20
COPY --from=eclipse-temurin:21-jre /opt/java/openjdk /opt/java/21
ARG DEBIAN_FRONTEND=noninteractive
@@ -46,7 +46,7 @@ ENTRYPOINT ["dotnet", "/app/Phantom.Agent.dll"]
# +--------------------------+
# | Phantom Controller image |
# +--------------------------+
FROM mcr.microsoft.com/dotnet/nightly/runtime:8.0 AS phantom-controller
FROM mcr.microsoft.com/dotnet/nightly/runtime:9.0 AS phantom-controller
RUN mkdir /data && chmod 777 /data
WORKDIR /data
@@ -59,7 +59,7 @@ ENTRYPOINT ["dotnet", "/app/Phantom.Controller.dll"]
# +-------------------+
# | Phantom Web image |
# +-------------------+
FROM mcr.microsoft.com/dotnet/nightly/aspnet:8.0 AS phantom-web
FROM mcr.microsoft.com/dotnet/nightly/aspnet:9.0 AS phantom-web
RUN mkdir /data && chmod 777 /data
WORKDIR /data

View File

@@ -18,8 +18,6 @@ Project("{FAE04EC0-301F-11D3-BF4B-00C04F79EFBC}") = "Phantom.Agent", "Agent\Phan
EndProject
Project("{FAE04EC0-301F-11D3-BF4B-00C04F79EFBC}") = "Phantom.Agent.Minecraft", "Agent\Phantom.Agent.Minecraft\Phantom.Agent.Minecraft.csproj", "{9FE000D0-91AC-4CB4-8956-91CCC0270015}"
EndProject
Project("{FAE04EC0-301F-11D3-BF4B-00C04F79EFBC}") = "Phantom.Agent.Rpc", "Agent\Phantom.Agent.Rpc\Phantom.Agent.Rpc.csproj", "{665C7B87-0165-48BC-B6A6-17A3812A70C9}"
EndProject
Project("{FAE04EC0-301F-11D3-BF4B-00C04F79EFBC}") = "Phantom.Agent.Services", "Agent\Phantom.Agent.Services\Phantom.Agent.Services.csproj", "{AEE8B77E-AB07-423F-9981-8CD829ACB834}"
EndProject
Project("{FAE04EC0-301F-11D3-BF4B-00C04F79EFBC}") = "Phantom.Common.Data", "Common\Phantom.Common.Data\Phantom.Common.Data.csproj", "{6C3DB1E5-F695-4D70-8F3A-78C2957274BE}"
@@ -76,10 +74,6 @@ Global
{9FE000D0-91AC-4CB4-8956-91CCC0270015}.Debug|Any CPU.Build.0 = Debug|Any CPU
{9FE000D0-91AC-4CB4-8956-91CCC0270015}.Release|Any CPU.ActiveCfg = Release|Any CPU
{9FE000D0-91AC-4CB4-8956-91CCC0270015}.Release|Any CPU.Build.0 = Release|Any CPU
{665C7B87-0165-48BC-B6A6-17A3812A70C9}.Debug|Any CPU.ActiveCfg = Debug|Any CPU
{665C7B87-0165-48BC-B6A6-17A3812A70C9}.Debug|Any CPU.Build.0 = Debug|Any CPU
{665C7B87-0165-48BC-B6A6-17A3812A70C9}.Release|Any CPU.ActiveCfg = Release|Any CPU
{665C7B87-0165-48BC-B6A6-17A3812A70C9}.Release|Any CPU.Build.0 = Release|Any CPU
{AEE8B77E-AB07-423F-9981-8CD829ACB834}.Debug|Any CPU.ActiveCfg = Debug|Any CPU
{AEE8B77E-AB07-423F-9981-8CD829ACB834}.Debug|Any CPU.Build.0 = Debug|Any CPU
{AEE8B77E-AB07-423F-9981-8CD829ACB834}.Release|Any CPU.ActiveCfg = Release|Any CPU
@@ -164,7 +158,6 @@ Global
GlobalSection(NestedProjects) = preSolution
{418BE1BF-9F63-4B46-B4E4-DF64C3B3DDA7} = {F5878792-64C8-4ECF-A075-66341FF97127}
{9FE000D0-91AC-4CB4-8956-91CCC0270015} = {F5878792-64C8-4ECF-A075-66341FF97127}
{665C7B87-0165-48BC-B6A6-17A3812A70C9} = {F5878792-64C8-4ECF-A075-66341FF97127}
{AEE8B77E-AB07-423F-9981-8CD829ACB834} = {F5878792-64C8-4ECF-A075-66341FF97127}
{6C3DB1E5-F695-4D70-8F3A-78C2957274BE} = {01CB1A81-8950-471C-BFDF-F135FDDB2C18}
{95B55357-F8F0-48C2-A1C2-5EA997651783} = {01CB1A81-8950-471C-BFDF-F135FDDB2C18}

View File

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

View File

@@ -35,21 +35,33 @@ public static class PhantomLogger {
}
public static ILogger Create<T>(string name) {
return Create(typeof(T).Name, name);
return Create(ConcatNames(typeof(T).Name, name));
}
public static ILogger Create<T>(string name1, string name2) {
return Create(typeof(T).Name, ConcatNames(name1, name2));
return Create(ConcatNames(typeof(T).Name, name1, name2));
}
public static ILogger Create<T1, T2>() {
return Create(typeof(T1).Name, typeof(T2).Name);
return Create(ConcatNames(typeof(T1).Name, typeof(T2).Name));
}
private static string ConcatNames(string name1, string name2) {
public static ILogger Create<T1, T2>(string name) {
return Create(ConcatNames(typeof(T1).Name, typeof(T2).Name, name));
}
public static ILogger Create<T1, T2>(string name1, string name2) {
return Create(ConcatNames(typeof(T1).Name, typeof(T2).Name, ConcatNames(name1, name2)));
}
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 void Dispose() {
Root.Dispose();
Base.Dispose();

View File

@@ -0,0 +1,30 @@
using System.Collections.Immutable;
using System.Security.Cryptography;
namespace Phantom.Utils.Rpc;
public sealed class AuthToken {
public const int Length = 12;
public ImmutableArray<byte> Bytes { get; }
public AuthToken(ImmutableArray<byte> bytes) {
if (bytes.Length != Length) {
throw new ArgumentOutOfRangeException(nameof(bytes), "Invalid token length: " + bytes.Length + ". Token length must be exactly " + Length + " bytes.");
}
this.Bytes = bytes;
}
public bool FixedTimeEquals(AuthToken providedAuthToken) {
return FixedTimeEquals(providedAuthToken.Bytes.AsSpan());
}
public bool FixedTimeEquals(ReadOnlySpan<byte> other) {
return CryptographicOperations.FixedTimeEquals(Bytes.AsSpan(), other);
}
public static AuthToken Generate() {
return new AuthToken([..RandomNumberGenerator.GetBytes(Length)]);
}
}

View File

@@ -0,0 +1,51 @@
using Phantom.Utils.Rpc.Frame.Types;
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;
static readonly ReadOnlyMemory<byte> TypePing = new ([TypePingId]);
static readonly ReadOnlyMemory<byte> TypeMessage = new ([TypeMessageId]);
static readonly ReadOnlyMemory<byte> TypeReply = new ([TypeReplyId]);
static readonly ReadOnlyMemory<byte> TypeError = new ([TypeErrorId]);
internal static async Task ReadFrom(Stream stream, IFrameReader reader, CancellationToken cancellationToken) {
byte[] oneByteBuffer = new byte[1];
while (!cancellationToken.IsCancellationRequested) {
await stream.ReadExactlyAsync(oneByteBuffer, cancellationToken);
switch (oneByteBuffer[0]) {
case TypePingId:
break;
case TypeMessageId:
var messageFrame = await MessageFrame.Read(stream, cancellationToken);
await reader.OnMessageFrame(messageFrame, cancellationToken);
break;
case TypeReplyId:
var replyFrame = await ReplyFrame.Read(stream, cancellationToken);
reader.OnReplyFrame(replyFrame);
break;
case TypeErrorId:
var errorFrame = await ErrorFrame.Read(stream, cancellationToken);
reader.OnErrorFrame(errorFrame);
break;
default:
reader.OnUnknownFrameStart(oneByteBuffer[0]);
break;
}
}
}
ReadOnlyMemory<byte> FrameType { get; }
Task Write(Stream stream, CancellationToken cancellationToken = default);
}

View File

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

View File

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

View File

@@ -0,0 +1,39 @@
using Phantom.Utils.Rpc.Runtime;
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> FrameType => IFrame.TypeMessage;
public async Task Write(Stream stream, CancellationToken cancellationToken) {
int messageLength = SerializedMessage.Length;
CheckMessageLength(messageLength);
await Serialization.WriteUnsignedInt(MessageId, stream, cancellationToken);
await Serialization.WriteUnsignedShort(RegistryCode, stream, cancellationToken);
await Serialization.WriteSignedInt(messageLength, stream, cancellationToken);
await stream.WriteAsync(SerializedMessage, cancellationToken);
}
public static async Task<MessageFrame> Read(Stream stream, CancellationToken cancellationToken) {
var messageId = await Serialization.ReadUnsignedInt(stream, cancellationToken);
var registryCode = await Serialization.ReadUnsignedShort(stream, cancellationToken);
var essageLength = await Serialization.ReadSignedInt(stream, cancellationToken);
CheckMessageLength(essageLength);
var serializedMessage = await Serialization.ReadBytes(essageLength, stream, cancellationToken);
return new MessageFrame(messageId, registryCode, serializedMessage);
}
private static void CheckMessageLength(int messageLength) {
if (messageLength < 0) {
throw new RpcErrorException("Message length is negative.", RpcError.InvalidData);
}
if (messageLength > MaxMessageBytes) {
throw new RpcErrorException("Message is too large: " + messageLength + " > " + MaxMessageBytes + " bytes", RpcError.MessageTooLarge);
}
}
}

View File

@@ -0,0 +1,11 @@
namespace Phantom.Utils.Rpc.Frame.Types;
sealed record PingFrame : IFrame {
public static PingFrame Instance { get; } = new PingFrame();
public ReadOnlyMemory<byte> FrameType => IFrame.TypePing;
public Task Write(Stream stream, CancellationToken cancellationToken) {
return Task.CompletedTask;
}
}

View File

@@ -0,0 +1,37 @@
using Phantom.Utils.Rpc.Runtime;
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> FrameType => IFrame.TypeReply;
public async Task Write(Stream stream, CancellationToken cancellationToken) {
int replyLength = SerializedReply.Length;
CheckReplyLength(replyLength);
await Serialization.WriteUnsignedInt(ReplyingToMessageId, stream, cancellationToken);
await Serialization.WriteSignedInt(replyLength, stream, cancellationToken);
await stream.WriteAsync(SerializedReply, cancellationToken);
}
public static async Task<ReplyFrame> Read(Stream stream, CancellationToken cancellationToken) {
var replyingToMessageId = await Serialization.ReadUnsignedInt(stream, cancellationToken);
var replyLength = await Serialization.ReadSignedInt(stream, cancellationToken);
CheckReplyLength(replyLength);
var reply = await Serialization.ReadBytes(replyLength, stream, cancellationToken);
return new ReplyFrame(replyingToMessageId, reply);
}
private static void CheckReplyLength(int replyLength) {
if (replyLength < 0) {
throw new RpcErrorException("Reply length is negative.", RpcError.InvalidData);
}
if (replyLength > MaxReplyBytes) {
throw new RpcErrorException("Reply is too large: " + replyLength + " > " + MaxReplyBytes + " bytes", RpcError.MessageTooLarge);
}
}
}

View File

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

View File

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

View File

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

View File

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

View File

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

View File

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

View File

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

View File

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

View File

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

View File

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

View File

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

View File

@@ -0,0 +1,79 @@
using Phantom.Utils.Actor;
using Phantom.Utils.Logging;
using Phantom.Utils.Rpc.Frame;
using Phantom.Utils.Rpc.Message;
using Serilog;
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) {
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 RpcClientToServerConnection connection;
public RpcSendChannel<TClientToServerMessage> SendChannel { get; }
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 RpcClientToServerConnection(loggerName, connector, connection);
this.SendChannel = new RpcSendChannel<TClientToServerMessage>(loggerName, connectionParameters.Common, this.connection, messageDefinitions.ToServer);
}
public async Task Listen(ActorRef<TServerToClientMessage> actor) {
var handler = new MessageHandlerImpl(SendChannel, actor);
try {
await connection.ReadConnection(stream => Receive(stream, handler));
} catch (OperationCanceledException) {
// Ignore.
}
}
private async Task Receive(Stream stream, MessageHandlerImpl handler) {
await IFrame.ReadFrom(stream, new RpcFrameReader<TClientToServerMessage, TServerToClientMessage>(loggerName, serverToClientMessageRegistry, handler, SendChannel), CancellationToken.None);
}
private sealed class MessageHandlerImpl(RpcSendChannel<TClientToServerMessage> sendChannel, ActorRef<TServerToClientMessage> actor) : MessageHandler<TServerToClientMessage> {
public ActorRef<TServerToClientMessage> Actor => actor;
public ValueTask OnReply<TMessage, TReply>(uint messageId, TReply reply, CancellationToken cancellationToken) where TMessage : TServerToClientMessage, ICanReply<TReply> {
return sendChannel.SendReply(messageId, reply, cancellationToken);
}
public ValueTask OnError(uint messageId, RpcError error, CancellationToken cancellationToken) {
return sendChannel.SendError(messageId, error, cancellationToken);
}
}
public async Task Shutdown() {
logger.Information("Shutting down client...");
try {
await SendChannel.Close();
} catch (Exception e) {
logger.Error(e, "Caught exception while closing send channel.");
}
try {
connection.Close();
} catch (Exception e) {
logger.Error(e, "Caught exception while closing connection.");
}
logger.Information("Client shut down.");
}
public void Dispose() {
connection.Dispose();
SendChannel.Dispose();
}
}

View File

@@ -0,0 +1,15 @@
using Phantom.Utils.Rpc.Runtime.Tls;
namespace Phantom.Utils.Rpc.Runtime.Client;
public readonly record struct RpcClientConnectionParameters(
string Host,
ushort Port,
string DistinguishedName,
RpcCertificateThumbprint CertificateThumbprint,
AuthToken AuthToken,
ushort SendQueueCapacity,
TimeSpan PingInterval
) {
internal RpcCommonConnectionParameters Common => new (SendQueueCapacity, PingInterval);
}

View File

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

View File

@@ -0,0 +1,190 @@
using System.Net.Security;
using System.Net.Sockets;
using System.Security.Authentication;
using System.Security.Cryptography.X509Certificates;
using Phantom.Utils.Collections;
using Phantom.Utils.Logging;
using Phantom.Utils.Rpc.Runtime.Tls;
using Serilog;
namespace Phantom.Utils.Rpc.Runtime.Client;
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);
private readonly ILogger logger;
private readonly Guid sessionId;
private readonly RpcClientConnectionParameters parameters;
private readonly SslClientAuthenticationOptions sslOptions;
private bool loggedCertificateValidationError = false;
public RpcClientToServerConnector(string loggerName, RpcClientConnectionParameters parameters) {
this.logger = PhantomLogger.Create<RpcClientToServerConnector>(loggerName);
this.sessionId = Guid.NewGuid();
this.parameters = parameters;
this.sslOptions = new SslClientAuthenticationOptions {
AllowRenegotiation = false,
AllowTlsResume = true,
CertificateRevocationCheckMode = X509RevocationMode.NoCheck,
EnabledSslProtocols = TlsSupport.SupportedProtocols,
EncryptionPolicy = EncryptionPolicy.RequireEncryption,
RemoteCertificateValidationCallback = ValidateServerCertificate,
TargetHost = parameters.DistinguishedName,
};
}
internal async Task<Connection> EstablishNewConnectionWithRetry(CancellationToken cancellationToken) {
TimeSpan nextAttemptDelay = InitialRetryDelay;
while (true) {
Connection? newConnection;
try {
newConnection = await EstablishNewConnection(cancellationToken);
} catch (Exception e) {
logger.Error(e, "Caught unhandled exception while connecting.");
newConnection = null;
}
if (newConnection != null) {
return newConnection;
}
logger.Warning("Failed to connect to server, trying again in {}.", nextAttemptDelay.TotalSeconds.ToString("F1"));
await Task.Delay(nextAttemptDelay, cancellationToken);
nextAttemptDelay = Comparables.Min(nextAttemptDelay.Multiply(1.5), MaximumRetryDelay);
}
}
internal async Task<Connection?> EstablishNewConnection(CancellationToken cancellationToken) {
logger.Information("Connecting to {Host}:{Port}...", parameters.Host, parameters.Port);
Socket clientSocket = new Socket(SocketType.Stream, ProtocolType.Tcp);
try {
await clientSocket.ConnectAsync(parameters.Host, parameters.Port, cancellationToken);
} catch (Exception e) {
logger.Error(e, "Could not connect.");
throw;
}
SslStream? stream;
try {
stream = new SslStream(new NetworkStream(clientSocket, ownsSocket: false), leaveInnerStreamOpen: false);
if (await FinalizeConnection(stream, cancellationToken)) {
logger.Information("Connected to {Host}:{Port}.", parameters.Host, parameters.Port);
return new Connection(clientSocket, stream);
}
} catch (Exception e) {
logger.Error(e, "Caught unhandled exception.");
stream = null;
}
try {
await DisconnectSocket(clientSocket, stream);
} finally {
clientSocket.Close();
}
return null;
}
private async Task<bool> FinalizeConnection(SslStream stream, CancellationToken cancellationToken) {
try {
loggedCertificateValidationError = false;
await stream.AuthenticateAsClientAsync(sslOptions, cancellationToken);
} catch (AuthenticationException e) {
if (!loggedCertificateValidationError) {
logger.Error(e, "Could not establish a secure connection.");
}
return false;
}
logger.Information("Established a secure connection.");
try {
if (!await PerformApplicationHandshake(stream, cancellationToken)) {
return false;
}
} catch (EndOfStreamException) {
logger.Warning("Could not perform application handshake, connection lost.");
} catch (Exception e) {
logger.Warning(e, "Could not perform application handshake.");
}
return true;
}
private async Task<bool> PerformApplicationHandshake(Stream stream, CancellationToken cancellationToken) {
await Serialization.WriteAuthToken(parameters.AuthToken, stream, cancellationToken);
await Serialization.WriteGuid(sessionId, stream, cancellationToken);
var result = (RpcHandshakeResult) await Serialization.ReadByte(stream, cancellationToken);
switch (result) {
case RpcHandshakeResult.Success:
return true;
case RpcHandshakeResult.InvalidAuthToken:
logger.Error("Server rejected authorization token.");
return false;
default:
logger.Error("Server rejected client due to unknown error: {ErrorId}", result);
return false;
}
}
private bool ValidateServerCertificate(object sender, X509Certificate? certificate, X509Chain? chain, SslPolicyErrors sslPolicyErrors) {
if (certificate == null || sslPolicyErrors.HasFlag(SslPolicyErrors.RemoteCertificateNotAvailable)) {
logger.Error("Could not establish a secure connection, server did not provide a certificate.");
}
else if (sslPolicyErrors.HasFlag(SslPolicyErrors.RemoteCertificateNameMismatch)) {
logger.Error("Could not establish a secure connection, server certificate has the wrong name: {Name}", certificate.Subject);
}
else if (!parameters.CertificateThumbprint.Check(certificate)) {
logger.Error("Could not establish a secure connection, server certificate does not match.");
}
else if (TlsSupport.CheckAlgorithm((X509Certificate2) certificate) is {} error) {
logger.Error("Could not establish a secure connection, server certificate rejected because it uses {ActualAlgorithmName} instead of {ExpectedAlgorithmName}.", error.ActualAlgorithmName, error.ExpectedAlgorithmName);
}
else if ((sslPolicyErrors & ~SslPolicyErrors.RemoteCertificateChainErrors) != SslPolicyErrors.None) {
logger.Error("Could not establish a secure connection, server certificate validation failed.");
}
else {
return true;
}
loggedCertificateValidationError = true;
return false;
}
private static async Task DisconnectSocket(Socket socket, Stream? stream) {
if (stream != null) {
await stream.DisposeAsync();
}
using CancellationTokenSource timeoutTokenSource = new CancellationTokenSource(DisconnectTimeout);
await socket.DisconnectAsync(reuseSocket: false, timeoutTokenSource.Token);
}
internal sealed record Connection(Socket Socket, Stream Stream) : IDisposable {
public async Task Disconnect() {
await DisconnectSocket(Socket, Stream);
}
public async ValueTask Shutdown() {
await Stream.DisposeAsync();
Socket.Shutdown(SocketShutdown.Both);
}
public void Dispose() {
Stream.Dispose();
Socket.Close();
}
}
}

View File

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

View File

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

View File

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

View File

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

View File

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

View File

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

View File

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

View File

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

Some files were not shown because too many files have changed in this diff Show More