dohertj2/mxaccessgw
1
0
Fork 0
Code Issues Pull Requests 1 Actions Packages Projects Releases Wiki Activity

Point the .NET client at the StreamAlarms alarm feed

Browse Source 
Replace the client's QueryActiveAlarmsAsync with StreamAlarmsAsync —
a session-less subscription to the gateway's central alarm feed that
yields the active-alarm snapshot followed by live transitions.
AcknowledgeAlarm is session-less (AcknowledgeAlarmRequest no longer
carries a session id). Updates the transport interface, the gRPC
transport, the test fake, and the alarm tests; the .NET client
solution builds and its alarm tests pass.

Co-Authored-By: Claude Opus 4.7 (1M context) <noreply@anthropic.com>
This commit is contained in:
Joseph Doherty
2026-05-21 16:30:50 -04:00
parent 40ca4b6908
commit ac12c150c3
5 changed files with 58 additions and 63 deletions
Download Patch File Download Diff File Expand all files Collapse all files
clients/dotnet/MxGateway.Client.Tests/FakeGatewayTransport.cs
+11 -9
View File
@@ -47,9 +47,9 @@ internal sealed class FakeGatewayTransport(MxGatewayClientOptions options) : IMx
public List<(AcknowledgeAlarmRequest Request, CallOptions CallOptions)> AcknowledgeAlarmCalls { get; } = [];
/// <summary>
/// Gets the list of captured QueryActiveAlarmsAsync calls.
/// Gets the list of captured StreamAlarmsAsync calls.
/// </summary>
public List<(QueryActiveAlarmsRequest Request, CallOptions CallOptions)> QueryActiveAlarmsCalls { get; } = [];
public List<(StreamAlarmsRequest Request, CallOptions CallOptions)> StreamAlarmsCalls { get; } = [];
/// <summary>
/// Gets the queue of exceptions to throw from AcknowledgeAlarmAsync.
@@ -223,7 +223,6 @@ internal sealed class FakeGatewayTransport(MxGatewayClientOptions options) : IMx
? _acknowledgeReplies.Dequeue()
: new AcknowledgeAlarmReply
{
SessionId = request.SessionId,
CorrelationId = request.ClientCorrelationId,
ProtocolStatus = new ProtocolStatus { Code = ProtocolStatusCode.Ok },
Status = new MxStatusProxy { Success = 1, Category = MxStatusCategory.Ok },
@@ -231,20 +230,23 @@ internal sealed class FakeGatewayTransport(MxGatewayClientOptions options) : IMx
}
/// <summary>
/// Records the query call and yields each enqueued snapshot.
/// Records the call and yields each enqueued snapshot as an active-alarm
/// feed message, then a snapshot-complete sentinel.
/// </summary>
public async IAsyncEnumerable<ActiveAlarmSnapshot> QueryActiveAlarmsAsync(
QueryActiveAlarmsRequest request,
public async IAsyncEnumerable<AlarmFeedMessage> StreamAlarmsAsync(
StreamAlarmsRequest request,
CallOptions callOptions)
{
QueryActiveAlarmsCalls.Add((request, callOptions));
StreamAlarmsCalls.Add((request, callOptions));
foreach (ActiveAlarmSnapshot snapshot in _activeAlarmSnapshots)
{
callOptions.CancellationToken.ThrowIfCancellationRequested();
await Task.Yield();
yield return snapshot;
yield return new AlarmFeedMessage { ActiveAlarm = snapshot };
}
yield return new AlarmFeedMessage { SnapshotComplete = true };
}
/// <summary>Enqueues an acknowledge reply.</summary>
@@ -253,7 +255,7 @@ internal sealed class FakeGatewayTransport(MxGatewayClientOptions options) : IMx
_acknowledgeReplies.Enqueue(reply);
}
/// <summary>Enqueues a snapshot to be yielded from QueryActiveAlarmsAsync.</summary>
/// <summary>Enqueues a snapshot yielded from StreamAlarmsAsync as an active-alarm message.</summary>
public void AddActiveAlarmSnapshot(ActiveAlarmSnapshot snapshot)
{
_activeAlarmSnapshots.Add(snapshot);
clients/dotnet/MxGateway.Client.Tests/MxGatewayClientAlarmsTests.cs
+19 -27
View File
@@ -5,9 +5,9 @@ using MxGateway.Contracts.Proto;
namespace MxGateway.Client.Tests;
/// <summary>
/// PR E.2 — pins the .NET SDK surface for the new alarm RPCs:
/// Pins the .NET SDK surface for the alarm RPCs:
/// <see cref="MxGatewayClient.AcknowledgeAlarmAsync"/> and
/// <see cref="MxGatewayClient.QueryActiveAlarmsAsync"/>.
/// <see cref="MxGatewayClient.StreamAlarmsAsync"/>.
/// </summary>
public sealed class MxGatewayClientAlarmsTests
{
@@ -17,7 +17,6 @@ public sealed class MxGatewayClientAlarmsTests
FakeGatewayTransport transport = CreateTransport();
transport.AddAcknowledgeReply(new AcknowledgeAlarmReply
{
SessionId = "session-fixture",
CorrelationId = "corr-1",
ProtocolStatus = new ProtocolStatus { Code = ProtocolStatusCode.Ok },
Status = new MxStatusProxy
@@ -31,7 +30,6 @@ public sealed class MxGatewayClientAlarmsTests
AcknowledgeAlarmReply reply = await client.AcknowledgeAlarmAsync(new AcknowledgeAlarmRequest
{
SessionId = "session-fixture",
ClientCorrelationId = "corr-1",
AlarmFullReference = "Tank01.Level.HiHi",
Comment = "investigating",
@@ -64,7 +62,6 @@ public sealed class MxGatewayClientAlarmsTests
client.AcknowledgeAlarmAsync(
new AcknowledgeAlarmRequest
{
SessionId = "session-fixture",
AlarmFullReference = "Tank01.Level.HiHi",
Comment = string.Empty,
OperatorUser = "alice",
@@ -87,7 +84,6 @@ public sealed class MxGatewayClientAlarmsTests
var ex = await Assert.ThrowsAsync<RpcException>(
() => client.AcknowledgeAlarmAsync(new AcknowledgeAlarmRequest
{
SessionId = "session-fixture",
AlarmFullReference = "Tank01.Level.HiHi",
Comment = string.Empty,
OperatorUser = "alice",
@@ -113,7 +109,6 @@ public sealed class MxGatewayClientAlarmsTests
var ex = await Assert.ThrowsAsync<MxGatewayAuthenticationException>(
() => client.AcknowledgeAlarmAsync(new AcknowledgeAlarmRequest
{
SessionId = "session-fixture",
AlarmFullReference = "Tank01.Level.HiHi",
Comment = string.Empty,
OperatorUser = "alice",
@@ -122,50 +117,47 @@ public sealed class MxGatewayClientAlarmsTests
}
[Fact]
public async Task QueryActiveAlarmsAsync_StreamsEnqueuedSnapshots()
public async Task StreamAlarmsAsync_StreamsSnapshotThenSnapshotComplete()
{
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
List<AlarmFeedMessage> messages = [];
await foreach (AlarmFeedMessage message in client.StreamAlarmsAsync(new StreamAlarmsRequest()))
{
SessionId = "session-fixture",
}))
{
snapshots.Add(snapshot);
messages.Add(message);
}
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);
Assert.Equal(3, messages.Count);
Assert.Equal("Tank01.Level.HiHi", messages[0].ActiveAlarm.AlarmFullReference);
Assert.Equal(AlarmConditionState.Active, messages[0].ActiveAlarm.CurrentState);
Assert.Equal(AlarmConditionState.ActiveAcked, messages[1].ActiveAlarm.CurrentState);
Assert.True(messages[2].SnapshotComplete);
Assert.Single(transport.StreamAlarmsCalls);
}
[Fact]
public async Task QueryActiveAlarmsAsync_PassesFilterPrefix()
public async Task StreamAlarmsAsync_PassesFilterPrefix()
{
FakeGatewayTransport transport = CreateTransport();
await using MxGatewayClient client = CreateClient(transport);
await foreach (ActiveAlarmSnapshot _ in client.QueryActiveAlarmsAsync(new QueryActiveAlarmsRequest
await foreach (AlarmFeedMessage _ in client.StreamAlarmsAsync(new StreamAlarmsRequest
{
SessionId = "session-fixture",
AlarmFilterPrefix = "Tank01.",
}))
{
// no snapshots enqueued; just verifying the request passes through
// only the snapshot-complete sentinel; verifying the request passes through
}
var call = Assert.Single(transport.QueryActiveAlarmsCalls);
var call = Assert.Single(transport.StreamAlarmsCalls);
Assert.Equal("Tank01.", call.Request.AlarmFilterPrefix);
}
[Fact]
public async Task QueryActiveAlarmsAsync_HonorsCancellationDuringEnumeration()
public async Task StreamAlarmsAsync_HonorsCancellationDuringEnumeration()
{
FakeGatewayTransport transport = CreateTransport();
transport.AddActiveAlarmSnapshot(MakeSnapshot("Tank01.Level.HiHi", AlarmConditionState.Active));
@@ -175,8 +167,8 @@ public sealed class MxGatewayClientAlarmsTests
using CancellationTokenSource cancellation = new();
await Assert.ThrowsAsync<OperationCanceledException>(async () =>
{
await foreach (ActiveAlarmSnapshot _ in client.QueryActiveAlarmsAsync(
new QueryActiveAlarmsRequest { SessionId = "session-fixture" },
await foreach (AlarmFeedMessage _ in client.StreamAlarmsAsync(
new StreamAlarmsRequest(),
cancellation.Token))
{
cancellation.Cancel();
clients/dotnet/MxGateway.Client/GrpcMxGatewayClientTransport.cs
+10 -10
View File
@@ -134,8 +134,8 @@ internal sealed class GrpcMxGatewayClientTransport(
}
/// <inheritdoc />
public async IAsyncEnumerable<ActiveAlarmSnapshot> QueryActiveAlarmsAsync(
QueryActiveAlarmsRequest request,
public async IAsyncEnumerable<AlarmFeedMessage> StreamAlarmsAsync(
StreamAlarmsRequest request,
CallOptions callOptions,
[System.Runtime.CompilerServices.EnumeratorCancellation] CancellationToken cancellationToken = default)
{
@@ -143,12 +143,12 @@ internal sealed class GrpcMxGatewayClientTransport(
? cancellationToken
: callOptions.CancellationToken;
using AsyncServerStreamingCall<ActiveAlarmSnapshot> call = RawClient.QueryActiveAlarms(request, callOptions);
using AsyncServerStreamingCall<AlarmFeedMessage> call = RawClient.StreamAlarms(request, callOptions);
IAsyncStreamReader<ActiveAlarmSnapshot> responseStream = call.ResponseStream;
IAsyncStreamReader<AlarmFeedMessage> responseStream = call.ResponseStream;
while (true)
{
ActiveAlarmSnapshot? snapshot;
AlarmFeedMessage? message;
try
{
if (!await responseStream.MoveNext(effectiveCancellationToken).ConfigureAwait(false))
@@ -156,22 +156,22 @@ internal sealed class GrpcMxGatewayClientTransport(
break;
}
snapshot = responseStream.Current;
message = responseStream.Current;
}
catch (RpcException exception)
{
throw RpcExceptionMapper.Map(exception, effectiveCancellationToken);
}
yield return snapshot;
yield return message;
}
}
/// <inheritdoc />
IAsyncEnumerable<ActiveAlarmSnapshot> IMxGatewayClientTransport.QueryActiveAlarmsAsync(
QueryActiveAlarmsRequest request,
IAsyncEnumerable<AlarmFeedMessage> IMxGatewayClientTransport.StreamAlarmsAsync(
StreamAlarmsRequest request,
CallOptions callOptions)
{
return QueryActiveAlarmsAsync(request, callOptions);
return StreamAlarmsAsync(request, callOptions);
}
}
clients/dotnet/MxGateway.Client/IMxGatewayClientTransport.cs
+6 -6
View File
@@ -66,13 +66,13 @@ internal interface IMxGatewayClientTransport
CallOptions callOptions);
/// <summary>
/// Streams a snapshot of all alarms currently in Active or ActiveAcked state — the
/// ConditionRefresh equivalent for the gateway.
/// Attaches to the gateway's central alarm feed — the current active-alarm
/// snapshot followed by live transitions.
/// </summary>
/// <param name="request">The query request, optionally scoped by alarm-reference prefix.</param>
/// <param name="request">The stream 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,
/// <returns>An async enumerable of alarm feed messages.</returns>
IAsyncEnumerable<AlarmFeedMessage> StreamAlarmsAsync(
StreamAlarmsRequest request,
CallOptions callOptions);
}
clients/dotnet/MxGateway.Client/MxGatewayClient.cs
+12 -11
View File
@@ -205,24 +205,25 @@ public sealed class MxGatewayClient : IAsyncDisposable
}
/// <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.
/// Attaches to the gateway's central alarm feed. The stream opens with one
/// <see cref="AlarmFeedMessage"/> per currently-active alarm (the
/// ConditionRefresh snapshot), then a single <c>snapshot_complete</c>, then a
/// <c>transition</c> for every subsequent raise / acknowledge / clear. Served
/// by the gateway's always-on alarm monitor — no worker session is opened, so
/// any number of clients may attach. Optionally scoped by alarm-reference
/// prefix (<see cref="StreamAlarmsRequest.AlarmFilterPrefix"/>).
/// </summary>
/// <param name="request">The query request, optionally scoped by alarm-reference prefix.</param>
/// <param name="request">The stream 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,
/// <returns>An async enumerable of alarm feed messages.</returns>
public IAsyncEnumerable<AlarmFeedMessage> StreamAlarmsAsync(
StreamAlarmsRequest request,
CancellationToken cancellationToken = default)
{
ArgumentNullException.ThrowIfNull(request);
ThrowIfDisposed();
return _transport.QueryActiveAlarmsAsync(request, CreateStreamCallOptions(cancellationToken));
return _transport.StreamAlarmsAsync(request, CreateStreamCallOptions(cancellationToken));
}
/// <summary>