Compare commits

..

22 Commits

Author SHA1 Message Date
Joseph Doherty af42891d5a Issue #47: scaffold Java Gradle build 2026-04-26 20:36:27 -04:00
dohertj2 01a51df053 Merge pull request #95 from agent-2/issue-44-implement-rust-client-session-values-errors-and-cli
Issue #44: implement Rust client session values errors and CLI
2026-04-26 20:34:28 -04:00
Joseph Doherty 89a8fb876a Issue #44: implement Rust client session values errors and CLI 2026-04-26 20:30:04 -04:00
dohertj2 c58358fad9 Merge pull request #94 from agent-3/issue-45-scaffold-python-package
Issue #45: scaffold Python package
2026-04-26 20:28:13 -04:00
dohertj2 8d312a6d2e Merge pull request #93 from agent-1/issue-40-implement-dotnet-values-status-errors-and-cli
Issue #40: implement .NET values status errors and CLI
2026-04-26 20:22:58 -04:00
Joseph Doherty f861a8b3b8 Issue #45: scaffold Python package 2026-04-26 20:22:35 -04:00
Joseph Doherty 499708b2a2 Issue #40: implement .NET values status errors and CLI 2026-04-26 20:17:02 -04:00
dohertj2 191b724f95 Merge pull request #92 from agent-3/issue-42-implement-go-client-session-values-errors-and-cli
Issue #42: implement Go client session values errors and CLI
2026-04-26 20:14:56 -04:00
Joseph Doherty 8793011838 Issue #42: implement Go client session values errors and CLI 2026-04-26 20:09:58 -04:00
dohertj2 b275eedb44 Merge pull request #91 from agent-2/issue-34-worker-live-mxaccess-smoke-test
Issue #34: Worker Live MXAccess Smoke Test
2026-04-26 20:06:58 -04:00
Joseph Doherty a9ef6d10d4 Issue #34: handle worktree roots in live smoke tests 2026-04-26 20:03:21 -04:00
Joseph Doherty 0f17a1d1d9 Add live MXAccess worker smoke test 2026-04-26 19:58:33 -04:00
dohertj2 160343aff4 Merge pull request #90 from agent-3/issue-43-scaffold-rust-workspace
Issue #43: scaffold Rust workspace
2026-04-26 19:52:33 -04:00
dohertj2 8ef98b8beb Merge pull request #89 from agent-1/issue-39-implement-dotnet-gatewayclient-and-session
Issue #39: implement .NET GatewayClient and session
2026-04-26 19:50:37 -04:00
Joseph Doherty f049d3e603 Merge remote-tracking branch 'origin/main' into agent-3/issue-43-scaffold-rust-workspace 2026-04-26 19:49:30 -04:00
Joseph Doherty ee88f9d647 Scaffold Rust client workspace 2026-04-26 19:47:26 -04:00
Joseph Doherty 6e34efd1a5 Merge remote-tracking branch 'origin/main' into agent-1/issue-39-implement-dotnet-gatewayclient-and-session 2026-04-26 19:47:17 -04:00
Joseph Doherty 01d6c33156 Implement .NET gateway client sessions 2026-04-26 19:45:43 -04:00
dohertj2 ec4e2f687e Merge pull request #88 from agent-2/issue-33-implement-graceful-shutdown
Issue #33: implement graceful shutdown
2026-04-26 19:44:00 -04:00
Joseph Doherty f7929cc12f Merge remote-tracking branch 'origin/main' into agent-2/issue-33-implement-graceful-shutdown
# Conflicts:
#	src/MxGateway.Worker.Tests/Ipc/WorkerPipeSessionTests.cs
#	src/MxGateway.Worker/Ipc/WorkerPipeClient.cs
#	src/MxGateway.Worker/Ipc/WorkerPipeSession.cs
2026-04-26 19:41:04 -04:00
Joseph Doherty d890eff862 Implement graceful worker shutdown 2026-04-26 19:36:22 -04:00
dohertj2 9dcd4baff2 Merge pull request #86 from agent-3/issue-41-scaffold-go-module
Issue #41: scaffold Go module
2026-04-26 19:33:44 -04:00
110 changed files with 78518 additions and 83 deletions
@@ -0,0 +1,117 @@
using System.Globalization;
namespace MxGateway.Client.Cli;
internal sealed class CliArguments
{
private readonly Dictionary<string, string> _values = new(StringComparer.OrdinalIgnoreCase);
private readonly HashSet<string> _flags = new(StringComparer.OrdinalIgnoreCase);
public CliArguments(IEnumerable<string> args)
{
string? pendingName = null;
foreach (string arg in args)
{
if (arg.StartsWith("--", StringComparison.Ordinal))
{
if (pendingName is not null)
{
_flags.Add(pendingName);
}
pendingName = arg[2..];
continue;
}
if (pendingName is null)
{
throw new ArgumentException($"Unexpected argument '{arg}'.");
}
_values[pendingName] = arg;
pendingName = null;
}
if (pendingName is not null)
{
_flags.Add(pendingName);
}
}
public bool HasFlag(string name)
{
return _flags.Contains(name);
}
public string? GetOptional(string name)
{
return _values.TryGetValue(name, out string? value)
? value
: null;
}
public string GetRequired(string name)
{
string? value = GetOptional(name);
if (string.IsNullOrWhiteSpace(value))
{
throw new ArgumentException($"Missing required option --{name}.");
}
return value;
}
public int GetInt32(string name, int? defaultValue = null)
{
string? value = GetOptional(name);
if (string.IsNullOrWhiteSpace(value))
{
if (defaultValue.HasValue)
{
return defaultValue.Value;
}
throw new ArgumentException($"Missing required option --{name}.");
}
return int.Parse(value, CultureInfo.InvariantCulture);
}
public uint GetUInt32(string name, uint defaultValue)
{
string? value = GetOptional(name);
return string.IsNullOrWhiteSpace(value)
? defaultValue
: uint.Parse(value, CultureInfo.InvariantCulture);
}
public ulong GetUInt64(string name, ulong defaultValue)
{
string? value = GetOptional(name);
return string.IsNullOrWhiteSpace(value)
? defaultValue
: ulong.Parse(value, CultureInfo.InvariantCulture);
}
public TimeSpan GetDuration(string name, TimeSpan defaultValue)
{
string? value = GetOptional(name);
if (string.IsNullOrWhiteSpace(value))
{
return defaultValue;
}
if (value.EndsWith("ms", StringComparison.OrdinalIgnoreCase))
{
return TimeSpan.FromMilliseconds(double.Parse(value[..^2], CultureInfo.InvariantCulture));
}
if (value.EndsWith('s'))
{
return TimeSpan.FromSeconds(double.Parse(value[..^1], CultureInfo.InvariantCulture));
}
return TimeSpan.Parse(value, CultureInfo.InvariantCulture);
}
}
@@ -0,0 +1,22 @@
using MxGateway.Contracts.Proto;
namespace MxGateway.Client.Cli;
public interface IMxGatewayCliClient : IAsyncDisposable
{
Task<OpenSessionReply> OpenSessionAsync(
OpenSessionRequest request,
CancellationToken cancellationToken);
Task<CloseSessionReply> CloseSessionAsync(
CloseSessionRequest request,
CancellationToken cancellationToken);
Task<MxCommandReply> InvokeAsync(
MxCommandRequest request,
CancellationToken cancellationToken);
IAsyncEnumerable<MxEvent> StreamEventsAsync(
StreamEventsRequest request,
CancellationToken cancellationToken);
}
@@ -0,0 +1,40 @@
using MxGateway.Client;
using MxGateway.Contracts.Proto;
namespace MxGateway.Client.Cli;
internal sealed class MxGatewayCliClientAdapter(MxGatewayClient client) : IMxGatewayCliClient
{
public Task<OpenSessionReply> OpenSessionAsync(
OpenSessionRequest request,
CancellationToken cancellationToken)
{
return client.OpenSessionRawAsync(request, cancellationToken);
}
public Task<CloseSessionReply> CloseSessionAsync(
CloseSessionRequest request,
CancellationToken cancellationToken)
{
return client.CloseSessionRawAsync(request, cancellationToken);
}
public Task<MxCommandReply> InvokeAsync(
MxCommandRequest request,
CancellationToken cancellationToken)
{
return client.InvokeAsync(request, cancellationToken);
}
public IAsyncEnumerable<MxEvent> StreamEventsAsync(
StreamEventsRequest request,
CancellationToken cancellationToken)
{
return client.StreamEventsAsync(request, cancellationToken);
}
public ValueTask DisposeAsync()
{
return client.DisposeAsync();
}
}
@@ -0,0 +1,14 @@
namespace MxGateway.Client.Cli;
internal static class MxGatewayCliSecretRedactor
{
public static string Redact(string value, string? apiKey)
{
if (string.IsNullOrEmpty(value) || string.IsNullOrEmpty(apiKey))
{
return value;
}
return value.Replace(apiKey, "[redacted]", StringComparison.Ordinal);
}
}
@@ -1,34 +1,702 @@
using System.Globalization;
using System.Text.Json;
using Google.Protobuf;
using MxGateway.Client;
using MxGateway.Contracts.Proto;
namespace MxGateway.Client.Cli;
public static class MxGatewayClientCli
{
private static readonly JsonFormatter ProtobufJsonFormatter = JsonFormatter.Default;
private static readonly JsonSerializerOptions JsonOptions = new(JsonSerializerDefaults.Web);
public static int Run(
string[] args,
TextWriter standardOutput,
TextWriter standardError)
{
return RunAsync(args, standardOutput, standardError)
.GetAwaiter()
.GetResult();
}
public static Task<int> RunAsync(
string[] args,
TextWriter standardOutput,
TextWriter standardError,
Func<MxGatewayClientOptions, IMxGatewayCliClient>? clientFactory = null)
{
ArgumentNullException.ThrowIfNull(args);
ArgumentNullException.ThrowIfNull(standardOutput);
ArgumentNullException.ThrowIfNull(standardError);
return RunCoreAsync(
args,
standardOutput,
standardError,
clientFactory ?? CreateDefaultClient);
}
private static async Task<int> RunCoreAsync(
string[] args,
TextWriter standardOutput,
TextWriter standardError,
Func<MxGatewayClientOptions, IMxGatewayCliClient> clientFactory)
{
if (args.Length is 0 || IsHelp(args[0]))
{
WriteUsage(standardOutput);
return 0;
}
if (string.Equals(args[0], "version", StringComparison.OrdinalIgnoreCase))
string command = args[0].ToLowerInvariant();
CliArguments arguments = new(args.Skip(1));
try
{
standardOutput.WriteLine(
$"gateway-protocol={MxGatewayClientContractInfo.GatewayProtocolVersion}");
standardOutput.WriteLine(
$"worker-protocol={MxGatewayClientContractInfo.WorkerProtocolVersion}");
return 0;
if (command is "version")
{
WriteVersion(arguments, standardOutput);
return 0;
}
if (!IsKnownGatewayCommand(command))
{
return WriteUnknownCommand(command, standardError);
}
await using IMxGatewayCliClient client = clientFactory(CreateOptions(arguments));
using CancellationTokenSource cancellation = CreateCancellation(arguments);
return command switch
{
"open-session" => await OpenSessionAsync(arguments, client, standardOutput, cancellation.Token)
.ConfigureAwait(false),
"close-session" => await CloseSessionAsync(arguments, client, standardOutput, cancellation.Token)
.ConfigureAwait(false),
"ping" => await PingAsync(arguments, client, standardOutput, cancellation.Token)
.ConfigureAwait(false),
"register" => await RegisterAsync(arguments, client, standardOutput, cancellation.Token)
.ConfigureAwait(false),
"add-item" => await AddItemAsync(arguments, client, standardOutput, cancellation.Token)
.ConfigureAwait(false),
"advise" => await AdviseAsync(arguments, client, standardOutput, cancellation.Token)
.ConfigureAwait(false),
"stream-events" => await StreamEventsAsync(arguments, client, standardOutput, cancellation.Token)
.ConfigureAwait(false),
"write" => await WriteAsync(arguments, client, standardOutput, cancellation.Token)
.ConfigureAwait(false),
"write2" => await Write2Async(arguments, client, standardOutput, cancellation.Token)
.ConfigureAwait(false),
"smoke" => await SmokeAsync(arguments, client, standardOutput, cancellation.Token)
.ConfigureAwait(false),
_ => WriteUnknownCommand(command, standardError),
};
}
catch (Exception exception) when (exception is not OperationCanceledException)
{
string? apiKey = arguments.GetOptional("api-key");
string message = MxGatewayCliSecretRedactor.Redact(exception.Message, apiKey);
if (arguments.HasFlag("json"))
{
standardError.WriteLine(JsonSerializer.Serialize(
new { error = message, type = exception.GetType().Name },
JsonOptions));
}
else
{
standardError.WriteLine(message);
}
return 1;
}
}
private static IMxGatewayCliClient CreateDefaultClient(MxGatewayClientOptions options)
{
return new MxGatewayCliClientAdapter(MxGatewayClient.Create(options));
}
private static MxGatewayClientOptions CreateOptions(CliArguments arguments)
{
string endpoint = arguments.GetOptional("endpoint")
?? Environment.GetEnvironmentVariable("MXGATEWAY_ENDPOINT")
?? "http://localhost:5000";
string apiKey = ResolveApiKey(arguments);
return new MxGatewayClientOptions
{
Endpoint = new Uri(endpoint, UriKind.Absolute),
ApiKey = apiKey,
UseTls = arguments.HasFlag("tls")
|| endpoint.StartsWith("https://", StringComparison.OrdinalIgnoreCase),
DefaultCallTimeout = arguments.GetDuration("timeout", TimeSpan.FromSeconds(30)),
ConnectTimeout = arguments.GetDuration("connect-timeout", TimeSpan.FromSeconds(10)),
CaCertificatePath = arguments.GetOptional("ca-file"),
ServerNameOverride = arguments.GetOptional("server-name"),
};
}
private static string ResolveApiKey(CliArguments arguments)
{
string? apiKey = arguments.GetOptional("api-key");
if (!string.IsNullOrWhiteSpace(apiKey))
{
return apiKey;
}
standardError.WriteLine($"Unknown command '{args[0]}'.");
string apiKeyEnvironmentName = arguments.GetOptional("api-key-env")
?? "MXGATEWAY_API_KEY";
apiKey = Environment.GetEnvironmentVariable(apiKeyEnvironmentName);
if (!string.IsNullOrWhiteSpace(apiKey))
{
return apiKey;
}
throw new ArgumentException(
$"Gateway API key is required. Pass --api-key or set {apiKeyEnvironmentName}.");
}
private static CancellationTokenSource CreateCancellation(CliArguments arguments)
{
var cancellation = new CancellationTokenSource();
TimeSpan timeout = arguments.GetDuration("timeout", TimeSpan.FromSeconds(30));
cancellation.CancelAfter(timeout);
return cancellation;
}
private static Task<int> OpenSessionAsync(
CliArguments arguments,
IMxGatewayCliClient client,
TextWriter output,
CancellationToken cancellationToken)
{
return WriteReplyAsync(
client.OpenSessionAsync(
new OpenSessionRequest
{
ClientSessionName = arguments.GetOptional("client-name") ?? "mxgw-dotnet-cli",
ClientCorrelationId = CreateCorrelationId(),
RequestedBackend = arguments.GetOptional("backend") ?? string.Empty,
},
cancellationToken),
arguments,
output);
}
private static Task<int> CloseSessionAsync(
CliArguments arguments,
IMxGatewayCliClient client,
TextWriter output,
CancellationToken cancellationToken)
{
return WriteReplyAsync(
client.CloseSessionAsync(
new CloseSessionRequest
{
SessionId = arguments.GetRequired("session-id"),
ClientCorrelationId = CreateCorrelationId(),
},
cancellationToken),
arguments,
output);
}
private static Task<int> PingAsync(
CliArguments arguments,
IMxGatewayCliClient client,
TextWriter output,
CancellationToken cancellationToken)
{
return InvokeAndWriteAsync(
arguments,
client,
output,
new MxCommand
{
Kind = MxCommandKind.Ping,
Ping = new PingCommand { Message = arguments.GetOptional("message") ?? "ping" },
},
cancellationToken);
}
private static Task<int> RegisterAsync(
CliArguments arguments,
IMxGatewayCliClient client,
TextWriter output,
CancellationToken cancellationToken)
{
return InvokeAndWriteAsync(
arguments,
client,
output,
new MxCommand
{
Kind = MxCommandKind.Register,
Register = new RegisterCommand { ClientName = arguments.GetOptional("client-name") ?? "mxgw-dotnet-cli" },
},
cancellationToken);
}
private static Task<int> AddItemAsync(
CliArguments arguments,
IMxGatewayCliClient client,
TextWriter output,
CancellationToken cancellationToken)
{
return InvokeAndWriteAsync(
arguments,
client,
output,
new MxCommand
{
Kind = MxCommandKind.AddItem,
AddItem = new AddItemCommand
{
ServerHandle = arguments.GetInt32("server-handle"),
ItemDefinition = arguments.GetRequired("item"),
},
},
cancellationToken);
}
private static Task<int> AdviseAsync(
CliArguments arguments,
IMxGatewayCliClient client,
TextWriter output,
CancellationToken cancellationToken)
{
return InvokeAndWriteAsync(
arguments,
client,
output,
new MxCommand
{
Kind = MxCommandKind.Advise,
Advise = new AdviseCommand
{
ServerHandle = arguments.GetInt32("server-handle"),
ItemHandle = arguments.GetInt32("item-handle"),
},
},
cancellationToken);
}
private static Task<int> WriteAsync(
CliArguments arguments,
IMxGatewayCliClient client,
TextWriter output,
CancellationToken cancellationToken)
{
return InvokeAndWriteAsync(
arguments,
client,
output,
new MxCommand
{
Kind = MxCommandKind.Write,
Write = new WriteCommand
{
ServerHandle = arguments.GetInt32("server-handle"),
ItemHandle = arguments.GetInt32("item-handle"),
UserId = arguments.GetInt32("user-id", 0),
Value = ParseValue(arguments),
},
},
cancellationToken);
}
private static Task<int> Write2Async(
CliArguments arguments,
IMxGatewayCliClient client,
TextWriter output,
CancellationToken cancellationToken)
{
return InvokeAndWriteAsync(
arguments,
client,
output,
new MxCommand
{
Kind = MxCommandKind.Write2,
Write2 = new Write2Command
{
ServerHandle = arguments.GetInt32("server-handle"),
ItemHandle = arguments.GetInt32("item-handle"),
UserId = arguments.GetInt32("user-id", 0),
Value = ParseValue(arguments),
TimestampValue = ParseTimestampValue(arguments),
},
},
cancellationToken);
}
private static async Task<int> StreamEventsAsync(
CliArguments arguments,
IMxGatewayCliClient client,
TextWriter output,
CancellationToken cancellationToken)
{
var events = new List<MxEvent>();
uint maxEvents = arguments.GetUInt32("max-events", 0);
uint eventCount = 0;
var request = new StreamEventsRequest
{
SessionId = arguments.GetRequired("session-id"),
AfterWorkerSequence = arguments.GetUInt64("after-worker-sequence", 0),
};
await foreach (MxEvent gatewayEvent in client.StreamEventsAsync(request, cancellationToken)
.WithCancellation(cancellationToken)
.ConfigureAwait(false))
{
if (arguments.HasFlag("json"))
{
events.Add(gatewayEvent);
}
else
{
output.WriteLine(ProtobufJsonFormatter.Format(gatewayEvent));
}
eventCount++;
if (maxEvents > 0 && eventCount >= maxEvents)
{
break;
}
}
if (arguments.HasFlag("json"))
{
output.WriteLine(JsonSerializer.Serialize(
new { events = events.Select(EventToJsonElement).ToArray() },
JsonOptions));
}
return 0;
}
private static async Task<int> SmokeAsync(
CliArguments arguments,
IMxGatewayCliClient client,
TextWriter output,
CancellationToken cancellationToken)
{
OpenSessionReply? openReply = null;
CloseSessionReply? closeReply = null;
var commandReplies = new List<MxCommandReply>();
var events = new List<MxEvent>();
try
{
openReply = await client.OpenSessionAsync(
new OpenSessionRequest
{
ClientSessionName = arguments.GetOptional("client-name") ?? "mxgw-dotnet-smoke",
ClientCorrelationId = CreateCorrelationId(),
},
cancellationToken)
.ConfigureAwait(false);
int serverHandle = await InvokeForHandleAsync(
arguments,
client,
openReply.SessionId,
new MxCommand
{
Kind = MxCommandKind.Register,
Register = new RegisterCommand { ClientName = arguments.GetOptional("client-name") ?? "mxgw-dotnet-smoke" },
},
reply => reply.Register?.ServerHandle ?? reply.ReturnValue.Int32Value,
commandReplies,
cancellationToken)
.ConfigureAwait(false);
int itemHandle = await InvokeForHandleAsync(
arguments,
client,
openReply.SessionId,
new MxCommand
{
Kind = MxCommandKind.AddItem,
AddItem = new AddItemCommand
{
ServerHandle = serverHandle,
ItemDefinition = arguments.GetRequired("item"),
},
},
reply => reply.AddItem?.ItemHandle ?? reply.ReturnValue.Int32Value,
commandReplies,
cancellationToken)
.ConfigureAwait(false);
commandReplies.Add(await InvokeAndEnsureAsync(
client,
CreateCommandRequest(
openReply.SessionId,
new MxCommand
{
Kind = MxCommandKind.Advise,
Advise = new AdviseCommand
{
ServerHandle = serverHandle,
ItemHandle = itemHandle,
},
}),
cancellationToken)
.ConfigureAwait(false));
if (arguments.GetOptional("value") is not null)
{
commandReplies.Add(await InvokeAndEnsureAsync(
client,
CreateCommandRequest(
openReply.SessionId,
new MxCommand
{
Kind = MxCommandKind.Write,
Write = new WriteCommand
{
ServerHandle = serverHandle,
ItemHandle = itemHandle,
UserId = arguments.GetInt32("user-id", 0),
Value = ParseValue(arguments),
},
}),
cancellationToken)
.ConfigureAwait(false));
}
using CancellationTokenSource streamCancellation = CancellationTokenSource
.CreateLinkedTokenSource(cancellationToken);
streamCancellation.CancelAfter(arguments.GetDuration(
"event-timeout",
TimeSpan.FromSeconds(2)));
try
{
await foreach (MxEvent gatewayEvent in client.StreamEventsAsync(
new StreamEventsRequest { SessionId = openReply.SessionId },
streamCancellation.Token)
.WithCancellation(streamCancellation.Token)
.ConfigureAwait(false))
{
events.Add(gatewayEvent);
if (events.Count >= arguments.GetUInt32("max-events", 1))
{
break;
}
}
}
catch (OperationCanceledException) when (!cancellationToken.IsCancellationRequested)
{
}
}
finally
{
if (openReply is not null)
{
closeReply = await client.CloseSessionAsync(
new CloseSessionRequest
{
SessionId = openReply.SessionId,
ClientCorrelationId = CreateCorrelationId(),
},
CancellationToken.None)
.ConfigureAwait(false);
}
}
WriteSmokeResult(arguments, output, openReply, closeReply, commandReplies, events);
return 0;
}
private static async Task<int> InvokeAndWriteAsync(
CliArguments arguments,
IMxGatewayCliClient client,
TextWriter output,
MxCommand command,
CancellationToken cancellationToken)
{
MxCommandReply reply = await InvokeAndEnsureAsync(
client,
CreateCommandRequest(arguments.GetRequired("session-id"), command),
cancellationToken)
.ConfigureAwait(false);
WriteMessage(arguments, output, reply);
return 0;
}
private static async Task<int> InvokeForHandleAsync(
CliArguments arguments,
IMxGatewayCliClient client,
string sessionId,
MxCommand command,
Func<MxCommandReply, int> handleSelector,
List<MxCommandReply> replies,
CancellationToken cancellationToken)
{
MxCommandReply reply = await InvokeAndEnsureAsync(
client,
CreateCommandRequest(sessionId, command),
cancellationToken)
.ConfigureAwait(false);
replies.Add(reply);
return handleSelector(reply);
}
private static async Task<MxCommandReply> InvokeAndEnsureAsync(
IMxGatewayCliClient client,
MxCommandRequest request,
CancellationToken cancellationToken)
{
MxCommandReply reply = await client.InvokeAsync(request, cancellationToken)
.ConfigureAwait(false);
reply.EnsureProtocolSuccess().EnsureMxAccessSuccess();
return reply;
}
private static MxCommandRequest CreateCommandRequest(
string sessionId,
MxCommand command)
{
return new MxCommandRequest
{
SessionId = sessionId,
ClientCorrelationId = CreateCorrelationId(),
Command = command,
};
}
private static async Task<int> WriteReplyAsync<TReply>(
Task<TReply> replyTask,
CliArguments arguments,
TextWriter output)
where TReply : IMessage
{
TReply reply = await replyTask.ConfigureAwait(false);
WriteMessage(arguments, output, reply);
return 0;
}
private static void WriteVersion(CliArguments arguments, TextWriter output)
{
if (arguments.HasFlag("json"))
{
output.WriteLine(JsonSerializer.Serialize(
new
{
gatewayProtocolVersion = MxGatewayClientContractInfo.GatewayProtocolVersion,
workerProtocolVersion = MxGatewayClientContractInfo.WorkerProtocolVersion,
},
JsonOptions));
return;
}
output.WriteLine(
$"gateway-protocol={MxGatewayClientContractInfo.GatewayProtocolVersion}");
output.WriteLine(
$"worker-protocol={MxGatewayClientContractInfo.WorkerProtocolVersion}");
}
private static void WriteMessage(
CliArguments arguments,
TextWriter output,
IMessage message)
{
output.WriteLine(arguments.HasFlag("json")
? ProtobufJsonFormatter.Format(message)
: message.ToString());
}
private static void WriteSmokeResult(
CliArguments arguments,
TextWriter output,
OpenSessionReply? openReply,
CloseSessionReply? closeReply,
IReadOnlyList<MxCommandReply> commandReplies,
IReadOnlyList<MxEvent> events)
{
if (!arguments.HasFlag("json"))
{
output.WriteLine($"session-id={openReply?.SessionId}");
output.WriteLine($"commands={commandReplies.Count}");
output.WriteLine($"events={events.Count}");
output.WriteLine($"closed={closeReply is not null}");
return;
}
output.WriteLine(JsonSerializer.Serialize(
new
{
sessionId = openReply?.SessionId,
closed = closeReply is not null,
commandReplies = commandReplies.Select(CommandReplyToJsonElement).ToArray(),
events = events.Select(EventToJsonElement).ToArray(),
},
JsonOptions));
}
private static JsonElement CommandReplyToJsonElement(MxCommandReply reply)
{
return JsonDocument.Parse(ProtobufJsonFormatter.Format(reply)).RootElement.Clone();
}
private static JsonElement EventToJsonElement(MxEvent gatewayEvent)
{
return JsonDocument.Parse(ProtobufJsonFormatter.Format(gatewayEvent)).RootElement.Clone();
}
private static MxValue ParseValue(CliArguments arguments)
{
string type = arguments.GetRequired("type").ToLowerInvariant();
string value = arguments.GetRequired("value");
string[] values = value.Split(',', StringSplitOptions.TrimEntries);
return type switch
{
"bool" or "boolean" => bool.Parse(value).ToMxValue(),
"bool-array" or "boolean-array" => values.Select(bool.Parse).ToArray().ToMxValue(),
"int32" or "integer" => int.Parse(value, CultureInfo.InvariantCulture).ToMxValue(),
"int32-array" or "integer-array" => values.Select(item => int.Parse(item, CultureInfo.InvariantCulture)).ToArray().ToMxValue(),
"int64" => long.Parse(value, CultureInfo.InvariantCulture).ToMxValue(),
"int64-array" => values.Select(item => long.Parse(item, CultureInfo.InvariantCulture)).ToArray().ToMxValue(),
"float" => float.Parse(value, CultureInfo.InvariantCulture).ToMxValue(),
"float-array" => values.Select(item => float.Parse(item, CultureInfo.InvariantCulture)).ToArray().ToMxValue(),
"double" => double.Parse(value, CultureInfo.InvariantCulture).ToMxValue(),
"double-array" => values.Select(item => double.Parse(item, CultureInfo.InvariantCulture)).ToArray().ToMxValue(),
"string" => value.ToMxValue(),
"string-array" => values.ToMxValue(),
"time" or "timestamp" => DateTimeOffset.Parse(value, CultureInfo.InvariantCulture, DateTimeStyles.AssumeUniversal).ToMxValue(),
"time-array" or "timestamp-array" => values
.Select(item => DateTimeOffset.Parse(item, CultureInfo.InvariantCulture, DateTimeStyles.AssumeUniversal))
.ToArray()
.ToMxValue(),
_ => throw new ArgumentException($"Unsupported MX value type '{type}'."),
};
}
private static MxValue ParseTimestampValue(CliArguments arguments)
{
string timestamp = arguments.GetOptional("timestamp")
?? DateTimeOffset.UtcNow.ToString("O", CultureInfo.InvariantCulture);
return DateTimeOffset.Parse(
timestamp,
CultureInfo.InvariantCulture,
DateTimeStyles.AssumeUniversal)
.ToMxValue();
}
private static int WriteUnknownCommand(string command, TextWriter standardError)
{
standardError.WriteLine($"Unknown command '{command}'.");
WriteUsage(standardError);
return 2;
}
@@ -40,9 +708,37 @@ public static class MxGatewayClientCli
|| string.Equals(value, "help", StringComparison.OrdinalIgnoreCase);
}
private static bool IsKnownGatewayCommand(string command)
{
return command is "open-session"
or "close-session"
or "ping"
or "register"
or "add-item"
or "advise"
or "stream-events"
or "write"
or "write2"
or "smoke";
}
private static string CreateCorrelationId()
{
return Guid.NewGuid().ToString("N");
}
private static void WriteUsage(TextWriter writer)
{
writer.WriteLine("mxgw-dotnet version");
writer.WriteLine("mxgw-dotnet --help");
writer.WriteLine("mxgw-dotnet version [--json]");
writer.WriteLine("mxgw-dotnet ping --session-id <id> [--json]");
writer.WriteLine("mxgw-dotnet open-session [--client-name <name>] [--json]");
writer.WriteLine("mxgw-dotnet close-session --session-id <id> [--json]");
writer.WriteLine("mxgw-dotnet register --session-id <id> --client-name <name> [--json]");
writer.WriteLine("mxgw-dotnet add-item --session-id <id> --server-handle <n> --item <ref> [--json]");
writer.WriteLine("mxgw-dotnet advise --session-id <id> --server-handle <n> --item-handle <n> [--json]");
writer.WriteLine("mxgw-dotnet stream-events --session-id <id> [--max-events <n>] [--json]");
writer.WriteLine("mxgw-dotnet write --session-id <id> --server-handle <n> --item-handle <n> --type <type> --value <value> [--json]");
writer.WriteLine("mxgw-dotnet write2 --session-id <id> --server-handle <n> --item-handle <n> --type <type> --value <value> [--timestamp <iso>] [--json]");
writer.WriteLine("mxgw-dotnet smoke --item <ref> [--value <value> --type <type>] [--json]");
}
}
@@ -1,3 +1,3 @@
using MxGateway.Client.Cli;
return MxGatewayClientCli.Run(args, Console.Out, Console.Error);
return await MxGatewayClientCli.RunAsync(args, Console.Out, Console.Error);
@@ -0,0 +1,86 @@
using Grpc.Core;
using MxGateway.Contracts.Proto;
namespace MxGateway.Client.Tests;
internal sealed class FakeGatewayTransport(MxGatewayClientOptions options) : IMxGatewayClientTransport
{
private readonly Queue<MxCommandReply> _invokeReplies = new();
private readonly List<MxEvent> _events = [];
public MxGatewayClientOptions Options { get; } = options;
public MxAccessGateway.MxAccessGatewayClient? RawClient => null;
public List<(OpenSessionRequest Request, CallOptions CallOptions)> OpenSessionCalls { get; } = [];
public List<(CloseSessionRequest Request, CallOptions CallOptions)> CloseSessionCalls { get; } = [];
public List<(MxCommandRequest Request, CallOptions CallOptions)> InvokeCalls { get; } = [];
public List<(StreamEventsRequest Request, CallOptions CallOptions)> StreamEventsCalls { get; } = [];
public OpenSessionReply OpenSessionReply { get; set; } = new()
{
SessionId = "session-fixture",
BackendName = "mxaccess-worker",
GatewayProtocolVersion = 1,
WorkerProtocolVersion = 1,
ProtocolStatus = new ProtocolStatus { Code = ProtocolStatusCode.Ok },
};
public CloseSessionReply CloseSessionReply { get; set; } = new()
{
SessionId = "session-fixture",
FinalState = SessionState.Closed,
ProtocolStatus = new ProtocolStatus { Code = ProtocolStatusCode.Ok },
};
public Task<OpenSessionReply> OpenSessionAsync(
OpenSessionRequest request,
CallOptions callOptions)
{
OpenSessionCalls.Add((request, callOptions));
return Task.FromResult(OpenSessionReply);
}
public Task<CloseSessionReply> CloseSessionAsync(
CloseSessionRequest request,
CallOptions callOptions)
{
CloseSessionCalls.Add((request, callOptions));
return Task.FromResult(CloseSessionReply);
}
public Task<MxCommandReply> InvokeAsync(
MxCommandRequest request,
CallOptions callOptions)
{
InvokeCalls.Add((request, callOptions));
return Task.FromResult(_invokeReplies.Dequeue());
}
public async IAsyncEnumerable<MxEvent> StreamEventsAsync(
StreamEventsRequest request,
CallOptions callOptions)
{
StreamEventsCalls.Add((request, callOptions));
foreach (MxEvent gatewayEvent in _events)
{
callOptions.CancellationToken.ThrowIfCancellationRequested();
await Task.Yield();
yield return gatewayEvent;
}
}
public void AddInvokeReply(MxCommandReply reply)
{
_invokeReplies.Enqueue(reply);
}
public void AddEvent(MxEvent gatewayEvent)
{
_events.Add(gatewayEvent);
}
}
@@ -0,0 +1,78 @@
using Google.Protobuf;
using MxGateway.Client;
using MxGateway.Contracts.Proto;
namespace MxGateway.Client.Tests;
public sealed class MxCommandReplyExtensionsTests
{
[Fact]
public void EnsureSuccess_WithRegisterFixture_ReturnsReply()
{
MxCommandReply reply = ReadReplyFixture("register.ok.reply.json");
Assert.Same(reply, reply.EnsureProtocolSuccess());
Assert.Same(reply, reply.EnsureMxAccessSuccess());
}
[Fact]
public void EnsureMxAccessSuccess_WithFailureFixture_PreservesHResultAndStatuses()
{
MxCommandReply reply = ReadReplyFixture("write.mxaccess-failure.reply.json");
reply.EnsureProtocolSuccess();
MxAccessException exception = Assert.Throws<MxAccessException>(
reply.EnsureMxAccessSuccess);
Assert.Equal(-2147220992, exception.HResultCode);
Assert.Equal(reply.Statuses.Count, exception.Statuses.Count);
Assert.Equal(reply, exception.Reply);
Assert.Contains("0x80040200", exception.Message);
}
[Fact]
public void EnsureProtocolSuccess_WithSessionFailure_ThrowsSessionException()
{
MxCommandReply reply = new()
{
SessionId = "session-missing",
CorrelationId = "correlation",
ProtocolStatus = new ProtocolStatus
{
Code = ProtocolStatusCode.SessionNotFound,
Message = "Session was not found.",
},
};
MxGatewaySessionException exception = Assert.Throws<MxGatewaySessionException>(
reply.EnsureProtocolSuccess);
Assert.Equal("session-missing", exception.SessionId);
Assert.Equal(ProtocolStatusCode.SessionNotFound, exception.ProtocolStatus?.Code);
}
private static MxCommandReply ReadReplyFixture(string fileName)
{
DirectoryInfo directory = new(AppContext.BaseDirectory);
while (directory is not null)
{
string path = Path.Combine(
directory.FullName,
"clients",
"proto",
"fixtures",
"behavior",
"command-replies",
fileName);
if (File.Exists(path))
{
return JsonParser.Default.Parse<MxCommandReply>(File.ReadAllText(path));
}
directory = directory.Parent!;
}
throw new FileNotFoundException(fileName);
}
}
@@ -1,4 +1,5 @@
using MxGateway.Client.Cli;
using MxGateway.Contracts.Proto;
namespace MxGateway.Client.Tests;
@@ -17,4 +18,224 @@ public sealed class MxGatewayClientCliTests
Assert.Contains("worker-protocol=1", output.ToString());
Assert.Equal(string.Empty, error.ToString());
}
[Fact]
public async Task RunAsync_VersionJson_PrintsJsonProtocolVersions()
{
using var output = new StringWriter();
using var error = new StringWriter();
int exitCode = await MxGatewayClientCli.RunAsync(["version", "--json"], output, error);
Assert.Equal(0, exitCode);
Assert.Contains("\"gatewayProtocolVersion\":1", output.ToString());
Assert.Equal(string.Empty, error.ToString());
}
[Fact]
public async Task RunAsync_Write_BuildsWriteCommandAndPrintsJsonReply()
{
using var output = new StringWriter();
using var error = new StringWriter();
FakeCliClient fakeClient = new();
fakeClient.InvokeReplies.Enqueue(new MxCommandReply
{
SessionId = "session-fixture",
Kind = MxCommandKind.Write,
ProtocolStatus = new ProtocolStatus { Code = ProtocolStatusCode.Ok },
});
int exitCode = await MxGatewayClientCli.RunAsync(
[
"write",
"--endpoint",
"http://localhost:5000",
"--api-key",
"test-api-key",
"--session-id",
"session-fixture",
"--server-handle",
"12",
"--item-handle",
"34",
"--type",
"int32",
"--value",
"123",
"--json",
],
output,
error,
_ => fakeClient);
Assert.Equal(0, exitCode);
MxCommandRequest request = Assert.Single(fakeClient.InvokeRequests);
Assert.Equal(MxCommandKind.Write, request.Command.Kind);
Assert.Equal(123, request.Command.Write.Value.Int32Value);
Assert.Contains("MX_COMMAND_KIND_WRITE", output.ToString());
Assert.Equal(string.Empty, error.ToString());
}
[Fact]
public async Task RunAsync_ErrorOutput_RedactsApiKey()
{
using var output = new StringWriter();
using var error = new StringWriter();
int exitCode = await MxGatewayClientCli.RunAsync(
[
"open-session",
"--endpoint",
"http://localhost:5000",
"--api-key",
"secret-api-key",
],
output,
error,
_ => throw new InvalidOperationException("boom secret-api-key"));
Assert.Equal(1, exitCode);
Assert.DoesNotContain("secret-api-key", error.ToString());
Assert.Contains("[redacted]", error.ToString());
}
[Fact]
public async Task RunAsync_StreamEvents_WithMaxEventsStopsNonJsonOutput()
{
using var output = new StringWriter();
using var error = new StringWriter();
FakeCliClient fakeClient = new();
fakeClient.Events.Add(new MxEvent
{
SessionId = "session-fixture",
Family = MxEventFamily.OnDataChange,
WorkerSequence = 1,
});
fakeClient.Events.Add(new MxEvent
{
SessionId = "session-fixture",
Family = MxEventFamily.OnWriteComplete,
WorkerSequence = 2,
});
int exitCode = await MxGatewayClientCli.RunAsync(
[
"stream-events",
"--endpoint",
"http://localhost:5000",
"--api-key",
"test-api-key",
"--session-id",
"session-fixture",
"--max-events",
"1",
],
output,
error,
_ => fakeClient);
Assert.Equal(0, exitCode);
Assert.Contains("workerSequence", output.ToString());
Assert.DoesNotContain("ON_WRITE_COMPLETE", output.ToString());
}
[Fact]
public async Task RunAsync_Smoke_WhenCommandFails_ClosesOpenedSession()
{
using var output = new StringWriter();
using var error = new StringWriter();
FakeCliClient fakeClient = new()
{
InvokeFailure = new InvalidOperationException("register failed"),
};
int exitCode = await MxGatewayClientCli.RunAsync(
[
"smoke",
"--endpoint",
"http://localhost:5000",
"--api-key",
"test-api-key",
"--item",
"Area001.Pump001.Speed",
"--json",
],
output,
error,
_ => fakeClient);
Assert.Equal(1, exitCode);
CloseSessionRequest closeRequest = Assert.Single(fakeClient.CloseSessionRequests);
Assert.Equal("session-fixture", closeRequest.SessionId);
}
private sealed class FakeCliClient : IMxGatewayCliClient
{
public Queue<MxCommandReply> InvokeReplies { get; } = new();
public List<MxCommandRequest> InvokeRequests { get; } = [];
public List<CloseSessionRequest> CloseSessionRequests { get; } = [];
public List<MxEvent> Events { get; } = [];
public Exception? InvokeFailure { get; init; }
public ValueTask DisposeAsync()
{
return ValueTask.CompletedTask;
}
public Task<OpenSessionReply> OpenSessionAsync(
OpenSessionRequest request,
CancellationToken cancellationToken)
{
return Task.FromResult(new OpenSessionReply
{
SessionId = "session-fixture",
ProtocolStatus = new ProtocolStatus { Code = ProtocolStatusCode.Ok },
GatewayProtocolVersion = 1,
WorkerProtocolVersion = 1,
});
}
public Task<CloseSessionReply> CloseSessionAsync(
CloseSessionRequest request,
CancellationToken cancellationToken)
{
CloseSessionRequests.Add(request);
return Task.FromResult(new CloseSessionReply
{
SessionId = request.SessionId,
ProtocolStatus = new ProtocolStatus { Code = ProtocolStatusCode.Ok },
FinalState = SessionState.Closed,
});
}
public Task<MxCommandReply> InvokeAsync(
MxCommandRequest request,
CancellationToken cancellationToken)
{
InvokeRequests.Add(request);
if (InvokeFailure is not null)
{
throw InvokeFailure;
}
return Task.FromResult(InvokeReplies.Dequeue());
}
public async IAsyncEnumerable<MxEvent> StreamEventsAsync(
StreamEventsRequest request,
[System.Runtime.CompilerServices.EnumeratorCancellation] CancellationToken cancellationToken)
{
foreach (MxEvent gatewayEvent in Events)
{
cancellationToken.ThrowIfCancellationRequested();
await Task.Yield();
yield return gatewayEvent;
}
}
}
}
@@ -0,0 +1,217 @@
using MxGateway.Contracts.Proto;
namespace MxGateway.Client.Tests;
public sealed class MxGatewayClientSessionTests
{
[Fact]
public async Task OpenSessionRawAsync_AttachesApiKeyMetadataAndCancellation()
{
using CancellationTokenSource cancellation = new();
FakeGatewayTransport transport = CreateTransport();
await using MxGatewayClient client = CreateClient(transport);
await client.OpenSessionRawAsync(new OpenSessionRequest(), cancellation.Token);
var call = Assert.Single(transport.OpenSessionCalls);
Assert.Equal("Bearer test-api-key", call.CallOptions.Headers?.GetValue("authorization"));
Assert.Equal(cancellation.Token, call.CallOptions.CancellationToken);
}
[Fact]
public async Task OpenSessionAsync_ReturnsSessionWithRawOpenReply()
{
FakeGatewayTransport transport = CreateTransport();
transport.OpenSessionReply.WorkerProcessId = 1234;
await using MxGatewayClient client = CreateClient(transport);
MxGatewaySession session = await client.OpenSessionAsync();
Assert.Equal("session-fixture", session.SessionId);
Assert.Same(transport.OpenSessionReply, session.OpenSessionReply);
Assert.Equal(1234, session.OpenSessionReply.WorkerProcessId);
}
[Fact]
public async Task RegisterAsync_BuildsRegisterCommandAndReturnsServerHandle()
{
FakeGatewayTransport transport = CreateTransport();
transport.AddInvokeReply(new MxCommandReply
{
SessionId = "session-fixture",
Kind = MxCommandKind.Register,
ProtocolStatus = new ProtocolStatus { Code = ProtocolStatusCode.Ok },
Register = new RegisterReply { ServerHandle = 12 },
});
await using MxGatewayClient client = CreateClient(transport);
MxGatewaySession session = await client.OpenSessionAsync();
int serverHandle = await session.RegisterAsync("fixture-client");
Assert.Equal(12, serverHandle);
var call = Assert.Single(transport.InvokeCalls);
Assert.Equal("session-fixture", call.Request.SessionId);
Assert.False(string.IsNullOrWhiteSpace(call.Request.ClientCorrelationId));
Assert.Equal(MxCommandKind.Register, call.Request.Command.Kind);
Assert.Equal("fixture-client", call.Request.Command.Register.ClientName);
}
[Fact]
public async Task AddItem2Async_BuildsAddItem2CommandWithContext()
{
FakeGatewayTransport transport = CreateTransport();
transport.AddInvokeReply(new MxCommandReply
{
SessionId = "session-fixture",
Kind = MxCommandKind.AddItem2,
ProtocolStatus = new ProtocolStatus { Code = ProtocolStatusCode.Ok },
AddItem2 = new AddItem2Reply { ItemHandle = 34 },
});
await using MxGatewayClient client = CreateClient(transport);
MxGatewaySession session = await client.OpenSessionAsync();
int itemHandle = await session.AddItem2Async(12, "Area001.Pump001.Speed", "runtime");
Assert.Equal(34, itemHandle);
MxCommandRequest request = Assert.Single(transport.InvokeCalls).Request;
Assert.Equal(MxCommandKind.AddItem2, request.Command.Kind);
Assert.Equal(12, request.Command.AddItem2.ServerHandle);
Assert.Equal("Area001.Pump001.Speed", request.Command.AddItem2.ItemDefinition);
Assert.Equal("runtime", request.Command.AddItem2.ItemContext);
}
[Fact]
public async Task WriteRawAsync_BuildsWriteCommandWithRawValue()
{
FakeGatewayTransport transport = CreateTransport();
transport.AddInvokeReply(new MxCommandReply
{
SessionId = "session-fixture",
Kind = MxCommandKind.Write,
ProtocolStatus = new ProtocolStatus { Code = ProtocolStatusCode.Ok },
});
await using MxGatewayClient client = CreateClient(transport);
MxGatewaySession session = await client.OpenSessionAsync();
MxValue value = new()
{
DataType = MxDataType.Integer,
VariantType = "VT_I4",
Int32Value = 123,
};
MxCommandReply reply = await session.WriteRawAsync(12, 34, value, 56);
Assert.Equal(MxCommandKind.Write, reply.Kind);
MxCommandRequest request = Assert.Single(transport.InvokeCalls).Request;
Assert.Equal(MxCommandKind.Write, request.Command.Kind);
Assert.Equal(12, request.Command.Write.ServerHandle);
Assert.Equal(34, request.Command.Write.ItemHandle);
Assert.Same(value, request.Command.Write.Value);
Assert.Equal(56, request.Command.Write.UserId);
}
[Fact]
public async Task Write2RawAsync_BuildsWrite2CommandWithValueAndTimestamp()
{
FakeGatewayTransport transport = CreateTransport();
transport.AddInvokeReply(new MxCommandReply
{
SessionId = "session-fixture",
Kind = MxCommandKind.Write2,
ProtocolStatus = new ProtocolStatus { Code = ProtocolStatusCode.Ok },
});
await using MxGatewayClient client = CreateClient(transport);
MxGatewaySession session = await client.OpenSessionAsync();
MxValue value = 123.ToMxValue();
MxValue timestampValue = DateTimeOffset.Parse("2026-01-01T00:00:00Z").ToMxValue();
MxCommandReply reply = await session.Write2RawAsync(12, 34, value, timestampValue, 56);
Assert.Equal(MxCommandKind.Write2, reply.Kind);
MxCommandRequest request = Assert.Single(transport.InvokeCalls).Request;
Assert.Equal(MxCommandKind.Write2, request.Command.Kind);
Assert.Equal(12, request.Command.Write2.ServerHandle);
Assert.Equal(34, request.Command.Write2.ItemHandle);
Assert.Same(value, request.Command.Write2.Value);
Assert.Same(timestampValue, request.Command.Write2.TimestampValue);
Assert.Equal(56, request.Command.Write2.UserId);
}
[Fact]
public async Task StreamEventsAsync_YieldsEventsInGatewayOrder()
{
FakeGatewayTransport transport = CreateTransport();
transport.AddEvent(new MxEvent
{
SessionId = "session-fixture",
Family = MxEventFamily.OnDataChange,
WorkerSequence = 1,
});
transport.AddEvent(new MxEvent
{
SessionId = "session-fixture",
Family = MxEventFamily.OnWriteComplete,
WorkerSequence = 2,
});
await using MxGatewayClient client = CreateClient(transport);
MxGatewaySession session = await client.OpenSessionAsync();
List<ulong> sequences = [];
await foreach (MxEvent gatewayEvent in session.StreamEventsAsync(afterWorkerSequence: 0))
{
sequences.Add(gatewayEvent.WorkerSequence);
}
Assert.Equal([1UL, 2UL], sequences);
StreamEventsRequest request = Assert.Single(transport.StreamEventsCalls).Request;
Assert.Equal("session-fixture", request.SessionId);
}
[Fact]
public async Task CloseAsync_IsExplicitAndIdempotent()
{
FakeGatewayTransport transport = CreateTransport();
await using MxGatewayClient client = CreateClient(transport);
MxGatewaySession session = await client.OpenSessionAsync();
CloseSessionReply first = await session.CloseAsync();
CloseSessionReply second = await session.CloseAsync();
Assert.Same(first, second);
var call = Assert.Single(transport.CloseSessionCalls);
Assert.Equal("session-fixture", call.Request.SessionId);
}
[Fact]
public async Task InvokeHelpers_PassCancellationTokenToTransport()
{
using CancellationTokenSource cancellation = new();
FakeGatewayTransport transport = CreateTransport();
transport.AddInvokeReply(new MxCommandReply
{
SessionId = "session-fixture",
Kind = MxCommandKind.Advise,
ProtocolStatus = new ProtocolStatus { Code = ProtocolStatusCode.Ok },
});
await using MxGatewayClient client = CreateClient(transport);
MxGatewaySession session = await client.OpenSessionAsync();
await session.AdviseAsync(12, 34, cancellation.Token);
Assert.Equal(cancellation.Token, Assert.Single(transport.InvokeCalls).CallOptions.CancellationToken);
}
private static MxGatewayClient CreateClient(FakeGatewayTransport transport)
{
return new MxGatewayClient(transport.Options, transport);
}
private static FakeGatewayTransport CreateTransport()
{
return new FakeGatewayTransport(new MxGatewayClientOptions
{
Endpoint = new Uri("http://localhost:5000"),
ApiKey = "test-api-key",
});
}
}
@@ -0,0 +1,57 @@
using System.Text.Json;
using Google.Protobuf;
using MxGateway.Client;
using MxGateway.Contracts.Proto;
namespace MxGateway.Client.Tests;
public sealed class MxStatusProxyExtensionsTests
{
[Fact]
public void FixtureStatuses_ProjectSuccessAndPreserveRawFields()
{
using JsonDocument document = JsonDocument.Parse(ReadFixture(
"statuses",
"status-conversion-cases.json"));
foreach (JsonElement testCase in document.RootElement.GetProperty("cases").EnumerateArray())
{
MxStatusProxy status = JsonParser.Default.Parse<MxStatusProxy>(
testCase.GetProperty("status").GetRawText());
int success = testCase.GetProperty("status").GetProperty("success").GetInt32();
Assert.Equal(success != 0 && status.Category is MxStatusCategory.Ok, status.IsSuccess());
Assert.Equal(
testCase.GetProperty("status").GetProperty("rawCategory").GetInt32(),
status.RawCategory);
Assert.Equal(
testCase.GetProperty("status").GetProperty("rawDetectedBy").GetInt32(),
status.RawDetectedBy);
}
}
private static string ReadFixture(string category, string fileName)
{
DirectoryInfo directory = new(AppContext.BaseDirectory);
while (directory is not null)
{
string path = Path.Combine(
directory.FullName,
"clients",
"proto",
"fixtures",
"behavior",
category,
fileName);
if (File.Exists(path))
{
return File.ReadAllText(path);
}
directory = directory.Parent!;
}
throw new FileNotFoundException(fileName);
}
}
@@ -0,0 +1,79 @@
using System.Text.Json;
using Google.Protobuf;
using MxGateway.Client;
using MxGateway.Contracts.Proto;
namespace MxGateway.Client.Tests;
public sealed class MxValueExtensionsTests
{
[Fact]
public void ToMxValue_WithScalarValues_CreatesTypedProtobufValues()
{
Assert.Equal(MxValue.KindOneofCase.BoolValue, true.ToMxValue().KindCase);
Assert.Equal(MxValue.KindOneofCase.Int32Value, 123.ToMxValue().KindCase);
Assert.Equal(MxValue.KindOneofCase.Int64Value, 123L.ToMxValue().KindCase);
Assert.Equal(MxValue.KindOneofCase.FloatValue, 1.25F.ToMxValue().KindCase);
Assert.Equal(MxValue.KindOneofCase.DoubleValue, 2.5D.ToMxValue().KindCase);
Assert.Equal(MxValue.KindOneofCase.StringValue, "alpha".ToMxValue().KindCase);
}
[Fact]
public void ToMxValue_WithArrays_CreatesTypedArrayProtobufValues()
{
MxValue value = new[] { "alpha", "beta" }.ToMxValue();
Assert.Equal(MxValue.KindOneofCase.ArrayValue, value.KindCase);
Assert.Equal(MxArray.ValuesOneofCase.StringValues, value.ArrayValue.ValuesCase);
Assert.Equal(["alpha", "beta"], value.ArrayValue.StringValues.Values);
Assert.Equal([2U], value.ArrayValue.Dimensions);
}
[Fact]
public void FixtureValues_ProjectExpectedKindsAndPreserveRawMetadata()
{
using JsonDocument document = JsonDocument.Parse(ReadFixture(
"values",
"value-conversion-cases.json"));
foreach (JsonElement testCase in document.RootElement.GetProperty("cases").EnumerateArray())
{
string expectedKind = testCase.GetProperty("expectedKind").GetString()!;
MxValue value = JsonParser.Default.Parse<MxValue>(
testCase.GetProperty("value").GetRawText());
Assert.Equal(expectedKind, value.GetProjectionKind());
if (testCase.GetProperty("id").GetString() is "raw-fallback.variant")
{
Assert.Equal(32767, value.RawDataType);
Assert.Equal([1, 2, 3, 4, 5], Assert.IsType<byte[]>(value.ToClrValue()));
}
}
}
private static string ReadFixture(string category, string fileName)
{
DirectoryInfo directory = new(AppContext.BaseDirectory);
while (directory is not null)
{
string path = Path.Combine(
directory.FullName,
"clients",
"proto",
"fixtures",
"behavior",
category,
fileName);
if (File.Exists(path))
{
return File.ReadAllText(path);
}
directory = directory.Parent!;
}
throw new FileNotFoundException(fileName);
}
}
@@ -0,0 +1,117 @@
using Grpc.Core;
using MxGateway.Contracts.Proto;
namespace MxGateway.Client;
internal sealed class GrpcMxGatewayClientTransport(
MxGatewayClientOptions options,
MxAccessGateway.MxAccessGatewayClient rawClient) : IMxGatewayClientTransport
{
public MxGatewayClientOptions Options { get; } = options;
public MxAccessGateway.MxAccessGatewayClient RawClient { get; } = rawClient;
MxAccessGateway.MxAccessGatewayClient? IMxGatewayClientTransport.RawClient => RawClient;
public async Task<OpenSessionReply> OpenSessionAsync(
OpenSessionRequest request,
CallOptions callOptions)
{
try
{
return await RawClient.OpenSessionAsync(request, callOptions)
.ResponseAsync
.ConfigureAwait(false);
}
catch (RpcException exception)
{
throw MapRpcException(exception);
}
}
public async Task<CloseSessionReply> CloseSessionAsync(
CloseSessionRequest request,
CallOptions callOptions)
{
try
{
return await RawClient.CloseSessionAsync(request, callOptions)
.ResponseAsync
.ConfigureAwait(false);
}
catch (RpcException exception)
{
throw MapRpcException(exception);
}
}
public async Task<MxCommandReply> InvokeAsync(
MxCommandRequest request,
CallOptions callOptions)
{
try
{
return await RawClient.InvokeAsync(request, callOptions)
.ResponseAsync
.ConfigureAwait(false);
}
catch (RpcException exception)
{
throw MapRpcException(exception);
}
}
public async IAsyncEnumerable<MxEvent> StreamEventsAsync(
StreamEventsRequest request,
CallOptions callOptions,
[System.Runtime.CompilerServices.EnumeratorCancellation] CancellationToken cancellationToken = default)
{
CancellationToken effectiveCancellationToken = cancellationToken.CanBeCanceled
? cancellationToken
: callOptions.CancellationToken;
using AsyncServerStreamingCall<MxEvent> call = RawClient.StreamEvents(request, callOptions);
IAsyncStreamReader<MxEvent> responseStream = call.ResponseStream;
while (true)
{
MxEvent? gatewayEvent;
try
{
if (!await responseStream.MoveNext(effectiveCancellationToken).ConfigureAwait(false))
{
break;
}
gatewayEvent = responseStream.Current;
}
catch (RpcException exception)
{
throw MapRpcException(exception);
}
yield return gatewayEvent;
}
}
IAsyncEnumerable<MxEvent> IMxGatewayClientTransport.StreamEventsAsync(
StreamEventsRequest request,
CallOptions callOptions)
{
return StreamEventsAsync(request, callOptions);
}
private static MxGatewayException MapRpcException(RpcException exception)
{
return exception.StatusCode switch
{
StatusCode.Unauthenticated => new MxGatewayAuthenticationException(
exception.Status.Detail,
innerException: exception),
StatusCode.PermissionDenied => new MxGatewayAuthorizationException(
exception.Status.Detail,
innerException: exception),
_ => new MxGatewayException(exception.Status.Detail, exception),
};
}
}
@@ -0,0 +1,27 @@
using Grpc.Core;
using MxGateway.Contracts.Proto;
namespace MxGateway.Client;
internal interface IMxGatewayClientTransport
{
MxGatewayClientOptions Options { get; }
MxAccessGateway.MxAccessGatewayClient? RawClient { get; }
Task<OpenSessionReply> OpenSessionAsync(
OpenSessionRequest request,
CallOptions callOptions);
Task<CloseSessionReply> CloseSessionAsync(
CloseSessionRequest request,
CallOptions callOptions);
Task<MxCommandReply> InvokeAsync(
MxCommandRequest request,
CallOptions callOptions);
IAsyncEnumerable<MxEvent> StreamEventsAsync(
StreamEventsRequest request,
CallOptions callOptions);
}
@@ -0,0 +1,24 @@
using MxGateway.Contracts.Proto;
namespace MxGateway.Client;
public sealed class MxAccessException : MxGatewayCommandException
{
public MxAccessException(
string message,
MxCommandReply reply,
Exception? innerException = null)
: base(
message,
reply.SessionId,
reply.CorrelationId,
reply.ProtocolStatus,
reply.HasHresult ? reply.Hresult : null,
reply.Statuses.ToArray(),
innerException)
{
Reply = reply;
}
public MxCommandReply Reply { get; }
}
@@ -0,0 +1,96 @@
using MxGateway.Contracts.Proto;
namespace MxGateway.Client;
public static class MxCommandReplyExtensions
{
public static MxCommandReply EnsureProtocolSuccess(this MxCommandReply reply)
{
ArgumentNullException.ThrowIfNull(reply);
ProtocolStatusCode code = reply.ProtocolStatus?.Code
?? ProtocolStatusCode.Unspecified;
if (code is ProtocolStatusCode.Ok or ProtocolStatusCode.MxaccessFailure)
{
return reply;
}
throw CreateProtocolException(reply, code);
}
public static MxCommandReply EnsureMxAccessSuccess(this MxCommandReply reply)
{
ArgumentNullException.ThrowIfNull(reply);
bool mxAccessFailure = reply.ProtocolStatus?.Code is ProtocolStatusCode.MxaccessFailure;
bool hResultFailure = reply.HasHresult && reply.Hresult != 0;
bool statusFailure = reply.Statuses.Any(status => !status.IsSuccess());
if (!mxAccessFailure && !hResultFailure && !statusFailure)
{
return reply;
}
throw new MxAccessException(CreateMxAccessMessage(reply), reply);
}
private static MxGatewayException CreateProtocolException(
MxCommandReply reply,
ProtocolStatusCode code)
{
string message = CreateProtocolMessage(reply);
int? hResult = reply.HasHresult ? reply.Hresult : null;
MxStatusProxy[] statuses = reply.Statuses.ToArray();
return code switch
{
ProtocolStatusCode.SessionNotFound or ProtocolStatusCode.SessionNotReady
=> new MxGatewaySessionException(
message,
reply.SessionId,
reply.CorrelationId,
reply.ProtocolStatus,
hResult,
statuses),
ProtocolStatusCode.WorkerUnavailable
=> new MxGatewayWorkerException(
message,
reply.SessionId,
reply.CorrelationId,
reply.ProtocolStatus,
hResult,
statuses),
_
=> new MxGatewayCommandException(
message,
reply.SessionId,
reply.CorrelationId,
reply.ProtocolStatus,
hResult,
statuses),
};
}
private static string CreateProtocolMessage(MxCommandReply reply)
{
string statusMessage = string.IsNullOrWhiteSpace(reply.ProtocolStatus?.Message)
? "Gateway protocol failure."
: reply.ProtocolStatus.Message;
return $"{statusMessage} code={reply.ProtocolStatus?.Code}; session={reply.SessionId}; correlation={reply.CorrelationId}";
}
private static string CreateMxAccessMessage(MxCommandReply reply)
{
string statusSummary = reply.Statuses.Count is 0
? "no MXSTATUS_PROXY entries"
: string.Join("; ", reply.Statuses.Select(status => status.ToDiagnosticSummary()));
string hResult = reply.HasHresult
? $"0x{reply.Hresult:X8}"
: "none";
return $"MXAccess command failed. kind={reply.Kind}; hresult={hResult}; statuses={statusSummary}";
}
}
@@ -0,0 +1,25 @@
using MxGateway.Contracts.Proto;
namespace MxGateway.Client;
public sealed class MxGatewayAuthenticationException : MxGatewayException
{
public MxGatewayAuthenticationException(
string message,
string? sessionId = null,
string? correlationId = null,
ProtocolStatus? protocolStatus = null,
int? hResult = null,
IReadOnlyList<MxStatusProxy>? statuses = null,
Exception? innerException = null)
: base(
message,
sessionId,
correlationId,
protocolStatus,
hResult,
statuses ?? [],
innerException)
{
}
}
@@ -0,0 +1,25 @@
using MxGateway.Contracts.Proto;
namespace MxGateway.Client;
public sealed class MxGatewayAuthorizationException : MxGatewayException
{
public MxGatewayAuthorizationException(
string message,
string? sessionId = null,
string? correlationId = null,
ProtocolStatus? protocolStatus = null,
int? hResult = null,
IReadOnlyList<MxStatusProxy>? statuses = null,
Exception? innerException = null)
: base(
message,
sessionId,
correlationId,
protocolStatus,
hResult,
statuses ?? [],
innerException)
{
}
}
@@ -1,22 +1,44 @@
using Grpc.Core;
using Grpc.Net.Client;
using MxGateway.Contracts.Proto;
namespace MxGateway.Client;
/// <summary>
/// Provides the initial .NET client entry point and raw generated gRPC client.
/// Provides the .NET client entry point for the public MXAccess Gateway gRPC API.
/// </summary>
public sealed class MxGatewayClient : IAsyncDisposable
{
private readonly GrpcChannel _channel;
private readonly IMxGatewayClientTransport _transport;
private bool _disposed;
private MxGatewayClient(GrpcChannel channel)
internal MxGatewayClient(
MxGatewayClientOptions options,
IMxGatewayClientTransport transport)
{
_channel = channel;
RawClient = new MxAccessGateway.MxAccessGatewayClient(channel);
ArgumentNullException.ThrowIfNull(options);
options.Validate();
Options = options;
_transport = transport ?? throw new ArgumentNullException(nameof(transport));
_channel = null!;
}
public MxAccessGateway.MxAccessGatewayClient RawClient { get; }
private MxGatewayClient(
GrpcChannel channel,
IMxGatewayClientTransport transport)
{
_channel = channel;
_transport = transport;
Options = transport.Options;
}
public MxGatewayClientOptions Options { get; }
public MxAccessGateway.MxAccessGatewayClient RawClient =>
_transport.RawClient
?? throw new InvalidOperationException("The raw generated gRPC client is not available for this client instance.");
public static MxGatewayClient Create(MxGatewayClientOptions options)
{
@@ -30,12 +52,92 @@ public sealed class MxGatewayClient : IAsyncDisposable
LoggerFactory = options.LoggerFactory,
});
return new MxGatewayClient(channel);
return new MxGatewayClient(
channel,
new GrpcMxGatewayClientTransport(
options,
new MxAccessGateway.MxAccessGatewayClient(channel)));
}
public async Task<MxGatewaySession> OpenSessionAsync(
OpenSessionRequest? request = null,
CancellationToken cancellationToken = default)
{
OpenSessionReply reply = await OpenSessionRawAsync(
request ?? new OpenSessionRequest(),
cancellationToken)
.ConfigureAwait(false);
return new MxGatewaySession(this, reply);
}
public Task<OpenSessionReply> OpenSessionRawAsync(
OpenSessionRequest request,
CancellationToken cancellationToken = default)
{
ArgumentNullException.ThrowIfNull(request);
ThrowIfDisposed();
return _transport.OpenSessionAsync(request, CreateCallOptions(cancellationToken));
}
public Task<CloseSessionReply> CloseSessionRawAsync(
CloseSessionRequest request,
CancellationToken cancellationToken = default)
{
ArgumentNullException.ThrowIfNull(request);
ThrowIfDisposed();
return _transport.CloseSessionAsync(request, CreateCallOptions(cancellationToken));
}
public Task<MxCommandReply> InvokeAsync(
MxCommandRequest request,
CancellationToken cancellationToken = default)
{
ArgumentNullException.ThrowIfNull(request);
ThrowIfDisposed();
return _transport.InvokeAsync(request, CreateCallOptions(cancellationToken));
}
public IAsyncEnumerable<MxEvent> StreamEventsAsync(
StreamEventsRequest request,
CancellationToken cancellationToken = default)
{
ArgumentNullException.ThrowIfNull(request);
ThrowIfDisposed();
return _transport.StreamEventsAsync(request, CreateCallOptions(cancellationToken));
}
public ValueTask DisposeAsync()
{
_channel.Dispose();
if (_disposed)
{
return ValueTask.CompletedTask;
}
_disposed = true;
_channel?.Dispose();
return ValueTask.CompletedTask;
}
internal CallOptions CreateCallOptions(CancellationToken cancellationToken)
{
Metadata headers = new()
{
{ "authorization", $"Bearer {Options.ApiKey}" },
};
return new CallOptions(
headers,
DateTime.UtcNow.Add(Options.DefaultCallTimeout),
cancellationToken);
}
private void ThrowIfDisposed()
{
ObjectDisposedException.ThrowIf(_disposed, this);
}
}
@@ -0,0 +1,25 @@
using MxGateway.Contracts.Proto;
namespace MxGateway.Client;
public class MxGatewayCommandException : MxGatewayException
{
public MxGatewayCommandException(
string message,
string? sessionId = null,
string? correlationId = null,
ProtocolStatus? protocolStatus = null,
int? hResult = null,
IReadOnlyList<MxStatusProxy>? statuses = null,
Exception? innerException = null)
: base(
message,
sessionId,
correlationId,
protocolStatus,
hResult,
statuses ?? [],
innerException)
{
}
}
@@ -0,0 +1,45 @@
using MxGateway.Contracts.Proto;
namespace MxGateway.Client;
public class MxGatewayException : Exception
{
public MxGatewayException(string message)
: base(message)
{
Statuses = [];
}
public MxGatewayException(string message, Exception? innerException)
: base(message, innerException)
{
Statuses = [];
}
public MxGatewayException(
string message,
string? sessionId,
string? correlationId,
ProtocolStatus? protocolStatus,
int? hResult,
IReadOnlyList<MxStatusProxy> statuses,
Exception? innerException = null)
: base(message, innerException)
{
SessionId = sessionId;
CorrelationId = correlationId;
ProtocolStatus = protocolStatus;
HResultCode = hResult;
Statuses = statuses;
}
public string? SessionId { get; }
public string? CorrelationId { get; }
public ProtocolStatus? ProtocolStatus { get; }
public int? HResultCode { get; }
public IReadOnlyList<MxStatusProxy> Statuses { get; }
}
@@ -0,0 +1,300 @@
using MxGateway.Contracts.Proto;
namespace MxGateway.Client;
/// <summary>
/// Represents one gateway-backed MXAccess session.
/// </summary>
public sealed class MxGatewaySession : IAsyncDisposable
{
private readonly MxGatewayClient _client;
private readonly SemaphoreSlim _closeLock = new(1, 1);
private CloseSessionReply? _closeReply;
internal MxGatewaySession(
MxGatewayClient client,
OpenSessionReply openSessionReply)
{
_client = client ?? throw new ArgumentNullException(nameof(client));
OpenSessionReply = openSessionReply ?? throw new ArgumentNullException(nameof(openSessionReply));
}
public string SessionId => OpenSessionReply.SessionId;
public OpenSessionReply OpenSessionReply { get; }
public async Task<CloseSessionReply> CloseAsync(CancellationToken cancellationToken = default)
{
if (_closeReply is not null)
{
return _closeReply;
}
await _closeLock.WaitAsync(cancellationToken).ConfigureAwait(false);
try
{
if (_closeReply is not null)
{
return _closeReply;
}
_closeReply = await _client.CloseSessionRawAsync(
new CloseSessionRequest { SessionId = SessionId },
cancellationToken)
.ConfigureAwait(false);
return _closeReply;
}
finally
{
_closeLock.Release();
}
}
public async Task<int> RegisterAsync(
string clientName,
CancellationToken cancellationToken = default)
{
MxCommandReply reply = await RegisterRawAsync(clientName, cancellationToken)
.ConfigureAwait(false);
reply.EnsureProtocolSuccess().EnsureMxAccessSuccess();
return reply.Register?.ServerHandle ?? reply.ReturnValue.Int32Value;
}
public Task<MxCommandReply> RegisterRawAsync(
string clientName,
CancellationToken cancellationToken = default)
{
ArgumentException.ThrowIfNullOrWhiteSpace(clientName);
return InvokeCommandAsync(
new MxCommand
{
Kind = MxCommandKind.Register,
Register = new RegisterCommand { ClientName = clientName },
},
cancellationToken);
}
public async Task<int> AddItemAsync(
int serverHandle,
string itemDefinition,
CancellationToken cancellationToken = default)
{
MxCommandReply reply = await AddItemRawAsync(
serverHandle,
itemDefinition,
cancellationToken)
.ConfigureAwait(false);
reply.EnsureProtocolSuccess().EnsureMxAccessSuccess();
return reply.AddItem?.ItemHandle ?? reply.ReturnValue.Int32Value;
}
public Task<MxCommandReply> AddItemRawAsync(
int serverHandle,
string itemDefinition,
CancellationToken cancellationToken = default)
{
ArgumentException.ThrowIfNullOrWhiteSpace(itemDefinition);
return InvokeCommandAsync(
new MxCommand
{
Kind = MxCommandKind.AddItem,
AddItem = new AddItemCommand
{
ServerHandle = serverHandle,
ItemDefinition = itemDefinition,
},
},
cancellationToken);
}
public async Task<int> AddItem2Async(
int serverHandle,
string itemDefinition,
string itemContext,
CancellationToken cancellationToken = default)
{
MxCommandReply reply = await AddItem2RawAsync(
serverHandle,
itemDefinition,
itemContext,
cancellationToken)
.ConfigureAwait(false);
reply.EnsureProtocolSuccess().EnsureMxAccessSuccess();
return reply.AddItem2?.ItemHandle ?? reply.ReturnValue.Int32Value;
}
public Task<MxCommandReply> AddItem2RawAsync(
int serverHandle,
string itemDefinition,
string itemContext,
CancellationToken cancellationToken = default)
{
ArgumentException.ThrowIfNullOrWhiteSpace(itemDefinition);
return InvokeCommandAsync(
new MxCommand
{
Kind = MxCommandKind.AddItem2,
AddItem2 = new AddItem2Command
{
ServerHandle = serverHandle,
ItemDefinition = itemDefinition,
ItemContext = itemContext ?? string.Empty,
},
},
cancellationToken);
}
public async Task AdviseAsync(
int serverHandle,
int itemHandle,
CancellationToken cancellationToken = default)
{
MxCommandReply reply = await AdviseRawAsync(serverHandle, itemHandle, cancellationToken)
.ConfigureAwait(false);
reply.EnsureProtocolSuccess().EnsureMxAccessSuccess();
}
public Task<MxCommandReply> AdviseRawAsync(
int serverHandle,
int itemHandle,
CancellationToken cancellationToken = default)
{
return InvokeCommandAsync(
new MxCommand
{
Kind = MxCommandKind.Advise,
Advise = new AdviseCommand
{
ServerHandle = serverHandle,
ItemHandle = itemHandle,
},
},
cancellationToken);
}
public async Task WriteAsync(
int serverHandle,
int itemHandle,
MxValue value,
int userId,
CancellationToken cancellationToken = default)
{
MxCommandReply reply = await WriteRawAsync(serverHandle, itemHandle, value, userId, cancellationToken)
.ConfigureAwait(false);
reply.EnsureProtocolSuccess().EnsureMxAccessSuccess();
}
public Task<MxCommandReply> WriteRawAsync(
int serverHandle,
int itemHandle,
MxValue value,
int userId,
CancellationToken cancellationToken = default)
{
ArgumentNullException.ThrowIfNull(value);
return InvokeCommandAsync(
new MxCommand
{
Kind = MxCommandKind.Write,
Write = new WriteCommand
{
ServerHandle = serverHandle,
ItemHandle = itemHandle,
Value = value,
UserId = userId,
},
},
cancellationToken);
}
public async Task Write2Async(
int serverHandle,
int itemHandle,
MxValue value,
MxValue timestampValue,
int userId,
CancellationToken cancellationToken = default)
{
MxCommandReply reply = await Write2RawAsync(
serverHandle,
itemHandle,
value,
timestampValue,
userId,
cancellationToken)
.ConfigureAwait(false);
reply.EnsureProtocolSuccess().EnsureMxAccessSuccess();
}
public Task<MxCommandReply> Write2RawAsync(
int serverHandle,
int itemHandle,
MxValue value,
MxValue timestampValue,
int userId,
CancellationToken cancellationToken = default)
{
ArgumentNullException.ThrowIfNull(value);
ArgumentNullException.ThrowIfNull(timestampValue);
return InvokeCommandAsync(
new MxCommand
{
Kind = MxCommandKind.Write2,
Write2 = new Write2Command
{
ServerHandle = serverHandle,
ItemHandle = itemHandle,
Value = value,
TimestampValue = timestampValue,
UserId = userId,
},
},
cancellationToken);
}
public Task<MxCommandReply> InvokeAsync(
MxCommandRequest request,
CancellationToken cancellationToken = default)
{
ArgumentNullException.ThrowIfNull(request);
return _client.InvokeAsync(request, cancellationToken);
}
public IAsyncEnumerable<MxEvent> StreamEventsAsync(
ulong afterWorkerSequence = 0,
CancellationToken cancellationToken = default)
{
return _client.StreamEventsAsync(
new StreamEventsRequest
{
SessionId = SessionId,
AfterWorkerSequence = afterWorkerSequence,
},
cancellationToken);
}
public async ValueTask DisposeAsync()
{
await CloseAsync().ConfigureAwait(false);
_closeLock.Dispose();
}
private Task<MxCommandReply> InvokeCommandAsync(
MxCommand command,
CancellationToken cancellationToken)
{
return _client.InvokeAsync(
new MxCommandRequest
{
SessionId = SessionId,
ClientCorrelationId = Guid.NewGuid().ToString("N"),
Command = command,
},
cancellationToken);
}
}
@@ -0,0 +1,25 @@
using MxGateway.Contracts.Proto;
namespace MxGateway.Client;
public sealed class MxGatewaySessionException : MxGatewayException
{
public MxGatewaySessionException(
string message,
string? sessionId = null,
string? correlationId = null,
ProtocolStatus? protocolStatus = null,
int? hResult = null,
IReadOnlyList<MxStatusProxy>? statuses = null,
Exception? innerException = null)
: base(
message,
sessionId,
correlationId,
protocolStatus,
hResult,
statuses ?? [],
innerException)
{
}
}
@@ -0,0 +1,25 @@
using MxGateway.Contracts.Proto;
namespace MxGateway.Client;
public sealed class MxGatewayWorkerException : MxGatewayException
{
public MxGatewayWorkerException(
string message,
string? sessionId = null,
string? correlationId = null,
ProtocolStatus? protocolStatus = null,
int? hResult = null,
IReadOnlyList<MxStatusProxy>? statuses = null,
Exception? innerException = null)
: base(
message,
sessionId,
correlationId,
protocolStatus,
hResult,
statuses ?? [],
innerException)
{
}
}
@@ -0,0 +1,25 @@
using MxGateway.Contracts.Proto;
namespace MxGateway.Client;
public static class MxStatusProxyExtensions
{
public static bool IsSuccess(this MxStatusProxy status)
{
ArgumentNullException.ThrowIfNull(status);
return status.Success != 0
&& status.Category is MxStatusCategory.Ok;
}
public static string ToDiagnosticSummary(this MxStatusProxy status)
{
ArgumentNullException.ThrowIfNull(status);
string diagnosticText = string.IsNullOrWhiteSpace(status.DiagnosticText)
? "no diagnostic text"
: status.DiagnosticText;
return $"{status.Category} by {status.DetectedBy}; detail={status.Detail}; {diagnosticText}";
}
}
@@ -0,0 +1,284 @@
using Google.Protobuf;
using Google.Protobuf.WellKnownTypes;
using MxGateway.Contracts.Proto;
namespace MxGateway.Client;
/// <summary>
/// Creates and projects gateway MXAccess values without hiding the raw
/// protobuf value carried by command replies and events.
/// </summary>
public static class MxValueExtensions
{
public static MxValue ToMxValue(this bool value)
{
return new MxValue
{
DataType = MxDataType.Boolean,
VariantType = "VT_BOOL",
BoolValue = value,
};
}
public static MxValue ToMxValue(this int value)
{
return new MxValue
{
DataType = MxDataType.Integer,
VariantType = "VT_I4",
Int32Value = value,
};
}
public static MxValue ToMxValue(this long value)
{
return new MxValue
{
DataType = MxDataType.Integer,
VariantType = "VT_I8",
Int64Value = value,
};
}
public static MxValue ToMxValue(this float value)
{
return new MxValue
{
DataType = MxDataType.Float,
VariantType = "VT_R4",
FloatValue = value,
};
}
public static MxValue ToMxValue(this double value)
{
return new MxValue
{
DataType = MxDataType.Double,
VariantType = "VT_R8",
DoubleValue = value,
};
}
public static MxValue ToMxValue(this string value)
{
ArgumentNullException.ThrowIfNull(value);
return new MxValue
{
DataType = MxDataType.String,
VariantType = "VT_BSTR",
StringValue = value,
};
}
public static MxValue ToMxValue(this DateTimeOffset value)
{
return new MxValue
{
DataType = MxDataType.Time,
VariantType = "VT_DATE",
TimestampValue = Timestamp.FromDateTimeOffset(value),
};
}
public static MxValue ToMxValue(this DateTime value)
{
return new DateTimeOffset(
value.Kind == DateTimeKind.Unspecified
? DateTime.SpecifyKind(value, DateTimeKind.Utc)
: value.ToUniversalTime())
.ToMxValue();
}
public static MxValue ToMxValue(this IReadOnlyList<bool> values)
{
ArgumentNullException.ThrowIfNull(values);
var array = new BoolArray();
array.Values.Add(values);
return CreateArrayValue(MxDataType.Boolean, "VT_ARRAY|VT_BOOL", values.Count, new MxArray
{
ElementDataType = MxDataType.Boolean,
VariantType = "VT_ARRAY|VT_BOOL",
BoolValues = array,
});
}
public static MxValue ToMxValue(this IReadOnlyList<int> values)
{
ArgumentNullException.ThrowIfNull(values);
var array = new Int32Array();
array.Values.Add(values);
return CreateArrayValue(MxDataType.Integer, "VT_ARRAY|VT_I4", values.Count, new MxArray
{
ElementDataType = MxDataType.Integer,
VariantType = "VT_ARRAY|VT_I4",
Int32Values = array,
});
}
public static MxValue ToMxValue(this IReadOnlyList<long> values)
{
ArgumentNullException.ThrowIfNull(values);
var array = new Int64Array();
array.Values.Add(values);
return CreateArrayValue(MxDataType.Integer, "VT_ARRAY|VT_I8", values.Count, new MxArray
{
ElementDataType = MxDataType.Integer,
VariantType = "VT_ARRAY|VT_I8",
Int64Values = array,
});
}
public static MxValue ToMxValue(this IReadOnlyList<float> values)
{
ArgumentNullException.ThrowIfNull(values);
var array = new FloatArray();
array.Values.Add(values);
return CreateArrayValue(MxDataType.Float, "VT_ARRAY|VT_R4", values.Count, new MxArray
{
ElementDataType = MxDataType.Float,
VariantType = "VT_ARRAY|VT_R4",
FloatValues = array,
});
}
public static MxValue ToMxValue(this IReadOnlyList<double> values)
{
ArgumentNullException.ThrowIfNull(values);
var array = new DoubleArray();
array.Values.Add(values);
return CreateArrayValue(MxDataType.Double, "VT_ARRAY|VT_R8", values.Count, new MxArray
{
ElementDataType = MxDataType.Double,
VariantType = "VT_ARRAY|VT_R8",
DoubleValues = array,
});
}
public static MxValue ToMxValue(this IReadOnlyList<string> values)
{
ArgumentNullException.ThrowIfNull(values);
var array = new StringArray();
array.Values.Add(values);
return CreateArrayValue(MxDataType.String, "VT_ARRAY|VT_BSTR", values.Count, new MxArray
{
ElementDataType = MxDataType.String,
VariantType = "VT_ARRAY|VT_BSTR",
StringValues = array,
});
}
public static MxValue ToMxValue(this IReadOnlyList<DateTimeOffset> values)
{
ArgumentNullException.ThrowIfNull(values);
var array = new TimestampArray();
array.Values.Add(values.Select(Timestamp.FromDateTimeOffset));
return CreateArrayValue(MxDataType.Time, "VT_ARRAY|VT_DATE", values.Count, new MxArray
{
ElementDataType = MxDataType.Time,
VariantType = "VT_ARRAY|VT_DATE",
TimestampValues = array,
});
}
public static string GetProjectionKind(this MxValue value)
{
ArgumentNullException.ThrowIfNull(value);
return value.KindCase switch
{
MxValue.KindOneofCase.BoolValue => "boolValue",
MxValue.KindOneofCase.Int32Value => "int32Value",
MxValue.KindOneofCase.Int64Value => "int64Value",
MxValue.KindOneofCase.FloatValue => "floatValue",
MxValue.KindOneofCase.DoubleValue => "doubleValue",
MxValue.KindOneofCase.StringValue => "stringValue",
MxValue.KindOneofCase.TimestampValue => "timestampValue",
MxValue.KindOneofCase.ArrayValue => "arrayValue",
MxValue.KindOneofCase.RawValue => "rawValue",
_ => value.IsNull ? "nullValue" : "unspecified",
};
}
public static object? ToClrValue(this MxValue value)
{
ArgumentNullException.ThrowIfNull(value);
return value.KindCase switch
{
MxValue.KindOneofCase.BoolValue => value.BoolValue,
MxValue.KindOneofCase.Int32Value => value.Int32Value,
MxValue.KindOneofCase.Int64Value => value.Int64Value,
MxValue.KindOneofCase.FloatValue => value.FloatValue,
MxValue.KindOneofCase.DoubleValue => value.DoubleValue,
MxValue.KindOneofCase.StringValue => value.StringValue,
MxValue.KindOneofCase.TimestampValue => value.TimestampValue.ToDateTimeOffset(),
MxValue.KindOneofCase.ArrayValue => value.ArrayValue.ToClrArrayValue(),
MxValue.KindOneofCase.RawValue => value.RawValue.ToByteArray(),
_ => value.IsNull ? null : value,
};
}
public static object? ToClrArrayValue(this MxArray array)
{
ArgumentNullException.ThrowIfNull(array);
return array.ValuesCase switch
{
MxArray.ValuesOneofCase.BoolValues => array.BoolValues.Values.ToArray(),
MxArray.ValuesOneofCase.Int32Values => array.Int32Values.Values.ToArray(),
MxArray.ValuesOneofCase.Int64Values => array.Int64Values.Values.ToArray(),
MxArray.ValuesOneofCase.FloatValues => array.FloatValues.Values.ToArray(),
MxArray.ValuesOneofCase.DoubleValues => array.DoubleValues.Values.ToArray(),
MxArray.ValuesOneofCase.StringValues => array.StringValues.Values.ToArray(),
MxArray.ValuesOneofCase.TimestampValues => array.TimestampValues.Values
.Select(timestamp => timestamp.ToDateTimeOffset())
.ToArray(),
MxArray.ValuesOneofCase.RawValues => array.RawValues.Values
.Select(value => value.ToByteArray())
.ToArray(),
_ => null,
};
}
public static MxValue ToRawMxValue(
byte[] value,
string variantType,
string rawDiagnostic,
int rawDataType = 0)
{
ArgumentNullException.ThrowIfNull(value);
return new MxValue
{
DataType = MxDataType.Unknown,
VariantType = variantType,
RawDiagnostic = rawDiagnostic,
RawDataType = rawDataType,
RawValue = ByteString.CopyFrom(value),
};
}
private static MxValue CreateArrayValue(
MxDataType dataType,
string variantType,
int length,
MxArray array)
{
array.Dimensions.Add((uint)length);
return new MxValue
{
DataType = dataType,
VariantType = variantType,
ArrayValue = array,
};
}
}
@@ -0,0 +1,3 @@
using System.Runtime.CompilerServices;
[assembly: InternalsVisibleTo("MxGateway.Client.Tests")]
+89 -2
View File
@@ -7,9 +7,9 @@ CLI, and unit tests.
| Project | Purpose |
|---------|---------|
| `MxGateway.Client` | .NET 10 library entry point and raw gRPC client access. |
| `MxGateway.Client` | .NET 10 library entry point, raw gRPC calls, and session helpers. |
| `MxGateway.Client.Cli` | Test CLI for smoke and diagnostic commands. |
| `MxGateway.Client.Tests` | Unit tests for the scaffold and generated contract wiring. |
| `MxGateway.Client.Tests` | Unit tests for client options, generated contract wiring, auth metadata, session helpers, cancellation, and event streaming. |
The projects reference `src/MxGateway.Contracts/MxGateway.Contracts.csproj` so
the client compiles against the same generated protobuf and gRPC types as the
@@ -22,3 +22,90 @@ future client build switches to client-local `Grpc.Tools` generation.
dotnet build clients/dotnet/MxGateway.Client.sln
dotnet test clients/dotnet/MxGateway.Client.sln --no-build
```
## Client Usage
`MxGatewayClient` opens a gRPC channel to the gateway and attaches the API key
to every unary and streaming call as `authorization: Bearer <api-key>`.
Cancellation tokens passed to the public methods flow to the generated gRPC
call. Client-side cancellation stops waiting for the gateway response; it does
not abort an MXAccess COM call that is already executing inside a worker.
```csharp
await using MxGatewayClient client = MxGatewayClient.Create(
new MxGatewayClientOptions
{
Endpoint = new Uri("http://localhost:5000"),
ApiKey = apiKey,
});
MxGatewaySession session = await client.OpenSessionAsync();
try
{
int serverHandle = await session.RegisterAsync("sample-client");
int itemHandle = await session.AddItemAsync(
serverHandle,
"Area001.Pump001.Speed");
await session.AdviseAsync(serverHandle, itemHandle);
}
finally
{
await session.CloseAsync();
}
```
Use `OpenSessionRawAsync`, `CloseSessionRawAsync`, `InvokeAsync`, and
`StreamEventsAsync` when tests or parity tools need direct generated protobuf
messages. `MxGatewaySession.OpenSessionReply` keeps the raw session-open reply
available, and command helpers have `*RawAsync` variants when callers need the
complete `MxCommandReply`.
`MxGatewaySession.CloseAsync` is explicit and idempotent. Repeated calls return
the first `CloseSessionReply` instead of sending another close request.
## Values, Status, And Errors
The client provides extension helpers for generated protobuf values. Use
`ToMxValue()` on .NET scalar values and typed arrays to create `MxValue`
instances for `Write` and `Write2`. Use `ToClrValue()` and
`GetProjectionKind()` when test or diagnostic code needs to inspect generated
`MxValue` replies while preserving `rawDiagnostic`, raw data type fields, and
raw byte payloads.
`MxStatusProxy.IsSuccess()` and `ToDiagnosticSummary()` expose MXAccess status
arrays without collapsing them into a single gateway success flag. Command
reply helpers follow the same split:
```csharp
reply.EnsureProtocolSuccess();
reply.EnsureMxAccessSuccess();
```
`EnsureProtocolSuccess()` raises gateway, session, worker, or command
exceptions for gateway-level failures. It leaves
`PROTOCOL_STATUS_CODE_MXACCESS_FAILURE` to `EnsureMxAccessSuccess()` so callers
can keep the full `MxCommandReply`, HRESULT, and status array when MXAccess
itself rejects a command. `MxAccessException.Reply` contains the raw generated
reply.
## CLI Usage
The test CLI supports deterministic JSON output for automation:
```powershell
dotnet run --project clients/dotnet/MxGateway.Client.Cli -- version --json
dotnet run --project clients/dotnet/MxGateway.Client.Cli -- open-session --endpoint http://localhost:5000 --api-key-env MXGATEWAY_API_KEY --json
dotnet run --project clients/dotnet/MxGateway.Client.Cli -- register --session-id <id> --client-name mxgw-dotnet-cli --json
dotnet run --project clients/dotnet/MxGateway.Client.Cli -- add-item --session-id <id> --server-handle 1 --item Area001.Pump001.Speed --json
dotnet run --project clients/dotnet/MxGateway.Client.Cli -- advise --session-id <id> --server-handle 1 --item-handle 1 --json
dotnet run --project clients/dotnet/MxGateway.Client.Cli -- write --session-id <id> --server-handle 1 --item-handle 1 --type int32 --value 123 --json
dotnet run --project clients/dotnet/MxGateway.Client.Cli -- write2 --session-id <id> --server-handle 1 --item-handle 1 --type int32 --value 123 --timestamp 2026-01-01T00:00:00Z --json
dotnet run --project clients/dotnet/MxGateway.Client.Cli -- stream-events --session-id <id> --max-events 1 --json
dotnet run --project clients/dotnet/MxGateway.Client.Cli -- smoke --endpoint http://localhost:5000 --api-key-env MXGATEWAY_API_KEY --item Area001.Pump001.Speed --json
```
`smoke` opens a session, registers a client, adds one item, advises it,
optionally writes a value when `--type` and `--value` are supplied, reads a
bounded event stream, and closes the session in a `finally` block. CLI error
output redacts API keys supplied through `--api-key`.
+34 -6
View File
@@ -37,19 +37,47 @@ Run the Go module checks from `clients/go`:
```powershell
go test ./...
go build ./...
go vet ./...
```
The scaffold tests parse the shared golden JSON fixtures with the generated Go
types. Later client implementation tests add fake gRPC services, auth metadata,
streaming, value conversion, and CLI behavior.
The tests parse the shared JSON fixtures, exercise value and status conversion,
use `bufconn` for fake gateway auth and streaming behavior, and cover CLI JSON
redaction.
## Client API
Use `mxgateway.Dial` with `mxgateway.Options` to configure plaintext or TLS
transport, API-key metadata, dial timeout, and per-call timeout:
```go
client, err := mxgateway.Dial(ctx, mxgateway.Options{
Endpoint: "localhost:5000",
APIKey: os.Getenv("MXGATEWAY_API_KEY"),
Plaintext: true,
})
```
`Client.OpenSession` returns a `Session` with helpers for `Register`,
`AddItem`, `AddItem2`, `Advise`, `Write`, `Events`, and `Close`. Raw protobuf
messages remain available through the `mxgateway` package aliases and the
`Raw` helper methods. Typed errors support `errors.As` for `GatewayError`,
`CommandError`, and `MxAccessError`; command errors preserve the raw reply.
## CLI
The scaffold CLI exposes version information:
The `mxgw-go` CLI emits JSON with redacted API keys for commands that connect to
the gateway:
```powershell
go run ./cmd/mxgw-go version -json
go run ./cmd/mxgw-go open-session -endpoint localhost:5000 -plaintext -json
go run ./cmd/mxgw-go register -session-id <id> -client-name mxgw-go -plaintext -json
go run ./cmd/mxgw-go add-item -session-id <id> -server-handle 1 -item Area001.Tag.Value -plaintext -json
go run ./cmd/mxgw-go advise -session-id <id> -server-handle 1 -item-handle 1 -plaintext -json
go run ./cmd/mxgw-go write -session-id <id> -server-handle 1 -item-handle 1 -type int32 -value 123 -plaintext -json
go run ./cmd/mxgw-go stream-events -session-id <id> -plaintext -json
go run ./cmd/mxgw-go smoke -item Area001.Tag.Value -plaintext -json
```
Additional commands are implemented with the client/session wrapper work.
Use `-api-key-env MXGATEWAY_API_KEY` or `-api-key <key>` when authentication is
enabled. CLI output redacts the key value and never writes the raw secret.
+478 -11
View File
@@ -1,12 +1,19 @@
package main
import (
"context"
"encoding/json"
"errors"
"flag"
"fmt"
"io"
"os"
"strconv"
"time"
"gitea.dohertylan.com/dohertj2/mxaccessgw/clients/go/mxgateway"
"google.golang.org/protobuf/encoding/protojson"
"google.golang.org/protobuf/reflect/protoreflect"
)
type versionOutput struct {
@@ -15,29 +22,76 @@ type versionOutput struct {
WorkerProtocolVersion uint32 `json:"workerProtocolVersion"`
}
type commonOptions struct {
Endpoint string `json:"endpoint"`
APIKey string `json:"apiKey"`
APIKeyEnv string `json:"apiKeyEnv,omitempty"`
Plaintext bool `json:"plaintext"`
CACertFile string `json:"caCertFile,omitempty"`
ServerName string `json:"serverNameOverride,omitempty"`
CallTimeout string `json:"callTimeout,omitempty"`
apiKeyValue string
timeout time.Duration
}
type openSessionOutput struct {
Command string `json:"command"`
Options commonOptions `json:"options"`
Reply json.RawMessage `json:"reply"`
}
type commandReplyOutput struct {
Command string `json:"command"`
Options commonOptions `json:"options"`
Reply json.RawMessage `json:"reply"`
}
func main() {
if err := run(os.Args[1:]); err != nil {
if err := runWithIO(context.Background(), os.Args[1:], os.Stdout, os.Stderr); err != nil {
fmt.Fprintln(os.Stderr, err)
os.Exit(2)
}
}
func run(args []string) error {
return runWithIO(context.Background(), args, os.Stdout, os.Stderr)
}
func runWithIO(ctx context.Context, args []string, stdout, stderr io.Writer) error {
if len(args) == 0 {
return fmt.Errorf("usage: mxgw-go version [-json]")
writeUsage(stderr)
return errors.New("missing command")
}
switch args[0] {
case "version":
return runVersion(args[1:])
return runVersion(args[1:], stdout, stderr)
case "open-session":
return runOpenSession(ctx, args[1:], stdout, stderr)
case "close-session":
return runCloseSession(ctx, args[1:], stdout, stderr)
case "register":
return runRegister(ctx, args[1:], stdout, stderr)
case "add-item":
return runAddItem(ctx, args[1:], stdout, stderr)
case "advise":
return runAdvise(ctx, args[1:], stdout, stderr)
case "write":
return runWrite(ctx, args[1:], stdout, stderr)
case "stream-events":
return runStreamEvents(ctx, args[1:], stdout, stderr)
case "smoke":
return runSmoke(ctx, args[1:], stdout, stderr)
default:
writeUsage(stderr)
return fmt.Errorf("unknown command %q", args[0])
}
}
func runVersion(args []string) error {
func runVersion(args []string, stdout, stderr io.Writer) error {
flags := flag.NewFlagSet("version", flag.ContinueOnError)
flags.SetOutput(os.Stderr)
flags.SetOutput(stderr)
jsonOutput := flags.Bool("json", false, "write JSON output")
if err := flags.Parse(args); err != nil {
@@ -51,13 +105,426 @@ func runVersion(args []string) error {
}
if *jsonOutput {
encoder := json.NewEncoder(os.Stdout)
encoder.SetIndent("", " ")
return encoder.Encode(output)
return writeJSON(stdout, output)
}
fmt.Fprintf(os.Stdout, "mxgw-go %s\n", output.ClientVersion)
fmt.Fprintf(os.Stdout, "gateway protocol %d\n", output.GatewayProtocolVersion)
fmt.Fprintf(os.Stdout, "worker protocol %d\n", output.WorkerProtocolVersion)
fmt.Fprintf(stdout, "mxgw-go %s\n", output.ClientVersion)
fmt.Fprintf(stdout, "gateway protocol %d\n", output.GatewayProtocolVersion)
fmt.Fprintf(stdout, "worker protocol %d\n", output.WorkerProtocolVersion)
return nil
}
func runOpenSession(ctx context.Context, args []string, stdout, stderr io.Writer) error {
flags := flag.NewFlagSet("open-session", flag.ContinueOnError)
flags.SetOutput(stderr)
common := bindCommonFlags(flags)
jsonOutput := flags.Bool("json", false, "write JSON output")
clientName := flags.String("client-session-name", "", "client session name")
backend := flags.String("backend", "", "requested backend")
if err := flags.Parse(args); err != nil {
return err
}
client, options, err := dialForCommand(ctx, common)
if err != nil {
return err
}
defer client.Close()
reply, err := client.OpenSessionRaw(ctx, (&mxgateway.OpenSessionOptions{
RequestedBackend: *backend,
ClientSessionName: *clientName,
}).Request())
if err != nil {
return err
}
if *jsonOutput {
return writeJSON(stdout, openSessionOutput{
Command: "open-session",
Options: options,
Reply: mustMarshalProto(reply),
})
}
fmt.Fprintln(stdout, reply.GetSessionId())
return nil
}
func runCloseSession(ctx context.Context, args []string, stdout, stderr io.Writer) error {
flags := flag.NewFlagSet("close-session", flag.ContinueOnError)
flags.SetOutput(stderr)
common := bindCommonFlags(flags)
jsonOutput := flags.Bool("json", false, "write JSON output")
sessionID := flags.String("session-id", "", "gateway session id")
if err := flags.Parse(args); err != nil {
return err
}
if *sessionID == "" {
return errors.New("session-id is required")
}
client, options, err := dialForCommand(ctx, common)
if err != nil {
return err
}
defer client.Close()
reply, err := client.CloseSessionRaw(ctx, &mxgateway.CloseSessionRequest{SessionId: *sessionID})
if err != nil {
return err
}
if *jsonOutput {
return writeJSON(stdout, commandReplyOutput{
Command: "close-session",
Options: options,
Reply: mustMarshalProto(reply),
})
}
fmt.Fprintln(stdout, reply.GetFinalState())
return nil
}
func runRegister(ctx context.Context, args []string, stdout, stderr io.Writer) error {
flags := flag.NewFlagSet("register", flag.ContinueOnError)
flags.SetOutput(stderr)
common := bindCommonFlags(flags)
jsonOutput := flags.Bool("json", false, "write JSON output")
sessionID := flags.String("session-id", "", "gateway session id")
clientName := flags.String("client-name", "", "MXAccess client name")
if err := flags.Parse(args); err != nil {
return err
}
if *sessionID == "" || *clientName == "" {
return errors.New("session-id and client-name are required")
}
client, options, err := dialForCommand(ctx, common)
if err != nil {
return err
}
defer client.Close()
session := mxgateway.NewSessionForID(client, *sessionID)
reply, err := session.RegisterRaw(ctx, *clientName)
return writeCommandOutput(stdout, *jsonOutput, "register", options, reply, err)
}
func runAddItem(ctx context.Context, args []string, stdout, stderr io.Writer) error {
flags := flag.NewFlagSet("add-item", flag.ContinueOnError)
flags.SetOutput(stderr)
common := bindCommonFlags(flags)
jsonOutput := flags.Bool("json", false, "write JSON output")
sessionID := flags.String("session-id", "", "gateway session id")
serverHandle := flags.Int("server-handle", 0, "MXAccess server handle")
item := flags.String("item", "", "item definition")
if err := flags.Parse(args); err != nil {
return err
}
if *sessionID == "" || *item == "" {
return errors.New("session-id and item are required")
}
client, options, err := dialForCommand(ctx, common)
if err != nil {
return err
}
defer client.Close()
session := mxgateway.NewSessionForID(client, *sessionID)
reply, err := session.AddItemRaw(ctx, int32(*serverHandle), *item)
return writeCommandOutput(stdout, *jsonOutput, "add-item", options, reply, err)
}
func runAdvise(ctx context.Context, args []string, stdout, stderr io.Writer) error {
flags := flag.NewFlagSet("advise", flag.ContinueOnError)
flags.SetOutput(stderr)
common := bindCommonFlags(flags)
jsonOutput := flags.Bool("json", false, "write JSON output")
sessionID := flags.String("session-id", "", "gateway session id")
serverHandle := flags.Int("server-handle", 0, "MXAccess server handle")
itemHandle := flags.Int("item-handle", 0, "MXAccess item handle")
if err := flags.Parse(args); err != nil {
return err
}
if *sessionID == "" {
return errors.New("session-id is required")
}
client, options, err := dialForCommand(ctx, common)
if err != nil {
return err
}
defer client.Close()
session := mxgateway.NewSessionForID(client, *sessionID)
reply, err := session.AdviseRaw(ctx, int32(*serverHandle), int32(*itemHandle))
return writeCommandOutput(stdout, *jsonOutput, "advise", options, reply, err)
}
func runWrite(ctx context.Context, args []string, stdout, stderr io.Writer) error {
flags := flag.NewFlagSet("write", flag.ContinueOnError)
flags.SetOutput(stderr)
common := bindCommonFlags(flags)
jsonOutput := flags.Bool("json", false, "write JSON output")
sessionID := flags.String("session-id", "", "gateway session id")
serverHandle := flags.Int("server-handle", 0, "MXAccess server handle")
itemHandle := flags.Int("item-handle", 0, "MXAccess item handle")
valueType := flags.String("type", "string", "value type: bool, int32, int64, float, double, string")
valueText := flags.String("value", "", "value text")
userID := flags.Int("user-id", 0, "MXAccess user id")
if err := flags.Parse(args); err != nil {
return err
}
if *sessionID == "" {
return errors.New("session-id is required")
}
value, err := parseValue(*valueType, *valueText)
if err != nil {
return err
}
client, options, err := dialForCommand(ctx, common)
if err != nil {
return err
}
defer client.Close()
session := mxgateway.NewSessionForID(client, *sessionID)
reply, err := session.WriteRaw(ctx, int32(*serverHandle), int32(*itemHandle), value, int32(*userID))
return writeCommandOutput(stdout, *jsonOutput, "write", options, reply, err)
}
func runStreamEvents(ctx context.Context, args []string, stdout, stderr io.Writer) error {
flags := flag.NewFlagSet("stream-events", flag.ContinueOnError)
flags.SetOutput(stderr)
common := bindCommonFlags(flags)
jsonOutput := flags.Bool("json", false, "write JSON output")
sessionID := flags.String("session-id", "", "gateway session id")
after := flags.Uint64("after-worker-sequence", 0, "first worker sequence to read after")
limit := flags.Int("limit", 0, "maximum events to read; 0 means unbounded")
if err := flags.Parse(args); err != nil {
return err
}
if *sessionID == "" {
return errors.New("session-id is required")
}
client, _, err := dialForCommand(ctx, common)
if err != nil {
return err
}
defer client.Close()
session := mxgateway.NewSessionForID(client, *sessionID)
streamCtx, cancelStream := context.WithCancel(ctx)
defer cancelStream()
events, err := session.EventsAfter(streamCtx, *after)
if err != nil {
return err
}
count := 0
for result := range events {
if result.Err != nil {
return result.Err
}
if *jsonOutput {
fmt.Fprintln(stdout, string(mustMarshalProto(result.Event)))
} else {
fmt.Fprintf(stdout, "%d %s\n", result.Event.GetWorkerSequence(), result.Event.GetFamily())
}
count++
if *limit > 0 && count >= *limit {
cancelStream()
return nil
}
}
return nil
}
func runSmoke(ctx context.Context, args []string, stdout, stderr io.Writer) error {
flags := flag.NewFlagSet("smoke", flag.ContinueOnError)
flags.SetOutput(stderr)
common := bindCommonFlags(flags)
jsonOutput := flags.Bool("json", false, "write JSON output")
clientName := flags.String("client-name", "mxgw-go-smoke", "MXAccess client name")
item := flags.String("item", "", "item definition")
if err := flags.Parse(args); err != nil {
return err
}
if *item == "" {
return errors.New("item is required")
}
client, options, err := dialForCommand(ctx, common)
if err != nil {
return err
}
defer client.Close()
session, err := client.OpenSession(ctx, mxgateway.OpenSessionOptions{ClientSessionName: *clientName})
if err != nil {
return err
}
defer session.Close(context.Background())
serverHandle, err := session.Register(ctx, *clientName)
if err != nil {
return err
}
itemHandle, err := session.AddItem(ctx, serverHandle, *item)
if err != nil {
return err
}
if err := session.Advise(ctx, serverHandle, itemHandle); err != nil {
return err
}
output := map[string]any{
"command": "smoke",
"options": options,
"sessionId": session.ID(),
"serverHandle": serverHandle,
"itemHandle": itemHandle,
}
if *jsonOutput {
return writeJSON(stdout, output)
}
fmt.Fprintf(stdout, "session=%s server=%d item=%d\n", session.ID(), serverHandle, itemHandle)
return nil
}
func bindCommonFlags(flags *flag.FlagSet) *commonOptions {
common := &commonOptions{}
flags.StringVar(&common.Endpoint, "endpoint", "localhost:5000", "gateway endpoint")
flags.StringVar(&common.APIKey, "api-key", "", "gateway API key")
flags.StringVar(&common.APIKeyEnv, "api-key-env", "MXGATEWAY_API_KEY", "environment variable containing the API key")
flags.BoolVar(&common.Plaintext, "plaintext", false, "use plaintext transport")
flags.StringVar(&common.CACertFile, "ca-cert", "", "CA certificate file")
flags.StringVar(&common.ServerName, "server-name-override", "", "TLS server name override")
flags.StringVar(&common.CallTimeout, "call-timeout", "30s", "per-call timeout")
return common
}
func dialForCommand(ctx context.Context, common *commonOptions) (*mxgateway.Client, commonOptions, error) {
options, err := common.resolved()
if err != nil {
return nil, options, err
}
client, err := mxgateway.Dial(ctx, mxgateway.Options{
Endpoint: options.Endpoint,
APIKey: options.apiKeyValue,
Plaintext: options.Plaintext,
CACertFile: options.CACertFile,
ServerNameOverride: options.ServerName,
CallTimeout: options.timeout,
})
return client, options, err
}
func (o *commonOptions) resolved() (commonOptions, error) {
resolved := *o
if resolved.APIKey == "" && resolved.APIKeyEnv != "" {
resolved.apiKeyValue = os.Getenv(resolved.APIKeyEnv)
} else {
resolved.apiKeyValue = resolved.APIKey
}
resolved.APIKey = mxgateway.RedactAPIKey(resolved.apiKeyValue)
if resolved.CallTimeout != "" {
timeout, err := time.ParseDuration(resolved.CallTimeout)
if err != nil {
return resolved, err
}
resolved.timeout = timeout
}
return resolved, nil
}
func parseValue(valueType, valueText string) (*mxgateway.MxValue, error) {
switch valueType {
case "bool":
value, err := strconv.ParseBool(valueText)
if err != nil {
return nil, err
}
return mxgateway.BoolValue(value), nil
case "int32":
value, err := strconv.ParseInt(valueText, 10, 32)
if err != nil {
return nil, err
}
return mxgateway.Int32Value(int32(value)), nil
case "int64":
value, err := strconv.ParseInt(valueText, 10, 64)
if err != nil {
return nil, err
}
return mxgateway.Int64Value(value), nil
case "float":
value, err := strconv.ParseFloat(valueText, 32)
if err != nil {
return nil, err
}
return mxgateway.FloatValue(float32(value)), nil
case "double":
value, err := strconv.ParseFloat(valueText, 64)
if err != nil {
return nil, err
}
return mxgateway.DoubleValue(value), nil
case "string":
return mxgateway.StringValue(valueText), nil
default:
return nil, fmt.Errorf("unsupported value type %q", valueType)
}
}
func writeCommandOutput(stdout io.Writer, jsonOutput bool, command string, options commonOptions, reply *mxgateway.MxCommandReply, err error) error {
if err != nil {
return err
}
if jsonOutput {
return writeJSON(stdout, commandReplyOutput{
Command: command,
Options: options,
Reply: mustMarshalProto(reply),
})
}
fmt.Fprintln(stdout, reply.GetKind())
return nil
}
func writeJSON(writer io.Writer, value any) error {
encoder := json.NewEncoder(writer)
encoder.SetIndent("", " ")
return encoder.Encode(value)
}
func mustMarshalProto(message protojsonMessage) json.RawMessage {
data, err := protojson.MarshalOptions{UseProtoNames: false}.Marshal(message)
if err != nil {
panic(err)
}
return data
}
type protojsonMessage interface {
ProtoReflect() protoreflect.Message
}
func writeUsage(writer io.Writer) {
fmt.Fprintln(writer, "usage: mxgw-go <version|open-session|close-session|register|add-item|advise|write|stream-events|smoke>")
}
+58
View File
@@ -0,0 +1,58 @@
package main
import (
"bytes"
"encoding/json"
"strings"
"testing"
)
func TestRunVersionJSON(t *testing.T) {
var stdout bytes.Buffer
var stderr bytes.Buffer
if err := runWithIO(t.Context(), []string{"version", "-json"}, &stdout, &stderr); err != nil {
t.Fatalf("runWithIO() error = %v; stderr = %s", err, stderr.String())
}
var output versionOutput
if err := json.Unmarshal(stdout.Bytes(), &output); err != nil {
t.Fatalf("parse JSON: %v", err)
}
if output.GatewayProtocolVersion == 0 || output.WorkerProtocolVersion == 0 {
t.Fatalf("protocol versions were not populated: %+v", output)
}
}
func TestCommonOptionsRedactsAPIKey(t *testing.T) {
options, err := (&commonOptions{
Endpoint: "localhost:5000",
APIKey: "mxgw_super_secret",
Plaintext: true,
CallTimeout: "2s",
}).resolved()
if err != nil {
t.Fatalf("resolved() error = %v", err)
}
data, err := json.Marshal(options)
if err != nil {
t.Fatalf("marshal options: %v", err)
}
if strings.Contains(string(data), "super_secret") {
t.Fatalf("redacted JSON leaked API key: %s", data)
}
if !strings.Contains(string(data), "mxgw") {
t.Fatalf("redacted JSON did not preserve key shape: %s", data)
}
}
func TestParseValueBuildsTypedValue(t *testing.T) {
value, err := parseValue("int32", "123")
if err != nil {
t.Fatalf("parseValue() error = %v", err)
}
if got := value.GetInt32Value(); got != 123 {
t.Fatalf("int32 value = %d, want 123", got)
}
}
+30
View File
@@ -0,0 +1,30 @@
package mxgateway
import (
"context"
"google.golang.org/grpc"
"google.golang.org/grpc/metadata"
)
const authorizationHeader = "authorization"
func unaryAuthInterceptor(apiKey string) grpc.UnaryClientInterceptor {
return func(ctx context.Context, method string, req, reply any, cc *grpc.ClientConn, invoker grpc.UnaryInvoker, opts ...grpc.CallOption) error {
return invoker(authContext(ctx, apiKey), method, req, reply, cc, opts...)
}
}
func streamAuthInterceptor(apiKey string) grpc.StreamClientInterceptor {
return func(ctx context.Context, desc *grpc.StreamDesc, cc *grpc.ClientConn, method string, streamer grpc.Streamer, opts ...grpc.CallOption) (grpc.ClientStream, error) {
return streamer(authContext(ctx, apiKey), desc, cc, method, opts...)
}
}
func authContext(ctx context.Context, apiKey string) context.Context {
if apiKey == "" {
return ctx
}
return metadata.AppendToOutgoingContext(ctx, authorizationHeader, "Bearer "+apiKey)
}
+236
View File
@@ -0,0 +1,236 @@
package mxgateway
import (
"context"
"crypto/tls"
"errors"
"time"
pb "gitea.dohertylan.com/dohertj2/mxaccessgw/clients/go/internal/generated"
"google.golang.org/grpc"
"google.golang.org/grpc/credentials"
"google.golang.org/grpc/credentials/insecure"
"google.golang.org/protobuf/types/known/durationpb"
)
const (
defaultDialTimeout = 10 * time.Second
defaultCallTimeout = 30 * time.Second
)
// Client owns a gateway gRPC connection and exposes session-oriented helpers.
type Client struct {
conn *grpc.ClientConn
raw pb.MxAccessGatewayClient
opts Options
}
// Dial opens a gRPC connection to the gateway and configures auth metadata,
// transport security, and blocking dial cancellation from ctx.
func Dial(ctx context.Context, opts Options) (*Client, error) {
if opts.Endpoint == "" {
return nil, errors.New("mxgateway: endpoint is required")
}
dialCtx := ctx
cancel := func() {}
if opts.DialTimeout > 0 {
dialCtx, cancel = context.WithTimeout(ctx, opts.DialTimeout)
} else if _, ok := ctx.Deadline(); !ok {
dialCtx, cancel = context.WithTimeout(ctx, defaultDialTimeout)
}
defer cancel()
transportCredentials, err := resolveTransportCredentials(opts)
if err != nil {
return nil, err
}
dialOptions := []grpc.DialOption{
grpc.WithTransportCredentials(transportCredentials),
grpc.WithUnaryInterceptor(unaryAuthInterceptor(opts.APIKey)),
grpc.WithStreamInterceptor(streamAuthInterceptor(opts.APIKey)),
grpc.WithBlock(),
}
dialOptions = append(dialOptions, opts.DialOptions...)
conn, err := grpc.DialContext(dialCtx, opts.Endpoint, dialOptions...)
if err != nil {
return nil, &GatewayError{Op: "dial", Err: err}
}
return NewClient(conn, opts), nil
}
// NewClient wraps an existing gRPC connection. The caller owns closing conn
// unless it calls Close on the returned Client.
func NewClient(conn *grpc.ClientConn, opts Options) *Client {
return &Client{
conn: conn,
raw: pb.NewMxAccessGatewayClient(conn),
opts: opts,
}
}
// RawClient returns the generated gRPC client for command-specific parity tests.
func (c *Client) RawClient() RawGatewayClient {
return c.raw
}
// OpenSession creates a gateway-backed MXAccess session.
func (c *Client) OpenSession(ctx context.Context, opts OpenSessionOptions) (*Session, error) {
reply, err := c.OpenSessionRaw(ctx, opts.Request())
if err != nil {
return nil, err
}
return newSession(c, reply), nil
}
// OpenSessionRaw sends a raw OpenSession request and validates protocol status.
func (c *Client) OpenSessionRaw(ctx context.Context, req *OpenSessionRequest) (*OpenSessionReply, error) {
if req == nil {
return nil, errors.New("mxgateway: open session request is required")
}
callCtx, cancel := c.callContext(ctx)
defer cancel()
reply, err := c.raw.OpenSession(callCtx, req)
if err != nil {
return nil, &GatewayError{Op: "open session", Err: err}
}
if err := EnsureProtocolSuccess("open session", reply.GetProtocolStatus(), nil); err != nil {
return reply, err
}
return reply, nil
}
// Invoke sends a raw MXAccess command request and validates protocol and
// MXAccess status fields while preserving the raw reply on typed errors.
func (c *Client) Invoke(ctx context.Context, req *MxCommandRequest) (*MxCommandReply, error) {
if req == nil {
return nil, errors.New("mxgateway: command request is required")
}
callCtx, cancel := c.callContext(ctx)
defer cancel()
reply, err := c.raw.Invoke(callCtx, req)
if err != nil {
return nil, &GatewayError{Op: "invoke", Err: err}
}
if err := EnsureProtocolSuccess("invoke", reply.GetProtocolStatus(), reply); err != nil {
return reply, err
}
if err := EnsureMxAccessSuccess("invoke", reply); err != nil {
return reply, err
}
return reply, nil
}
// CloseSessionRaw sends a raw CloseSession request and validates protocol
// status.
func (c *Client) CloseSessionRaw(ctx context.Context, req *CloseSessionRequest) (*CloseSessionReply, error) {
if req == nil {
return nil, errors.New("mxgateway: close session request is required")
}
callCtx, cancel := c.callContext(ctx)
defer cancel()
reply, err := c.raw.CloseSession(callCtx, req)
if err != nil {
return nil, &GatewayError{Op: "close session", Err: err}
}
if err := EnsureProtocolSuccess("close session", reply.GetProtocolStatus(), nil); err != nil {
return reply, err
}
return reply, nil
}
// StreamEventsRaw starts the generated event stream for callers that need direct
// control over Recv.
func (c *Client) StreamEventsRaw(ctx context.Context, req *StreamEventsRequest) (RawEventStream, error) {
if req == nil {
return nil, errors.New("mxgateway: stream events request is required")
}
stream, err := c.raw.StreamEvents(ctx, req)
if err != nil {
return nil, &GatewayError{Op: "stream events", Err: err}
}
return stream, nil
}
// Close closes the underlying gRPC connection.
func (c *Client) Close() error {
if c == nil || c.conn == nil {
return nil
}
return c.conn.Close()
}
func (c *Client) callContext(ctx context.Context) (context.Context, context.CancelFunc) {
timeout := c.opts.CallTimeout
if timeout == 0 {
timeout = defaultCallTimeout
}
if timeout < 0 {
return ctx, func() {}
}
if _, ok := ctx.Deadline(); ok {
return ctx, func() {}
}
return context.WithTimeout(ctx, timeout)
}
func resolveTransportCredentials(opts Options) (credentials.TransportCredentials, error) {
if opts.TransportCredentials != nil {
return opts.TransportCredentials, nil
}
if opts.Plaintext {
return insecure.NewCredentials(), nil
}
if opts.CACertFile != "" {
return credentials.NewClientTLSFromFile(opts.CACertFile, opts.ServerNameOverride)
}
if opts.TLSConfig != nil {
cfg := opts.TLSConfig.Clone()
if opts.ServerNameOverride != "" {
cfg.ServerName = opts.ServerNameOverride
}
return credentials.NewTLS(cfg), nil
}
return credentials.NewTLS(&tls.Config{
MinVersion: tls.VersionTLS12,
ServerName: opts.ServerNameOverride,
}), nil
}
// OpenSessionOptions describes fields used to create an OpenSessionRequest.
type OpenSessionOptions struct {
RequestedBackend string
ClientSessionName string
ClientCorrelationID string
CommandTimeout time.Duration
}
// Request returns the raw protobuf OpenSessionRequest for these options.
func (o OpenSessionOptions) Request() *OpenSessionRequest {
req := &OpenSessionRequest{
RequestedBackend: o.RequestedBackend,
ClientSessionName: o.ClientSessionName,
ClientCorrelationId: o.ClientCorrelationID,
}
if o.CommandTimeout > 0 {
req.CommandTimeout = durationpb.New(o.CommandTimeout)
}
return req
}
+261
View File
@@ -0,0 +1,261 @@
package mxgateway
import (
"context"
"errors"
"io"
"net"
"testing"
"time"
pb "gitea.dohertylan.com/dohertj2/mxaccessgw/clients/go/internal/generated"
"google.golang.org/grpc"
"google.golang.org/grpc/metadata"
"google.golang.org/grpc/test/bufconn"
)
const bufSize = 1024 * 1024
func TestDialAttachesAuthMetadataToUnaryCalls(t *testing.T) {
fake := &fakeGatewayServer{
openReply: &pb.OpenSessionReply{
SessionId: "session-1",
GatewayProtocolVersion: GatewayProtocolVersion,
WorkerProtocolVersion: WorkerProtocolVersion,
ProtocolStatus: &pb.ProtocolStatus{
Code: pb.ProtocolStatusCode_PROTOCOL_STATUS_CODE_OK,
},
},
}
client, cleanup := newBufconnClient(t, fake)
defer cleanup()
_, err := client.OpenSession(context.Background(), OpenSessionOptions{ClientSessionName: "fixture"})
if err != nil {
t.Fatalf("OpenSession() error = %v", err)
}
if got := fake.openAuth; got != "Bearer test-api-key" {
t.Fatalf("authorization metadata = %q, want %q", got, "Bearer test-api-key")
}
}
func TestStreamEventsAttachesAuthMetadataAndClosesOnCancellation(t *testing.T) {
fake := &fakeGatewayServer{
streamStarted: make(chan struct{}),
}
client, cleanup := newBufconnClient(t, fake)
defer cleanup()
session := NewSessionForID(client, "session-1")
ctx, cancel := context.WithCancel(context.Background())
events, err := session.Events(ctx)
if err != nil {
t.Fatalf("Events() error = %v", err)
}
<-fake.streamStarted
first := <-events
if first.Err != nil {
t.Fatalf("first event error = %v", first.Err)
}
if first.Event.GetWorkerSequence() != 1 {
t.Fatalf("worker sequence = %d, want 1", first.Event.GetWorkerSequence())
}
if got := fake.streamAuth; got != "Bearer test-api-key" {
t.Fatalf("stream authorization metadata = %q, want %q", got, "Bearer test-api-key")
}
cancel()
select {
case _, ok := <-events:
if ok {
t.Fatal("events channel produced an extra item after cancellation")
}
case <-time.After(2 * time.Second):
t.Fatal("events channel did not close after cancellation")
}
}
func TestSessionHelpersBuildCommandsAndExposeRawReply(t *testing.T) {
fake := &fakeGatewayServer{
invokeReply: &pb.MxCommandReply{
SessionId: "session-1",
Kind: pb.MxCommandKind_MX_COMMAND_KIND_ADD_ITEM2,
ProtocolStatus: &pb.ProtocolStatus{
Code: pb.ProtocolStatusCode_PROTOCOL_STATUS_CODE_OK,
},
Payload: &pb.MxCommandReply_AddItem2{
AddItem2: &pb.AddItem2Reply{ItemHandle: 42},
},
},
}
client, cleanup := newBufconnClient(t, fake)
defer cleanup()
session := NewSessionForID(client, "session-1")
itemHandle, err := session.AddItem2(context.Background(), 12, "Area001.Pump001.Speed", "runtime")
if err != nil {
t.Fatalf("AddItem2() error = %v", err)
}
if itemHandle != 42 {
t.Fatalf("item handle = %d, want 42", itemHandle)
}
req := fake.invokeRequest
if req.GetSessionId() != "session-1" {
t.Fatalf("session id = %q, want session-1", req.GetSessionId())
}
if req.GetClientCorrelationId() == "" {
t.Fatal("client correlation id is empty")
}
if req.GetCommand().GetKind() != pb.MxCommandKind_MX_COMMAND_KIND_ADD_ITEM2 {
t.Fatalf("command kind = %s", req.GetCommand().GetKind())
}
if req.GetCommand().GetAddItem2().GetItemContext() != "runtime" {
t.Fatalf("item context = %q, want runtime", req.GetCommand().GetAddItem2().GetItemContext())
}
}
func TestInvokeReturnsTypedMxAccessErrorWithRawReply(t *testing.T) {
hresult := int32(-2147467259)
fake := &fakeGatewayServer{
invokeReply: &pb.MxCommandReply{
SessionId: "session-1",
Kind: pb.MxCommandKind_MX_COMMAND_KIND_ADVISE,
Hresult: &hresult,
DiagnosticMessage: "native failure",
ProtocolStatus: &pb.ProtocolStatus{Code: pb.ProtocolStatusCode_PROTOCOL_STATUS_CODE_MXACCESS_FAILURE, Message: "MXAccess failed"},
Statuses: []*pb.MxStatusProxy{{Success: 0, DiagnosticText: "failed"}},
},
}
client, cleanup := newBufconnClient(t, fake)
defer cleanup()
session := NewSessionForID(client, "session-1")
err := session.Advise(context.Background(), 12, 34)
var mxErr *MxAccessError
if !errors.As(err, &mxErr) {
t.Fatalf("error %T does not support errors.As(*MxAccessError)", err)
}
if mxErr.Reply.GetHresult() != hresult {
t.Fatalf("raw reply HRESULT = %d, want %d", mxErr.Reply.GetHresult(), hresult)
}
var commandErr *CommandError
if !errors.As(err, &commandErr) {
t.Fatalf("error %T does not support errors.As(*CommandError)", err)
}
if commandErr.Reply.GetDiagnosticMessage() != "native failure" {
t.Fatalf("raw diagnostic = %q", commandErr.Reply.GetDiagnosticMessage())
}
}
func newBufconnClient(t *testing.T, fake *fakeGatewayServer) (*Client, func()) {
t.Helper()
listener := bufconn.Listen(bufSize)
server := grpc.NewServer()
pb.RegisterMxAccessGatewayServer(server, fake)
go func() {
if err := server.Serve(listener); err != nil && !errors.Is(err, grpc.ErrServerStopped) {
t.Errorf("bufconn server failed: %v", err)
}
}()
dialer := func(ctx context.Context, _ string) (net.Conn, error) {
return listener.DialContext(ctx)
}
client, err := Dial(context.Background(), Options{
Endpoint: "bufnet",
APIKey: "test-api-key",
Plaintext: true,
DialOptions: []grpc.DialOption{
grpc.WithContextDialer(dialer),
},
})
if err != nil {
t.Fatalf("Dial() error = %v", err)
}
return client, func() {
client.Close()
server.Stop()
listener.Close()
}
}
type fakeGatewayServer struct {
pb.UnimplementedMxAccessGatewayServer
openReply *pb.OpenSessionReply
openAuth string
streamAuth string
streamStarted chan struct{}
invokeReply *pb.MxCommandReply
invokeRequest *pb.MxCommandRequest
}
func (s *fakeGatewayServer) OpenSession(ctx context.Context, req *pb.OpenSessionRequest) (*pb.OpenSessionReply, error) {
s.openAuth = authorizationFromContext(ctx)
if s.openReply != nil {
return s.openReply, nil
}
return &pb.OpenSessionReply{
SessionId: "session-1",
ProtocolStatus: &pb.ProtocolStatus{
Code: pb.ProtocolStatusCode_PROTOCOL_STATUS_CODE_OK,
},
}, nil
}
func (s *fakeGatewayServer) CloseSession(ctx context.Context, req *pb.CloseSessionRequest) (*pb.CloseSessionReply, error) {
return &pb.CloseSessionReply{
SessionId: req.GetSessionId(),
ProtocolStatus: &pb.ProtocolStatus{
Code: pb.ProtocolStatusCode_PROTOCOL_STATUS_CODE_OK,
},
}, nil
}
func (s *fakeGatewayServer) Invoke(ctx context.Context, req *pb.MxCommandRequest) (*pb.MxCommandReply, error) {
s.invokeRequest = req
if s.invokeReply != nil {
return s.invokeReply, nil
}
return &pb.MxCommandReply{
SessionId: req.GetSessionId(),
Kind: req.GetCommand().GetKind(),
ProtocolStatus: &pb.ProtocolStatus{
Code: pb.ProtocolStatusCode_PROTOCOL_STATUS_CODE_OK,
},
}, nil
}
func (s *fakeGatewayServer) StreamEvents(req *pb.StreamEventsRequest, stream grpc.ServerStreamingServer[pb.MxEvent]) error {
s.streamAuth = authorizationFromContext(stream.Context())
if s.streamStarted != nil {
close(s.streamStarted)
}
if err := stream.Send(&pb.MxEvent{
SessionId: req.GetSessionId(),
Family: pb.MxEventFamily_MX_EVENT_FAMILY_ON_DATA_CHANGE,
WorkerSequence: 1,
}); err != nil {
return err
}
<-stream.Context().Done()
return io.EOF
}
func authorizationFromContext(ctx context.Context) string {
md, ok := metadata.FromIncomingContext(ctx)
if !ok {
return ""
}
values := md.Get(authorizationHeader)
if len(values) == 0 {
return ""
}
return values[0]
}
+73
View File
@@ -0,0 +1,73 @@
package mxgateway
import (
"encoding/json"
"os"
"path/filepath"
"testing"
pb "gitea.dohertylan.com/dohertj2/mxaccessgw/clients/go/internal/generated"
"google.golang.org/protobuf/encoding/protojson"
)
func TestValueConversionFixtures(t *testing.T) {
data, err := os.ReadFile(filepath.Join("..", "..", "proto", "fixtures", "behavior", "values", "value-conversion-cases.json"))
if err != nil {
t.Fatalf("read fixture: %v", err)
}
var fixture struct {
Cases []struct {
ID string `json:"id"`
ExpectedKind string `json:"expectedKind"`
Value json.RawMessage `json:"value"`
} `json:"cases"`
}
if err := json.Unmarshal(data, &fixture); err != nil {
t.Fatalf("parse fixture manifest: %v", err)
}
for _, tc := range fixture.Cases {
t.Run(tc.ID, func(t *testing.T) {
var value pb.MxValue
if err := protojson.Unmarshal(tc.Value, &value); err != nil {
t.Fatalf("parse value: %v", err)
}
if _, err := NativeValue(&value); err != nil {
t.Fatalf("NativeValue() error = %v", err)
}
if got := value.ProtoReflect().WhichOneof(value.ProtoReflect().Descriptor().Oneofs().ByName("kind")).JSONName(); got != tc.ExpectedKind {
t.Fatalf("kind = %q, want %q", got, tc.ExpectedKind)
}
})
}
}
func TestStatusConversionFixtures(t *testing.T) {
data, err := os.ReadFile(filepath.Join("..", "..", "proto", "fixtures", "behavior", "statuses", "status-conversion-cases.json"))
if err != nil {
t.Fatalf("read fixture: %v", err)
}
var fixture struct {
Cases []struct {
ID string `json:"id"`
Status json.RawMessage `json:"status"`
} `json:"cases"`
}
if err := json.Unmarshal(data, &fixture); err != nil {
t.Fatalf("parse fixture manifest: %v", err)
}
for _, tc := range fixture.Cases {
t.Run(tc.ID, func(t *testing.T) {
var status pb.MxStatusProxy
if err := protojson.Unmarshal(tc.Status, &status); err != nil {
t.Fatalf("parse status: %v", err)
}
if got, want := StatusSucceeded(&status), status.GetSuccess() != 0; got != want {
t.Fatalf("StatusSucceeded() = %v, want %v", got, want)
}
})
}
}
+118
View File
@@ -0,0 +1,118 @@
package mxgateway
import (
"fmt"
pb "gitea.dohertylan.com/dohertj2/mxaccessgw/clients/go/internal/generated"
)
// GatewayError wraps transport-level gRPC failures.
type GatewayError struct {
Op string
Err error
}
func (e *GatewayError) Error() string {
if e == nil {
return ""
}
if e.Op == "" {
return fmt.Sprintf("mxgateway: %v", e.Err)
}
return fmt.Sprintf("mxgateway: %s failed: %v", e.Op, e.Err)
}
func (e *GatewayError) Unwrap() error {
if e == nil {
return nil
}
return e.Err
}
// CommandError reports a non-OK gateway protocol status and keeps the raw
// command reply when one exists.
type CommandError struct {
Op string
Status *ProtocolStatus
Reply *MxCommandReply
}
func (e *CommandError) Error() string {
if e == nil {
return ""
}
status := e.Status
if status == nil {
return fmt.Sprintf("mxgateway: %s failed with missing protocol status", e.Op)
}
if status.GetMessage() == "" {
return fmt.Sprintf("mxgateway: %s failed with protocol status %s", e.Op, status.GetCode())
}
return fmt.Sprintf("mxgateway: %s failed with protocol status %s: %s", e.Op, status.GetCode(), status.GetMessage())
}
// MxAccessError reports HRESULT or MXSTATUS_PROXY failures returned by MXAccess.
type MxAccessError struct {
Command *CommandError
Reply *MxCommandReply
}
func (e *MxAccessError) Error() string {
if e == nil {
return ""
}
if e.Command != nil && e.Command.Status != nil && e.Command.Status.GetMessage() != "" {
return e.Command.Error()
}
if e.Reply != nil && e.Reply.GetDiagnosticMessage() != "" {
return fmt.Sprintf("mxgateway: MXAccess command %s failed: %s", e.Reply.GetKind(), e.Reply.GetDiagnosticMessage())
}
if e.Reply != nil && e.Reply.Hresult != nil {
return fmt.Sprintf("mxgateway: MXAccess command %s failed with HRESULT 0x%08X", e.Reply.GetKind(), uint32(e.Reply.GetHresult()))
}
return "mxgateway: MXAccess command failed"
}
func (e *MxAccessError) Unwrap() error {
if e == nil {
return nil
}
return e.Command
}
// EnsureProtocolSuccess returns a typed CommandError when status is non-OK.
func EnsureProtocolSuccess(op string, status *ProtocolStatus, reply *MxCommandReply) error {
if status == nil || status.GetCode() == pb.ProtocolStatusCode_PROTOCOL_STATUS_CODE_OK {
return nil
}
commandError := &CommandError{
Op: op,
Status: status,
Reply: reply,
}
if status.GetCode() == pb.ProtocolStatusCode_PROTOCOL_STATUS_CODE_MXACCESS_FAILURE {
return &MxAccessError{
Command: commandError,
Reply: reply,
}
}
return commandError
}
// EnsureMxAccessSuccess returns a typed MxAccessError for failing HRESULTs or
// MXSTATUS_PROXY entries.
func EnsureMxAccessSuccess(op string, reply *MxCommandReply) error {
if reply == nil {
return nil
}
if reply.Hresult != nil && reply.GetHresult() != 0 {
return &MxAccessError{Reply: reply}
}
for _, status := range reply.GetStatuses() {
if !StatusSucceeded(status) {
return &MxAccessError{Reply: reply}
}
}
return nil
}
+19 -7
View File
@@ -1,14 +1,26 @@
package mxgateway
import "strings"
import (
"crypto/tls"
"strings"
"time"
// Options configures future gateway connections.
"google.golang.org/grpc"
"google.golang.org/grpc/credentials"
)
// Options configures gateway connections.
type Options struct {
Endpoint string
APIKey string
Plaintext bool
CACertFile string
ServerNameOverride string
Endpoint string
APIKey string
Plaintext bool
CACertFile string
ServerNameOverride string
DialTimeout time.Duration
CallTimeout time.Duration
TLSConfig *tls.Config
TransportCredentials credentials.TransportCredentials
DialOptions []grpc.DialOption
}
// RedactedAPIKey returns a display-safe representation of the configured API
+260
View File
@@ -0,0 +1,260 @@
package mxgateway
import (
"context"
"crypto/rand"
"encoding/hex"
"errors"
"io"
"sync"
pb "gitea.dohertylan.com/dohertj2/mxaccessgw/clients/go/internal/generated"
"google.golang.org/grpc/codes"
"google.golang.org/grpc/status"
)
// EventResult carries either the next ordered event or a terminal stream error.
type EventResult struct {
Event *MxEvent
Err error
}
// Session represents one gateway-backed MXAccess session.
type Session struct {
client *Client
openReply *OpenSessionReply
closeMu sync.Mutex
closeReply *CloseSessionReply
}
func newSession(client *Client, openReply *OpenSessionReply) *Session {
return &Session{
client: client,
openReply: openReply,
}
}
// NewSessionForID creates a session wrapper for commands against an existing
// gateway session id.
func NewSessionForID(client *Client, sessionID string) *Session {
return newSession(client, &pb.OpenSessionReply{SessionId: sessionID})
}
// ID returns the gateway session identifier.
func (s *Session) ID() string {
return s.openReply.GetSessionId()
}
// OpenReply returns the raw OpenSession reply.
func (s *Session) OpenReply() *OpenSessionReply {
return s.openReply
}
// Close closes the gateway session once and returns the raw close reply.
func (s *Session) Close(ctx context.Context) (*CloseSessionReply, error) {
s.closeMu.Lock()
defer s.closeMu.Unlock()
if s.closeReply != nil {
return s.closeReply, nil
}
reply, err := s.client.CloseSessionRaw(ctx, &pb.CloseSessionRequest{SessionId: s.ID()})
if err != nil {
return reply, err
}
s.closeReply = reply
return reply, nil
}
// Register invokes MXAccess Register and returns the server handle.
func (s *Session) Register(ctx context.Context, clientName string) (int32, error) {
reply, err := s.RegisterRaw(ctx, clientName)
if err != nil {
return 0, err
}
if reply.GetRegister() != nil {
return reply.GetRegister().GetServerHandle(), nil
}
return reply.GetReturnValue().GetInt32Value(), nil
}
// RegisterRaw invokes MXAccess Register and returns the raw reply.
func (s *Session) RegisterRaw(ctx context.Context, clientName string) (*MxCommandReply, error) {
if clientName == "" {
return nil, errors.New("mxgateway: client name is required")
}
return s.invokeCommand(ctx, &pb.MxCommand{
Kind: pb.MxCommandKind_MX_COMMAND_KIND_REGISTER,
Payload: &pb.MxCommand_Register{
Register: &pb.RegisterCommand{ClientName: clientName},
},
})
}
// Unregister invokes MXAccess Unregister.
func (s *Session) Unregister(ctx context.Context, serverHandle int32) error {
_, err := s.invokeCommand(ctx, &pb.MxCommand{
Kind: pb.MxCommandKind_MX_COMMAND_KIND_UNREGISTER,
Payload: &pb.MxCommand_Unregister{
Unregister: &pb.UnregisterCommand{ServerHandle: serverHandle},
},
})
return err
}
// AddItem invokes MXAccess AddItem and returns the item handle.
func (s *Session) AddItem(ctx context.Context, serverHandle int32, itemDefinition string) (int32, error) {
reply, err := s.AddItemRaw(ctx, serverHandle, itemDefinition)
if err != nil {
return 0, err
}
if reply.GetAddItem() != nil {
return reply.GetAddItem().GetItemHandle(), nil
}
return reply.GetReturnValue().GetInt32Value(), nil
}
// AddItemRaw invokes MXAccess AddItem and returns the raw reply.
func (s *Session) AddItemRaw(ctx context.Context, serverHandle int32, itemDefinition string) (*MxCommandReply, error) {
if itemDefinition == "" {
return nil, errors.New("mxgateway: item definition is required")
}
return s.invokeCommand(ctx, &pb.MxCommand{
Kind: pb.MxCommandKind_MX_COMMAND_KIND_ADD_ITEM,
Payload: &pb.MxCommand_AddItem{
AddItem: &pb.AddItemCommand{
ServerHandle: serverHandle,
ItemDefinition: itemDefinition,
},
},
})
}
// AddItem2 invokes MXAccess AddItem2 and returns the item handle.
func (s *Session) AddItem2(ctx context.Context, serverHandle int32, itemDefinition, itemContext string) (int32, error) {
reply, err := s.AddItem2Raw(ctx, serverHandle, itemDefinition, itemContext)
if err != nil {
return 0, err
}
if reply.GetAddItem2() != nil {
return reply.GetAddItem2().GetItemHandle(), nil
}
return reply.GetReturnValue().GetInt32Value(), nil
}
// AddItem2Raw invokes MXAccess AddItem2 and returns the raw reply.
func (s *Session) AddItem2Raw(ctx context.Context, serverHandle int32, itemDefinition, itemContext string) (*MxCommandReply, error) {
if itemDefinition == "" {
return nil, errors.New("mxgateway: item definition is required")
}
return s.invokeCommand(ctx, &pb.MxCommand{
Kind: pb.MxCommandKind_MX_COMMAND_KIND_ADD_ITEM2,
Payload: &pb.MxCommand_AddItem2{
AddItem2: &pb.AddItem2Command{
ServerHandle: serverHandle,
ItemDefinition: itemDefinition,
ItemContext: itemContext,
},
},
})
}
// Advise invokes MXAccess Advise.
func (s *Session) Advise(ctx context.Context, serverHandle, itemHandle int32) error {
_, err := s.AdviseRaw(ctx, serverHandle, itemHandle)
return err
}
// AdviseRaw invokes MXAccess Advise and returns the raw reply.
func (s *Session) AdviseRaw(ctx context.Context, serverHandle, itemHandle int32) (*MxCommandReply, error) {
return s.invokeCommand(ctx, &pb.MxCommand{
Kind: pb.MxCommandKind_MX_COMMAND_KIND_ADVISE,
Payload: &pb.MxCommand_Advise{
Advise: &pb.AdviseCommand{
ServerHandle: serverHandle,
ItemHandle: itemHandle,
},
},
})
}
// Write invokes MXAccess Write.
func (s *Session) Write(ctx context.Context, serverHandle, itemHandle int32, value *MxValue, userID int32) error {
_, err := s.WriteRaw(ctx, serverHandle, itemHandle, value, userID)
return err
}
// WriteRaw invokes MXAccess Write and returns the raw reply.
func (s *Session) WriteRaw(ctx context.Context, serverHandle, itemHandle int32, value *MxValue, userID int32) (*MxCommandReply, error) {
if value == nil {
return nil, errors.New("mxgateway: write value is required")
}
return s.invokeCommand(ctx, &pb.MxCommand{
Kind: pb.MxCommandKind_MX_COMMAND_KIND_WRITE,
Payload: &pb.MxCommand_Write{
Write: &pb.WriteCommand{
ServerHandle: serverHandle,
ItemHandle: itemHandle,
Value: value,
UserId: userID,
},
},
})
}
// Events streams ordered session events until the server ends the stream,
// context cancellation stops Recv, or a terminal error is sent.
func (s *Session) Events(ctx context.Context) (<-chan EventResult, error) {
return s.EventsAfter(ctx, 0)
}
// EventsAfter streams ordered session events after the given worker sequence.
func (s *Session) EventsAfter(ctx context.Context, afterWorkerSequence uint64) (<-chan EventResult, error) {
stream, err := s.client.StreamEventsRaw(ctx, &pb.StreamEventsRequest{
SessionId: s.ID(),
AfterWorkerSequence: afterWorkerSequence,
})
if err != nil {
return nil, err
}
results := make(chan EventResult, 16)
go func() {
defer close(results)
for {
event, err := stream.Recv()
if err == nil {
results <- EventResult{Event: event}
continue
}
if err == io.EOF || status.Code(err) == codes.Canceled || ctx.Err() != nil {
return
}
results <- EventResult{Err: &GatewayError{Op: "stream events", Err: err}}
return
}
}()
return results, nil
}
func (s *Session) invokeCommand(ctx context.Context, command *MxCommand) (*MxCommandReply, error) {
return s.client.Invoke(ctx, &pb.MxCommandRequest{
SessionId: s.ID(),
ClientCorrelationId: newCorrelationID(),
Command: command,
})
}
func newCorrelationID() string {
var buffer [16]byte
if _, err := rand.Read(buffer[:]); err != nil {
return ""
}
return hex.EncodeToString(buffer[:])
}
+6
View File
@@ -0,0 +1,6 @@
package mxgateway
// StatusSucceeded reports whether an MXSTATUS_PROXY entry represents success.
func StatusSucceeded(status *MxStatusProxy) bool {
return status == nil || status.GetSuccess() != 0
}
+70
View File
@@ -0,0 +1,70 @@
package mxgateway
import pb "gitea.dohertylan.com/dohertj2/mxaccessgw/clients/go/internal/generated"
// RawGatewayClient is the generated gRPC client interface exposed for callers
// that need direct contract access.
type RawGatewayClient = pb.MxAccessGatewayClient
// RawEventStream is the generated StreamEvents client stream.
type RawEventStream = pb.MxAccessGateway_StreamEventsClient
// Generated protobuf aliases keep raw contract access available from the public
// mxgateway package while generated code remains under internal/generated.
type (
OpenSessionRequest = pb.OpenSessionRequest
OpenSessionReply = pb.OpenSessionReply
CloseSessionRequest = pb.CloseSessionRequest
CloseSessionReply = pb.CloseSessionReply
StreamEventsRequest = pb.StreamEventsRequest
MxCommandRequest = pb.MxCommandRequest
MxCommandReply = pb.MxCommandReply
MxCommand = pb.MxCommand
MxEvent = pb.MxEvent
MxValue = pb.MxValue
Value = pb.MxValue
MxArray = pb.MxArray
MxStatusProxy = pb.MxStatusProxy
ProtocolStatus = pb.ProtocolStatus
RegisterCommand = pb.RegisterCommand
UnregisterCommand = pb.UnregisterCommand
AddItemCommand = pb.AddItemCommand
AddItem2Command = pb.AddItem2Command
AdviseCommand = pb.AdviseCommand
WriteCommand = pb.WriteCommand
Write2Command = pb.Write2Command
RegisterReply = pb.RegisterReply
AddItemReply = pb.AddItemReply
AddItem2Reply = pb.AddItem2Reply
)
type (
MxCommandKind = pb.MxCommandKind
MxDataType = pb.MxDataType
MxEventFamily = pb.MxEventFamily
MxStatusCategory = pb.MxStatusCategory
MxStatusSource = pb.MxStatusSource
ProtocolStatusCode = pb.ProtocolStatusCode
SessionState = pb.SessionState
)
const (
CommandKindRegister = pb.MxCommandKind_MX_COMMAND_KIND_REGISTER
CommandKindUnregister = pb.MxCommandKind_MX_COMMAND_KIND_UNREGISTER
CommandKindAddItem = pb.MxCommandKind_MX_COMMAND_KIND_ADD_ITEM
CommandKindAddItem2 = pb.MxCommandKind_MX_COMMAND_KIND_ADD_ITEM2
CommandKindAdvise = pb.MxCommandKind_MX_COMMAND_KIND_ADVISE
CommandKindWrite = pb.MxCommandKind_MX_COMMAND_KIND_WRITE
CommandKindWrite2 = pb.MxCommandKind_MX_COMMAND_KIND_WRITE2
DataTypeUnknown = pb.MxDataType_MX_DATA_TYPE_UNKNOWN
DataTypeBoolean = pb.MxDataType_MX_DATA_TYPE_BOOLEAN
DataTypeInteger = pb.MxDataType_MX_DATA_TYPE_INTEGER
DataTypeFloat = pb.MxDataType_MX_DATA_TYPE_FLOAT
DataTypeDouble = pb.MxDataType_MX_DATA_TYPE_DOUBLE
DataTypeString = pb.MxDataType_MX_DATA_TYPE_STRING
DataTypeTime = pb.MxDataType_MX_DATA_TYPE_TIME
ProtocolStatusOK = pb.ProtocolStatusCode_PROTOCOL_STATUS_CODE_OK
ProtocolStatusMxAccessFailure = pb.ProtocolStatusCode_PROTOCOL_STATUS_CODE_MXACCESS_FAILURE
)
+148
View File
@@ -0,0 +1,148 @@
package mxgateway
import (
"fmt"
"time"
pb "gitea.dohertylan.com/dohertj2/mxaccessgw/clients/go/internal/generated"
"google.golang.org/protobuf/types/known/timestamppb"
)
// BoolValue builds an MXAccess Boolean value.
func BoolValue(value bool) *MxValue {
return &pb.MxValue{
DataType: pb.MxDataType_MX_DATA_TYPE_BOOLEAN,
VariantType: "VT_BOOL",
Kind: &pb.MxValue_BoolValue{BoolValue: value},
}
}
// Int32Value builds an MXAccess Int32 value.
func Int32Value(value int32) *MxValue {
return &pb.MxValue{
DataType: pb.MxDataType_MX_DATA_TYPE_INTEGER,
VariantType: "VT_I4",
Kind: &pb.MxValue_Int32Value{Int32Value: value},
}
}
// Int64Value builds an MXAccess Int64 value.
func Int64Value(value int64) *MxValue {
return &pb.MxValue{
DataType: pb.MxDataType_MX_DATA_TYPE_INTEGER,
VariantType: "VT_I8",
Kind: &pb.MxValue_Int64Value{Int64Value: value},
}
}
// FloatValue builds an MXAccess Float value.
func FloatValue(value float32) *MxValue {
return &pb.MxValue{
DataType: pb.MxDataType_MX_DATA_TYPE_FLOAT,
VariantType: "VT_R4",
Kind: &pb.MxValue_FloatValue{FloatValue: value},
}
}
// DoubleValue builds an MXAccess Double value.
func DoubleValue(value float64) *MxValue {
return &pb.MxValue{
DataType: pb.MxDataType_MX_DATA_TYPE_DOUBLE,
VariantType: "VT_R8",
Kind: &pb.MxValue_DoubleValue{DoubleValue: value},
}
}
// StringValue builds an MXAccess String value.
func StringValue(value string) *MxValue {
return &pb.MxValue{
DataType: pb.MxDataType_MX_DATA_TYPE_STRING,
VariantType: "VT_BSTR",
Kind: &pb.MxValue_StringValue{StringValue: value},
}
}
// TimestampValue builds an MXAccess timestamp value from a Go time.
func TimestampValue(value time.Time) *MxValue {
return &pb.MxValue{
DataType: pb.MxDataType_MX_DATA_TYPE_TIME,
VariantType: "VT_DATE",
Kind: &pb.MxValue_TimestampValue{TimestampValue: timestamppb.New(value)},
}
}
// NativeValue converts a protobuf MxValue to the closest Go representation
// without discarding raw fallback data.
func NativeValue(value *MxValue) (any, error) {
if value == nil || value.GetIsNull() {
return nil, nil
}
switch kind := value.GetKind().(type) {
case *pb.MxValue_BoolValue:
return kind.BoolValue, nil
case *pb.MxValue_Int32Value:
return kind.Int32Value, nil
case *pb.MxValue_Int64Value:
return kind.Int64Value, nil
case *pb.MxValue_FloatValue:
return kind.FloatValue, nil
case *pb.MxValue_DoubleValue:
return kind.DoubleValue, nil
case *pb.MxValue_StringValue:
return kind.StringValue, nil
case *pb.MxValue_TimestampValue:
if kind.TimestampValue == nil {
return nil, nil
}
return kind.TimestampValue.AsTime(), nil
case *pb.MxValue_ArrayValue:
return NativeArray(kind.ArrayValue)
case *pb.MxValue_RawValue:
return append([]byte(nil), kind.RawValue...), nil
default:
return nil, fmt.Errorf("mxgateway: unsupported value kind %T", kind)
}
}
// NativeArray converts a protobuf MxArray to the closest Go slice
// representation.
func NativeArray(array *MxArray) (any, error) {
if array == nil {
return nil, nil
}
switch values := array.GetValues().(type) {
case *pb.MxArray_BoolValues:
return append([]bool(nil), values.BoolValues.GetValues()...), nil
case *pb.MxArray_Int32Values:
return append([]int32(nil), values.Int32Values.GetValues()...), nil
case *pb.MxArray_Int64Values:
return append([]int64(nil), values.Int64Values.GetValues()...), nil
case *pb.MxArray_FloatValues:
return append([]float32(nil), values.FloatValues.GetValues()...), nil
case *pb.MxArray_DoubleValues:
return append([]float64(nil), values.DoubleValues.GetValues()...), nil
case *pb.MxArray_StringValues:
return append([]string(nil), values.StringValues.GetValues()...), nil
case *pb.MxArray_TimestampValues:
result := make([]time.Time, 0, len(values.TimestampValues.GetValues()))
for _, value := range values.TimestampValues.GetValues() {
if value == nil {
result = append(result, time.Time{})
continue
}
result = append(result, value.AsTime())
}
return result, nil
case *pb.MxArray_RawValues:
rawValues := values.RawValues.GetValues()
result := make([][]byte, 0, len(rawValues))
for _, value := range rawValues {
result = append(result, append([]byte(nil), value...))
}
return result, nil
default:
return nil, fmt.Errorf("mxgateway: unsupported array value kind %T", values)
}
}
+41
View File
@@ -0,0 +1,41 @@
# Java Client
The Java client workspace contains the Gradle scaffold for the MXAccess Gateway
client library, generated protobuf/gRPC bindings, a test CLI project, and JUnit
tests.
## Layout
```text
clients/java/
settings.gradle
build.gradle
src/main/generated/
mxgateway-client/
mxgateway-cli/
```
`mxgateway-client` generates Java protobuf and gRPC sources from
`../../src/MxGateway.Contracts/Protos`. The Gradle protobuf plugin writes those
generated sources under `src/main/generated`, which matches the client proto
manifest in `../proto/proto-inputs.json`. Do not edit generated files by hand.
`mxgateway-cli` depends on `mxgateway-client` and provides the `mxgw-java`
application entry point used by later CLI implementation work.
## Build And Test
Run the Java checks from `clients/java`:
```powershell
gradle test
```
The build uses the Java 21 Gradle toolchain, compiles generated protobuf/gRPC
code, and runs JUnit 5 tests for the scaffold and CLI entry point.
## Related Documentation
- [Client Proto Generation](../../docs/client-proto-generation.md)
- [Java Client Detailed Design](../../docs/clients-java-design.md)
- [Java Style Guide](../../docs/style-guides/JavaStyleGuide.md)
+38
View File
@@ -0,0 +1,38 @@
plugins {
id 'base'
}
ext {
grpcVersion = '1.76.0'
junitVersion = '5.14.1'
picocliVersion = '4.7.7'
protobufVersion = '4.33.1'
}
subprojects {
group = 'com.dohertylan.mxgateway'
version = '0.1.0'
pluginManager.withPlugin('java') {
java {
toolchain {
languageVersion = JavaLanguageVersion.of(21)
}
}
tasks.withType(JavaCompile).configureEach {
options.encoding = 'UTF-8'
options.release = 21
}
tasks.withType(Test).configureEach {
useJUnitPlatform()
}
dependencies {
testImplementation platform("org.junit:junit-bom:${junitVersion}")
testImplementation 'org.junit.jupiter:junit-jupiter'
testRuntimeOnly 'org.junit.platform:junit-platform-launcher'
}
}
}
+12
View File
@@ -0,0 +1,12 @@
plugins {
id 'application'
}
dependencies {
implementation project(':mxgateway-client')
implementation "info.picocli:picocli:${picocliVersion}"
}
application {
mainClass = 'com.dohertylan.mxgateway.cli.MxGatewayCli'
}
@@ -0,0 +1,53 @@
package com.dohertylan.mxgateway.cli;
import com.dohertylan.mxgateway.client.MxGatewayClientVersion;
import java.io.PrintWriter;
import java.util.concurrent.Callable;
import picocli.CommandLine;
import picocli.CommandLine.Command;
import picocli.CommandLine.Model.CommandSpec;
import picocli.CommandLine.Spec;
@Command(
name = "mxgw-java",
mixinStandardHelpOptions = true,
description = "MXAccess Gateway Java test CLI.",
subcommands = MxGatewayCli.VersionCommand.class)
public final class MxGatewayCli implements Callable<Integer> {
@Spec
private CommandSpec spec;
public static void main(String[] args) {
int exitCode = new CommandLine(new MxGatewayCli()).execute(args);
System.exit(exitCode);
}
public static int execute(PrintWriter out, PrintWriter err, String... args) {
CommandLine commandLine = new CommandLine(new MxGatewayCli());
commandLine.setOut(out);
commandLine.setErr(err);
return commandLine.execute(args);
}
@Override
public Integer call() {
spec.commandLine().usage(spec.commandLine().getOut());
return 0;
}
@Command(name = "version", description = "Prints the Java client scaffold version.")
public static final class VersionCommand implements Callable<Integer> {
@Spec
private CommandSpec spec;
@Override
public Integer call() {
spec.commandLine().getOut().printf(
"mxgateway-java %s gatewayProtocolVersion=%d workerProtocolVersion=%d%n",
MxGatewayClientVersion.clientVersion(),
MxGatewayClientVersion.gatewayProtocolVersion(),
MxGatewayClientVersion.workerProtocolVersion());
return 0;
}
}
}
@@ -0,0 +1,27 @@
package com.dohertylan.mxgateway.cli;
import static org.junit.jupiter.api.Assertions.assertEquals;
import static org.junit.jupiter.api.Assertions.assertTrue;
import java.io.PrintWriter;
import java.io.StringWriter;
import org.junit.jupiter.api.Test;
final class MxGatewayCliTests {
@Test
void versionCommandPrintsProtocolVersions() {
StringWriter output = new StringWriter();
StringWriter errors = new StringWriter();
int exitCode = MxGatewayCli.execute(
new PrintWriter(output, true),
new PrintWriter(errors, true),
"version");
assertEquals(0, exitCode);
assertEquals("", errors.toString());
assertTrue(output.toString().contains("mxgateway-java 0.1.0"));
assertTrue(output.toString().contains("gatewayProtocolVersion=1"));
assertTrue(output.toString().contains("workerProtocolVersion=1"));
}
}
@@ -0,0 +1,46 @@
plugins {
id 'java-library'
id 'com.google.protobuf'
}
dependencies {
api "com.google.protobuf:protobuf-java:${protobufVersion}"
api "io.grpc:grpc-protobuf:${grpcVersion}"
api "io.grpc:grpc-stub:${grpcVersion}"
implementation "io.grpc:grpc-netty-shaded:${grpcVersion}"
compileOnly 'javax.annotation:javax.annotation-api:1.3.2'
}
sourceSets {
main {
proto {
srcDir rootProject.file('../../src/MxGateway.Contracts/Protos')
include 'mxaccess_gateway.proto'
include 'mxaccess_worker.proto'
}
}
}
protobuf {
protoc {
artifact = "com.google.protobuf:protoc:${protobufVersion}"
}
plugins {
grpc {
artifact = "io.grpc:protoc-gen-grpc-java:${grpcVersion}"
}
}
generatedFilesBaseDir = rootProject.file('src/main/generated').absolutePath
generateProtoTasks {
all().configureEach {
plugins {
grpc {}
}
}
}
}
@@ -0,0 +1,22 @@
package com.dohertylan.mxgateway.client;
public final class MxGatewayClientVersion {
private static final int GATEWAY_PROTOCOL_VERSION = 1;
private static final int WORKER_PROTOCOL_VERSION = 1;
private static final String CLIENT_VERSION = "0.1.0";
private MxGatewayClientVersion() {
}
public static String clientVersion() {
return CLIENT_VERSION;
}
public static int gatewayProtocolVersion() {
return GATEWAY_PROTOCOL_VERSION;
}
public static int workerProtocolVersion() {
return WORKER_PROTOCOL_VERSION;
}
}
@@ -0,0 +1,29 @@
package com.dohertylan.mxgateway.client;
import static org.junit.jupiter.api.Assertions.assertEquals;
import mxaccess_gateway.v1.MxAccessGatewayGrpc;
import mxaccess_gateway.v1.MxaccessGateway.OpenSessionRequest;
import mxaccess_worker.v1.MxaccessWorker.WorkerEnvelope;
import org.junit.jupiter.api.Test;
final class GeneratedContractSmokeTests {
@Test
void generatedGatewayAndWorkerContractsCompile() {
OpenSessionRequest request = OpenSessionRequest.newBuilder()
.setClientSessionName("junit")
.build();
WorkerEnvelope envelope = WorkerEnvelope.newBuilder()
.setProtocolVersion(MxGatewayClientVersion.workerProtocolVersion())
.build();
assertEquals("junit", request.getClientSessionName());
assertEquals("mxaccess_gateway.v1.MxAccessGateway", MxAccessGatewayGrpc.SERVICE_NAME);
assertEquals(1, envelope.getProtocolVersion());
}
@Test
void javaTwentyOneToolchainRunsTests() {
assertEquals(21, Runtime.version().feature());
}
}
+22
View File
@@ -0,0 +1,22 @@
pluginManagement {
repositories {
gradlePluginPortal()
mavenCentral()
}
plugins {
id 'com.google.protobuf' version '0.9.5'
}
}
dependencyResolutionManagement {
repositoriesMode.set(RepositoriesMode.FAIL_ON_PROJECT_REPOS)
repositories {
mavenCentral()
}
}
rootProject.name = 'mxaccessgw-java'
include 'mxgateway-client'
include 'mxgateway-cli'
@@ -0,0 +1,588 @@
package mxaccess_gateway.v1;
import static io.grpc.MethodDescriptor.generateFullMethodName;
/**
* <pre>
* Public client API for MXAccess sessions hosted by the gateway.
* </pre>
*/
@io.grpc.stub.annotations.GrpcGenerated
public final class MxAccessGatewayGrpc {
private MxAccessGatewayGrpc() {}
public static final java.lang.String SERVICE_NAME = "mxaccess_gateway.v1.MxAccessGateway";
// Static method descriptors that strictly reflect the proto.
private static volatile io.grpc.MethodDescriptor<mxaccess_gateway.v1.MxaccessGateway.OpenSessionRequest,
mxaccess_gateway.v1.MxaccessGateway.OpenSessionReply> getOpenSessionMethod;
@io.grpc.stub.annotations.RpcMethod(
fullMethodName = SERVICE_NAME + '/' + "OpenSession",
requestType = mxaccess_gateway.v1.MxaccessGateway.OpenSessionRequest.class,
responseType = mxaccess_gateway.v1.MxaccessGateway.OpenSessionReply.class,
methodType = io.grpc.MethodDescriptor.MethodType.UNARY)
public static io.grpc.MethodDescriptor<mxaccess_gateway.v1.MxaccessGateway.OpenSessionRequest,
mxaccess_gateway.v1.MxaccessGateway.OpenSessionReply> getOpenSessionMethod() {
io.grpc.MethodDescriptor<mxaccess_gateway.v1.MxaccessGateway.OpenSessionRequest, mxaccess_gateway.v1.MxaccessGateway.OpenSessionReply> getOpenSessionMethod;
if ((getOpenSessionMethod = MxAccessGatewayGrpc.getOpenSessionMethod) == null) {
synchronized (MxAccessGatewayGrpc.class) {
if ((getOpenSessionMethod = MxAccessGatewayGrpc.getOpenSessionMethod) == null) {
MxAccessGatewayGrpc.getOpenSessionMethod = getOpenSessionMethod =
io.grpc.MethodDescriptor.<mxaccess_gateway.v1.MxaccessGateway.OpenSessionRequest, mxaccess_gateway.v1.MxaccessGateway.OpenSessionReply>newBuilder()
.setType(io.grpc.MethodDescriptor.MethodType.UNARY)
.setFullMethodName(generateFullMethodName(SERVICE_NAME, "OpenSession"))
.setSampledToLocalTracing(true)
.setRequestMarshaller(io.grpc.protobuf.ProtoUtils.marshaller(
mxaccess_gateway.v1.MxaccessGateway.OpenSessionRequest.getDefaultInstance()))
.setResponseMarshaller(io.grpc.protobuf.ProtoUtils.marshaller(
mxaccess_gateway.v1.MxaccessGateway.OpenSessionReply.getDefaultInstance()))
.setSchemaDescriptor(new MxAccessGatewayMethodDescriptorSupplier("OpenSession"))
.build();
}
}
}
return getOpenSessionMethod;
}
private static volatile io.grpc.MethodDescriptor<mxaccess_gateway.v1.MxaccessGateway.CloseSessionRequest,
mxaccess_gateway.v1.MxaccessGateway.CloseSessionReply> getCloseSessionMethod;
@io.grpc.stub.annotations.RpcMethod(
fullMethodName = SERVICE_NAME + '/' + "CloseSession",
requestType = mxaccess_gateway.v1.MxaccessGateway.CloseSessionRequest.class,
responseType = mxaccess_gateway.v1.MxaccessGateway.CloseSessionReply.class,
methodType = io.grpc.MethodDescriptor.MethodType.UNARY)
public static io.grpc.MethodDescriptor<mxaccess_gateway.v1.MxaccessGateway.CloseSessionRequest,
mxaccess_gateway.v1.MxaccessGateway.CloseSessionReply> getCloseSessionMethod() {
io.grpc.MethodDescriptor<mxaccess_gateway.v1.MxaccessGateway.CloseSessionRequest, mxaccess_gateway.v1.MxaccessGateway.CloseSessionReply> getCloseSessionMethod;
if ((getCloseSessionMethod = MxAccessGatewayGrpc.getCloseSessionMethod) == null) {
synchronized (MxAccessGatewayGrpc.class) {
if ((getCloseSessionMethod = MxAccessGatewayGrpc.getCloseSessionMethod) == null) {
MxAccessGatewayGrpc.getCloseSessionMethod = getCloseSessionMethod =
io.grpc.MethodDescriptor.<mxaccess_gateway.v1.MxaccessGateway.CloseSessionRequest, mxaccess_gateway.v1.MxaccessGateway.CloseSessionReply>newBuilder()
.setType(io.grpc.MethodDescriptor.MethodType.UNARY)
.setFullMethodName(generateFullMethodName(SERVICE_NAME, "CloseSession"))
.setSampledToLocalTracing(true)
.setRequestMarshaller(io.grpc.protobuf.ProtoUtils.marshaller(
mxaccess_gateway.v1.MxaccessGateway.CloseSessionRequest.getDefaultInstance()))
.setResponseMarshaller(io.grpc.protobuf.ProtoUtils.marshaller(
mxaccess_gateway.v1.MxaccessGateway.CloseSessionReply.getDefaultInstance()))
.setSchemaDescriptor(new MxAccessGatewayMethodDescriptorSupplier("CloseSession"))
.build();
}
}
}
return getCloseSessionMethod;
}
private static volatile io.grpc.MethodDescriptor<mxaccess_gateway.v1.MxaccessGateway.MxCommandRequest,
mxaccess_gateway.v1.MxaccessGateway.MxCommandReply> getInvokeMethod;
@io.grpc.stub.annotations.RpcMethod(
fullMethodName = SERVICE_NAME + '/' + "Invoke",
requestType = mxaccess_gateway.v1.MxaccessGateway.MxCommandRequest.class,
responseType = mxaccess_gateway.v1.MxaccessGateway.MxCommandReply.class,
methodType = io.grpc.MethodDescriptor.MethodType.UNARY)
public static io.grpc.MethodDescriptor<mxaccess_gateway.v1.MxaccessGateway.MxCommandRequest,
mxaccess_gateway.v1.MxaccessGateway.MxCommandReply> getInvokeMethod() {
io.grpc.MethodDescriptor<mxaccess_gateway.v1.MxaccessGateway.MxCommandRequest, mxaccess_gateway.v1.MxaccessGateway.MxCommandReply> getInvokeMethod;
if ((getInvokeMethod = MxAccessGatewayGrpc.getInvokeMethod) == null) {
synchronized (MxAccessGatewayGrpc.class) {
if ((getInvokeMethod = MxAccessGatewayGrpc.getInvokeMethod) == null) {
MxAccessGatewayGrpc.getInvokeMethod = getInvokeMethod =
io.grpc.MethodDescriptor.<mxaccess_gateway.v1.MxaccessGateway.MxCommandRequest, mxaccess_gateway.v1.MxaccessGateway.MxCommandReply>newBuilder()
.setType(io.grpc.MethodDescriptor.MethodType.UNARY)
.setFullMethodName(generateFullMethodName(SERVICE_NAME, "Invoke"))
.setSampledToLocalTracing(true)
.setRequestMarshaller(io.grpc.protobuf.ProtoUtils.marshaller(
mxaccess_gateway.v1.MxaccessGateway.MxCommandRequest.getDefaultInstance()))
.setResponseMarshaller(io.grpc.protobuf.ProtoUtils.marshaller(
mxaccess_gateway.v1.MxaccessGateway.MxCommandReply.getDefaultInstance()))
.setSchemaDescriptor(new MxAccessGatewayMethodDescriptorSupplier("Invoke"))
.build();
}
}
}
return getInvokeMethod;
}
private static volatile io.grpc.MethodDescriptor<mxaccess_gateway.v1.MxaccessGateway.StreamEventsRequest,
mxaccess_gateway.v1.MxaccessGateway.MxEvent> getStreamEventsMethod;
@io.grpc.stub.annotations.RpcMethod(
fullMethodName = SERVICE_NAME + '/' + "StreamEvents",
requestType = mxaccess_gateway.v1.MxaccessGateway.StreamEventsRequest.class,
responseType = mxaccess_gateway.v1.MxaccessGateway.MxEvent.class,
methodType = io.grpc.MethodDescriptor.MethodType.SERVER_STREAMING)
public static io.grpc.MethodDescriptor<mxaccess_gateway.v1.MxaccessGateway.StreamEventsRequest,
mxaccess_gateway.v1.MxaccessGateway.MxEvent> getStreamEventsMethod() {
io.grpc.MethodDescriptor<mxaccess_gateway.v1.MxaccessGateway.StreamEventsRequest, mxaccess_gateway.v1.MxaccessGateway.MxEvent> getStreamEventsMethod;
if ((getStreamEventsMethod = MxAccessGatewayGrpc.getStreamEventsMethod) == null) {
synchronized (MxAccessGatewayGrpc.class) {
if ((getStreamEventsMethod = MxAccessGatewayGrpc.getStreamEventsMethod) == null) {
MxAccessGatewayGrpc.getStreamEventsMethod = getStreamEventsMethod =
io.grpc.MethodDescriptor.<mxaccess_gateway.v1.MxaccessGateway.StreamEventsRequest, mxaccess_gateway.v1.MxaccessGateway.MxEvent>newBuilder()
.setType(io.grpc.MethodDescriptor.MethodType.SERVER_STREAMING)
.setFullMethodName(generateFullMethodName(SERVICE_NAME, "StreamEvents"))
.setSampledToLocalTracing(true)
.setRequestMarshaller(io.grpc.protobuf.ProtoUtils.marshaller(
mxaccess_gateway.v1.MxaccessGateway.StreamEventsRequest.getDefaultInstance()))
.setResponseMarshaller(io.grpc.protobuf.ProtoUtils.marshaller(
mxaccess_gateway.v1.MxaccessGateway.MxEvent.getDefaultInstance()))
.setSchemaDescriptor(new MxAccessGatewayMethodDescriptorSupplier("StreamEvents"))
.build();
}
}
}
return getStreamEventsMethod;
}
/**
* Creates a new async stub that supports all call types for the service
*/
public static MxAccessGatewayStub newStub(io.grpc.Channel channel) {
io.grpc.stub.AbstractStub.StubFactory<MxAccessGatewayStub> factory =
new io.grpc.stub.AbstractStub.StubFactory<MxAccessGatewayStub>() {
@java.lang.Override
public MxAccessGatewayStub newStub(io.grpc.Channel channel, io.grpc.CallOptions callOptions) {
return new MxAccessGatewayStub(channel, callOptions);
}
};
return MxAccessGatewayStub.newStub(factory, channel);
}
/**
* Creates a new blocking-style stub that supports all types of calls on the service
*/
public static MxAccessGatewayBlockingV2Stub newBlockingV2Stub(
io.grpc.Channel channel) {
io.grpc.stub.AbstractStub.StubFactory<MxAccessGatewayBlockingV2Stub> factory =
new io.grpc.stub.AbstractStub.StubFactory<MxAccessGatewayBlockingV2Stub>() {
@java.lang.Override
public MxAccessGatewayBlockingV2Stub newStub(io.grpc.Channel channel, io.grpc.CallOptions callOptions) {
return new MxAccessGatewayBlockingV2Stub(channel, callOptions);
}
};
return MxAccessGatewayBlockingV2Stub.newStub(factory, channel);
}
/**
* Creates a new blocking-style stub that supports unary and streaming output calls on the service
*/
public static MxAccessGatewayBlockingStub newBlockingStub(
io.grpc.Channel channel) {
io.grpc.stub.AbstractStub.StubFactory<MxAccessGatewayBlockingStub> factory =
new io.grpc.stub.AbstractStub.StubFactory<MxAccessGatewayBlockingStub>() {
@java.lang.Override
public MxAccessGatewayBlockingStub newStub(io.grpc.Channel channel, io.grpc.CallOptions callOptions) {
return new MxAccessGatewayBlockingStub(channel, callOptions);
}
};
return MxAccessGatewayBlockingStub.newStub(factory, channel);
}
/**
* Creates a new ListenableFuture-style stub that supports unary calls on the service
*/
public static MxAccessGatewayFutureStub newFutureStub(
io.grpc.Channel channel) {
io.grpc.stub.AbstractStub.StubFactory<MxAccessGatewayFutureStub> factory =
new io.grpc.stub.AbstractStub.StubFactory<MxAccessGatewayFutureStub>() {
@java.lang.Override
public MxAccessGatewayFutureStub newStub(io.grpc.Channel channel, io.grpc.CallOptions callOptions) {
return new MxAccessGatewayFutureStub(channel, callOptions);
}
};
return MxAccessGatewayFutureStub.newStub(factory, channel);
}
/**
* <pre>
* Public client API for MXAccess sessions hosted by the gateway.
* </pre>
*/
public interface AsyncService {
/**
*/
default void openSession(mxaccess_gateway.v1.MxaccessGateway.OpenSessionRequest request,
io.grpc.stub.StreamObserver<mxaccess_gateway.v1.MxaccessGateway.OpenSessionReply> responseObserver) {
io.grpc.stub.ServerCalls.asyncUnimplementedUnaryCall(getOpenSessionMethod(), responseObserver);
}
/**
*/
default void closeSession(mxaccess_gateway.v1.MxaccessGateway.CloseSessionRequest request,
io.grpc.stub.StreamObserver<mxaccess_gateway.v1.MxaccessGateway.CloseSessionReply> responseObserver) {
io.grpc.stub.ServerCalls.asyncUnimplementedUnaryCall(getCloseSessionMethod(), responseObserver);
}
/**
*/
default void invoke(mxaccess_gateway.v1.MxaccessGateway.MxCommandRequest request,
io.grpc.stub.StreamObserver<mxaccess_gateway.v1.MxaccessGateway.MxCommandReply> responseObserver) {
io.grpc.stub.ServerCalls.asyncUnimplementedUnaryCall(getInvokeMethod(), responseObserver);
}
/**
*/
default void streamEvents(mxaccess_gateway.v1.MxaccessGateway.StreamEventsRequest request,
io.grpc.stub.StreamObserver<mxaccess_gateway.v1.MxaccessGateway.MxEvent> responseObserver) {
io.grpc.stub.ServerCalls.asyncUnimplementedUnaryCall(getStreamEventsMethod(), responseObserver);
}
}
/**
* Base class for the server implementation of the service MxAccessGateway.
* <pre>
* Public client API for MXAccess sessions hosted by the gateway.
* </pre>
*/
public static abstract class MxAccessGatewayImplBase
implements io.grpc.BindableService, AsyncService {
@java.lang.Override public final io.grpc.ServerServiceDefinition bindService() {
return MxAccessGatewayGrpc.bindService(this);
}
}
/**
* A stub to allow clients to do asynchronous rpc calls to service MxAccessGateway.
* <pre>
* Public client API for MXAccess sessions hosted by the gateway.
* </pre>
*/
public static final class MxAccessGatewayStub
extends io.grpc.stub.AbstractAsyncStub<MxAccessGatewayStub> {
private MxAccessGatewayStub(
io.grpc.Channel channel, io.grpc.CallOptions callOptions) {
super(channel, callOptions);
}
@java.lang.Override
protected MxAccessGatewayStub build(
io.grpc.Channel channel, io.grpc.CallOptions callOptions) {
return new MxAccessGatewayStub(channel, callOptions);
}
/**
*/
public void openSession(mxaccess_gateway.v1.MxaccessGateway.OpenSessionRequest request,
io.grpc.stub.StreamObserver<mxaccess_gateway.v1.MxaccessGateway.OpenSessionReply> responseObserver) {
io.grpc.stub.ClientCalls.asyncUnaryCall(
getChannel().newCall(getOpenSessionMethod(), getCallOptions()), request, responseObserver);
}
/**
*/
public void closeSession(mxaccess_gateway.v1.MxaccessGateway.CloseSessionRequest request,
io.grpc.stub.StreamObserver<mxaccess_gateway.v1.MxaccessGateway.CloseSessionReply> responseObserver) {
io.grpc.stub.ClientCalls.asyncUnaryCall(
getChannel().newCall(getCloseSessionMethod(), getCallOptions()), request, responseObserver);
}
/**
*/
public void invoke(mxaccess_gateway.v1.MxaccessGateway.MxCommandRequest request,
io.grpc.stub.StreamObserver<mxaccess_gateway.v1.MxaccessGateway.MxCommandReply> responseObserver) {
io.grpc.stub.ClientCalls.asyncUnaryCall(
getChannel().newCall(getInvokeMethod(), getCallOptions()), request, responseObserver);
}
/**
*/
public void streamEvents(mxaccess_gateway.v1.MxaccessGateway.StreamEventsRequest request,
io.grpc.stub.StreamObserver<mxaccess_gateway.v1.MxaccessGateway.MxEvent> responseObserver) {
io.grpc.stub.ClientCalls.asyncServerStreamingCall(
getChannel().newCall(getStreamEventsMethod(), getCallOptions()), request, responseObserver);
}
}
/**
* A stub to allow clients to do synchronous rpc calls to service MxAccessGateway.
* <pre>
* Public client API for MXAccess sessions hosted by the gateway.
* </pre>
*/
public static final class MxAccessGatewayBlockingV2Stub
extends io.grpc.stub.AbstractBlockingStub<MxAccessGatewayBlockingV2Stub> {
private MxAccessGatewayBlockingV2Stub(
io.grpc.Channel channel, io.grpc.CallOptions callOptions) {
super(channel, callOptions);
}
@java.lang.Override
protected MxAccessGatewayBlockingV2Stub build(
io.grpc.Channel channel, io.grpc.CallOptions callOptions) {
return new MxAccessGatewayBlockingV2Stub(channel, callOptions);
}
/**
*/
public mxaccess_gateway.v1.MxaccessGateway.OpenSessionReply openSession(mxaccess_gateway.v1.MxaccessGateway.OpenSessionRequest request) throws io.grpc.StatusException {
return io.grpc.stub.ClientCalls.blockingV2UnaryCall(
getChannel(), getOpenSessionMethod(), getCallOptions(), request);
}
/**
*/
public mxaccess_gateway.v1.MxaccessGateway.CloseSessionReply closeSession(mxaccess_gateway.v1.MxaccessGateway.CloseSessionRequest request) throws io.grpc.StatusException {
return io.grpc.stub.ClientCalls.blockingV2UnaryCall(
getChannel(), getCloseSessionMethod(), getCallOptions(), request);
}
/**
*/
public mxaccess_gateway.v1.MxaccessGateway.MxCommandReply invoke(mxaccess_gateway.v1.MxaccessGateway.MxCommandRequest request) throws io.grpc.StatusException {
return io.grpc.stub.ClientCalls.blockingV2UnaryCall(
getChannel(), getInvokeMethod(), getCallOptions(), request);
}
/**
*/
@io.grpc.ExperimentalApi("https://github.com/grpc/grpc-java/issues/10918")
public io.grpc.stub.BlockingClientCall<?, mxaccess_gateway.v1.MxaccessGateway.MxEvent>
streamEvents(mxaccess_gateway.v1.MxaccessGateway.StreamEventsRequest request) {
return io.grpc.stub.ClientCalls.blockingV2ServerStreamingCall(
getChannel(), getStreamEventsMethod(), getCallOptions(), request);
}
}
/**
* A stub to allow clients to do limited synchronous rpc calls to service MxAccessGateway.
* <pre>
* Public client API for MXAccess sessions hosted by the gateway.
* </pre>
*/
public static final class MxAccessGatewayBlockingStub
extends io.grpc.stub.AbstractBlockingStub<MxAccessGatewayBlockingStub> {
private MxAccessGatewayBlockingStub(
io.grpc.Channel channel, io.grpc.CallOptions callOptions) {
super(channel, callOptions);
}
@java.lang.Override
protected MxAccessGatewayBlockingStub build(
io.grpc.Channel channel, io.grpc.CallOptions callOptions) {
return new MxAccessGatewayBlockingStub(channel, callOptions);
}
/**
*/
public mxaccess_gateway.v1.MxaccessGateway.OpenSessionReply openSession(mxaccess_gateway.v1.MxaccessGateway.OpenSessionRequest request) {
return io.grpc.stub.ClientCalls.blockingUnaryCall(
getChannel(), getOpenSessionMethod(), getCallOptions(), request);
}
/**
*/
public mxaccess_gateway.v1.MxaccessGateway.CloseSessionReply closeSession(mxaccess_gateway.v1.MxaccessGateway.CloseSessionRequest request) {
return io.grpc.stub.ClientCalls.blockingUnaryCall(
getChannel(), getCloseSessionMethod(), getCallOptions(), request);
}
/**
*/
public mxaccess_gateway.v1.MxaccessGateway.MxCommandReply invoke(mxaccess_gateway.v1.MxaccessGateway.MxCommandRequest request) {
return io.grpc.stub.ClientCalls.blockingUnaryCall(
getChannel(), getInvokeMethod(), getCallOptions(), request);
}
/**
*/
public java.util.Iterator<mxaccess_gateway.v1.MxaccessGateway.MxEvent> streamEvents(
mxaccess_gateway.v1.MxaccessGateway.StreamEventsRequest request) {
return io.grpc.stub.ClientCalls.blockingServerStreamingCall(
getChannel(), getStreamEventsMethod(), getCallOptions(), request);
}
}
/**
* A stub to allow clients to do ListenableFuture-style rpc calls to service MxAccessGateway.
* <pre>
* Public client API for MXAccess sessions hosted by the gateway.
* </pre>
*/
public static final class MxAccessGatewayFutureStub
extends io.grpc.stub.AbstractFutureStub<MxAccessGatewayFutureStub> {
private MxAccessGatewayFutureStub(
io.grpc.Channel channel, io.grpc.CallOptions callOptions) {
super(channel, callOptions);
}
@java.lang.Override
protected MxAccessGatewayFutureStub build(
io.grpc.Channel channel, io.grpc.CallOptions callOptions) {
return new MxAccessGatewayFutureStub(channel, callOptions);
}
/**
*/
public com.google.common.util.concurrent.ListenableFuture<mxaccess_gateway.v1.MxaccessGateway.OpenSessionReply> openSession(
mxaccess_gateway.v1.MxaccessGateway.OpenSessionRequest request) {
return io.grpc.stub.ClientCalls.futureUnaryCall(
getChannel().newCall(getOpenSessionMethod(), getCallOptions()), request);
}
/**
*/
public com.google.common.util.concurrent.ListenableFuture<mxaccess_gateway.v1.MxaccessGateway.CloseSessionReply> closeSession(
mxaccess_gateway.v1.MxaccessGateway.CloseSessionRequest request) {
return io.grpc.stub.ClientCalls.futureUnaryCall(
getChannel().newCall(getCloseSessionMethod(), getCallOptions()), request);
}
/**
*/
public com.google.common.util.concurrent.ListenableFuture<mxaccess_gateway.v1.MxaccessGateway.MxCommandReply> invoke(
mxaccess_gateway.v1.MxaccessGateway.MxCommandRequest request) {
return io.grpc.stub.ClientCalls.futureUnaryCall(
getChannel().newCall(getInvokeMethod(), getCallOptions()), request);
}
}
private static final int METHODID_OPEN_SESSION = 0;
private static final int METHODID_CLOSE_SESSION = 1;
private static final int METHODID_INVOKE = 2;
private static final int METHODID_STREAM_EVENTS = 3;
private static final class MethodHandlers<Req, Resp> implements
io.grpc.stub.ServerCalls.UnaryMethod<Req, Resp>,
io.grpc.stub.ServerCalls.ServerStreamingMethod<Req, Resp>,
io.grpc.stub.ServerCalls.ClientStreamingMethod<Req, Resp>,
io.grpc.stub.ServerCalls.BidiStreamingMethod<Req, Resp> {
private final AsyncService serviceImpl;
private final int methodId;
MethodHandlers(AsyncService serviceImpl, int methodId) {
this.serviceImpl = serviceImpl;
this.methodId = methodId;
}
@java.lang.Override
@java.lang.SuppressWarnings("unchecked")
public void invoke(Req request, io.grpc.stub.StreamObserver<Resp> responseObserver) {
switch (methodId) {
case METHODID_OPEN_SESSION:
serviceImpl.openSession((mxaccess_gateway.v1.MxaccessGateway.OpenSessionRequest) request,
(io.grpc.stub.StreamObserver<mxaccess_gateway.v1.MxaccessGateway.OpenSessionReply>) responseObserver);
break;
case METHODID_CLOSE_SESSION:
serviceImpl.closeSession((mxaccess_gateway.v1.MxaccessGateway.CloseSessionRequest) request,
(io.grpc.stub.StreamObserver<mxaccess_gateway.v1.MxaccessGateway.CloseSessionReply>) responseObserver);
break;
case METHODID_INVOKE:
serviceImpl.invoke((mxaccess_gateway.v1.MxaccessGateway.MxCommandRequest) request,
(io.grpc.stub.StreamObserver<mxaccess_gateway.v1.MxaccessGateway.MxCommandReply>) responseObserver);
break;
case METHODID_STREAM_EVENTS:
serviceImpl.streamEvents((mxaccess_gateway.v1.MxaccessGateway.StreamEventsRequest) request,
(io.grpc.stub.StreamObserver<mxaccess_gateway.v1.MxaccessGateway.MxEvent>) responseObserver);
break;
default:
throw new AssertionError();
}
}
@java.lang.Override
@java.lang.SuppressWarnings("unchecked")
public io.grpc.stub.StreamObserver<Req> invoke(
io.grpc.stub.StreamObserver<Resp> responseObserver) {
switch (methodId) {
default:
throw new AssertionError();
}
}
}
public static final io.grpc.ServerServiceDefinition bindService(AsyncService service) {
return io.grpc.ServerServiceDefinition.builder(getServiceDescriptor())
.addMethod(
getOpenSessionMethod(),
io.grpc.stub.ServerCalls.asyncUnaryCall(
new MethodHandlers<
mxaccess_gateway.v1.MxaccessGateway.OpenSessionRequest,
mxaccess_gateway.v1.MxaccessGateway.OpenSessionReply>(
service, METHODID_OPEN_SESSION)))
.addMethod(
getCloseSessionMethod(),
io.grpc.stub.ServerCalls.asyncUnaryCall(
new MethodHandlers<
mxaccess_gateway.v1.MxaccessGateway.CloseSessionRequest,
mxaccess_gateway.v1.MxaccessGateway.CloseSessionReply>(
service, METHODID_CLOSE_SESSION)))
.addMethod(
getInvokeMethod(),
io.grpc.stub.ServerCalls.asyncUnaryCall(
new MethodHandlers<
mxaccess_gateway.v1.MxaccessGateway.MxCommandRequest,
mxaccess_gateway.v1.MxaccessGateway.MxCommandReply>(
service, METHODID_INVOKE)))
.addMethod(
getStreamEventsMethod(),
io.grpc.stub.ServerCalls.asyncServerStreamingCall(
new MethodHandlers<
mxaccess_gateway.v1.MxaccessGateway.StreamEventsRequest,
mxaccess_gateway.v1.MxaccessGateway.MxEvent>(
service, METHODID_STREAM_EVENTS)))
.build();
}
private static abstract class MxAccessGatewayBaseDescriptorSupplier
implements io.grpc.protobuf.ProtoFileDescriptorSupplier, io.grpc.protobuf.ProtoServiceDescriptorSupplier {
MxAccessGatewayBaseDescriptorSupplier() {}
@java.lang.Override
public com.google.protobuf.Descriptors.FileDescriptor getFileDescriptor() {
return mxaccess_gateway.v1.MxaccessGateway.getDescriptor();
}
@java.lang.Override
public com.google.protobuf.Descriptors.ServiceDescriptor getServiceDescriptor() {
return getFileDescriptor().findServiceByName("MxAccessGateway");
}
}
private static final class MxAccessGatewayFileDescriptorSupplier
extends MxAccessGatewayBaseDescriptorSupplier {
MxAccessGatewayFileDescriptorSupplier() {}
}
private static final class MxAccessGatewayMethodDescriptorSupplier
extends MxAccessGatewayBaseDescriptorSupplier
implements io.grpc.protobuf.ProtoMethodDescriptorSupplier {
private final java.lang.String methodName;
MxAccessGatewayMethodDescriptorSupplier(java.lang.String methodName) {
this.methodName = methodName;
}
@java.lang.Override
public com.google.protobuf.Descriptors.MethodDescriptor getMethodDescriptor() {
return getServiceDescriptor().findMethodByName(methodName);
}
}
private static volatile io.grpc.ServiceDescriptor serviceDescriptor;
public static io.grpc.ServiceDescriptor getServiceDescriptor() {
io.grpc.ServiceDescriptor result = serviceDescriptor;
if (result == null) {
synchronized (MxAccessGatewayGrpc.class) {
result = serviceDescriptor;
if (result == null) {
serviceDescriptor = result = io.grpc.ServiceDescriptor.newBuilder(SERVICE_NAME)
.setSchemaDescriptor(new MxAccessGatewayFileDescriptorSupplier())
.addMethod(getOpenSessionMethod())
.addMethod(getCloseSessionMethod())
.addMethod(getInvokeMethod())
.addMethod(getStreamEventsMethod())
.build();
}
}
}
return result;
}
}
File diff suppressed because it is too large Load Diff
File diff suppressed because it is too large Load Diff
+57
View File
@@ -0,0 +1,57 @@
# Python Client
The Python client package contains generated MXAccess Gateway protobuf
bindings, the `mxgateway` package scaffold, and the `mxgw-py` test CLI
scaffold. The package uses the shared proto inputs documented in
`../../docs/client-proto-generation.md` so gateway and client contracts stay in
sync.
## Layout
```text
clients/python/
pyproject.toml
generate-proto.ps1
src/mxgateway/
src/mxgateway/generated/
src/mxgateway_cli/
tests/
```
`src/mxgateway/generated` contains code produced by `grpc_tools.protoc`. Do not
edit generated files by hand.
## Regenerating Protobuf Bindings
Run generation after the shared `.proto` files or the Python output path
changes:
```powershell
./generate-proto.ps1
```
The script uses the Python tool path recorded in
`../../docs/toolchain-links.md`.
## Build And Test
Run the Python checks from `clients/python`:
```powershell
python -m pip install -e ".[dev]"
python -m pytest
python -m pip wheel . --no-deps --wheel-dir "$env:TEMP\mxgateway-python-wheel"
```
The scaffold tests import the generated gateway and worker stubs and exercise
the deterministic CLI version output.
## CLI
The scaffold CLI exposes version information:
```powershell
mxgw-py version --json
```
Additional commands are implemented with the async client/session wrapper work.
+22
View File
@@ -0,0 +1,22 @@
Set-StrictMode -Version Latest
$ErrorActionPreference = 'Stop'
$repoRoot = Resolve-Path (Join-Path $PSScriptRoot '..\..')
$protoRoot = Join-Path $repoRoot 'src\MxGateway.Contracts\Protos'
$outputRoot = Join-Path $PSScriptRoot 'src\mxgateway\generated'
$python = 'C:\Users\dohertj2\AppData\Local\Programs\Python\Python312\python.exe'
if (-not (Test-Path $python)) {
throw "Python was not found at $python. See docs/toolchain-links.md."
}
New-Item -ItemType Directory -Path $outputRoot -Force | Out-Null
Get-ChildItem -Path (Join-Path $outputRoot '*_pb2.py') -File | Remove-Item
Get-ChildItem -Path (Join-Path $outputRoot '*_pb2_grpc.py') -File | Remove-Item
& $python -m grpc_tools.protoc `
"-I$protoRoot" `
"--python_out=$outputRoot" `
"--grpc_python_out=$outputRoot" `
mxaccess_gateway.proto `
mxaccess_worker.proto
+33
View File
@@ -0,0 +1,33 @@
[build-system]
requires = ["setuptools>=69", "wheel"]
build-backend = "setuptools.build_meta"
[project]
name = "mxaccess-gateway-client"
version = "0.1.0"
description = "Async Python client scaffold for MXAccess Gateway."
readme = "README.md"
requires-python = ">=3.12"
dependencies = [
"click>=8.3,<9",
"grpcio>=1.80,<2",
"protobuf>=6.33,<7",
]
[project.optional-dependencies]
dev = [
"grpcio-tools>=1.80,<2",
"pytest>=9,<10",
"pytest-asyncio>=1.3,<2",
]
[project.scripts]
mxgw-py = "mxgateway_cli.commands:main"
[tool.setuptools.packages.find]
where = ["src"]
[tool.pytest.ini_options]
addopts = "-ra"
pythonpath = ["src"]
testpaths = ["tests"]
+5
View File
@@ -0,0 +1,5 @@
"""MXAccess Gateway Python client package."""
from .version import __version__
__all__ = ["__version__"]
@@ -0,0 +1,29 @@
"""Generated protobuf and gRPC modules for MXAccess Gateway.
The Python protobuf generator emits absolute imports between generated modules.
This package initializer registers package-local aliases so callers can import
the generated stubs through `mxgateway.generated` without moving the modules to
the top-level import namespace.
"""
from importlib import import_module
import sys
mxaccess_gateway_pb2 = import_module(f"{__name__}.mxaccess_gateway_pb2")
sys.modules.setdefault("mxaccess_gateway_pb2", mxaccess_gateway_pb2)
mxaccess_gateway_pb2_grpc = import_module(f"{__name__}.mxaccess_gateway_pb2_grpc")
sys.modules.setdefault("mxaccess_gateway_pb2_grpc", mxaccess_gateway_pb2_grpc)
mxaccess_worker_pb2 = import_module(f"{__name__}.mxaccess_worker_pb2")
sys.modules.setdefault("mxaccess_worker_pb2", mxaccess_worker_pb2)
mxaccess_worker_pb2_grpc = import_module(f"{__name__}.mxaccess_worker_pb2_grpc")
sys.modules.setdefault("mxaccess_worker_pb2_grpc", mxaccess_worker_pb2_grpc)
__all__ = [
"mxaccess_gateway_pb2",
"mxaccess_gateway_pb2_grpc",
"mxaccess_worker_pb2",
"mxaccess_worker_pb2_grpc",
]
File diff suppressed because one or more lines are too long
@@ -0,0 +1,229 @@
# Generated by the gRPC Python protocol compiler plugin. DO NOT EDIT!
"""Client and server classes corresponding to protobuf-defined services."""
import grpc
import warnings
import mxaccess_gateway_pb2 as mxaccess__gateway__pb2
GRPC_GENERATED_VERSION = '1.80.0'
GRPC_VERSION = grpc.__version__
_version_not_supported = False
try:
from grpc._utilities import first_version_is_lower
_version_not_supported = first_version_is_lower(GRPC_VERSION, GRPC_GENERATED_VERSION)
except ImportError:
_version_not_supported = True
if _version_not_supported:
raise RuntimeError(
f'The grpc package installed is at version {GRPC_VERSION},'
+ ' but the generated code in mxaccess_gateway_pb2_grpc.py depends on'
+ f' grpcio>={GRPC_GENERATED_VERSION}.'
+ f' Please upgrade your grpc module to grpcio>={GRPC_GENERATED_VERSION}'
+ f' or downgrade your generated code using grpcio-tools<={GRPC_VERSION}.'
)
class MxAccessGatewayStub(object):
"""Public client API for MXAccess sessions hosted by the gateway.
"""
def __init__(self, channel):
"""Constructor.
Args:
channel: A grpc.Channel.
"""
self.OpenSession = channel.unary_unary(
'/mxaccess_gateway.v1.MxAccessGateway/OpenSession',
request_serializer=mxaccess__gateway__pb2.OpenSessionRequest.SerializeToString,
response_deserializer=mxaccess__gateway__pb2.OpenSessionReply.FromString,
_registered_method=True)
self.CloseSession = channel.unary_unary(
'/mxaccess_gateway.v1.MxAccessGateway/CloseSession',
request_serializer=mxaccess__gateway__pb2.CloseSessionRequest.SerializeToString,
response_deserializer=mxaccess__gateway__pb2.CloseSessionReply.FromString,
_registered_method=True)
self.Invoke = channel.unary_unary(
'/mxaccess_gateway.v1.MxAccessGateway/Invoke',
request_serializer=mxaccess__gateway__pb2.MxCommandRequest.SerializeToString,
response_deserializer=mxaccess__gateway__pb2.MxCommandReply.FromString,
_registered_method=True)
self.StreamEvents = channel.unary_stream(
'/mxaccess_gateway.v1.MxAccessGateway/StreamEvents',
request_serializer=mxaccess__gateway__pb2.StreamEventsRequest.SerializeToString,
response_deserializer=mxaccess__gateway__pb2.MxEvent.FromString,
_registered_method=True)
class MxAccessGatewayServicer(object):
"""Public client API for MXAccess sessions hosted by the gateway.
"""
def OpenSession(self, request, context):
"""Missing associated documentation comment in .proto file."""
context.set_code(grpc.StatusCode.UNIMPLEMENTED)
context.set_details('Method not implemented!')
raise NotImplementedError('Method not implemented!')
def CloseSession(self, request, context):
"""Missing associated documentation comment in .proto file."""
context.set_code(grpc.StatusCode.UNIMPLEMENTED)
context.set_details('Method not implemented!')
raise NotImplementedError('Method not implemented!')
def Invoke(self, request, context):
"""Missing associated documentation comment in .proto file."""
context.set_code(grpc.StatusCode.UNIMPLEMENTED)
context.set_details('Method not implemented!')
raise NotImplementedError('Method not implemented!')
def StreamEvents(self, request, context):
"""Missing associated documentation comment in .proto file."""
context.set_code(grpc.StatusCode.UNIMPLEMENTED)
context.set_details('Method not implemented!')
raise NotImplementedError('Method not implemented!')
def add_MxAccessGatewayServicer_to_server(servicer, server):
rpc_method_handlers = {
'OpenSession': grpc.unary_unary_rpc_method_handler(
servicer.OpenSession,
request_deserializer=mxaccess__gateway__pb2.OpenSessionRequest.FromString,
response_serializer=mxaccess__gateway__pb2.OpenSessionReply.SerializeToString,
),
'CloseSession': grpc.unary_unary_rpc_method_handler(
servicer.CloseSession,
request_deserializer=mxaccess__gateway__pb2.CloseSessionRequest.FromString,
response_serializer=mxaccess__gateway__pb2.CloseSessionReply.SerializeToString,
),
'Invoke': grpc.unary_unary_rpc_method_handler(
servicer.Invoke,
request_deserializer=mxaccess__gateway__pb2.MxCommandRequest.FromString,
response_serializer=mxaccess__gateway__pb2.MxCommandReply.SerializeToString,
),
'StreamEvents': grpc.unary_stream_rpc_method_handler(
servicer.StreamEvents,
request_deserializer=mxaccess__gateway__pb2.StreamEventsRequest.FromString,
response_serializer=mxaccess__gateway__pb2.MxEvent.SerializeToString,
),
}
generic_handler = grpc.method_handlers_generic_handler(
'mxaccess_gateway.v1.MxAccessGateway', rpc_method_handlers)
server.add_generic_rpc_handlers((generic_handler,))
server.add_registered_method_handlers('mxaccess_gateway.v1.MxAccessGateway', rpc_method_handlers)
# This class is part of an EXPERIMENTAL API.
class MxAccessGateway(object):
"""Public client API for MXAccess sessions hosted by the gateway.
"""
@staticmethod
def OpenSession(request,
target,
options=(),
channel_credentials=None,
call_credentials=None,
insecure=False,
compression=None,
wait_for_ready=None,
timeout=None,
metadata=None):
return grpc.experimental.unary_unary(
request,
target,
'/mxaccess_gateway.v1.MxAccessGateway/OpenSession',
mxaccess__gateway__pb2.OpenSessionRequest.SerializeToString,
mxaccess__gateway__pb2.OpenSessionReply.FromString,
options,
channel_credentials,
insecure,
call_credentials,
compression,
wait_for_ready,
timeout,
metadata,
_registered_method=True)
@staticmethod
def CloseSession(request,
target,
options=(),
channel_credentials=None,
call_credentials=None,
insecure=False,
compression=None,
wait_for_ready=None,
timeout=None,
metadata=None):
return grpc.experimental.unary_unary(
request,
target,
'/mxaccess_gateway.v1.MxAccessGateway/CloseSession',
mxaccess__gateway__pb2.CloseSessionRequest.SerializeToString,
mxaccess__gateway__pb2.CloseSessionReply.FromString,
options,
channel_credentials,
insecure,
call_credentials,
compression,
wait_for_ready,
timeout,
metadata,
_registered_method=True)
@staticmethod
def Invoke(request,
target,
options=(),
channel_credentials=None,
call_credentials=None,
insecure=False,
compression=None,
wait_for_ready=None,
timeout=None,
metadata=None):
return grpc.experimental.unary_unary(
request,
target,
'/mxaccess_gateway.v1.MxAccessGateway/Invoke',
mxaccess__gateway__pb2.MxCommandRequest.SerializeToString,
mxaccess__gateway__pb2.MxCommandReply.FromString,
options,
channel_credentials,
insecure,
call_credentials,
compression,
wait_for_ready,
timeout,
metadata,
_registered_method=True)
@staticmethod
def StreamEvents(request,
target,
options=(),
channel_credentials=None,
call_credentials=None,
insecure=False,
compression=None,
wait_for_ready=None,
timeout=None,
metadata=None):
return grpc.experimental.unary_stream(
request,
target,
'/mxaccess_gateway.v1.MxAccessGateway/StreamEvents',
mxaccess__gateway__pb2.StreamEventsRequest.SerializeToString,
mxaccess__gateway__pb2.MxEvent.FromString,
options,
channel_credentials,
insecure,
call_credentials,
compression,
wait_for_ready,
timeout,
metadata,
_registered_method=True)
@@ -0,0 +1,66 @@
# -*- coding: utf-8 -*-
# Generated by the protocol buffer compiler. DO NOT EDIT!
# NO CHECKED-IN PROTOBUF GENCODE
# source: mxaccess_worker.proto
# Protobuf Python Version: 6.31.1
"""Generated protocol buffer code."""
from google.protobuf import descriptor as _descriptor
from google.protobuf import descriptor_pool as _descriptor_pool
from google.protobuf import runtime_version as _runtime_version
from google.protobuf import symbol_database as _symbol_database
from google.protobuf.internal import builder as _builder
_runtime_version.ValidateProtobufRuntimeVersion(
_runtime_version.Domain.PUBLIC,
6,
31,
1,
'',
'mxaccess_worker.proto'
)
# @@protoc_insertion_point(imports)
_sym_db = _symbol_database.Default()
from google.protobuf import duration_pb2 as google_dot_protobuf_dot_duration__pb2
from google.protobuf import timestamp_pb2 as google_dot_protobuf_dot_timestamp__pb2
import mxaccess_gateway_pb2 as mxaccess__gateway__pb2
DESCRIPTOR = _descriptor_pool.Default().AddSerializedFile(b'\n\x15mxaccess_worker.proto\x12\x12mxaccess_worker.v1\x1a\x1egoogle/protobuf/duration.proto\x1a\x1fgoogle/protobuf/timestamp.proto\x1a\x16mxaccess_gateway.proto\"\x95\x06\n\x0eWorkerEnvelope\x12\x18\n\x10protocol_version\x18\x01 \x01(\r\x12\x12\n\nsession_id\x18\x02 \x01(\t\x12\x10\n\x08sequence\x18\x03 \x01(\x04\x12\x16\n\x0e\x63orrelation_id\x18\x04 \x01(\t\x12\x39\n\rgateway_hello\x18\n \x01(\x0b\x32 .mxaccess_worker.v1.GatewayHelloH\x00\x12\x37\n\x0cworker_hello\x18\x0b \x01(\x0b\x32\x1f.mxaccess_worker.v1.WorkerHelloH\x00\x12\x37\n\x0cworker_ready\x18\x0c \x01(\x0b\x32\x1f.mxaccess_worker.v1.WorkerReadyH\x00\x12;\n\x0eworker_command\x18\r \x01(\x0b\x32!.mxaccess_worker.v1.WorkerCommandH\x00\x12\x46\n\x14worker_command_reply\x18\x0e \x01(\x0b\x32&.mxaccess_worker.v1.WorkerCommandReplyH\x00\x12\x39\n\rworker_cancel\x18\x0f \x01(\x0b\x32 .mxaccess_worker.v1.WorkerCancelH\x00\x12=\n\x0fworker_shutdown\x18\x10 \x01(\x0b\x32\".mxaccess_worker.v1.WorkerShutdownH\x00\x12\x44\n\x13worker_shutdown_ack\x18\x11 \x01(\x0b\x32%.mxaccess_worker.v1.WorkerShutdownAckH\x00\x12\x37\n\x0cworker_event\x18\x12 \x01(\x0b\x32\x1f.mxaccess_worker.v1.WorkerEventH\x00\x12?\n\x10worker_heartbeat\x18\x13 \x01(\x0b\x32#.mxaccess_worker.v1.WorkerHeartbeatH\x00\x12\x37\n\x0cworker_fault\x18\x14 \x01(\x0b\x32\x1f.mxaccess_worker.v1.WorkerFaultH\x00\x42\x06\n\x04\x62ody\"Z\n\x0cGatewayHello\x12\"\n\x1asupported_protocol_version\x18\x01 \x01(\r\x12\r\n\x05nonce\x18\x02 \x01(\t\x12\x17\n\x0fgateway_version\x18\x03 \x01(\t\"i\n\x0bWorkerHello\x12\x18\n\x10protocol_version\x18\x01 \x01(\r\x12\r\n\x05nonce\x18\x02 \x01(\t\x12\x19\n\x11worker_process_id\x18\x03 \x01(\x05\x12\x16\n\x0eworker_version\x18\x04 \x01(\t\"\x8e\x01\n\x0bWorkerReady\x12\x19\n\x11worker_process_id\x18\x01 \x01(\x05\x12\x17\n\x0fmxaccess_progid\x18\x02 \x01(\t\x12\x16\n\x0emxaccess_clsid\x18\x03 \x01(\t\x12\x33\n\x0fready_timestamp\x18\x04 \x01(\x0b\x32\x1a.google.protobuf.Timestamp\"w\n\rWorkerCommand\x12/\n\x07\x63ommand\x18\x01 \x01(\x0b\x32\x1e.mxaccess_gateway.v1.MxCommand\x12\x35\n\x11\x65nqueue_timestamp\x18\x02 \x01(\x0b\x32\x1a.google.protobuf.Timestamp\"\x81\x01\n\x12WorkerCommandReply\x12\x32\n\x05reply\x18\x01 \x01(\x0b\x32#.mxaccess_gateway.v1.MxCommandReply\x12\x37\n\x13\x63ompleted_timestamp\x18\x02 \x01(\x0b\x32\x1a.google.protobuf.Timestamp\"\x1e\n\x0cWorkerCancel\x12\x0e\n\x06reason\x18\x01 \x01(\t\"Q\n\x0eWorkerShutdown\x12/\n\x0cgrace_period\x18\x01 \x01(\x0b\x32\x19.google.protobuf.Duration\x12\x0e\n\x06reason\x18\x02 \x01(\t\"H\n\x11WorkerShutdownAck\x12\x33\n\x06status\x18\x01 \x01(\x0b\x32#.mxaccess_gateway.v1.ProtocolStatus\":\n\x0bWorkerEvent\x12+\n\x05\x65vent\x18\x01 \x01(\x0b\x32\x1c.mxaccess_gateway.v1.MxEvent\"\xa5\x02\n\x0fWorkerHeartbeat\x12\x19\n\x11worker_process_id\x18\x01 \x01(\x05\x12.\n\x05state\x18\x02 \x01(\x0e\x32\x1f.mxaccess_worker.v1.WorkerState\x12?\n\x1blast_sta_activity_timestamp\x18\x03 \x01(\x0b\x32\x1a.google.protobuf.Timestamp\x12\x1d\n\x15pending_command_count\x18\x04 \x01(\r\x12\"\n\x1aoutbound_event_queue_depth\x18\x05 \x01(\r\x12\x1b\n\x13last_event_sequence\x18\x06 \x01(\x04\x12&\n\x1e\x63urrent_command_correlation_id\x18\x07 \x01(\t\"\xf4\x01\n\x0bWorkerFault\x12\x39\n\x08\x63\x61tegory\x18\x01 \x01(\x0e\x32\'.mxaccess_worker.v1.WorkerFaultCategory\x12\x16\n\x0e\x63ommand_method\x18\x02 \x01(\t\x12\x14\n\x07hresult\x18\x03 \x01(\x05H\x00\x88\x01\x01\x12\x16\n\x0e\x65xception_type\x18\x04 \x01(\t\x12\x1a\n\x12\x64iagnostic_message\x18\x05 \x01(\t\x12<\n\x0fprotocol_status\x18\x06 \x01(\x0b\x32#.mxaccess_gateway.v1.ProtocolStatusB\n\n\x08_hresult*\x97\x02\n\x0bWorkerState\x12\x1c\n\x18WORKER_STATE_UNSPECIFIED\x10\x00\x12\x19\n\x15WORKER_STATE_STARTING\x10\x01\x12\x1c\n\x18WORKER_STATE_HANDSHAKING\x10\x02\x12!\n\x1dWORKER_STATE_INITIALIZING_STA\x10\x03\x12\x16\n\x12WORKER_STATE_READY\x10\x04\x12\"\n\x1eWORKER_STATE_EXECUTING_COMMAND\x10\x05\x12\x1e\n\x1aWORKER_STATE_SHUTTING_DOWN\x10\x06\x12\x18\n\x14WORKER_STATE_STOPPED\x10\x07\x12\x18\n\x14WORKER_STATE_FAULTED\x10\x08*\xc7\x04\n\x13WorkerFaultCategory\x12%\n!WORKER_FAULT_CATEGORY_UNSPECIFIED\x10\x00\x12+\n\'WORKER_FAULT_CATEGORY_INVALID_ARGUMENTS\x10\x01\x12\x37\n3WORKER_FAULT_CATEGORY_GATEWAY_AUTHENTICATION_FAILED\x10\x02\x12+\n\'WORKER_FAULT_CATEGORY_PROTOCOL_MISMATCH\x10\x03\x12,\n(WORKER_FAULT_CATEGORY_PROTOCOL_VIOLATION\x10\x04\x12+\n\'WORKER_FAULT_CATEGORY_PIPE_DISCONNECTED\x10\x05\x12\x32\n.WORKER_FAULT_CATEGORY_MXACCESS_CREATION_FAILED\x10\x06\x12\x31\n-WORKER_FAULT_CATEGORY_MXACCESS_COMMAND_FAILED\x10\x07\x12:\n6WORKER_FAULT_CATEGORY_MXACCESS_EVENT_CONVERSION_FAILED\x10\x08\x12\"\n\x1eWORKER_FAULT_CATEGORY_STA_HUNG\x10\t\x12(\n$WORKER_FAULT_CATEGORY_QUEUE_OVERFLOW\x10\n\x12*\n&WORKER_FAULT_CATEGORY_SHUTDOWN_TIMEOUT\x10\x0b\x42\x1c\xaa\x02\x19MxGateway.Contracts.Protob\x06proto3')
_globals = globals()
_builder.BuildMessageAndEnumDescriptors(DESCRIPTOR, _globals)
_builder.BuildTopDescriptorsAndMessages(DESCRIPTOR, 'mxaccess_worker_pb2', _globals)
if not _descriptor._USE_C_DESCRIPTORS:
_globals['DESCRIPTOR']._loaded_options = None
_globals['DESCRIPTOR']._serialized_options = b'\252\002\031MxGateway.Contracts.Proto'
_globals['_WORKERSTATE']._serialized_start=2316
_globals['_WORKERSTATE']._serialized_end=2595
_globals['_WORKERFAULTCATEGORY']._serialized_start=2598
_globals['_WORKERFAULTCATEGORY']._serialized_end=3181
_globals['_WORKERENVELOPE']._serialized_start=135
_globals['_WORKERENVELOPE']._serialized_end=924
_globals['_GATEWAYHELLO']._serialized_start=926
_globals['_GATEWAYHELLO']._serialized_end=1016
_globals['_WORKERHELLO']._serialized_start=1018
_globals['_WORKERHELLO']._serialized_end=1123
_globals['_WORKERREADY']._serialized_start=1126
_globals['_WORKERREADY']._serialized_end=1268
_globals['_WORKERCOMMAND']._serialized_start=1270
_globals['_WORKERCOMMAND']._serialized_end=1389
_globals['_WORKERCOMMANDREPLY']._serialized_start=1392
_globals['_WORKERCOMMANDREPLY']._serialized_end=1521
_globals['_WORKERCANCEL']._serialized_start=1523
_globals['_WORKERCANCEL']._serialized_end=1553
_globals['_WORKERSHUTDOWN']._serialized_start=1555
_globals['_WORKERSHUTDOWN']._serialized_end=1636
_globals['_WORKERSHUTDOWNACK']._serialized_start=1638
_globals['_WORKERSHUTDOWNACK']._serialized_end=1710
_globals['_WORKEREVENT']._serialized_start=1712
_globals['_WORKEREVENT']._serialized_end=1770
_globals['_WORKERHEARTBEAT']._serialized_start=1773
_globals['_WORKERHEARTBEAT']._serialized_end=2066
_globals['_WORKERFAULT']._serialized_start=2069
_globals['_WORKERFAULT']._serialized_end=2313
# @@protoc_insertion_point(module_scope)
@@ -0,0 +1,24 @@
# Generated by the gRPC Python protocol compiler plugin. DO NOT EDIT!
"""Client and server classes corresponding to protobuf-defined services."""
import grpc
import warnings
GRPC_GENERATED_VERSION = '1.80.0'
GRPC_VERSION = grpc.__version__
_version_not_supported = False
try:
from grpc._utilities import first_version_is_lower
_version_not_supported = first_version_is_lower(GRPC_VERSION, GRPC_GENERATED_VERSION)
except ImportError:
_version_not_supported = True
if _version_not_supported:
raise RuntimeError(
f'The grpc package installed is at version {GRPC_VERSION},'
+ ' but the generated code in mxaccess_worker_pb2_grpc.py depends on'
+ f' grpcio>={GRPC_GENERATED_VERSION}.'
+ f' Please upgrade your grpc module to grpcio>={GRPC_GENERATED_VERSION}'
+ f' or downgrade your generated code using grpcio-tools<={GRPC_VERSION}.'
)
+3
View File
@@ -0,0 +1,3 @@
"""Package version information."""
__version__ = "0.1.0"
@@ -0,0 +1 @@
"""Command-line entry points for the MXAccess Gateway Python client."""
@@ -0,0 +1,6 @@
"""Module execution entry point for `python -m mxgateway_cli`."""
from .commands import main
if __name__ == "__main__":
main()
@@ -0,0 +1,29 @@
"""CLI scaffold for the MXAccess Gateway Python client."""
import json
import click
from mxgateway import __version__
@click.group()
def main() -> None:
"""MXAccess Gateway Python test CLI."""
@main.command()
@click.option("--json", "output_json", is_flag=True, help="Emit JSON output.")
def version(output_json: bool) -> None:
"""Print client package version information."""
payload = {
"client": "mxgw-py",
"package": "mxaccess-gateway-client",
"version": __version__,
}
if output_json:
click.echo(json.dumps(payload, sort_keys=True))
return
click.echo(f"mxgw-py {__version__}")
+21
View File
@@ -0,0 +1,21 @@
"""Tests for the Python CLI scaffold."""
import json
from click.testing import CliRunner
from mxgateway import __version__
from mxgateway_cli.commands import main
def test_version_json_is_deterministic() -> None:
runner = CliRunner()
result = runner.invoke(main, ["version", "--json"])
assert result.exit_code == 0
assert json.loads(result.output) == {
"client": "mxgw-py",
"package": "mxaccess-gateway-client",
"version": __version__,
}
@@ -0,0 +1,30 @@
"""Tests for generated protobuf and gRPC module importability."""
from mxgateway.generated import mxaccess_gateway_pb2
from mxgateway.generated import mxaccess_gateway_pb2_grpc
from mxgateway.generated import mxaccess_worker_pb2
def test_gateway_messages_import() -> None:
request = mxaccess_gateway_pb2.OpenSessionRequest(
client_session_name="pytest",
client_correlation_id="test-correlation",
)
assert request.client_session_name == "pytest"
assert hasattr(mxaccess_gateway_pb2_grpc, "MxAccessGatewayStub")
def test_worker_messages_import_gateway_types() -> None:
envelope = mxaccess_worker_pb2.WorkerEnvelope(
protocol_version=1,
session_id="test-session",
worker_command=mxaccess_worker_pb2.WorkerCommand(
command=mxaccess_gateway_pb2.MxCommand(
kind=mxaccess_gateway_pb2.MX_COMMAND_KIND_PING,
ping=mxaccess_gateway_pb2.PingCommand(message="hello"),
),
),
)
assert envelope.worker_command.command.ping.message == "hello"
+1438
View File
File diff suppressed because it is too large Load Diff
+46
View File
@@ -0,0 +1,46 @@
[package]
name = "mxgateway-client"
version = "0.1.0"
edition = "2021"
publish = false
build = "build.rs"
[workspace]
members = ["crates/mxgw-cli"]
resolver = "2"
[workspace.package]
edition = "2021"
version = "0.1.0"
publish = false
[workspace.dependencies]
clap = { version = "4.5.53", features = ["derive"] }
futures-core = "0.3.31"
futures-util = "0.3.31"
prost = "0.13.5"
prost-types = "0.13.5"
serde = { version = "1.0.228", features = ["derive"] }
serde_json = "1.0.145"
thiserror = "2.0.17"
tokio = { version = "1.48.0", features = ["macros", "rt-multi-thread", "sync", "time"] }
tokio-stream = { version = "0.1.17", features = ["net"] }
tonic = { version = "0.13.1", features = ["transport", "tls-ring"] }
tonic-build = "0.13.1"
[dependencies]
futures-core = { workspace = true }
futures-util = { workspace = true }
prost = { workspace = true }
prost-types = { workspace = true }
thiserror = { workspace = true }
tokio = { workspace = true }
tonic = { workspace = true }
[dev-dependencies]
serde_json = { workspace = true }
tokio = { workspace = true }
tokio-stream = { workspace = true }
[build-dependencies]
tonic-build = { workspace = true }
+90
View File
@@ -0,0 +1,90 @@
# Rust Client Workspace
The Rust client workspace contains the MXAccess Gateway client library, a
test CLI, and tests for generated contract wiring plus wrapper behavior. The
library uses
the shared protobuf inputs documented in
`../../docs/client-proto-generation.md` so the Rust bindings compile against
the same public gateway and worker contracts as the server.
## Layout
```text
clients/rust/
Cargo.toml
build.rs
src/
tests/
crates/mxgw-cli/
```
`build.rs` reads the `.proto` files from
`../../src/MxGateway.Contracts/Protos` and generates `tonic`/`prost` bindings
into Cargo build output. `src/generated.rs` declares the Rust modules that
include those generated files. `src/generated` remains reserved for checked-in
generator output if the crate later changes to source-tree generation.
## Build And Test
Run the Rust workspace checks from `clients/rust`:
```powershell
cargo fmt --all --check
cargo test --workspace
cargo check --workspace
cargo clippy --workspace --all-targets -- -D warnings
```
The build script uses `protoc` from `PATH` or the Windows path recorded in
`../../docs/toolchain-links.md`.
## CLI
The CLI exposes version, session, command, event stream, write, and smoke
commands over the same client wrapper used by tests:
```powershell
cargo run -p mxgw-cli -- version --json
cargo run -p mxgw-cli -- open-session --endpoint http://localhost:5000 --api-key-env MXGATEWAY_API_KEY --json
cargo run -p mxgw-cli -- register --session-id <session-id> --client-name mxgw-rust-cli --json
cargo run -p mxgw-cli -- add-item --session-id <session-id> --server-handle 1 --item TestChildObject.TestInt --json
cargo run -p mxgw-cli -- advise --session-id <session-id> --server-handle 1 --item-handle 1 --json
cargo run -p mxgw-cli -- stream-events --session-id <session-id> --max-events 1 --json
cargo run -p mxgw-cli -- write --session-id <session-id> --server-handle 1 --item-handle 1 --value-type int32 --value 123 --json
```
Use `--tls`, `--ca-file`, and `--server-name-override` for TLS endpoints. The
CLI reads the API key from `--api-key` or from `--api-key-env`, which defaults
to `MXGATEWAY_API_KEY`. API keys are redacted by the library option and secret
types.
## Library Surface
`ClientOptions` configures endpoint, API key, plaintext or TLS transport,
timeouts, custom CA files, and server name override. `GatewayClient::connect`
creates an authenticated `tonic` client and attaches `authorization: Bearer
<api-key>` metadata to unary and streaming calls.
`GatewayClient` exposes raw generated calls through `open_session_raw`,
`close_session_raw`, `invoke_raw`, `stream_events`, and `raw_client`. The
session helpers keep MXAccess handles visible:
```rust
let session = client.open_session(request).await?;
let server_handle = session.register("mxgw-rust").await?;
let item_handle = session.add_item(server_handle, "TestChildObject.TestInt").await?;
session.advise(server_handle, item_handle).await?;
let mut events = session.events().await?;
session.close().await?;
```
`MxValue`, `MxArrayValue`, and `MxStatus` wrap generated protobuf messages while
preserving the raw message for parity diagnostics. Command replies whose
protocol status is not `PROTOCOL_STATUS_CODE_OK` become `Error::Command` and
retain the raw `MxCommandReply`.
## Related Documentation
- [Client Proto Generation](../../docs/client-proto-generation.md)
- [Rust Client Detailed Design](../../docs/clients-rust-design.md)
- [Rust Style Guide](../../docs/style-guides/RustStyleGuide.md)
+59
View File
@@ -0,0 +1,59 @@
use std::env;
use std::error::Error;
use std::path::{Path, PathBuf};
fn main() -> Result<(), Box<dyn Error>> {
configure_protoc();
let manifest_dir = PathBuf::from(env::var("CARGO_MANIFEST_DIR")?);
let repo_root = manifest_dir
.parent()
.and_then(Path::parent)
.ok_or("clients/rust must live two levels below the repository root")?;
let proto_root = repo_root.join("src/MxGateway.Contracts/Protos");
let gateway_proto = proto_root.join("mxaccess_gateway.proto");
let worker_proto = proto_root.join("mxaccess_worker.proto");
let descriptor_path = PathBuf::from(env::var("OUT_DIR")?).join("mxaccessgw-client-v1.protoset");
println!("cargo:rerun-if-changed={}", gateway_proto.display());
println!("cargo:rerun-if-changed={}", worker_proto.display());
tonic_build::configure()
.build_server(true)
.build_client(true)
.file_descriptor_set_path(descriptor_path)
.compile_protos(
&[gateway_proto.as_path(), worker_proto.as_path()],
&[proto_root.as_path()],
)?;
Ok(())
}
fn configure_protoc() {
if env::var_os("PROTOC").is_some() {
return;
}
for candidate in protoc_candidates() {
if candidate.is_file() {
env::set_var("PROTOC", candidate);
return;
}
}
}
fn protoc_candidates() -> Vec<PathBuf> {
let mut candidates = Vec::new();
if cfg!(windows) {
if let Some(local_app_data) = env::var_os("LOCALAPPDATA") {
candidates.push(PathBuf::from(local_app_data).join(
"Microsoft/WinGet/Packages/Google.Protobuf_Microsoft.Winget.Source_8wekyb3d8bbwe/bin/protoc.exe",
));
}
}
candidates.push(PathBuf::from("protoc"));
candidates
}
+17
View File
@@ -0,0 +1,17 @@
[package]
name = "mxgw-cli"
version.workspace = true
edition.workspace = true
publish.workspace = true
[[bin]]
name = "mxgw"
path = "src/main.rs"
[dependencies]
clap = { workspace = true }
futures-util = { workspace = true }
mxgateway-client = { path = "../.." }
serde = { workspace = true }
serde_json = { workspace = true }
tokio = { workspace = true }
+568
View File
@@ -0,0 +1,568 @@
use std::env;
use std::path::PathBuf;
use std::process::ExitCode;
use std::time::Duration;
use clap::{Args, Parser, Subcommand, ValueEnum};
use futures_util::StreamExt;
use mxgateway_client::generated::mxaccess_gateway::v1::{
CloseSessionRequest, MxCommand, MxCommandKind, MxCommandRequest, OpenSessionRequest,
PingCommand, StreamEventsRequest,
};
use mxgateway_client::{
ApiKey, ClientOptions, Error, GatewayClient, MxValue, CLIENT_VERSION, GATEWAY_PROTOCOL_VERSION,
WORKER_PROTOCOL_VERSION,
};
use serde_json::json;
use serde_json::Value;
#[derive(Debug, Parser)]
#[command(name = "mxgw")]
#[command(about = "MXAccess Gateway Rust test CLI")]
struct Cli {
#[command(subcommand)]
command: Command,
}
#[derive(Debug, Subcommand)]
enum Command {
Version {
#[arg(long)]
json: bool,
},
Ping {
#[command(flatten)]
connection: ConnectionArgs,
#[arg(long, default_value = "ping")]
message: String,
#[arg(long)]
json: bool,
},
OpenSession {
#[command(flatten)]
connection: ConnectionArgs,
#[arg(long, default_value = "mxgw-rust-cli")]
client_name: String,
#[arg(long)]
json: bool,
},
CloseSession {
#[command(flatten)]
connection: ConnectionArgs,
#[arg(long)]
session_id: String,
#[arg(long)]
json: bool,
},
Register {
#[command(flatten)]
connection: ConnectionArgs,
#[arg(long)]
session_id: String,
#[arg(long, default_value = "mxgw-rust-cli")]
client_name: String,
#[arg(long)]
json: bool,
},
AddItem {
#[command(flatten)]
connection: ConnectionArgs,
#[arg(long)]
session_id: String,
#[arg(long)]
server_handle: i32,
#[arg(long)]
item: String,
#[arg(long)]
json: bool,
},
Advise {
#[command(flatten)]
connection: ConnectionArgs,
#[arg(long)]
session_id: String,
#[arg(long)]
server_handle: i32,
#[arg(long)]
item_handle: i32,
#[arg(long)]
json: bool,
},
StreamEvents {
#[command(flatten)]
connection: ConnectionArgs,
#[arg(long)]
session_id: String,
#[arg(long, default_value_t = 0)]
after_worker_sequence: u64,
#[arg(long, default_value_t = 1)]
max_events: usize,
#[arg(long)]
json: bool,
},
Write {
#[command(flatten)]
connection: ConnectionArgs,
#[arg(long)]
session_id: String,
#[arg(long)]
server_handle: i32,
#[arg(long)]
item_handle: i32,
#[arg(long, value_enum)]
value_type: CliValueType,
#[arg(long)]
value: String,
#[arg(long, default_value_t = 0)]
user_id: i32,
#[arg(long)]
json: bool,
},
Write2 {
#[command(flatten)]
connection: ConnectionArgs,
#[arg(long)]
session_id: String,
#[arg(long)]
server_handle: i32,
#[arg(long)]
item_handle: i32,
#[arg(long, value_enum)]
value_type: CliValueType,
#[arg(long)]
value: String,
#[arg(long)]
timestamp: String,
#[arg(long, default_value_t = 0)]
user_id: i32,
#[arg(long)]
json: bool,
},
Smoke {
#[command(flatten)]
connection: ConnectionArgs,
#[arg(long)]
item: String,
#[arg(long, default_value = "mxgw-rust-smoke")]
client_name: String,
#[arg(long)]
json: bool,
},
}
#[derive(Debug, Args, Clone)]
struct ConnectionArgs {
#[arg(long, default_value = "http://127.0.0.1:5000")]
endpoint: String,
#[arg(long)]
api_key: Option<String>,
#[arg(long, default_value = "MXGATEWAY_API_KEY")]
api_key_env: String,
#[arg(long)]
plaintext: bool,
#[arg(long)]
tls: bool,
#[arg(long)]
ca_file: Option<PathBuf>,
#[arg(long)]
server_name_override: Option<String>,
#[arg(long, default_value_t = 10)]
connect_timeout_seconds: u64,
#[arg(long, default_value_t = 30)]
call_timeout_seconds: u64,
}
impl ConnectionArgs {
fn options(&self) -> ClientOptions {
let mut options = ClientOptions::new(self.endpoint.clone())
.with_plaintext(!self.tls || self.plaintext)
.with_connect_timeout(Duration::from_secs(self.connect_timeout_seconds))
.with_call_timeout(Duration::from_secs(self.call_timeout_seconds));
if let Some(api_key) = self
.api_key
.clone()
.or_else(|| env::var(&self.api_key_env).ok())
.filter(|value| !value.is_empty())
{
options = options.with_api_key(ApiKey::new(api_key));
}
if let Some(ca_file) = &self.ca_file {
options = options.with_ca_file(ca_file);
}
if let Some(server_name_override) = &self.server_name_override {
options = options.with_server_name_override(server_name_override);
}
options
}
}
#[derive(Clone, Copy, Debug, ValueEnum)]
enum CliValueType {
Bool,
Int32,
Int64,
Float,
Double,
String,
}
#[tokio::main]
async fn main() -> ExitCode {
let cli = Cli::parse();
match run(cli).await {
Ok(()) => ExitCode::SUCCESS,
Err(error) => {
eprintln!("{error}");
ExitCode::FAILURE
}
}
}
async fn run(cli: Cli) -> Result<(), Error> {
match cli.command {
Command::Version { json } => print_version(json),
Command::Ping {
connection,
message,
json,
} => {
let client = connect(connection).await?;
let reply = client
.invoke(MxCommandRequest {
client_correlation_id: "rust-cli-ping".to_owned(),
command: Some(MxCommand {
kind: MxCommandKind::Ping as i32,
payload: Some(mxgateway_client::generated::mxaccess_gateway::v1::mx_command::Payload::Ping(
PingCommand { message },
)),
}),
..MxCommandRequest::default()
})
.await?;
print_command_reply("ping", &reply, json);
}
Command::OpenSession {
connection,
client_name,
json,
} => {
let client = connect(connection).await?;
let reply = client
.open_session_raw(OpenSessionRequest {
client_session_name: client_name,
..OpenSessionRequest::default()
})
.await?;
if json {
println!(
"{}",
json!({
"sessionId": reply.session_id,
"backendName": reply.backend_name,
"gatewayProtocolVersion": reply.gateway_protocol_version,
"workerProtocolVersion": reply.worker_protocol_version,
})
);
} else {
println!("{}", reply.session_id);
}
}
Command::CloseSession {
connection,
session_id,
json,
} => {
let client = connect(connection).await?;
let reply = client
.close_session_raw(CloseSessionRequest {
session_id,
client_correlation_id: "rust-cli-close-session".to_owned(),
})
.await?;
if json {
println!("{}", json!({ "sessionId": reply.session_id }));
} else {
println!("closed {}", reply.session_id);
}
}
Command::Register {
connection,
session_id,
client_name,
json,
} => {
let session = session_for(connection, session_id).await?;
let server_handle = session.register(&client_name).await?;
print_handle("serverHandle", server_handle, json);
}
Command::AddItem {
connection,
session_id,
server_handle,
item,
json,
} => {
let session = session_for(connection, session_id).await?;
let item_handle = session.add_item(server_handle, &item).await?;
print_handle("itemHandle", item_handle, json);
}
Command::Advise {
connection,
session_id,
server_handle,
item_handle,
json,
} => {
let session = session_for(connection, session_id).await?;
session.advise(server_handle, item_handle).await?;
print_ok("advise", json);
}
Command::StreamEvents {
connection,
session_id,
after_worker_sequence,
max_events,
json,
} => {
let client = connect(connection).await?;
let mut stream = client
.stream_events(StreamEventsRequest {
session_id,
after_worker_sequence,
})
.await?;
let mut events = Vec::new();
while events.len() < max_events {
let Some(event) = stream.next().await else {
break;
};
events.push(event?);
}
if json {
println!("{}", json!({ "eventCount": events.len() }));
} else {
for event in events {
println!("{} {}", event.worker_sequence, event.family);
}
}
}
Command::Write {
connection,
session_id,
server_handle,
item_handle,
value_type,
value,
user_id,
json,
} => {
let session = session_for(connection, session_id).await?;
session
.write(
server_handle,
item_handle,
parse_value(value_type, &value)?,
user_id,
)
.await?;
print_ok("write", json);
}
Command::Write2 {
connection,
session_id,
server_handle,
item_handle,
value_type,
value,
timestamp,
user_id,
json,
} => {
let session = session_for(connection, session_id).await?;
session
.write2(
server_handle,
item_handle,
parse_value(value_type, &value)?,
MxValue::string(timestamp),
user_id,
)
.await?;
print_ok("write2", json);
}
Command::Smoke {
connection,
item,
client_name,
json,
} => {
let client = connect(connection).await?;
let session = client
.open_session(OpenSessionRequest {
client_session_name: client_name.clone(),
..OpenSessionRequest::default()
})
.await?;
let result = async {
let server_handle = session.register(&client_name).await?;
let item_handle = session.add_item(server_handle, &item).await?;
session.advise(server_handle, item_handle).await?;
Ok::<_, Error>((server_handle, item_handle))
}
.await;
let close_result = session.close().await;
let (server_handle, item_handle) = result?;
close_result?;
if json {
println!(
"{}",
json!({
"sessionId": session.id(),
"serverHandle": server_handle,
"itemHandle": item_handle,
"closed": true,
})
);
} else {
println!(
"session {} registered server {server_handle}, item {item_handle}, closed",
session.id()
);
}
}
}
Ok(())
}
async fn connect(connection: ConnectionArgs) -> Result<GatewayClient, Error> {
GatewayClient::connect(connection.options()).await
}
async fn session_for(
connection: ConnectionArgs,
session_id: String,
) -> Result<mxgateway_client::Session, Error> {
let client = connect(connection).await?;
Ok(client.session(session_id))
}
fn print_version(use_json: bool) {
if use_json {
println!("{}", version_json());
return;
}
println!("mxgw {CLIENT_VERSION}");
println!("gateway protocol {GATEWAY_PROTOCOL_VERSION}");
println!("worker protocol {WORKER_PROTOCOL_VERSION}");
}
fn version_json() -> Value {
json!({
"clientVersion": CLIENT_VERSION,
"gatewayProtocolVersion": GATEWAY_PROTOCOL_VERSION,
"workerProtocolVersion": WORKER_PROTOCOL_VERSION,
})
}
fn print_command_reply(
operation: &str,
reply: &mxgateway_client::generated::mxaccess_gateway::v1::MxCommandReply,
use_json: bool,
) {
if use_json {
println!(
"{}",
json!({
"operation": operation,
"sessionId": reply.session_id,
"correlationId": reply.correlation_id,
"kind": reply.kind,
})
);
} else {
println!("{operation} completed");
}
}
fn print_handle(name: &str, handle: i32, use_json: bool) {
if use_json {
println!("{}", json!({ name: handle }));
} else {
println!("{handle}");
}
}
fn print_ok(operation: &str, use_json: bool) {
if use_json {
println!("{}", json!({ "operation": operation, "ok": true }));
} else {
println!("{operation} completed");
}
}
fn parse_value(value_type: CliValueType, value: &str) -> Result<MxValue, Error> {
let parsed = match value_type {
CliValueType::Bool => MxValue::bool(parse_cli_value(value)?),
CliValueType::Int32 => MxValue::int32(parse_cli_value(value)?),
CliValueType::Int64 => MxValue::int64(parse_cli_value(value)?),
CliValueType::Float => MxValue::float(parse_cli_value(value)?),
CliValueType::Double => MxValue::double(parse_cli_value(value)?),
CliValueType::String => MxValue::string(value),
};
Ok(parsed)
}
fn parse_cli_value<T>(value: &str) -> Result<T, Error>
where
T: std::str::FromStr,
T::Err: std::fmt::Display,
{
value.parse::<T>().map_err(|source| Error::InvalidArgument {
name: "value".to_owned(),
detail: source.to_string(),
})
}
#[cfg(test)]
mod tests {
use clap::Parser;
use super::Cli;
#[test]
fn parses_version_json_command() {
let parsed = Cli::try_parse_from(["mxgw", "version", "--json"]);
assert!(parsed.is_ok());
}
#[test]
fn parses_write_command() {
let parsed = Cli::try_parse_from([
"mxgw",
"write",
"--session-id",
"session-1",
"--server-handle",
"12",
"--item-handle",
"34",
"--value-type",
"int32",
"--value",
"123",
]);
assert!(parsed.is_ok());
}
#[test]
fn version_json_output_has_protocol_versions() {
let value = super::version_json();
assert_eq!(value["gatewayProtocolVersion"], 1);
assert_eq!(value["workerProtocolVersion"], 1);
}
}
+87
View File
@@ -0,0 +1,87 @@
use std::fmt;
use tonic::metadata::MetadataValue;
use tonic::service::Interceptor;
use tonic::{Request, Status};
/// API key wrapper that avoids exposing raw credentials in formatted output.
#[derive(Clone, Eq, PartialEq)]
pub struct ApiKey(String);
impl ApiKey {
pub fn new(value: impl Into<String>) -> Self {
Self(value.into())
}
pub fn expose_secret(&self) -> &str {
&self.0
}
}
impl fmt::Debug for ApiKey {
fn fmt(&self, formatter: &mut fmt::Formatter<'_>) -> fmt::Result {
formatter
.debug_tuple("ApiKey")
.field(&"<redacted>")
.finish()
}
}
impl fmt::Display for ApiKey {
fn fmt(&self, formatter: &mut fmt::Formatter<'_>) -> fmt::Result {
formatter.write_str("<redacted>")
}
}
/// `tonic` interceptor that attaches gateway API key metadata.
#[derive(Clone, Debug, Default)]
pub struct AuthInterceptor {
api_key: Option<ApiKey>,
}
impl AuthInterceptor {
pub fn new(api_key: Option<ApiKey>) -> Self {
Self { api_key }
}
}
impl Interceptor for AuthInterceptor {
fn call(&mut self, mut request: Request<()>) -> Result<Request<()>, Status> {
if let Some(api_key) = &self.api_key {
let header_value = format!("Bearer {}", api_key.expose_secret())
.parse::<MetadataValue<_>>()
.map_err(|_| Status::unauthenticated("invalid API key metadata"))?;
request.metadata_mut().insert("authorization", header_value);
}
Ok(request)
}
}
#[cfg(test)]
mod tests {
use tonic::service::Interceptor;
use tonic::Request;
use super::{ApiKey, AuthInterceptor};
#[test]
fn api_key_debug_is_redacted() {
let key = ApiKey::new("mxgw_visible_secret");
assert_eq!(format!("{key:?}"), "ApiKey(\"<redacted>\")");
assert!(!format!("{key:?}").contains("visible_secret"));
assert_eq!(key.to_string(), "<redacted>");
}
#[test]
fn interceptor_attaches_bearer_metadata() {
let mut interceptor = AuthInterceptor::new(Some(ApiKey::new("mxgw_fixture_secret")));
let request = interceptor.call(Request::new(())).unwrap();
assert_eq!(
request.metadata().get("authorization").unwrap(),
"Bearer mxgw_fixture_secret"
);
}
}
+123
View File
@@ -0,0 +1,123 @@
use std::fs;
use tonic::codegen::InterceptedService;
use tonic::transport::{Certificate, Channel, ClientTlsConfig};
use tonic::Request;
use crate::auth::AuthInterceptor;
use crate::error::{ensure_command_success, Error};
use crate::generated::mxaccess_gateway::v1::mx_access_gateway_client::MxAccessGatewayClient;
use crate::generated::mxaccess_gateway::v1::{
CloseSessionReply, CloseSessionRequest, MxCommandReply, MxCommandRequest, MxEvent,
OpenSessionReply, OpenSessionRequest, StreamEventsRequest,
};
use crate::options::ClientOptions;
use crate::session::Session;
pub type RawGatewayClient = MxAccessGatewayClient<InterceptedService<Channel, AuthInterceptor>>;
pub type EventStream =
std::pin::Pin<Box<dyn futures_core::Stream<Item = Result<MxEvent, Error>> + Send + 'static>>;
/// Thin owner for the generated gateway client.
#[derive(Clone)]
pub struct GatewayClient {
inner: RawGatewayClient,
call_timeout: std::time::Duration,
}
impl GatewayClient {
pub async fn connect(options: ClientOptions) -> Result<Self, Error> {
let mut endpoint =
Channel::from_shared(options.endpoint().to_owned()).map_err(|source| {
Error::InvalidEndpoint {
endpoint: options.endpoint().to_owned(),
detail: source.to_string(),
}
})?;
endpoint = endpoint.connect_timeout(options.connect_timeout());
if !options.plaintext() {
let mut tls = ClientTlsConfig::new();
if let Some(server_name) = options.server_name_override() {
tls = tls.domain_name(server_name.to_owned());
}
if let Some(ca_file) = options.ca_file() {
let certificate = fs::read(ca_file).map_err(|source| Error::InvalidEndpoint {
endpoint: options.endpoint().to_owned(),
detail: format!("failed to read CA file {}: {source}", ca_file.display()),
})?;
tls = tls.ca_certificate(Certificate::from_pem(certificate));
}
endpoint = endpoint.tls_config(tls)?;
}
let channel = endpoint.connect().await?;
let interceptor = AuthInterceptor::new(options.api_key().cloned());
Ok(Self {
inner: MxAccessGatewayClient::with_interceptor(channel, interceptor),
call_timeout: options.call_timeout(),
})
}
pub fn raw_client(&mut self) -> &mut RawGatewayClient {
&mut self.inner
}
pub fn into_inner(self) -> RawGatewayClient {
self.inner
}
pub fn session(&self, session_id: impl Into<String>) -> Session {
Session::new(session_id, self.clone())
}
pub async fn open_session_raw(
&self,
request: OpenSessionRequest,
) -> Result<OpenSessionReply, Error> {
let mut client = self.inner.clone();
let response = client.open_session(self.unary_request(request)).await?;
Ok(response.into_inner())
}
pub async fn open_session(&self, request: OpenSessionRequest) -> Result<Session, Error> {
let reply = self.open_session_raw(request).await?;
Ok(Session::new(reply.session_id, self.clone()))
}
pub async fn close_session_raw(
&self,
request: CloseSessionRequest,
) -> Result<CloseSessionReply, Error> {
let mut client = self.inner.clone();
let response = client.close_session(self.unary_request(request)).await?;
Ok(response.into_inner())
}
pub async fn invoke_raw(&self, request: MxCommandRequest) -> Result<MxCommandReply, Error> {
let mut client = self.inner.clone();
let response = client.invoke(self.unary_request(request)).await?;
Ok(response.into_inner())
}
pub async fn invoke(&self, request: MxCommandRequest) -> Result<MxCommandReply, Error> {
ensure_command_success(self.invoke_raw(request).await?)
}
pub async fn stream_events(&self, request: StreamEventsRequest) -> Result<EventStream, Error> {
let mut client = self.inner.clone();
let response = client.stream_events(self.unary_request(request)).await?;
let stream = futures_util::StreamExt::map(response.into_inner(), |result| {
result.map_err(Error::from)
});
Ok(Box::pin(stream))
}
fn unary_request<T>(&self, message: T) -> Request<T> {
let mut request = Request::new(message);
request.set_timeout(self.call_timeout);
request
}
}
+161
View File
@@ -0,0 +1,161 @@
use thiserror::Error as ThisError;
use tonic::Code;
use crate::generated::mxaccess_gateway::v1::{MxCommandReply, ProtocolStatusCode};
#[derive(Debug, ThisError)]
pub enum Error {
#[error("invalid gateway endpoint `{endpoint}`: {detail}")]
InvalidEndpoint { endpoint: String, detail: String },
#[error("invalid argument `{name}`: {detail}")]
InvalidArgument { name: String, detail: String },
#[error("gateway transport error: {0}")]
Transport(#[from] tonic::transport::Error),
#[error("authentication failed: {message}")]
Authentication {
message: String,
#[source]
status: Box<tonic::Status>,
},
#[error("authorization failed: {message}")]
Authorization {
message: String,
#[source]
status: Box<tonic::Status>,
},
#[error("gateway call timed out: {message}")]
Timeout {
message: String,
#[source]
status: Box<tonic::Status>,
},
#[error("gateway call cancelled: {message}")]
Cancelled {
message: String,
#[source]
status: Box<tonic::Status>,
},
#[error("gateway status error: {0}")]
Status(Box<tonic::Status>),
#[error("gateway command failed: {0}")]
Command(#[from] Box<CommandError>),
}
#[derive(Clone, Debug)]
pub struct CommandError {
reply: MxCommandReply,
}
impl CommandError {
pub fn new(reply: MxCommandReply) -> Self {
Self { reply }
}
pub fn reply(&self) -> &MxCommandReply {
&self.reply
}
pub fn into_reply(self) -> MxCommandReply {
self.reply
}
}
impl std::fmt::Display for CommandError {
fn fmt(&self, formatter: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
let status = self.reply.protocol_status.as_ref();
let code = status
.and_then(|status| ProtocolStatusCode::try_from(status.code).ok())
.unwrap_or(ProtocolStatusCode::Unspecified);
let message = status.map(|status| status.message.as_str()).unwrap_or("");
if message.is_empty() {
write!(formatter, "{code:?}")
} else {
write!(formatter, "{code:?}: {message}")
}
}
}
impl std::error::Error for CommandError {}
impl From<tonic::Status> for Error {
fn from(status: tonic::Status) -> Self {
let message = redact_credentials(status.message());
match status.code() {
Code::Unauthenticated => Self::Authentication {
message,
status: Box::new(status),
},
Code::PermissionDenied => Self::Authorization {
message,
status: Box::new(status),
},
Code::DeadlineExceeded => Self::Timeout {
message,
status: Box::new(status),
},
Code::Cancelled => Self::Cancelled {
message,
status: Box::new(status),
},
_ => Self::Status(Box::new(status)),
}
}
}
pub fn ensure_command_success(reply: MxCommandReply) -> Result<MxCommandReply, Error> {
let code = reply
.protocol_status
.as_ref()
.and_then(|status| ProtocolStatusCode::try_from(status.code).ok())
.unwrap_or(ProtocolStatusCode::Unspecified);
if code == ProtocolStatusCode::Ok {
Ok(reply)
} else {
Err(Box::new(CommandError::new(reply)).into())
}
}
fn redact_credentials(message: &str) -> String {
message
.split_whitespace()
.map(|part| {
if part.starts_with("mxgw_") || part.eq_ignore_ascii_case("bearer") {
"<redacted>"
} else {
part
}
})
.collect::<Vec<_>>()
.join(" ")
}
#[cfg(test)]
mod tests {
use tonic::{Code, Status};
use super::Error;
#[test]
fn classifies_authentication_status() {
let error = Error::from(Status::new(
Code::Unauthenticated,
"invalid API key mxgw_visible_secret",
));
let message = error.to_string();
assert!(matches!(error, Error::Authentication { .. }));
assert!(message.contains("<redacted>"));
assert!(!message.contains("visible_secret"));
}
}
+15
View File
@@ -0,0 +1,15 @@
pub mod mxaccess_gateway {
pub mod v1 {
#![allow(clippy::large_enum_variant)]
tonic::include_proto!("mxaccess_gateway.v1");
}
}
pub mod mxaccess_worker {
pub mod v1 {
#![allow(clippy::large_enum_variant)]
tonic::include_proto!("mxaccess_worker.v1");
}
}
+22
View File
@@ -0,0 +1,22 @@
//! Rust client scaffold for MXAccess Gateway.
//!
//! The crate compiles generated `tonic` bindings from the shared gateway
//! protobuf contracts and exposes a small handwritten surface for future client
//! implementation work.
pub mod auth;
pub mod client;
pub mod error;
pub mod generated;
pub mod options;
pub mod session;
pub mod value;
pub mod version;
pub use auth::{ApiKey, AuthInterceptor};
pub use client::{EventStream, GatewayClient};
pub use error::{CommandError, Error};
pub use options::ClientOptions;
pub use session::Session;
pub use value::{MxArrayProjection, MxArrayValue, MxStatus, MxValue, MxValueProjection};
pub use version::{CLIENT_VERSION, GATEWAY_PROTOCOL_VERSION, WORKER_PROTOCOL_VERSION};
+126
View File
@@ -0,0 +1,126 @@
use std::fmt;
use std::path::PathBuf;
use std::time::Duration;
use crate::auth::ApiKey;
#[derive(Clone)]
pub struct ClientOptions {
endpoint: String,
api_key: Option<ApiKey>,
plaintext: bool,
ca_file: Option<PathBuf>,
server_name_override: Option<String>,
connect_timeout: Duration,
call_timeout: Duration,
}
impl ClientOptions {
pub fn new(endpoint: impl Into<String>) -> Self {
Self {
endpoint: endpoint.into(),
api_key: None,
plaintext: true,
ca_file: None,
server_name_override: None,
connect_timeout: Duration::from_secs(10),
call_timeout: Duration::from_secs(30),
}
}
pub fn with_api_key(mut self, api_key: ApiKey) -> Self {
self.api_key = Some(api_key);
self
}
pub fn with_plaintext(mut self, plaintext: bool) -> Self {
self.plaintext = plaintext;
self
}
pub fn with_ca_file(mut self, ca_file: impl Into<PathBuf>) -> Self {
self.ca_file = Some(ca_file.into());
self
}
pub fn with_server_name_override(mut self, server_name_override: impl Into<String>) -> Self {
self.server_name_override = Some(server_name_override.into());
self
}
pub fn with_connect_timeout(mut self, connect_timeout: Duration) -> Self {
self.connect_timeout = connect_timeout;
self
}
pub fn with_call_timeout(mut self, call_timeout: Duration) -> Self {
self.call_timeout = call_timeout;
self
}
pub fn endpoint(&self) -> &str {
&self.endpoint
}
pub fn api_key(&self) -> Option<&ApiKey> {
self.api_key.as_ref()
}
pub fn plaintext(&self) -> bool {
self.plaintext
}
pub fn ca_file(&self) -> Option<&PathBuf> {
self.ca_file.as_ref()
}
pub fn server_name_override(&self) -> Option<&str> {
self.server_name_override.as_deref()
}
pub fn connect_timeout(&self) -> Duration {
self.connect_timeout
}
pub fn call_timeout(&self) -> Duration {
self.call_timeout
}
}
impl Default for ClientOptions {
fn default() -> Self {
Self::new("http://127.0.0.1:5000")
}
}
impl fmt::Debug for ClientOptions {
fn fmt(&self, formatter: &mut fmt::Formatter<'_>) -> fmt::Result {
formatter
.debug_struct("ClientOptions")
.field("endpoint", &self.endpoint)
.field("api_key", &self.api_key.as_ref().map(|_| "<redacted>"))
.field("plaintext", &self.plaintext)
.field("ca_file", &self.ca_file)
.field("server_name_override", &self.server_name_override)
.field("connect_timeout", &self.connect_timeout)
.field("call_timeout", &self.call_timeout)
.finish()
}
}
#[cfg(test)]
mod tests {
use super::ClientOptions;
use crate::auth::ApiKey;
#[test]
fn debug_redacts_api_key() {
let options =
ClientOptions::new("http://localhost:5000").with_api_key(ApiKey::new("mxgw_secret"));
let debug = format!("{options:?}");
assert!(debug.contains("<redacted>"));
assert!(!debug.contains("mxgw_secret"));
}
}
+234
View File
@@ -0,0 +1,234 @@
use crate::client::{EventStream, GatewayClient};
use crate::error::Error;
use crate::generated::mxaccess_gateway::v1::mx_command::Payload;
use crate::generated::mxaccess_gateway::v1::mx_command_reply;
use crate::generated::mxaccess_gateway::v1::{
AddItem2Command, AddItemCommand, AdviseCommand, CloseSessionRequest, MxCommand, MxCommandKind,
MxCommandReply, MxCommandRequest, MxValue as ProtoMxValue, OpenSessionRequest, RegisterCommand,
StreamEventsRequest, Write2Command, WriteCommand,
};
use crate::value::MxValue;
/// Session identifier returned by the gateway.
#[derive(Clone)]
pub struct Session {
id: String,
client: GatewayClient,
}
impl Session {
pub(crate) fn new(id: impl Into<String>, client: GatewayClient) -> Self {
Self {
id: id.into(),
client,
}
}
pub fn id(&self) -> &str {
&self.id
}
pub async fn open(client: GatewayClient, client_session_name: &str) -> Result<Self, Error> {
client
.open_session(OpenSessionRequest {
client_session_name: client_session_name.to_owned(),
..OpenSessionRequest::default()
})
.await
}
pub async fn close(&self) -> Result<(), Error> {
self.client
.close_session_raw(CloseSessionRequest {
session_id: self.id.clone(),
client_correlation_id: "rust-client-close-session".to_owned(),
})
.await?;
Ok(())
}
pub async fn register(&self, client_name: &str) -> Result<i32, Error> {
let reply = self
.invoke(
MxCommandKind::Register,
Payload::Register(RegisterCommand {
client_name: client_name.to_owned(),
}),
)
.await?;
Ok(register_server_handle(&reply))
}
pub async fn add_item(&self, server_handle: i32, item_definition: &str) -> Result<i32, Error> {
let reply = self
.invoke(
MxCommandKind::AddItem,
Payload::AddItem(AddItemCommand {
server_handle,
item_definition: item_definition.to_owned(),
}),
)
.await?;
Ok(add_item_handle(&reply))
}
pub async fn add_item2(
&self,
server_handle: i32,
item_definition: &str,
item_context: &str,
) -> Result<i32, Error> {
let reply = self
.invoke(
MxCommandKind::AddItem2,
Payload::AddItem2(AddItem2Command {
server_handle,
item_definition: item_definition.to_owned(),
item_context: item_context.to_owned(),
}),
)
.await?;
Ok(add_item2_handle(&reply))
}
pub async fn advise(&self, server_handle: i32, item_handle: i32) -> Result<(), Error> {
self.invoke(
MxCommandKind::Advise,
Payload::Advise(AdviseCommand {
server_handle,
item_handle,
}),
)
.await?;
Ok(())
}
pub async fn write(
&self,
server_handle: i32,
item_handle: i32,
value: MxValue,
user_id: i32,
) -> Result<(), Error> {
self.invoke(
MxCommandKind::Write,
Payload::Write(WriteCommand {
server_handle,
item_handle,
value: Some(value.into_proto()),
user_id,
}),
)
.await?;
Ok(())
}
pub async fn write2(
&self,
server_handle: i32,
item_handle: i32,
value: MxValue,
timestamp_value: MxValue,
user_id: i32,
) -> Result<(), Error> {
self.invoke(
MxCommandKind::Write2,
Payload::Write2(Write2Command {
server_handle,
item_handle,
value: Some(value.into_proto()),
timestamp_value: Some(timestamp_value.into_proto()),
user_id,
}),
)
.await?;
Ok(())
}
pub async fn events(&self) -> Result<EventStream, Error> {
self.events_after(0).await
}
pub async fn events_after(&self, after_worker_sequence: u64) -> Result<EventStream, Error> {
self.client
.stream_events(StreamEventsRequest {
session_id: self.id.clone(),
after_worker_sequence,
})
.await
}
pub async fn invoke_raw(
&self,
kind: MxCommandKind,
payload: Payload,
) -> Result<MxCommandReply, Error> {
self.client
.invoke_raw(self.command_request(kind, payload))
.await
}
pub async fn invoke(
&self,
kind: MxCommandKind,
payload: Payload,
) -> Result<MxCommandReply, Error> {
self.client
.invoke(self.command_request(kind, payload))
.await
}
fn command_request(&self, kind: MxCommandKind, payload: Payload) -> MxCommandRequest {
MxCommandRequest {
session_id: self.id.clone(),
client_correlation_id: format!("rust-client-{}", kind.as_str_name()),
command: Some(MxCommand {
kind: kind as i32,
payload: Some(payload),
}),
}
}
}
fn register_server_handle(reply: &MxCommandReply) -> i32 {
match reply.payload.as_ref() {
Some(mx_command_reply::Payload::Register(register)) => register.server_handle,
_ => reply
.return_value
.as_ref()
.and_then(int32_reply_value)
.unwrap_or_default(),
}
}
fn add_item_handle(reply: &MxCommandReply) -> i32 {
match reply.payload.as_ref() {
Some(mx_command_reply::Payload::AddItem(add_item)) => add_item.item_handle,
_ => reply
.return_value
.as_ref()
.and_then(int32_reply_value)
.unwrap_or_default(),
}
}
fn add_item2_handle(reply: &MxCommandReply) -> i32 {
match reply.payload.as_ref() {
Some(mx_command_reply::Payload::AddItem2(add_item)) => add_item.item_handle,
_ => reply
.return_value
.as_ref()
.and_then(int32_reply_value)
.unwrap_or_default(),
}
}
fn int32_reply_value(value: &ProtoMxValue) -> Option<i32> {
match value.kind.as_ref()? {
crate::generated::mxaccess_gateway::v1::mx_value::Kind::Int32Value(value) => Some(*value),
_ => None,
}
}
+239
View File
@@ -0,0 +1,239 @@
use crate::generated::mxaccess_gateway::v1::mx_array::Values;
use crate::generated::mxaccess_gateway::v1::mx_value::Kind;
use crate::generated::mxaccess_gateway::v1::{
BoolArray, DoubleArray, FloatArray, Int32Array, Int64Array, MxArray, MxDataType,
MxStatusCategory, MxStatusProxy, MxStatusSource, MxValue as ProtoMxValue, RawArray,
StringArray, TimestampArray,
};
#[derive(Clone, Debug, PartialEq)]
pub struct MxValue {
raw: ProtoMxValue,
projection: MxValueProjection,
}
impl MxValue {
pub fn from_proto(raw: ProtoMxValue) -> Self {
let projection = MxValueProjection::from_proto(&raw);
Self { raw, projection }
}
pub fn bool(value: bool) -> Self {
Self::from_proto(ProtoMxValue {
data_type: MxDataType::Boolean as i32,
variant_type: "VT_BOOL".to_owned(),
kind: Some(Kind::BoolValue(value)),
..ProtoMxValue::default()
})
}
pub fn int32(value: i32) -> Self {
Self::from_proto(ProtoMxValue {
data_type: MxDataType::Integer as i32,
variant_type: "VT_I4".to_owned(),
kind: Some(Kind::Int32Value(value)),
..ProtoMxValue::default()
})
}
pub fn int64(value: i64) -> Self {
Self::from_proto(ProtoMxValue {
data_type: MxDataType::Integer as i32,
variant_type: "VT_I8".to_owned(),
kind: Some(Kind::Int64Value(value)),
..ProtoMxValue::default()
})
}
pub fn float(value: f32) -> Self {
Self::from_proto(ProtoMxValue {
data_type: MxDataType::Float as i32,
variant_type: "VT_R4".to_owned(),
kind: Some(Kind::FloatValue(value)),
..ProtoMxValue::default()
})
}
pub fn double(value: f64) -> Self {
Self::from_proto(ProtoMxValue {
data_type: MxDataType::Double as i32,
variant_type: "VT_R8".to_owned(),
kind: Some(Kind::DoubleValue(value)),
..ProtoMxValue::default()
})
}
pub fn string(value: impl Into<String>) -> Self {
Self::from_proto(ProtoMxValue {
data_type: MxDataType::String as i32,
variant_type: "VT_BSTR".to_owned(),
kind: Some(Kind::StringValue(value.into())),
..ProtoMxValue::default()
})
}
pub fn raw(&self) -> &ProtoMxValue {
&self.raw
}
pub fn projection(&self) -> &MxValueProjection {
&self.projection
}
pub fn into_proto(self) -> ProtoMxValue {
self.raw
}
}
impl From<MxValue> for ProtoMxValue {
fn from(value: MxValue) -> Self {
value.into_proto()
}
}
impl From<ProtoMxValue> for MxValue {
fn from(value: ProtoMxValue) -> Self {
Self::from_proto(value)
}
}
#[derive(Clone, Debug, PartialEq)]
pub enum MxValueProjection {
Unset,
Null,
Bool(bool),
Int32(i32),
Int64(i64),
Float(f32),
Double(f64),
String(String),
Timestamp(prost_types::Timestamp),
Array(MxArrayValue),
Raw(Vec<u8>),
}
impl MxValueProjection {
fn from_proto(value: &ProtoMxValue) -> Self {
if value.is_null {
return Self::Null;
}
match value.kind.as_ref() {
Some(Kind::BoolValue(value)) => Self::Bool(*value),
Some(Kind::Int32Value(value)) => Self::Int32(*value),
Some(Kind::Int64Value(value)) => Self::Int64(*value),
Some(Kind::FloatValue(value)) => Self::Float(*value),
Some(Kind::DoubleValue(value)) => Self::Double(*value),
Some(Kind::StringValue(value)) => Self::String(value.clone()),
Some(Kind::TimestampValue(value)) => Self::Timestamp(*value),
Some(Kind::ArrayValue(value)) => Self::Array(MxArrayValue::from_proto(value.clone())),
Some(Kind::RawValue(value)) => Self::Raw(value.clone()),
None => Self::Unset,
}
}
}
#[derive(Clone, Debug, PartialEq)]
pub struct MxArrayValue {
raw: MxArray,
projection: MxArrayProjection,
}
impl MxArrayValue {
pub fn from_proto(raw: MxArray) -> Self {
let projection = MxArrayProjection::from_proto(&raw);
Self { raw, projection }
}
pub fn string(values: Vec<String>) -> Self {
Self::from_proto(MxArray {
element_data_type: MxDataType::String as i32,
variant_type: "VT_ARRAY|VT_BSTR".to_owned(),
dimensions: vec![values.len() as u32],
values: Some(Values::StringValues(StringArray { values })),
..MxArray::default()
})
}
pub fn raw(&self) -> &MxArray {
&self.raw
}
pub fn projection(&self) -> &MxArrayProjection {
&self.projection
}
}
#[derive(Clone, Debug, PartialEq)]
pub enum MxArrayProjection {
Unset,
Bool(Vec<bool>),
Int32(Vec<i32>),
Int64(Vec<i64>),
Float(Vec<f32>),
Double(Vec<f64>),
String(Vec<String>),
Timestamp(Vec<prost_types::Timestamp>),
Raw(Vec<Vec<u8>>),
}
impl MxArrayProjection {
fn from_proto(array: &MxArray) -> Self {
match array.values.as_ref() {
Some(Values::BoolValues(BoolArray { values })) => Self::Bool(values.clone()),
Some(Values::Int32Values(Int32Array { values })) => Self::Int32(values.clone()),
Some(Values::Int64Values(Int64Array { values })) => Self::Int64(values.clone()),
Some(Values::FloatValues(FloatArray { values })) => Self::Float(values.clone()),
Some(Values::DoubleValues(DoubleArray { values })) => Self::Double(values.clone()),
Some(Values::StringValues(StringArray { values })) => Self::String(values.clone()),
Some(Values::TimestampValues(TimestampArray { values })) => {
Self::Timestamp(values.clone())
}
Some(Values::RawValues(RawArray { values })) => Self::Raw(values.clone()),
None => Self::Unset,
}
}
}
#[derive(Clone, Debug, PartialEq)]
pub struct MxStatus {
raw: MxStatusProxy,
}
impl MxStatus {
pub fn from_proto(raw: MxStatusProxy) -> Self {
Self { raw }
}
pub fn raw(&self) -> &MxStatusProxy {
&self.raw
}
pub fn success(&self) -> i32 {
self.raw.success
}
pub fn category(&self) -> Option<MxStatusCategory> {
MxStatusCategory::try_from(self.raw.category).ok()
}
pub fn detected_by(&self) -> Option<MxStatusSource> {
MxStatusSource::try_from(self.raw.detected_by).ok()
}
pub fn detail(&self) -> i32 {
self.raw.detail
}
pub fn raw_category(&self) -> i32 {
self.raw.raw_category
}
pub fn raw_detected_by(&self) -> i32 {
self.raw.raw_detected_by
}
pub fn diagnostic_text(&self) -> &str {
&self.raw.diagnostic_text
}
}
+3
View File
@@ -0,0 +1,3 @@
pub const CLIENT_VERSION: &str = "0.1.0-dev";
pub const GATEWAY_PROTOCOL_VERSION: u32 = 1;
pub const WORKER_PROTOCOL_VERSION: u32 = 1;
+398
View File
@@ -0,0 +1,398 @@
use std::pin::Pin;
use std::sync::{
atomic::{AtomicBool, Ordering},
Arc,
};
use std::task::{Context, Poll};
use std::time::Duration;
use futures_core::Stream;
use futures_util::StreamExt;
use mxgateway_client::generated::mxaccess_gateway::v1::mx_access_gateway_server::{
MxAccessGateway, MxAccessGatewayServer,
};
use mxgateway_client::generated::mxaccess_gateway::v1::mx_command_reply;
use mxgateway_client::generated::mxaccess_gateway::v1::mx_value::Kind;
use mxgateway_client::generated::mxaccess_gateway::v1::{
AddItemReply, CloseSessionReply, CloseSessionRequest, MxCommandKind, MxCommandReply,
MxDataType, MxEvent, MxEventFamily, MxStatusCategory, MxStatusProxy, MxStatusSource, MxValue,
OpenSessionReply, OpenSessionRequest, ProtocolStatus, ProtocolStatusCode, SessionState,
StreamEventsRequest,
};
use mxgateway_client::{
ApiKey, ClientOptions, CommandError, Error, GatewayClient, MxStatus, MxValue as ClientMxValue,
MxValueProjection,
};
use serde_json::Value;
use tokio::net::TcpListener;
use tokio::sync::{mpsc, Mutex};
use tokio_stream::wrappers::{ReceiverStream, TcpListenerStream};
use tonic::transport::Server;
use tonic::{Request, Response, Status};
#[tokio::test]
async fn fake_server_receives_bearer_metadata_and_raw_client_is_reachable() {
let state = Arc::new(FakeState::default());
let endpoint = spawn_fake_gateway(state.clone()).await;
let mut client = GatewayClient::connect(
ClientOptions::new(endpoint).with_api_key(ApiKey::new("mxgw_fixture_secret")),
)
.await
.unwrap();
let _raw = client.raw_client();
let session = client
.open_session(OpenSessionRequest {
client_session_name: "rust-test".to_owned(),
..OpenSessionRequest::default()
})
.await
.unwrap();
assert_eq!(session.id(), "session-fixture");
assert_eq!(
state.authorization.lock().await.as_deref(),
Some("Bearer mxgw_fixture_secret")
);
}
#[tokio::test]
async fn session_helpers_build_commands_and_preserve_command_errors() {
let state = Arc::new(FakeState::default());
let endpoint = spawn_fake_gateway(state.clone()).await;
let client = GatewayClient::connect(ClientOptions::new(endpoint))
.await
.unwrap();
let session = client.session("session-fixture");
let item_handle = session.add_item(12, "Plant.Area.Tag").await.unwrap();
assert_eq!(item_handle, 34);
let last_command = state.last_command_kind.lock().await;
assert_eq!(*last_command, Some(MxCommandKind::AddItem as i32));
drop(last_command);
let error = session
.write(12, 34, ClientMxValue::int32(123), 0)
.await
.unwrap_err();
let Error::Command(error) = error else {
panic!("write failure should preserve the raw command reply: {error:?}");
};
assert_eq!(
error.reply().protocol_status.as_ref().unwrap().code,
ProtocolStatusCode::MxaccessFailure as i32
);
assert_eq!(error.reply().hresult, Some(-2147220992));
assert_eq!(error.reply().statuses.len(), 2);
}
#[tokio::test]
async fn event_stream_preserves_order_and_drop_cancels_server_stream() {
let state = Arc::new(FakeState::default());
let endpoint = spawn_fake_gateway(state.clone()).await;
let client = GatewayClient::connect(ClientOptions::new(endpoint))
.await
.unwrap();
let mut stream = client
.stream_events(StreamEventsRequest {
session_id: "session-fixture".to_owned(),
after_worker_sequence: 0,
})
.await
.unwrap();
assert_eq!(stream.next().await.unwrap().unwrap().worker_sequence, 1);
assert_eq!(stream.next().await.unwrap().unwrap().worker_sequence, 2);
drop(stream);
for _ in 0..20 {
if state.stream_dropped.load(Ordering::SeqCst) {
return;
}
tokio::time::sleep(Duration::from_millis(25)).await;
}
assert!(state.stream_dropped.load(Ordering::SeqCst));
}
#[test]
fn value_conversion_fixtures_keep_typed_projection_and_raw_metadata() {
let fixture = behavior_fixture("values/value-conversion-cases.json");
let cases = fixture["cases"].as_array().unwrap();
let int64_case = case_by_id(cases, "int64.large");
let int64_value = ClientMxValue::from_proto(MxValue {
data_type: MxDataType::Integer as i32,
variant_type: "VT_I8".to_owned(),
kind: Some(Kind::Int64Value(
int64_case["value"]["int64Value"]
.as_str()
.unwrap()
.parse()
.unwrap(),
)),
..MxValue::default()
});
assert_eq!(
int64_value.projection(),
&MxValueProjection::Int64(9_223_372_036_854_770_000)
);
let raw_case = case_by_id(cases, "raw-fallback.variant");
let raw_value = ClientMxValue::from_proto(MxValue {
data_type: MxDataType::Unknown as i32,
variant_type: "VT_RECORD".to_owned(),
raw_diagnostic: raw_case["value"]["rawDiagnostic"]
.as_str()
.unwrap()
.to_owned(),
raw_data_type: raw_case["value"]["rawDataType"].as_i64().unwrap() as i32,
kind: Some(Kind::RawValue(vec![1, 2, 3, 4, 5])),
..MxValue::default()
});
assert_eq!(
raw_value.projection(),
&MxValueProjection::Raw(vec![1, 2, 3, 4, 5])
);
assert_eq!(raw_value.raw().raw_data_type, 32767);
assert!(raw_value.raw().raw_diagnostic.contains("No lossless"));
}
#[test]
fn status_conversion_fixtures_preserve_raw_fields() {
let fixture = behavior_fixture("statuses/status-conversion-cases.json");
let cases = fixture["cases"].as_array().unwrap();
let raw_case = case_by_id(cases, "raw-unknown-category");
let status = MxStatus::from_proto(MxStatusProxy {
success: raw_case["status"]["success"].as_i64().unwrap() as i32,
category: MxStatusCategory::Unknown as i32,
detected_by: MxStatusSource::Unknown as i32,
detail: raw_case["status"]["detail"].as_i64().unwrap() as i32,
raw_category: raw_case["status"]["rawCategory"].as_i64().unwrap() as i32,
raw_detected_by: raw_case["status"]["rawDetectedBy"].as_i64().unwrap() as i32,
diagnostic_text: raw_case["status"]["diagnosticText"]
.as_str()
.unwrap()
.to_owned(),
});
assert_eq!(status.success(), 0);
assert_eq!(status.category(), Some(MxStatusCategory::Unknown));
assert_eq!(status.raw_category(), 99);
assert_eq!(status.raw_detected_by(), 77);
assert!(status.diagnostic_text().contains("preserved"));
}
#[test]
fn authentication_and_authorization_statuses_are_distinct_and_redacted() {
let auth = Error::from(Status::unauthenticated(
"invalid API key mxgw_visible_secret",
));
let denied = Error::from(Status::permission_denied("missing scope mxaccess.write"));
assert!(matches!(auth, Error::Authentication { .. }));
assert!(matches!(denied, Error::Authorization { .. }));
assert!(!auth.to_string().contains("visible_secret"));
}
#[test]
fn command_error_display_keeps_raw_reply_accessible() {
let reply = mxaccess_failure_reply();
let error = CommandError::new(reply.clone());
assert_eq!(error.reply().hresult, Some(-2147220992));
assert!(error.to_string().contains("MxaccessFailure"));
}
#[derive(Default)]
struct FakeState {
authorization: Mutex<Option<String>>,
last_command_kind: Mutex<Option<i32>>,
stream_dropped: Arc<AtomicBool>,
}
#[derive(Clone)]
struct FakeGateway {
state: Arc<FakeState>,
}
#[tonic::async_trait]
impl MxAccessGateway for FakeGateway {
async fn open_session(
&self,
request: Request<OpenSessionRequest>,
) -> Result<Response<OpenSessionReply>, Status> {
*self.state.authorization.lock().await = request
.metadata()
.get("authorization")
.and_then(|value| value.to_str().ok())
.map(str::to_owned);
Ok(Response::new(OpenSessionReply {
session_id: "session-fixture".to_owned(),
backend_name: "fake".to_owned(),
worker_process_id: 1234,
worker_protocol_version: 1,
gateway_protocol_version: 1,
protocol_status: Some(ok_status("opened")),
..OpenSessionReply::default()
}))
}
async fn close_session(
&self,
request: Request<CloseSessionRequest>,
) -> Result<Response<CloseSessionReply>, Status> {
Ok(Response::new(CloseSessionReply {
session_id: request.into_inner().session_id,
final_state: SessionState::Closed as i32,
protocol_status: Some(ok_status("closed")),
}))
}
async fn invoke(
&self,
request: Request<mxgateway_client::generated::mxaccess_gateway::v1::MxCommandRequest>,
) -> Result<Response<MxCommandReply>, Status> {
let request = request.into_inner();
let kind = request
.command
.as_ref()
.map(|command| command.kind)
.unwrap_or_default();
*self.state.last_command_kind.lock().await = Some(kind);
if kind == MxCommandKind::Write as i32 {
return Ok(Response::new(mxaccess_failure_reply()));
}
Ok(Response::new(MxCommandReply {
session_id: request.session_id,
correlation_id: "fake-correlation".to_owned(),
kind,
protocol_status: Some(ok_status("command ok")),
payload: Some(mx_command_reply::Payload::AddItem(AddItemReply {
item_handle: 34,
})),
..MxCommandReply::default()
}))
}
type StreamEventsStream = DropAwareStream;
async fn stream_events(
&self,
_request: Request<StreamEventsRequest>,
) -> Result<Response<Self::StreamEventsStream>, Status> {
let (sender, receiver) = mpsc::channel(4);
sender.send(Ok(event(1))).await.unwrap();
sender.send(Ok(event(2))).await.unwrap();
Ok(Response::new(DropAwareStream {
inner: ReceiverStream::new(receiver),
dropped: self.state.stream_dropped.clone(),
}))
}
}
struct DropAwareStream {
inner: ReceiverStream<Result<MxEvent, Status>>,
dropped: Arc<AtomicBool>,
}
impl Stream for DropAwareStream {
type Item = Result<MxEvent, Status>;
fn poll_next(mut self: Pin<&mut Self>, context: &mut Context<'_>) -> Poll<Option<Self::Item>> {
Pin::new(&mut self.inner).poll_next(context)
}
}
impl Drop for DropAwareStream {
fn drop(&mut self) {
self.dropped.store(true, Ordering::SeqCst);
}
}
async fn spawn_fake_gateway(state: Arc<FakeState>) -> String {
let listener = TcpListener::bind("127.0.0.1:0").await.unwrap();
let address = listener.local_addr().unwrap();
let incoming = TcpListenerStream::new(listener);
let service = MxAccessGatewayServer::new(FakeGateway { state });
tokio::spawn(async move {
Server::builder()
.add_service(service)
.serve_with_incoming(incoming)
.await
.unwrap();
});
format!("http://{address}")
}
fn ok_status(message: &str) -> ProtocolStatus {
ProtocolStatus {
code: ProtocolStatusCode::Ok as i32,
message: message.to_owned(),
}
}
fn mxaccess_failure_reply() -> MxCommandReply {
MxCommandReply {
session_id: "session-fixture".to_owned(),
correlation_id: "gateway-correlation-write-1".to_owned(),
kind: MxCommandKind::Write as i32,
protocol_status: Some(ProtocolStatus {
code: ProtocolStatusCode::MxaccessFailure as i32,
message: "MXAccess rejected the write.".to_owned(),
}),
hresult: Some(-2147220992),
statuses: vec![
MxStatusProxy {
success: 0,
category: MxStatusCategory::SecurityError as i32,
detected_by: MxStatusSource::RespondingLmx as i32,
detail: 321,
raw_category: 8,
raw_detected_by: 3,
diagnostic_text: "Write denied by provider security.".to_owned(),
},
MxStatusProxy {
success: 0,
category: MxStatusCategory::OperationalError as i32,
detected_by: MxStatusSource::RespondingNmx as i32,
detail: 902,
raw_category: 7,
raw_detected_by: 5,
diagnostic_text: "Provider rejected the item state.".to_owned(),
},
],
..MxCommandReply::default()
}
}
fn event(sequence: u64) -> MxEvent {
MxEvent {
family: MxEventFamily::OnDataChange as i32,
session_id: "session-fixture".to_owned(),
worker_sequence: sequence,
..MxEvent::default()
}
}
fn behavior_fixture(path: &str) -> Value {
let path = std::path::Path::new(env!("CARGO_MANIFEST_DIR"))
.join("../proto/fixtures/behavior")
.join(path);
let data = std::fs::read_to_string(&path).unwrap();
serde_json::from_str(&data).unwrap()
}
fn case_by_id<'a>(cases: &'a [Value], id: &str) -> &'a Value {
cases
.iter()
.find(|case| case["id"].as_str() == Some(id))
.unwrap_or_else(|| panic!("missing fixture case {id}"))
}
+144
View File
@@ -0,0 +1,144 @@
use std::fs;
use std::path::PathBuf;
use mxgateway_client::generated::mxaccess_gateway::v1::{
mx_command, mx_value, MxCommand, MxCommandKind, MxCommandRequest, MxDataType, MxEvent,
MxEventFamily, MxValue, OpenSessionReply, ProtocolStatusCode, RegisterCommand,
};
use mxgateway_client::{GATEWAY_PROTOCOL_VERSION, WORKER_PROTOCOL_VERSION};
use serde_json::Value;
#[test]
fn generated_golden_fixtures_are_available() {
for fixture_name in [
"open-session-reply.ok.json",
"register-command-request.json",
"on-data-change-event.json",
] {
let fixture = read_fixture(fixture_name);
assert!(
fixture.is_object(),
"{fixture_name} must remain a protobuf JSON object"
);
}
}
#[test]
fn open_session_fixture_matches_protocol_versions() {
let fixture = read_fixture("open-session-reply.ok.json");
let reply = OpenSessionReply {
session_id: string_field(&fixture, "sessionId"),
backend_name: string_field(&fixture, "backendName"),
worker_process_id: i32_field(&fixture, "workerProcessId"),
worker_protocol_version: u32_field(&fixture, "workerProtocolVersion"),
gateway_protocol_version: u32_field(&fixture, "gatewayProtocolVersion"),
protocol_status: Some(
mxgateway_client::generated::mxaccess_gateway::v1::ProtocolStatus {
code: ProtocolStatusCode::Ok as i32,
message: string_field(&fixture["protocolStatus"], "message"),
},
),
..OpenSessionReply::default()
};
assert_eq!(reply.gateway_protocol_version, GATEWAY_PROTOCOL_VERSION);
assert_eq!(reply.worker_protocol_version, WORKER_PROTOCOL_VERSION);
}
#[test]
fn register_fixture_can_build_generated_request() {
let fixture = read_fixture("register-command-request.json");
let command = &fixture["command"];
let request = MxCommandRequest {
session_id: string_field(&fixture, "sessionId"),
client_correlation_id: string_field(&fixture, "clientCorrelationId"),
command: Some(MxCommand {
kind: MxCommandKind::Register as i32,
payload: Some(mx_command::Payload::Register(RegisterCommand {
client_name: string_field(&command["register"], "clientName"),
})),
}),
};
assert_eq!(request.session_id, "session-fixture");
assert_eq!(
request.command.unwrap().kind,
MxCommandKind::Register as i32
);
}
#[test]
fn on_data_change_fixture_can_build_generated_event() {
let fixture = read_fixture("on-data-change-event.json");
let event = MxEvent {
family: MxEventFamily::OnDataChange as i32,
session_id: string_field(&fixture, "sessionId"),
server_handle: i32_field(&fixture, "serverHandle"),
item_handle: i32_field(&fixture, "itemHandle"),
value: Some(MxValue {
data_type: MxDataType::Integer as i32,
variant_type: string_field(&fixture["value"], "variantType"),
kind: Some(mx_value::Kind::Int32Value(i32_field(
&fixture["value"],
"int32Value",
))),
..MxValue::default()
}),
quality: i32_field(&fixture, "quality"),
worker_sequence: u64_field(&fixture, "workerSequence"),
..MxEvent::default()
};
assert_eq!(event.family, MxEventFamily::OnDataChange as i32);
assert_eq!(event.value.unwrap().data_type, MxDataType::Integer as i32);
}
fn read_fixture(name: &str) -> Value {
let path = fixture_root().join(name);
let data = fs::read_to_string(&path).unwrap_or_else(|error| {
panic!("failed to read fixture {}: {error}", path.display());
});
serde_json::from_str(&data).unwrap_or_else(|error| {
panic!("failed to parse fixture {}: {error}", path.display());
})
}
fn fixture_root() -> PathBuf {
PathBuf::from(env!("CARGO_MANIFEST_DIR")).join("../proto/fixtures/golden")
}
fn string_field(value: &Value, name: &str) -> String {
value[name]
.as_str()
.unwrap_or_else(|| panic!("missing string field {name}"))
.to_owned()
}
fn i32_field(value: &Value, name: &str) -> i32 {
value[name]
.as_i64()
.unwrap_or_else(|| panic!("missing i32 field {name}"))
.try_into()
.unwrap_or_else(|_| panic!("field {name} does not fit in i32"))
}
fn u32_field(value: &Value, name: &str) -> u32 {
value[name]
.as_u64()
.unwrap_or_else(|| panic!("missing u32 field {name}"))
.try_into()
.unwrap_or_else(|_| panic!("field {name} does not fit in u32"))
}
fn u64_field(value: &Value, name: &str) -> u64 {
if let Some(number) = value[name].as_u64() {
return number;
}
value[name]
.as_str()
.unwrap_or_else(|| panic!("missing u64 field {name}"))
.parse()
.unwrap_or_else(|_| panic!("field {name} does not parse as u64"))
}
+39
View File
@@ -35,6 +35,45 @@ inside the test.
`OpenSession`, `Register`, `AddItem`, `Advise`, one streamed `OnDataChange`
event, and `CloseSession` without loading MXAccess COM.
## Live MXAccess Smoke
`WorkerLiveMxAccessSmokeTests` in `src/MxGateway.IntegrationTests/` composes the
real gRPC service, `SessionManager`, `SessionWorkerClientFactory`,
`WorkerClient`, `WorkerProcessLauncher`, and `MxGateway.Worker.exe`. It is
skipped unless `MXGATEWAY_RUN_LIVE_MXACCESS_TESTS=1` is set because it creates
the installed MXAccess COM object and depends on live provider state.
The live smoke opens a gateway session, launches the x86 worker, runs
`Register`, `AddItem`, and `Advise`, waits a bounded time for one
`OnDataChange`, and closes the session in a `finally` block so the worker gets a
graceful shutdown request even when a command or event assertion fails.
Build the worker before running the smoke:
```bash
dotnet build src/MxGateway.Worker/MxGateway.Worker.csproj -p:Platform=x86
```
Run the smoke explicitly:
```bash
$env:MXGATEWAY_RUN_LIVE_MXACCESS_TESTS = "1"
dotnet test src/MxGateway.IntegrationTests/MxGateway.IntegrationTests.csproj --filter FullyQualifiedName~WorkerLiveMxAccessSmokeTests
```
Optional live smoke variables:
| Variable | Default | Description |
|----------|---------|-------------|
| `MXGATEWAY_LIVE_MXACCESS_WORKER_EXE` | First existing `MxGateway.Worker.exe` under `src/MxGateway.Worker/bin/...` | Worker executable path. Set this when running against a packaged worker or a non-default build output. |
| `MXGATEWAY_LIVE_MXACCESS_ITEM` | `TestChildObject.TestInt` | MXAccess item reference used by `AddItem`. |
| `MXGATEWAY_LIVE_MXACCESS_CLIENT_NAME` | `MxGateway.IntegrationTests` | Client name passed to `Register`. |
| `MXGATEWAY_LIVE_MXACCESS_EVENT_TIMEOUT_SECONDS` | `15` | Maximum wait for the first `OnDataChange`. |
The test output includes session id, worker process id, command status,
HRESULT/status diagnostics, event sequence and handles, close status, and worker
stdout/stderr lines emitted during the run.
## Focused Commands
Run the fake worker tests after changing gateway worker IPC, session startup, or
+32 -7
View File
@@ -111,18 +111,43 @@ The script maps both proto files into the internal Go package
the source `.proto` files do not carry Go-specific `go_package` options. This
keeps language-specific packaging outside the public contract files.
Rust clients should use `tonic-build` or the selected protobuf generator from
the Rust client build script, with generated modules placed under
`clients/rust/src/generated` or included from the build output according to the
client crate design.
Rust clients use `tonic-build` from `clients/rust/build.rs`. The build script
reads the shared `.proto` files and emits generated `tonic`/`prost` modules
into Cargo build output. `clients/rust/src/generated.rs` contains the module
declarations that include those generated files. `clients/rust/src/generated`
remains reserved for checked-in generator output if the crate later changes to
source-tree generation, and handwritten wrapper code stays outside that
directory.
Run the Rust workspace checks from `clients/rust`:
```powershell
cargo fmt --all --check
cargo test --workspace
cargo check --workspace
```
Python clients should use `grpc_tools.protoc` and write generated modules under
`clients/python/src/mxgateway/generated` so imports stay separate from
handwritten async wrappers.
Java clients should use the Gradle protobuf plugin and write generated sources
under `clients/java/src/main/generated`. The Java client scaffold owns the
Gradle plugin versions and source-set wiring.
The Python scaffold provides a repo-local generation script:
```powershell
clients/python/generate-proto.ps1
```
Java clients use the Gradle protobuf plugin from `clients/java`. The
`mxgateway-client` project reads the shared `.proto` files and writes generated
Java protobuf and gRPC sources under `clients/java/src/main/generated`, matching
the manifest output path. Handwritten client and CLI code stays in the
`mxgateway-client` and `mxgateway-cli` project source trees.
Run the Java workspace checks from `clients/java`:
```powershell
gradle test
```
## Golden Fixtures
+15
View File
@@ -17,6 +17,7 @@ Recommended Gradle multi-project layout:
clients/java/
settings.gradle
build.gradle
src/main/generated/
mxgateway-client/
build.gradle
src/main/java/com/dohertylan/mxgateway/client/
@@ -31,6 +32,7 @@ Alternative Maven layout is acceptable if the repo standardizes on Maven.
Target Java:
- Java 21 recommended.
- The Gradle scaffold uses the Java 21 toolchain for compilation and tests.
Expected dependencies:
@@ -189,3 +191,16 @@ Publish library and CLI separately:
Generated protobuf code should be produced during the build from shared proto
files and should not be hand-edited.
## Current Build
Run the Java scaffold checks from `clients/java`:
```powershell
gradle test
```
The `mxgateway-client` project generates the gateway and worker protobuf/gRPC
bindings into `src/main/generated`, compiles the generated contracts, and runs
JUnit 5 tests. The `mxgateway-cli` project builds a Picocli-based `mxgw-java`
entry point for later command implementation.
+11
View File
@@ -175,6 +175,12 @@ Behavior:
`CloseSession` should be idempotent. Closing an already closed session should
return a successful close result with the final known state.
`WorkerClient.ShutdownAsync` sends `WorkerShutdown`, waits for the worker read,
write, and heartbeat loops to stop, and waits for the launched worker process to
exit within the same shutdown timeout. If the pipe loops or process exit exceed
the timeout, the close operation fails with `ShutdownTimeout`; `GatewaySession`
then kills the worker process tree before surfacing the close failure.
### Invoke
`Invoke` forwards one MXAccess command to the worker that owns the session.
@@ -515,6 +521,11 @@ It handles:
The write loop should fail the session if a pipe write fails outside normal
shutdown.
During shutdown the worker client treats `WorkerShutdownAck` as the protocol
close signal, but the process handle remains authoritative for process lifetime.
The client waits for both the protocol close and process exit before reporting a
clean shutdown to `GatewaySession`.
## Command Correlation
Each command gets:
+38
View File
@@ -321,6 +321,13 @@ If COM creation fails, the worker should send a structured fault with:
when the exception exposes one, and does not send `WorkerReady` after a failed
COM creation attempt.
After `WorkerReady`, `WorkerPipeSession` continues reading gateway frames for
the lifetime of the process. `WorkerCommand` frames are dispatched to
`MxAccessStaSession`, replies are written as `WorkerCommandReply`, and queued
worker events are drained after command replies. `WorkerShutdown` starts the
graceful shutdown path and returns `WorkerShutdownAck` only after the STA
cleanup path completes.
## Event Sink
The worker must subscribe to every public MXAccess event family:
@@ -675,6 +682,29 @@ Graceful shutdown sequence:
If shutdown wedges, the gateway kills the process. The worker should be written
so process kill does not corrupt other sessions.
`MxAccessStaSession.ShutdownGracefullyAsync` implements the current cleanup
path. It first calls `StaCommandDispatcher.RequestShutdown()` so new commands
are rejected and queued commands that have not started receive
`ProtocolStatusCode.WorkerUnavailable`. The command already executing on the
STA is allowed to finish until the shutdown grace period expires.
After command dispatch is closed, cleanup runs on the STA in MXAccess handle
order:
1. one `UnAdvise` call per advised server/item pair,
2. `RemoveItem` for active item handles,
3. `Unregister` for active server handles,
4. event sink detach,
5. COM release.
Each cleanup call is best effort. A failed cleanup operation is recorded as an
`MxAccessShutdownFailure`, logged by `WorkerPipeSession`, and does not prevent
later cleanup calls from running. A shutdown with cleanup failures still returns
`WorkerShutdownAck` with `ProtocolStatusCode.Ok` because the worker reached the
controlled release path. If the grace period expires before cleanup can run or
finish, the worker reports `WorkerFaultCategory.ShutdownTimeout` when possible
and relies on the gateway to kill the process.
## Fault Handling
Worker fault categories:
@@ -777,6 +807,14 @@ tests. `AddItem` uses `TestChildObject.TestInt` by default and accepts an
override through `MXGATEWAY_LIVE_MXACCESS_ITEM`; `AddItem2` uses the captured
parity fixture shape `AddItem2("TestInt", "TestChildObject")`.
`WorkerLiveMxAccessSmokeTests` in `src/MxGateway.IntegrationTests/` uses the
same opt-in variable for the gateway-to-worker live smoke. It launches the x86
worker through `WorkerProcessLauncher`, opens a gateway session, runs
`Register`, `AddItem`, and `Advise`, waits for one `OnDataChange`, and closes
the session. The smoke accepts `MXGATEWAY_LIVE_MXACCESS_WORKER_EXE` for a
non-default worker executable path and
`MXGATEWAY_LIVE_MXACCESS_EVENT_TIMEOUT_SECONDS` for the bounded event wait.
## Initial Implementation Slice
The first worker slice should implement:
@@ -3,10 +3,92 @@ namespace MxGateway.IntegrationTests;
public static class IntegrationTestEnvironment
{
public const string LiveMxAccessVariableName = "MXGATEWAY_RUN_LIVE_MXACCESS_TESTS";
public const string LiveMxAccessWorkerExecutableVariableName = "MXGATEWAY_LIVE_MXACCESS_WORKER_EXE";
public const string LiveMxAccessItemVariableName = "MXGATEWAY_LIVE_MXACCESS_ITEM";
public const string LiveMxAccessClientNameVariableName = "MXGATEWAY_LIVE_MXACCESS_CLIENT_NAME";
public const string LiveMxAccessEventTimeoutSecondsVariableName = "MXGATEWAY_LIVE_MXACCESS_EVENT_TIMEOUT_SECONDS";
public static bool LiveMxAccessTestsEnabled =>
string.Equals(
Environment.GetEnvironmentVariable(LiveMxAccessVariableName),
"1",
StringComparison.Ordinal);
public static string LiveMxAccessItem =>
GetOptionalEnvironmentVariable(
LiveMxAccessItemVariableName,
"TestChildObject.TestInt");
public static string LiveMxAccessClientName =>
GetOptionalEnvironmentVariable(
LiveMxAccessClientNameVariableName,
"MxGateway.IntegrationTests");
public static TimeSpan LiveMxAccessEventTimeout =>
TimeSpan.FromSeconds(GetPositiveIntegerEnvironmentVariable(
LiveMxAccessEventTimeoutSecondsVariableName,
defaultValue: 15));
public static string ResolveLiveMxAccessWorkerExecutablePath()
{
string? configuredPath = Environment.GetEnvironmentVariable(LiveMxAccessWorkerExecutableVariableName);
if (!string.IsNullOrWhiteSpace(configuredPath))
{
return Path.GetFullPath(configuredPath);
}
string repositoryRoot = ResolveRepositoryRoot(AppContext.BaseDirectory);
string[] candidatePaths =
[
Path.Combine(repositoryRoot, "src", "MxGateway.Worker", "bin", "x86", "Debug", "net48", "MxGateway.Worker.exe"),
Path.Combine(repositoryRoot, "src", "MxGateway.Worker", "bin", "Debug", "net48", "MxGateway.Worker.exe"),
Path.Combine(repositoryRoot, "src", "MxGateway.Worker", "bin", "x86", "Release", "net48", "MxGateway.Worker.exe"),
Path.Combine(repositoryRoot, "src", "MxGateway.Worker", "bin", "Release", "net48", "MxGateway.Worker.exe"),
Path.Combine(repositoryRoot, "src", "MxGateway.Worker", "bin", "x86", "Release", "MxGateway.Worker.exe"),
];
return candidatePaths.FirstOrDefault(File.Exists)
?? candidatePaths[0];
}
private static string GetOptionalEnvironmentVariable(
string name,
string defaultValue)
{
string? value = Environment.GetEnvironmentVariable(name);
return string.IsNullOrWhiteSpace(value)
? defaultValue
: value;
}
private static int GetPositiveIntegerEnvironmentVariable(
string name,
int defaultValue)
{
string? value = Environment.GetEnvironmentVariable(name);
if (int.TryParse(value, out int parsed) && parsed > 0)
{
return parsed;
}
return defaultValue;
}
internal static string ResolveRepositoryRoot(string startDirectory)
{
DirectoryInfo? directory = new(startDirectory);
while (directory is not null)
{
if ((Directory.Exists(Path.Combine(directory.FullName, ".git"))
|| File.Exists(Path.Combine(directory.FullName, ".git")))
&& Directory.Exists(Path.Combine(directory.FullName, "src")))
{
return directory.FullName;
}
directory = directory.Parent;
}
return Directory.GetCurrentDirectory();
}
}
@@ -9,4 +9,37 @@ public sealed class IntegrationTestEnvironmentTests
"MXGATEWAY_RUN_LIVE_MXACCESS_TESTS",
IntegrationTestEnvironment.LiveMxAccessVariableName);
}
[Fact]
public void LiveMxAccessWorkerExecutable_UsesDocumentedEnvironmentVariable()
{
Assert.Equal(
"MXGATEWAY_LIVE_MXACCESS_WORKER_EXE",
IntegrationTestEnvironment.LiveMxAccessWorkerExecutableVariableName);
}
[Fact]
public void ResolveRepositoryRoot_AcceptsGitWorktreeFile()
{
string temporaryRoot = Path.Combine(Path.GetTempPath(), Path.GetRandomFileName());
string nestedDirectory = Path.Combine(temporaryRoot, "tests", "bin");
try
{
Directory.CreateDirectory(nestedDirectory);
Directory.CreateDirectory(Path.Combine(temporaryRoot, "src"));
File.WriteAllText(Path.Combine(temporaryRoot, ".git"), "gitdir: ../.git/worktrees/test");
string repositoryRoot = IntegrationTestEnvironment.ResolveRepositoryRoot(nestedDirectory);
Assert.Equal(temporaryRoot, repositoryRoot);
}
finally
{
if (Directory.Exists(temporaryRoot))
{
Directory.Delete(temporaryRoot, recursive: true);
}
}
}
}
@@ -0,0 +1,12 @@
namespace MxGateway.IntegrationTests;
public sealed class LiveMxAccessFactAttribute : FactAttribute
{
public LiveMxAccessFactAttribute()
{
if (!IntegrationTestEnvironment.LiveMxAccessTestsEnabled)
{
Skip = $"Set {IntegrationTestEnvironment.LiveMxAccessVariableName}=1 to run live MXAccess tests.";
}
}
}
@@ -18,6 +18,7 @@
<ItemGroup>
<ProjectReference Include="..\MxGateway.Contracts\MxGateway.Contracts.csproj" />
<ProjectReference Include="..\MxGateway.Server\MxGateway.Server.csproj" />
</ItemGroup>
</Project>
@@ -0,0 +1,517 @@
using System.Collections.Concurrent;
using System.Diagnostics;
using Google.Protobuf.WellKnownTypes;
using Grpc.Core;
using Microsoft.Extensions.Logging;
using Microsoft.Extensions.Options;
using MxGateway.Contracts;
using MxGateway.Contracts.Proto;
using MxGateway.Server.Configuration;
using MxGateway.Server.Grpc;
using MxGateway.Server.Metrics;
using MxGateway.Server.Security.Authorization;
using MxGateway.Server.Sessions;
using MxGateway.Server.Workers;
using Xunit.Abstractions;
namespace MxGateway.IntegrationTests;
public sealed class WorkerLiveMxAccessSmokeTests(ITestOutputHelper output)
{
private static readonly TimeSpan CommandTimeout = TimeSpan.FromSeconds(15);
private static readonly TimeSpan StreamShutdownTimeout = TimeSpan.FromSeconds(10);
[LiveMxAccessFact]
[Trait("Category", "LiveMxAccess")]
public async Task GatewaySession_WithLiveWorker_RegistersAdvisesStreamsDataAndCloses()
{
string workerExecutablePath = IntegrationTestEnvironment.ResolveLiveMxAccessWorkerExecutablePath();
Assert.True(
File.Exists(workerExecutablePath),
$"Live MXAccess worker executable was not found at {workerExecutablePath}. Build the worker or set {IntegrationTestEnvironment.LiveMxAccessWorkerExecutableVariableName}.");
TestWorkerProcessFactory processFactory = new(output);
await using GatewayServiceFixture fixture = new(workerExecutablePath, processFactory, output);
string? sessionId = null;
RecordingServerStreamWriter<MxEvent>? eventWriter = null;
Task? streamTask = null;
try
{
OpenSessionReply openReply = await fixture.Service.OpenSession(
new OpenSessionRequest
{
ClientSessionName = "live-mxaccess-smoke",
ClientCorrelationId = "live-open",
CommandTimeout = Duration.FromTimeSpan(CommandTimeout),
},
new TestServerCallContext()).ConfigureAwait(false);
sessionId = openReply.SessionId;
output.WriteLine($"OpenSession status={openReply.ProtocolStatus.Code} session={sessionId} worker_pid={openReply.WorkerProcessId}");
Assert.Equal(ProtocolStatusCode.Ok, openReply.ProtocolStatus.Code);
Assert.True(openReply.WorkerProcessId > 0);
eventWriter = new RecordingServerStreamWriter<MxEvent>();
streamTask = fixture.Service.StreamEvents(
new StreamEventsRequest { SessionId = sessionId },
eventWriter,
new TestServerCallContext());
MxCommandReply registerReply = await fixture.Service.Invoke(
CreateRegisterRequest(sessionId),
new TestServerCallContext()).ConfigureAwait(false);
LogReply("Register", registerReply);
Assert.Equal(ProtocolStatusCode.Ok, registerReply.ProtocolStatus.Code);
Assert.True(registerReply.Register.ServerHandle > 0);
MxCommandReply addItemReply = await fixture.Service.Invoke(
CreateAddItemRequest(sessionId, registerReply.Register.ServerHandle),
new TestServerCallContext()).ConfigureAwait(false);
LogReply("AddItem", addItemReply);
Assert.Equal(ProtocolStatusCode.Ok, addItemReply.ProtocolStatus.Code);
Assert.True(addItemReply.AddItem.ItemHandle > 0);
MxCommandReply adviseReply = await fixture.Service.Invoke(
CreateAdviseRequest(
sessionId,
registerReply.Register.ServerHandle,
addItemReply.AddItem.ItemHandle),
new TestServerCallContext()).ConfigureAwait(false);
LogReply("Advise", adviseReply);
Assert.Equal(ProtocolStatusCode.Ok, adviseReply.ProtocolStatus.Code);
MxEvent dataChange = await eventWriter
.WaitForFirstMessageAsync(IntegrationTestEnvironment.LiveMxAccessEventTimeout)
.ConfigureAwait(false);
LogEvent(dataChange);
Assert.Equal(MxEventFamily.OnDataChange, dataChange.Family);
Assert.Equal(sessionId, dataChange.SessionId);
Assert.Equal(registerReply.Register.ServerHandle, dataChange.ServerHandle);
Assert.Equal(addItemReply.AddItem.ItemHandle, dataChange.ItemHandle);
}
finally
{
try
{
if (!string.IsNullOrWhiteSpace(sessionId))
{
await CloseSessionAsync(fixture, sessionId).ConfigureAwait(false);
}
if (streamTask is not null)
{
await streamTask.WaitAsync(StreamShutdownTimeout).ConfigureAwait(false);
}
}
finally
{
await processFactory.WaitForProcessesAsync(StreamShutdownTimeout).ConfigureAwait(false);
}
}
}
private static MxCommandRequest CreateRegisterRequest(string sessionId)
{
return new MxCommandRequest
{
SessionId = sessionId,
ClientCorrelationId = "live-register",
Command = new MxCommand
{
Kind = MxCommandKind.Register,
Register = new RegisterCommand
{
ClientName = IntegrationTestEnvironment.LiveMxAccessClientName,
},
},
};
}
private static MxCommandRequest CreateAddItemRequest(
string sessionId,
int serverHandle)
{
return new MxCommandRequest
{
SessionId = sessionId,
ClientCorrelationId = "live-add-item",
Command = new MxCommand
{
Kind = MxCommandKind.AddItem,
AddItem = new AddItemCommand
{
ServerHandle = serverHandle,
ItemDefinition = IntegrationTestEnvironment.LiveMxAccessItem,
},
},
};
}
private static MxCommandRequest CreateAdviseRequest(
string sessionId,
int serverHandle,
int itemHandle)
{
return new MxCommandRequest
{
SessionId = sessionId,
ClientCorrelationId = "live-advise",
Command = new MxCommand
{
Kind = MxCommandKind.Advise,
Advise = new AdviseCommand
{
ServerHandle = serverHandle,
ItemHandle = itemHandle,
},
},
};
}
private async Task CloseSessionAsync(
GatewayServiceFixture fixture,
string sessionId)
{
CloseSessionReply closeReply = await fixture.Service.CloseSession(
new CloseSessionRequest
{
SessionId = sessionId,
ClientCorrelationId = "live-close",
},
new TestServerCallContext()).ConfigureAwait(false);
output.WriteLine($"CloseSession status={closeReply.ProtocolStatus.Code} final_state={closeReply.FinalState}");
}
private void LogReply(
string method,
MxCommandReply reply)
{
output.WriteLine(
$"{method} status={reply.ProtocolStatus.Code} hresult={reply.Hresult} diagnostic={reply.DiagnosticMessage}");
foreach (MxStatusProxy status in reply.Statuses)
{
output.WriteLine(
$"{method} mxstatus success={status.Success} category={status.Category} detail={status.Detail} text={status.DiagnosticText}");
}
}
private void LogEvent(MxEvent dataChange)
{
output.WriteLine(
$"Event family={dataChange.Family} worker_sequence={dataChange.WorkerSequence} server_handle={dataChange.ServerHandle} item_handle={dataChange.ItemHandle} quality={dataChange.Quality}");
output.WriteLine(
$"Event value_type={dataChange.Value?.DataType} raw_status={dataChange.RawStatus}");
}
private sealed class GatewayServiceFixture : IAsyncDisposable
{
private readonly GatewayMetrics _metrics = new();
private readonly SessionRegistry _registry = new();
private readonly ILoggerFactory _loggerFactory;
public GatewayServiceFixture(
string workerExecutablePath,
IWorkerProcessFactory processFactory,
ITestOutputHelper output)
{
IOptions<GatewayOptions> options = Options.Create(CreateOptions(workerExecutablePath));
_loggerFactory = LoggerFactory.Create(builder => builder.AddProvider(new TestOutputLoggerProvider(output)));
WorkerProcessLauncher launcher = new(
options,
processFactory,
new WorkerProcessStartedProbe(),
_metrics);
SessionWorkerClientFactory workerClientFactory = new(
launcher,
options,
_metrics,
_loggerFactory);
SessionManager sessionManager = new(
_registry,
workerClientFactory,
options,
_metrics,
logger: _loggerFactory.CreateLogger<SessionManager>());
MxAccessGrpcMapper mapper = new();
EventStreamService eventStreamService = new(
sessionManager,
options,
mapper,
_metrics,
_loggerFactory.CreateLogger<EventStreamService>());
Service = new MxAccessGatewayService(
sessionManager,
new GatewayRequestIdentityAccessor(),
new MxAccessGrpcRequestValidator(),
mapper,
eventStreamService,
_loggerFactory.CreateLogger<MxAccessGatewayService>());
}
public MxAccessGatewayService Service { get; }
public async ValueTask DisposeAsync()
{
foreach (GatewaySession session in _registry.Snapshot())
{
await session.DisposeAsync().ConfigureAwait(false);
}
_loggerFactory.Dispose();
_metrics.Dispose();
}
private static GatewayOptions CreateOptions(string workerExecutablePath)
{
return new GatewayOptions
{
Worker = new WorkerOptions
{
ExecutablePath = workerExecutablePath,
StartupTimeoutSeconds = 30,
ShutdownTimeoutSeconds = 15,
HeartbeatIntervalSeconds = 5,
HeartbeatGraceSeconds = 15,
MaxMessageBytes = WorkerFrameProtocolOptions.DefaultMaxMessageBytes,
RequiredArchitecture = WorkerArchitecture.X86,
},
Sessions = new SessionOptions
{
DefaultCommandTimeoutSeconds = 15,
MaxSessions = 1,
},
Events = new EventOptions
{
QueueCapacity = 32,
},
};
}
}
private sealed class RecordingServerStreamWriter<T> : IServerStreamWriter<T>
{
private readonly object syncRoot = new();
private readonly TaskCompletionSource<T> firstMessage = new(TaskCreationOptions.RunContinuationsAsynchronously);
private readonly List<T> messages = [];
public IReadOnlyList<T> Messages
{
get
{
lock (syncRoot)
{
return messages.ToArray();
}
}
}
public WriteOptions? WriteOptions { get; set; }
public Task WriteAsync(T message)
{
lock (syncRoot)
{
messages.Add(message);
}
firstMessage.TrySetResult(message);
return Task.CompletedTask;
}
public async Task<T> WaitForFirstMessageAsync(TimeSpan timeout)
{
return await firstMessage.Task.WaitAsync(timeout).ConfigureAwait(false);
}
}
private sealed class TestServerCallContext(CancellationToken cancellationToken = default) : ServerCallContext
{
private readonly Metadata requestHeaders = [];
private readonly Metadata responseTrailers = [];
private readonly Dictionary<object, object> userState = [];
private Status status;
private WriteOptions? writeOptions;
protected override string MethodCore => "/mxaccess_gateway.v1.MxAccessGateway/Test";
protected override string HostCore => "localhost";
protected override string PeerCore => "ipv4:127.0.0.1:5000";
protected override DateTime DeadlineCore => DateTime.UtcNow.AddMinutes(1);
protected override Metadata RequestHeadersCore => requestHeaders;
protected override CancellationToken CancellationTokenCore => cancellationToken;
protected override Metadata ResponseTrailersCore => responseTrailers;
protected override Status StatusCore
{
get => status;
set => status = value;
}
protected override WriteOptions? WriteOptionsCore
{
get => writeOptions;
set => writeOptions = value;
}
protected override AuthContext AuthContextCore { get; } = new(
string.Empty,
new Dictionary<string, List<AuthProperty>>(StringComparer.Ordinal));
protected override IDictionary<object, object> UserStateCore => userState;
protected override Task WriteResponseHeadersAsyncCore(Metadata responseHeaders)
{
return Task.CompletedTask;
}
protected override ContextPropagationToken CreatePropagationTokenCore(
ContextPropagationOptions? options)
{
throw new NotSupportedException();
}
}
private sealed class TestWorkerProcessFactory(ITestOutputHelper output) : IWorkerProcessFactory
{
private readonly ConcurrentBag<TestWorkerProcess> processes = [];
public IWorkerProcess Start(ProcessStartInfo startInfo)
{
startInfo.RedirectStandardError = true;
startInfo.RedirectStandardOutput = true;
startInfo.UseShellExecute = false;
Process process = new()
{
StartInfo = startInfo,
EnableRaisingEvents = true,
};
process.OutputDataReceived += (_, args) => WriteWorkerOutput("stdout", args.Data);
process.ErrorDataReceived += (_, args) => WriteWorkerOutput("stderr", args.Data);
if (!process.Start())
{
process.Dispose();
throw new InvalidOperationException("Worker process failed to start.");
}
process.BeginOutputReadLine();
process.BeginErrorReadLine();
TestWorkerProcess workerProcess = new(process);
processes.Add(workerProcess);
output.WriteLine($"WorkerProcess started pid={workerProcess.Id} path={startInfo.FileName}");
return workerProcess;
}
public async Task WaitForProcessesAsync(TimeSpan timeout)
{
foreach (TestWorkerProcess process in processes)
{
if (process.HasExited)
{
output.WriteLine($"WorkerProcess exited pid={process.Id} exit_code={process.ExitCode}");
continue;
}
using CancellationTokenSource timeoutCancellation = new(timeout);
await process.WaitForExitAsync(timeoutCancellation.Token).ConfigureAwait(false);
output.WriteLine($"WorkerProcess exited pid={process.Id} exit_code={process.ExitCode}");
}
}
private void WriteWorkerOutput(
string streamName,
string? line)
{
if (!string.IsNullOrWhiteSpace(line))
{
output.WriteLine($"worker_{streamName}: {line}");
}
}
}
private sealed class TestWorkerProcess(Process process) : IWorkerProcess
{
public int Id => process.Id;
public bool HasExited => process.HasExited;
public int? ExitCode => process.HasExited ? process.ExitCode : null;
public async ValueTask WaitForExitAsync(CancellationToken cancellationToken)
{
await process.WaitForExitAsync(cancellationToken).ConfigureAwait(false);
}
public void Kill(bool entireProcessTree)
{
process.Kill(entireProcessTree);
}
public void Dispose()
{
process.Dispose();
}
}
private sealed class TestOutputLoggerProvider(ITestOutputHelper output) : ILoggerProvider
{
public ILogger CreateLogger(string categoryName)
{
return new TestOutputLogger(output, categoryName);
}
public void Dispose()
{
}
}
private sealed class TestOutputLogger(
ITestOutputHelper output,
string categoryName) : ILogger
{
public IDisposable? BeginScope<TState>(TState state)
where TState : notnull
{
return null;
}
public bool IsEnabled(LogLevel logLevel)
{
return logLevel >= LogLevel.Information;
}
public void Log<TState>(
LogLevel logLevel,
EventId eventId,
TState state,
Exception? exception,
Func<TState, Exception?, string> formatter)
{
if (!IsEnabled(logLevel))
{
return;
}
output.WriteLine($"{logLevel} {categoryName}: {formatter(state, exception)}");
if (exception is not null)
{
output.WriteLine(exception.ToString());
}
}
}
}
@@ -227,6 +227,7 @@ public sealed class WorkerClient : IWorkerClient
try
{
await WaitForBackgroundTasksAsync(timeoutCts.Token).ConfigureAwait(false);
await WaitForProcessExitAsync(timeoutCts.Token).ConfigureAwait(false);
MarkClosed("shutdown");
}
catch (OperationCanceledException) when (!cancellationToken.IsCancellationRequested)
@@ -717,6 +718,17 @@ public sealed class WorkerClient : IWorkerClient
await Task.WhenAll(tasks).WaitAsync(cancellationToken).ConfigureAwait(false);
}
private async Task WaitForProcessExitAsync(CancellationToken cancellationToken)
{
WorkerProcessHandle? processHandle = _connection.ProcessHandle;
if (processHandle is null || processHandle.Process.HasExited)
{
return;
}
await processHandle.Process.WaitForExitAsync(cancellationToken).ConfigureAwait(false);
}
private void ThrowIfDisposed()
{
ObjectDisposedException.ThrowIf(_disposed, this);
@@ -86,6 +86,26 @@ public sealed class SessionManagerTests
Assert.Equal(0, metrics.GetSnapshot().OpenSessions);
}
[Fact]
public async Task CloseSessionAsync_WhenWorkerShutdownFails_KillsWorker()
{
FakeWorkerClient workerClient = new()
{
ShutdownException = new WorkerClientException(
WorkerClientErrorCode.ShutdownTimeout,
"Worker shutdown timed out."),
};
SessionManager manager = CreateManager(new FakeSessionWorkerClientFactory(workerClient));
GatewaySession session = await manager.OpenSessionAsync(CreateOpenRequest(), "client-1", CancellationToken.None);
SessionManagerException exception = await Assert.ThrowsAsync<SessionManagerException>(
async () => await manager.CloseSessionAsync(session.SessionId, CancellationToken.None));
Assert.Equal(SessionManagerErrorCode.CloseFailed, exception.ErrorCode);
Assert.Equal(1, workerClient.ShutdownCount);
Assert.Equal(1, workerClient.KillCount);
}
[Fact]
public async Task OpenSessionAsync_WhenWorkerCreationFails_RemovesSessionFromRegistry()
{
@@ -266,6 +286,8 @@ public sealed class SessionManagerTests
public int KillCount { get; private set; }
public Exception? ShutdownException { get; init; }
public Task StartAsync(CancellationToken cancellationToken)
{
return Task.CompletedTask;
@@ -302,6 +324,11 @@ public sealed class SessionManagerTests
CancellationToken cancellationToken)
{
ShutdownCount++;
if (ShutdownException is not null)
{
throw ShutdownException;
}
State = WorkerClientState.Closed;
return Task.CompletedTask;
}
@@ -142,6 +142,13 @@ public sealed class WorkerPipeClientTests
{
}
public Task<MxAccessShutdownResult> ShutdownGracefullyAsync(
TimeSpan timeout,
CancellationToken cancellationToken = default)
{
return Task.FromResult(new MxAccessShutdownResult(Array.Empty<MxAccessShutdownFailure>()));
}
public void Dispose()
{
}
@@ -555,6 +555,14 @@ public sealed class WorkerPipeSessionTests
releaseDispatch.Set();
}
public Task<MxAccessShutdownResult> ShutdownGracefullyAsync(
TimeSpan timeout,
CancellationToken cancellationToken = default)
{
releaseDispatch.Set();
return Task.FromResult(new MxAccessShutdownResult(Array.Empty<MxAccessShutdownFailure>()));
}
public void ReleaseDispatch()
{
releaseDispatch.Set();
@@ -1,4 +1,6 @@
using System;
using System.Collections.Generic;
using System.Linq;
using System.Runtime.InteropServices;
using System.Threading.Tasks;
using MxGateway.Contracts.Proto;
@@ -414,6 +416,57 @@ public sealed class MxAccessCommandExecutorTests
Assert.Equal(MxAccessAdviceKind.Plain, adviceHandle.AdviceKind);
}
[Fact]
public async Task ShutdownGracefullyAsync_CleansHandlesInAdviceItemServerOrder()
{
FakeMxAccessComObject fakeComObject = new(
registerHandle: 58,
addItemHandle: 510);
FakeMxAccessComObjectFactory factory = new(fakeComObject);
using StaRuntime runtime = CreateRuntime();
using MxAccessStaSession session = new(runtime, factory, new NoopEventSink());
await session.StartAsync(workerProcessId: 1234);
await session.DispatchAsync(CreateRegisterCommand("register-before-shutdown", "client-a"));
await session.DispatchAsync(CreateAddItemCommand("add-before-shutdown", 58, "Galaxy.Tag.Value"));
await session.DispatchAsync(CreateAdviseCommand("advise-before-shutdown", 58, 510));
await session.DispatchAsync(CreateAdviseSupervisoryCommand("supervisory-before-shutdown", 58, 510));
MxAccessShutdownResult result = await session.ShutdownGracefullyAsync(TimeSpan.FromSeconds(2));
Assert.True(result.Succeeded);
Assert.Equal(
new[] { "UnAdvise:58:510", "RemoveItem:58:510", "Unregister:58" },
fakeComObject.OperationNames.Where(name => name.StartsWith("Un", StringComparison.Ordinal)
|| name.StartsWith("Remove", StringComparison.Ordinal)));
}
[Fact]
public async Task ShutdownGracefullyAsync_RecordsCleanupFailuresAndContinues()
{
const int hresult = unchecked((int)0x80070057);
COMException cleanupException = new("Invalid handle.", hresult);
FakeMxAccessComObject fakeComObject = new(
registerHandle: 59,
addItemHandle: 511,
unregisterException: cleanupException,
removeItemException: cleanupException,
unAdviseException: cleanupException);
FakeMxAccessComObjectFactory factory = new(fakeComObject);
using StaRuntime runtime = CreateRuntime();
using MxAccessStaSession session = new(runtime, factory, new NoopEventSink());
await session.StartAsync(workerProcessId: 1234);
await session.DispatchAsync(CreateRegisterCommand("register-before-shutdown-failure", "client-a"));
await session.DispatchAsync(CreateAddItemCommand("add-before-shutdown-failure", 59, "Galaxy.Tag.Value"));
await session.DispatchAsync(CreateAdviseCommand("advise-before-shutdown-failure", 59, 511));
MxAccessShutdownResult result = await session.ShutdownGracefullyAsync(TimeSpan.FromSeconds(2));
Assert.False(result.Succeeded);
Assert.Equal(new[] { "UnAdvise", "RemoveItem", "Unregister" }, result.Failures.Select(failure => failure.Operation));
Assert.All(result.Failures, failure => Assert.Equal(hresult, failure.HResult));
Assert.Contains("Unregister:59", fakeComObject.OperationNames);
}
[Fact]
public async Task DispatchAsync_RegisterWithoutPayload_ReturnsInvalidRequest()
{
@@ -644,6 +697,7 @@ public sealed class MxAccessCommandExecutorTests
private readonly Exception? adviseException;
private readonly Exception? unAdviseException;
private readonly Exception? adviseSupervisoryException;
private readonly List<string> operationNames = new();
public FakeMxAccessComObject(
int registerHandle,
@@ -715,8 +769,11 @@ public sealed class MxAccessCommandExecutorTests
public int? AdviseSupervisoryThreadId { get; private set; }
public IReadOnlyList<string> OperationNames => operationNames.ToArray();
public int Register(string clientName)
{
operationNames.Add($"Register:{clientName}");
RegisteredClientName = clientName;
RegisterThreadId = Environment.CurrentManagedThreadId;
@@ -725,6 +782,7 @@ public sealed class MxAccessCommandExecutorTests
public void Unregister(int serverHandle)
{
operationNames.Add($"Unregister:{serverHandle}");
UnregisteredServerHandle = serverHandle;
UnregisterThreadId = Environment.CurrentManagedThreadId;
@@ -738,6 +796,7 @@ public sealed class MxAccessCommandExecutorTests
int serverHandle,
string itemDefinition)
{
operationNames.Add($"AddItem:{serverHandle}:{itemDefinition}");
AddItemServerHandle = serverHandle;
AddItemDefinition = itemDefinition;
AddItemThreadId = Environment.CurrentManagedThreadId;
@@ -755,6 +814,7 @@ public sealed class MxAccessCommandExecutorTests
string itemDefinition,
string itemContext)
{
operationNames.Add($"AddItem2:{serverHandle}:{itemDefinition}:{itemContext}");
AddItem2ServerHandle = serverHandle;
AddItem2Definition = itemDefinition;
AddItem2Context = itemContext;
@@ -772,6 +832,7 @@ public sealed class MxAccessCommandExecutorTests
int serverHandle,
int itemHandle)
{
operationNames.Add($"RemoveItem:{serverHandle}:{itemHandle}");
RemoveItemServerHandle = serverHandle;
RemovedItemHandle = itemHandle;
RemoveItemThreadId = Environment.CurrentManagedThreadId;
@@ -786,6 +847,7 @@ public sealed class MxAccessCommandExecutorTests
int serverHandle,
int itemHandle)
{
operationNames.Add($"Advise:{serverHandle}:{itemHandle}");
AdviseServerHandle = serverHandle;
AdvisedItemHandle = itemHandle;
AdviseThreadId = Environment.CurrentManagedThreadId;
@@ -800,6 +862,7 @@ public sealed class MxAccessCommandExecutorTests
int serverHandle,
int itemHandle)
{
operationNames.Add($"UnAdvise:{serverHandle}:{itemHandle}");
UnAdviseServerHandle = serverHandle;
UnAdvisedItemHandle = itemHandle;
UnAdviseThreadId = Environment.CurrentManagedThreadId;
@@ -814,6 +877,7 @@ public sealed class MxAccessCommandExecutorTests
int serverHandle,
int itemHandle)
{
operationNames.Add($"AdviseSupervisory:{serverHandle}:{itemHandle}");
AdviseSupervisoryServerHandle = serverHandle;
AdviseSupervisoryItemHandle = itemHandle;
AdviseSupervisoryThreadId = Environment.CurrentManagedThreadId;

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