clients/dotnet: SDK methods for AcknowledgeAlarm + QueryActiveAlarms (PR E.2) #106
@@ -41,6 +41,24 @@ internal sealed class FakeGatewayTransport(MxGatewayClientOptions options) : IMx
|
||||
/// </summary>
|
||||
public List<(StreamEventsRequest Request, CallOptions CallOptions)> StreamEventsCalls { get; } = [];
|
||||
|
||||
/// <summary>
|
||||
/// Gets the list of captured AcknowledgeAlarmAsync calls.
|
||||
/// </summary>
|
||||
public List<(AcknowledgeAlarmRequest Request, CallOptions CallOptions)> AcknowledgeAlarmCalls { get; } = [];
|
||||
|
||||
/// <summary>
|
||||
/// Gets the list of captured QueryActiveAlarmsAsync calls.
|
||||
/// </summary>
|
||||
public List<(QueryActiveAlarmsRequest Request, CallOptions CallOptions)> QueryActiveAlarmsCalls { get; } = [];
|
||||
|
||||
/// <summary>
|
||||
/// Gets the queue of exceptions to throw from AcknowledgeAlarmAsync.
|
||||
/// </summary>
|
||||
public Queue<Exception> AcknowledgeAlarmExceptions { get; } = new();
|
||||
|
||||
private readonly Queue<AcknowledgeAlarmReply> _acknowledgeReplies = new();
|
||||
private readonly List<ActiveAlarmSnapshot> _activeAlarmSnapshots = [];
|
||||
|
||||
/// <summary>
|
||||
/// Gets or sets the reply to return from OpenSessionAsync.
|
||||
/// </summary>
|
||||
@@ -168,4 +186,57 @@ internal sealed class FakeGatewayTransport(MxGatewayClientOptions options) : IMx
|
||||
{
|
||||
_events.Add(gatewayEvent);
|
||||
}
|
||||
|
||||
/// <summary>
|
||||
/// Records the acknowledge call and returns the next enqueued reply (or default).
|
||||
/// </summary>
|
||||
public Task<AcknowledgeAlarmReply> AcknowledgeAlarmAsync(
|
||||
AcknowledgeAlarmRequest request,
|
||||
CallOptions callOptions)
|
||||
{
|
||||
AcknowledgeAlarmCalls.Add((request, callOptions));
|
||||
if (AcknowledgeAlarmExceptions.TryDequeue(out Exception? exception))
|
||||
{
|
||||
throw exception;
|
||||
}
|
||||
|
||||
return Task.FromResult(_acknowledgeReplies.Count > 0
|
||||
? _acknowledgeReplies.Dequeue()
|
||||
: new AcknowledgeAlarmReply
|
||||
{
|
||||
SessionId = request.SessionId,
|
||||
CorrelationId = request.ClientCorrelationId,
|
||||
ProtocolStatus = new ProtocolStatus { Code = ProtocolStatusCode.Ok },
|
||||
Status = new MxStatusProxy { Success = 1, Category = MxStatusCategory.Ok },
|
||||
});
|
||||
}
|
||||
|
||||
/// <summary>
|
||||
/// Records the query call and yields each enqueued snapshot.
|
||||
/// </summary>
|
||||
public async IAsyncEnumerable<ActiveAlarmSnapshot> QueryActiveAlarmsAsync(
|
||||
QueryActiveAlarmsRequest request,
|
||||
CallOptions callOptions)
|
||||
{
|
||||
QueryActiveAlarmsCalls.Add((request, callOptions));
|
||||
|
||||
foreach (ActiveAlarmSnapshot snapshot in _activeAlarmSnapshots)
|
||||
{
|
||||
callOptions.CancellationToken.ThrowIfCancellationRequested();
|
||||
await Task.Yield();
|
||||
yield return snapshot;
|
||||
}
|
||||
}
|
||||
|
||||
/// <summary>Enqueues an acknowledge reply.</summary>
|
||||
public void AddAcknowledgeReply(AcknowledgeAlarmReply reply)
|
||||
{
|
||||
_acknowledgeReplies.Enqueue(reply);
|
||||
}
|
||||
|
||||
/// <summary>Enqueues a snapshot to be yielded from QueryActiveAlarmsAsync.</summary>
|
||||
public void AddActiveAlarmSnapshot(ActiveAlarmSnapshot snapshot)
|
||||
{
|
||||
_activeAlarmSnapshots.Add(snapshot);
|
||||
}
|
||||
}
|
||||
|
||||
@@ -0,0 +1,192 @@
|
||||
using Google.Protobuf.WellKnownTypes;
|
||||
using Grpc.Core;
|
||||
using MxGateway.Contracts.Proto;
|
||||
|
||||
namespace MxGateway.Client.Tests;
|
||||
|
||||
/// <summary>
|
||||
/// PR E.2 — pins the .NET SDK surface for the new alarm RPCs:
|
||||
/// <see cref="MxGatewayClient.AcknowledgeAlarmAsync"/> and
|
||||
/// <see cref="MxGatewayClient.QueryActiveAlarmsAsync"/>.
|
||||
/// </summary>
|
||||
public sealed class MxGatewayClientAlarmsTests
|
||||
{
|
||||
[Fact]
|
||||
public async Task AcknowledgeAlarmAsync_RecordsRequestShapeAndReturnsReply()
|
||||
{
|
||||
FakeGatewayTransport transport = CreateTransport();
|
||||
transport.AddAcknowledgeReply(new AcknowledgeAlarmReply
|
||||
{
|
||||
SessionId = "session-fixture",
|
||||
CorrelationId = "corr-1",
|
||||
ProtocolStatus = new ProtocolStatus { Code = ProtocolStatusCode.Ok },
|
||||
Status = new MxStatusProxy
|
||||
{
|
||||
Success = 1,
|
||||
Category = MxStatusCategory.Ok,
|
||||
DetectedBy = MxStatusSource.RespondingLmx,
|
||||
},
|
||||
});
|
||||
await using MxGatewayClient client = CreateClient(transport);
|
||||
|
||||
AcknowledgeAlarmReply reply = await client.AcknowledgeAlarmAsync(new AcknowledgeAlarmRequest
|
||||
{
|
||||
SessionId = "session-fixture",
|
||||
ClientCorrelationId = "corr-1",
|
||||
AlarmFullReference = "Tank01.Level.HiHi",
|
||||
Comment = "investigating",
|
||||
OperatorUser = "alice",
|
||||
});
|
||||
|
||||
Assert.Equal(ProtocolStatusCode.Ok, reply.ProtocolStatus.Code);
|
||||
Assert.Equal(MxStatusCategory.Ok, reply.Status.Category);
|
||||
|
||||
var call = Assert.Single(transport.AcknowledgeAlarmCalls);
|
||||
Assert.Equal("Tank01.Level.HiHi", call.Request.AlarmFullReference);
|
||||
Assert.Equal("investigating", call.Request.Comment);
|
||||
Assert.Equal("alice", call.Request.OperatorUser);
|
||||
Assert.Equal("Bearer test-api-key", call.CallOptions.Headers?.GetValue("authorization"));
|
||||
}
|
||||
|
||||
[Fact]
|
||||
public async Task AcknowledgeAlarmAsync_HonorsCancellation()
|
||||
{
|
||||
// Acks are routed through the safe-unary retry pipeline (idempotent at the
|
||||
// MxAccess level), so the transport-side cancellation token is a linked one
|
||||
// rather than the caller's original. Verify cancellation by tripping the source
|
||||
// and asserting the call observes it.
|
||||
using CancellationTokenSource cancellation = new();
|
||||
cancellation.Cancel();
|
||||
FakeGatewayTransport transport = CreateTransport();
|
||||
await using MxGatewayClient client = CreateClient(transport);
|
||||
|
||||
await Assert.ThrowsAnyAsync<OperationCanceledException>(() =>
|
||||
client.AcknowledgeAlarmAsync(
|
||||
new AcknowledgeAlarmRequest
|
||||
{
|
||||
SessionId = "session-fixture",
|
||||
AlarmFullReference = "Tank01.Level.HiHi",
|
||||
Comment = string.Empty,
|
||||
OperatorUser = "alice",
|
||||
},
|
||||
cancellation.Token));
|
||||
}
|
||||
|
||||
[Fact]
|
||||
public async Task AcknowledgeAlarmAsync_MapsUnauthenticated_RpcException_ToTypedException()
|
||||
{
|
||||
FakeGatewayTransport transport = CreateTransport();
|
||||
transport.AcknowledgeAlarmExceptions.Enqueue(
|
||||
new RpcException(new Status(StatusCode.Unauthenticated, "expired key")));
|
||||
await using MxGatewayClient client = CreateClient(transport);
|
||||
|
||||
// Note: the FakeGatewayTransport surfaces RpcException directly (it does not run
|
||||
// through GrpcMxGatewayClientTransport's mapping); the fake's contract here is to
|
||||
// pass the exception verbatim. RpcException → typed exception mapping is covered
|
||||
// in the GrpcMxGatewayClientTransport-level tests; the SDK-level test pins the
|
||||
// pass-through shape so a future migration to direct mapping won't silently
|
||||
// change observable behaviour.
|
||||
var ex = await Assert.ThrowsAsync<RpcException>(
|
||||
() => client.AcknowledgeAlarmAsync(new AcknowledgeAlarmRequest
|
||||
{
|
||||
SessionId = "session-fixture",
|
||||
AlarmFullReference = "Tank01.Level.HiHi",
|
||||
Comment = string.Empty,
|
||||
OperatorUser = "alice",
|
||||
}));
|
||||
Assert.Equal(StatusCode.Unauthenticated, ex.StatusCode);
|
||||
}
|
||||
|
||||
[Fact]
|
||||
public async Task QueryActiveAlarmsAsync_StreamsEnqueuedSnapshots()
|
||||
{
|
||||
FakeGatewayTransport transport = CreateTransport();
|
||||
transport.AddActiveAlarmSnapshot(MakeSnapshot("Tank01.Level.HiHi", AlarmConditionState.Active));
|
||||
transport.AddActiveAlarmSnapshot(MakeSnapshot("Tank02.Level.HiHi", AlarmConditionState.ActiveAcked));
|
||||
await using MxGatewayClient client = CreateClient(transport);
|
||||
|
||||
List<ActiveAlarmSnapshot> snapshots = [];
|
||||
await foreach (ActiveAlarmSnapshot snapshot in client.QueryActiveAlarmsAsync(new QueryActiveAlarmsRequest
|
||||
{
|
||||
SessionId = "session-fixture",
|
||||
}))
|
||||
{
|
||||
snapshots.Add(snapshot);
|
||||
}
|
||||
|
||||
Assert.Equal(2, snapshots.Count);
|
||||
Assert.Equal("Tank01.Level.HiHi", snapshots[0].AlarmFullReference);
|
||||
Assert.Equal(AlarmConditionState.Active, snapshots[0].CurrentState);
|
||||
Assert.Equal(AlarmConditionState.ActiveAcked, snapshots[1].CurrentState);
|
||||
Assert.Single(transport.QueryActiveAlarmsCalls);
|
||||
}
|
||||
|
||||
[Fact]
|
||||
public async Task QueryActiveAlarmsAsync_PassesFilterPrefix()
|
||||
{
|
||||
FakeGatewayTransport transport = CreateTransport();
|
||||
await using MxGatewayClient client = CreateClient(transport);
|
||||
|
||||
await foreach (ActiveAlarmSnapshot _ in client.QueryActiveAlarmsAsync(new QueryActiveAlarmsRequest
|
||||
{
|
||||
SessionId = "session-fixture",
|
||||
AlarmFilterPrefix = "Tank01.",
|
||||
}))
|
||||
{
|
||||
// no snapshots enqueued; just verifying the request passes through
|
||||
}
|
||||
|
||||
var call = Assert.Single(transport.QueryActiveAlarmsCalls);
|
||||
Assert.Equal("Tank01.", call.Request.AlarmFilterPrefix);
|
||||
}
|
||||
|
||||
[Fact]
|
||||
public async Task QueryActiveAlarmsAsync_HonorsCancellationDuringEnumeration()
|
||||
{
|
||||
FakeGatewayTransport transport = CreateTransport();
|
||||
transport.AddActiveAlarmSnapshot(MakeSnapshot("Tank01.Level.HiHi", AlarmConditionState.Active));
|
||||
transport.AddActiveAlarmSnapshot(MakeSnapshot("Tank02.Level.HiHi", AlarmConditionState.Active));
|
||||
await using MxGatewayClient client = CreateClient(transport);
|
||||
|
||||
using CancellationTokenSource cancellation = new();
|
||||
await Assert.ThrowsAsync<OperationCanceledException>(async () =>
|
||||
{
|
||||
await foreach (ActiveAlarmSnapshot _ in client.QueryActiveAlarmsAsync(
|
||||
new QueryActiveAlarmsRequest { SessionId = "session-fixture" },
|
||||
cancellation.Token))
|
||||
{
|
||||
cancellation.Cancel();
|
||||
}
|
||||
});
|
||||
}
|
||||
|
||||
private static ActiveAlarmSnapshot MakeSnapshot(string fullReference, AlarmConditionState state)
|
||||
{
|
||||
return new ActiveAlarmSnapshot
|
||||
{
|
||||
AlarmFullReference = fullReference,
|
||||
SourceObjectReference = fullReference.Split('.')[0],
|
||||
AlarmTypeName = "AnalogLimitAlarm.HiHi",
|
||||
Severity = 750,
|
||||
CurrentState = state,
|
||||
Category = "Process",
|
||||
Description = "Tank high-high level",
|
||||
OriginalRaiseTimestamp = Timestamp.FromDateTime(new DateTime(2026, 5, 1, 12, 0, 0, DateTimeKind.Utc)),
|
||||
LastTransitionTimestamp = Timestamp.FromDateTime(new DateTime(2026, 5, 1, 12, 0, 30, DateTimeKind.Utc)),
|
||||
};
|
||||
}
|
||||
|
||||
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",
|
||||
});
|
||||
}
|
||||
}
|
||||
@@ -18,7 +18,7 @@ public sealed class MxGatewayClientCliTests
|
||||
var exitCode = MxGatewayClientCli.Run(["version"], output, error);
|
||||
|
||||
Assert.Equal(0, exitCode);
|
||||
Assert.Contains("gateway-protocol=2", output.ToString());
|
||||
Assert.Contains("gateway-protocol=3", output.ToString());
|
||||
Assert.Contains("worker-protocol=1", output.ToString());
|
||||
Assert.Equal(string.Empty, error.ToString());
|
||||
}
|
||||
@@ -33,7 +33,7 @@ public sealed class MxGatewayClientCliTests
|
||||
int exitCode = await MxGatewayClientCli.RunAsync(["version", "--json"], output, error);
|
||||
|
||||
Assert.Equal(0, exitCode);
|
||||
Assert.Contains("\"gatewayProtocolVersion\":2", output.ToString());
|
||||
Assert.Contains("\"gatewayProtocolVersion\":3", output.ToString());
|
||||
Assert.Equal(string.Empty, error.ToString());
|
||||
}
|
||||
|
||||
|
||||
@@ -116,6 +116,65 @@ internal sealed class GrpcMxGatewayClientTransport(
|
||||
return StreamEventsAsync(request, callOptions);
|
||||
}
|
||||
|
||||
/// <inheritdoc />
|
||||
public async Task<AcknowledgeAlarmReply> AcknowledgeAlarmAsync(
|
||||
AcknowledgeAlarmRequest request,
|
||||
CallOptions callOptions)
|
||||
{
|
||||
try
|
||||
{
|
||||
return await RawClient.AcknowledgeAlarmAsync(request, callOptions)
|
||||
.ResponseAsync
|
||||
.ConfigureAwait(false);
|
||||
}
|
||||
catch (RpcException exception)
|
||||
{
|
||||
throw MapRpcException(exception, callOptions.CancellationToken);
|
||||
}
|
||||
}
|
||||
|
||||
/// <inheritdoc />
|
||||
public async IAsyncEnumerable<ActiveAlarmSnapshot> QueryActiveAlarmsAsync(
|
||||
QueryActiveAlarmsRequest request,
|
||||
CallOptions callOptions,
|
||||
[System.Runtime.CompilerServices.EnumeratorCancellation] CancellationToken cancellationToken = default)
|
||||
{
|
||||
CancellationToken effectiveCancellationToken = cancellationToken.CanBeCanceled
|
||||
? cancellationToken
|
||||
: callOptions.CancellationToken;
|
||||
|
||||
using AsyncServerStreamingCall<ActiveAlarmSnapshot> call = RawClient.QueryActiveAlarms(request, callOptions);
|
||||
|
||||
IAsyncStreamReader<ActiveAlarmSnapshot> responseStream = call.ResponseStream;
|
||||
while (true)
|
||||
{
|
||||
ActiveAlarmSnapshot? snapshot;
|
||||
try
|
||||
{
|
||||
if (!await responseStream.MoveNext(effectiveCancellationToken).ConfigureAwait(false))
|
||||
{
|
||||
break;
|
||||
}
|
||||
|
||||
snapshot = responseStream.Current;
|
||||
}
|
||||
catch (RpcException exception)
|
||||
{
|
||||
throw MapRpcException(exception, effectiveCancellationToken);
|
||||
}
|
||||
|
||||
yield return snapshot;
|
||||
}
|
||||
}
|
||||
|
||||
/// <inheritdoc />
|
||||
IAsyncEnumerable<ActiveAlarmSnapshot> IMxGatewayClientTransport.QueryActiveAlarmsAsync(
|
||||
QueryActiveAlarmsRequest request,
|
||||
CallOptions callOptions)
|
||||
{
|
||||
return QueryActiveAlarmsAsync(request, callOptions);
|
||||
}
|
||||
|
||||
private static Exception MapRpcException(
|
||||
RpcException exception,
|
||||
CancellationToken cancellationToken)
|
||||
|
||||
@@ -54,4 +54,25 @@ internal interface IMxGatewayClientTransport
|
||||
IAsyncEnumerable<MxEvent> StreamEventsAsync(
|
||||
StreamEventsRequest request,
|
||||
CallOptions callOptions);
|
||||
|
||||
/// <summary>
|
||||
/// Acknowledges an active MXAccess alarm condition.
|
||||
/// </summary>
|
||||
/// <param name="request">The acknowledge request.</param>
|
||||
/// <param name="callOptions">gRPC call options.</param>
|
||||
/// <returns>The acknowledge reply with native MxStatus.</returns>
|
||||
Task<AcknowledgeAlarmReply> AcknowledgeAlarmAsync(
|
||||
AcknowledgeAlarmRequest request,
|
||||
CallOptions callOptions);
|
||||
|
||||
/// <summary>
|
||||
/// Streams a snapshot of all alarms currently in Active or ActiveAcked state — the
|
||||
/// ConditionRefresh equivalent for the gateway.
|
||||
/// </summary>
|
||||
/// <param name="request">The query request, optionally scoped by alarm-reference prefix.</param>
|
||||
/// <param name="callOptions">gRPC call options.</param>
|
||||
/// <returns>An async enumerable of active-alarm snapshots.</returns>
|
||||
IAsyncEnumerable<ActiveAlarmSnapshot> QueryActiveAlarmsAsync(
|
||||
QueryActiveAlarmsRequest request,
|
||||
CallOptions callOptions);
|
||||
}
|
||||
|
||||
@@ -182,6 +182,48 @@ public sealed class MxGatewayClient : IAsyncDisposable
|
||||
return _transport.StreamEventsAsync(request, CreateStreamCallOptions(cancellationToken));
|
||||
}
|
||||
|
||||
/// <summary>
|
||||
/// Acknowledges an active MXAccess alarm condition through the gateway. The
|
||||
/// gateway authenticates the request against the API key's <c>invoke:alarm-ack</c>
|
||||
/// scope and forwards the acknowledge to the worker's MXAccess session;
|
||||
/// the resulting <see cref="MxStatusProxy"/> is returned in the reply.
|
||||
/// </summary>
|
||||
/// <param name="request">The acknowledge request — alarm reference, comment, operator user.</param>
|
||||
/// <param name="cancellationToken">Cancellation token for the operation.</param>
|
||||
/// <returns>The acknowledge reply with protocol + native MxStatus.</returns>
|
||||
public Task<AcknowledgeAlarmReply> AcknowledgeAlarmAsync(
|
||||
AcknowledgeAlarmRequest request,
|
||||
CancellationToken cancellationToken = default)
|
||||
{
|
||||
ArgumentNullException.ThrowIfNull(request);
|
||||
ThrowIfDisposed();
|
||||
|
||||
return ExecuteSafeUnaryAsync(
|
||||
token => _transport.AcknowledgeAlarmAsync(request, CreateCallOptions(token)),
|
||||
cancellationToken);
|
||||
}
|
||||
|
||||
/// <summary>
|
||||
/// Streams a snapshot of all alarms currently Active or ActiveAcked — the gateway's
|
||||
/// ConditionRefresh equivalent. Used after reconnect to seed the local Part 9 state
|
||||
/// machine, or to reconcile alarms that may have been missed during a transport
|
||||
/// blip. Optionally scoped by alarm-reference prefix
|
||||
/// (<see cref="QueryActiveAlarmsRequest.AlarmFilterPrefix"/>) so a partial refresh
|
||||
/// can target an equipment sub-tree.
|
||||
/// </summary>
|
||||
/// <param name="request">The query request, optionally scoped by alarm-reference prefix.</param>
|
||||
/// <param name="cancellationToken">Cancellation token for the stream.</param>
|
||||
/// <returns>An async enumerable of active-alarm snapshots.</returns>
|
||||
public IAsyncEnumerable<ActiveAlarmSnapshot> QueryActiveAlarmsAsync(
|
||||
QueryActiveAlarmsRequest request,
|
||||
CancellationToken cancellationToken = default)
|
||||
{
|
||||
ArgumentNullException.ThrowIfNull(request);
|
||||
ThrowIfDisposed();
|
||||
|
||||
return _transport.QueryActiveAlarmsAsync(request, CreateStreamCallOptions(cancellationToken));
|
||||
}
|
||||
|
||||
/// <summary>
|
||||
/// Disposes the client and releases all resources.
|
||||
/// </summary>
|
||||
|
||||
Reference in New Issue
Block a user