diff --git a/clients/dotnet/ZB.MOM.WW.MxGateway.Client.Cli/MxGatewayClientCli.cs b/clients/dotnet/ZB.MOM.WW.MxGateway.Client.Cli/MxGatewayClientCli.cs index 5320e3c..cdd6e7d 100644 --- a/clients/dotnet/ZB.MOM.WW.MxGateway.Client.Cli/MxGatewayClientCli.cs +++ b/clients/dotnet/ZB.MOM.WW.MxGateway.Client.Cli/MxGatewayClientCli.cs @@ -1418,14 +1418,16 @@ public static class MxGatewayClientCli .WithCancellation(cancellationToken) .ConfigureAwait(false)) { - if (jsonLines) - { - output.WriteLine(ProtobufJsonFormatter.Format(gatewayEvent)); - } - else if (json) + if (json && !jsonLines) { events.Add(gatewayEvent); } + else if (gatewayEvent.ReplayGap is { } replayGap) + { + // Render the ReplayGap sentinel as the typed cross-CLI row instead of the raw + // sentinel MxEvent (NEXT-02, mirroring the Go/Python/Rust CLIs). + output.WriteLine(FormatReplayGapRow(replayGap)); + } else { output.WriteLine(ProtobufJsonFormatter.Format(gatewayEvent)); @@ -1835,7 +1837,31 @@ public static class MxGatewayClientCli private static JsonElement EventToJsonElement(MxEvent gatewayEvent) { - return JsonDocument.Parse(ProtobufJsonFormatter.Format(gatewayEvent)).RootElement.Clone(); + string row = gatewayEvent.ReplayGap is { } replayGap + ? FormatReplayGapRow(replayGap) + : ProtobufJsonFormatter.Format(gatewayEvent); + return JsonDocument.Parse(row).RootElement.Clone(); + } + + /// + /// Formats the typed ReplayGap row shared by the CLIs (NEXT-02). Hand-built so the + /// cursors are JSON numbers like the Go/Python/Rust rows, not the protobuf JSON + /// formatter's quoted uint64 strings. + /// + /// Replay gap sentinel payload. + /// A single-line JSON row describing the gap. + private static string FormatReplayGapRow(ReplayGap replayGap) + { + return JsonSerializer.Serialize( + new + { + replayGap = new + { + requestedAfterSequence = replayGap.RequestedAfterSequence, + oldestAvailableSequence = replayGap.OldestAvailableSequence, + }, + }, + JsonOptions); } private static MxValue ParseValue(CliArguments arguments) diff --git a/clients/dotnet/ZB.MOM.WW.MxGateway.Client.Tests/MxGatewayClientCliTests.cs b/clients/dotnet/ZB.MOM.WW.MxGateway.Client.Tests/MxGatewayClientCliTests.cs index 2a98d75..726f591 100644 --- a/clients/dotnet/ZB.MOM.WW.MxGateway.Client.Tests/MxGatewayClientCliTests.cs +++ b/clients/dotnet/ZB.MOM.WW.MxGateway.Client.Tests/MxGatewayClientCliTests.cs @@ -1,3 +1,4 @@ +using System.Text.Json; using Google.Protobuf.WellKnownTypes; using ZB.MOM.WW.MxGateway.Client.Cli; using ZB.MOM.WW.MxGateway.Contracts.Proto; @@ -585,6 +586,84 @@ public sealed class MxGatewayClientCliTests Assert.DoesNotContain("ON_WRITE_COMPLETE", output.ToString()); } + /// + /// Verifies stream-events renders the ReplayGap sentinel as the typed cross-CLI row — + /// numeric cursors under a replayGap key — instead of the raw sentinel MxEvent (NEXT-02). + /// + /// A task that represents the asynchronous operation. + [Fact] + public async Task RunAsync_StreamEvents_RendersReplayGapAsTypedRow() + { + using var output = new StringWriter(); + using var error = new StringWriter(); + FakeCliClient fakeClient = new(); + fakeClient.Events.Add(new MxEvent + { + ReplayGap = new ReplayGap + { + RequestedAfterSequence = 7, + OldestAvailableSequence = 42, + }, + }); + fakeClient.Events.Add(new MxEvent + { + SessionId = "session-fixture", + Family = MxEventFamily.OnDataChange, + WorkerSequence = 43, + }); + + int exitCode = await MxGatewayClientCli.RunAsync( + [ + "stream-events", + "--endpoint", + "http://localhost:5000", + "--api-key", + "test-api-key", + "--session-id", + "session-fixture", + "--max-events", + "2", + ], + output, + error, + _ => fakeClient); + + Assert.Equal(0, exitCode); + string[] rows = output.ToString().Split(Environment.NewLine, StringSplitOptions.RemoveEmptyEntries); + Assert.Equal(2, rows.Length); + using JsonDocument gapRow = JsonDocument.Parse(rows[0]); + JsonElement gap = gapRow.RootElement.GetProperty("replayGap"); + Assert.Equal(7UL, gap.GetProperty("requestedAfterSequence").GetUInt64()); + Assert.Equal(42UL, gap.GetProperty("oldestAvailableSequence").GetUInt64()); + Assert.Equal(JsonValueKind.Number, gap.GetProperty("requestedAfterSequence").ValueKind); + Assert.DoesNotContain("MX_EVENT_FAMILY_UNSPECIFIED", rows[0], StringComparison.Ordinal); + Assert.Contains("workerSequence", rows[1], StringComparison.Ordinal); + + // The aggregate --json shape carries the same typed row inside the events array. + using var aggregateOutput = new StringWriter(); + int aggregateExit = await MxGatewayClientCli.RunAsync( + [ + "stream-events", + "--endpoint", + "http://localhost:5000", + "--api-key", + "test-api-key", + "--session-id", + "session-fixture", + "--max-events", + "2", + "--json", + ], + aggregateOutput, + error, + _ => fakeClient); + + Assert.Equal(0, aggregateExit); + using JsonDocument aggregate = JsonDocument.Parse(aggregateOutput.ToString()); + JsonElement firstRow = aggregate.RootElement.GetProperty("events")[0]; + Assert.Equal(42UL, firstRow.GetProperty("replayGap").GetProperty("oldestAvailableSequence").GetUInt64()); + } + /// Verifies that stream-alarms with --max-events stops output and distinguishes payload cases. /// A task that represents the asynchronous operation. diff --git a/clients/java/zb-mom-ww-mxgateway-cli/src/main/java/com/zb/mom/ww/mxgateway/cli/MxGatewayCli.java b/clients/java/zb-mom-ww-mxgateway-cli/src/main/java/com/zb/mom/ww/mxgateway/cli/MxGatewayCli.java index 9bb0f43..3fe632e 100644 --- a/clients/java/zb-mom-ww-mxgateway-cli/src/main/java/com/zb/mom/ww/mxgateway/cli/MxGatewayCli.java +++ b/clients/java/zb-mom-ww-mxgateway-cli/src/main/java/com/zb/mom/ww/mxgateway/cli/MxGatewayCli.java @@ -5,6 +5,7 @@ import com.zb.mom.ww.mxgateway.client.DeployEventStream; import com.zb.mom.ww.mxgateway.client.GalaxyRepositoryClient; import com.zb.mom.ww.mxgateway.client.LazyBrowseNode; import com.zb.mom.ww.mxgateway.client.MxEventStream; +import com.zb.mom.ww.mxgateway.client.MxEventStreamItem; import com.zb.mom.ww.mxgateway.client.MxGatewayAlarmFeedSubscription; import com.zb.mom.ww.mxgateway.client.MxGatewayClient; import com.zb.mom.ww.mxgateway.client.MxGatewayClientOptions; @@ -59,6 +60,7 @@ import mxaccess_gateway.v1.MxaccessGateway.MxValue; import mxaccess_gateway.v1.MxaccessGateway.OnAlarmTransitionEvent; import mxaccess_gateway.v1.MxaccessGateway.OpenSessionRequest; import mxaccess_gateway.v1.MxaccessGateway.PingCommand; +import mxaccess_gateway.v1.MxaccessGateway.ReplayGap; import mxaccess_gateway.v1.MxaccessGateway.StreamAlarmsRequest; import mxaccess_gateway.v1.MxaccessGateway.SubscribeResult; import mxaccess_gateway.v1.MxaccessGateway.Write2BulkEntry; @@ -1654,11 +1656,30 @@ public final class MxGatewayCli implements Callable { MxEventStream events = client.session(sessionId).streamEventsAfter(afterWorkerSequence)) { int count = 0; while (events.hasNext()) { - MxEvent event = events.next(); - if (json) { - client.out().println(protoJson(event)); + MxEventStreamItem item = events.nextItem(); + if (item.isReplayGap()) { + // Render the ReplayGap sentinel as the typed cross-CLI row (NEXT-02, + // mirroring the Go/Python/Rust/.NET CLIs) instead of the raw sentinel + // event, whose text form printed "0 MX_EVENT_FAMILY_UNSPECIFIED". + ReplayGap gap = item.replayGap(); + if (json) { + client.out().printf( + "{\"replayGap\":{\"requestedAfterSequence\":%s,\"oldestAvailableSequence\":%s}}%n", + Long.toUnsignedString(gap.getRequestedAfterSequence()), + Long.toUnsignedString(gap.getOldestAvailableSequence())); + } else { + client.out().printf( + "REPLAY_GAP requested_after=%s oldest_available=%s%n", + Long.toUnsignedString(gap.getRequestedAfterSequence()), + Long.toUnsignedString(gap.getOldestAvailableSequence())); + } } else { - client.out().printf("%d %s%n", event.getWorkerSequence(), event.getFamily()); + MxEvent event = item.event(); + if (json) { + client.out().println(protoJson(event)); + } else { + client.out().printf("%d %s%n", event.getWorkerSequence(), event.getFamily()); + } } count++; if (limit > 0 && count >= limit) { diff --git a/clients/java/zb-mom-ww-mxgateway-cli/src/test/java/com/zb/mom/ww/mxgateway/cli/MxGatewayCliTests.java b/clients/java/zb-mom-ww-mxgateway-cli/src/test/java/com/zb/mom/ww/mxgateway/cli/MxGatewayCliTests.java index 09610bd..c3dd35d 100644 --- a/clients/java/zb-mom-ww-mxgateway-cli/src/test/java/com/zb/mom/ww/mxgateway/cli/MxGatewayCliTests.java +++ b/clients/java/zb-mom-ww-mxgateway-cli/src/test/java/com/zb/mom/ww/mxgateway/cli/MxGatewayCliTests.java @@ -43,6 +43,7 @@ import mxaccess_gateway.v1.MxaccessGateway.OpenSessionRequest; import mxaccess_gateway.v1.MxaccessGateway.ProtocolStatus; import mxaccess_gateway.v1.MxaccessGateway.ProtocolStatusCode; import mxaccess_gateway.v1.MxaccessGateway.RegisterReply; +import mxaccess_gateway.v1.MxaccessGateway.ReplayGap; import mxaccess_gateway.v1.MxaccessGateway.SessionState; import mxaccess_gateway.v1.MxaccessGateway.StreamAlarmsRequest; import mxaccess_gateway.v1.MxaccessGateway.SubscribeResult; @@ -902,6 +903,59 @@ final class MxGatewayCliTests { } } + @Test + void streamEventsRendersReplayGapAsTypedRow() { + // NEXT-02: the ReplayGap sentinel must render as the typed cross-CLI + // row (numeric cursors under a replayGap key in --json, a REPLAY_GAP + // line in text mode), never as the raw sentinel event — text mode + // used to print "0 MX_EVENT_FAMILY_UNSPECIFIED". + MxEvent gap = MxEvent.newBuilder() + .setReplayGap(ReplayGap.newBuilder() + .setRequestedAfterSequence(7L) + .setOldestAvailableSequence(42L) + .build()) + .build(); + MxEvent dataChange = MxEvent.newBuilder() + .setFamily(MxEventFamily.MX_EVENT_FAMILY_ON_DATA_CHANGE) + .setSessionId("session-cli") + .setWorkerSequence(43L) + .build(); + + try (InProcessGatewayHarness harness = new InProcessGatewayHarness()) { + harness.setScriptedEvents(List.of(gap, dataChange)); + CliRun jsonRun = execute( + new HarnessClientFactory(harness), + "stream-events", + "--session-id", + "session-cli", + "--json"); + + assertEquals(0, jsonRun.exitCode(), "errors:\n" + jsonRun.errors()); + String jsonOut = jsonRun.output(); + assertTrue( + jsonOut.contains( + "{\"replayGap\":{\"requestedAfterSequence\":7,\"oldestAvailableSequence\":42}}"), + jsonOut); + assertTrue(jsonOut.contains("\"family\":\"MX_EVENT_FAMILY_ON_DATA_CHANGE\""), jsonOut); + assertFalse(jsonOut.contains("MX_EVENT_FAMILY_UNSPECIFIED"), jsonOut); + } + + try (InProcessGatewayHarness harness = new InProcessGatewayHarness()) { + harness.setScriptedEvents(List.of(gap, dataChange)); + CliRun textRun = execute( + new HarnessClientFactory(harness), + "stream-events", + "--session-id", + "session-cli"); + + assertEquals(0, textRun.exitCode(), "errors:\n" + textRun.errors()); + String textOut = textRun.output(); + assertTrue(textOut.contains("REPLAY_GAP requested_after=7 oldest_available=42"), textOut); + assertFalse(textOut.contains("MX_EVENT_FAMILY_UNSPECIFIED"), textOut); + assertTrue(textOut.contains("43 MX_EVENT_FAMILY_ON_DATA_CHANGE"), textOut); + } + } + // ---- galaxy-discover / galaxy-watch over the in-process harness (Task 6) ---- @Test diff --git a/docs/CrossLanguageSmokeMatrix.md b/docs/CrossLanguageSmokeMatrix.md index acd1a37..4a7e0bb 100644 --- a/docs/CrossLanguageSmokeMatrix.md +++ b/docs/CrossLanguageSmokeMatrix.md @@ -40,27 +40,24 @@ reports the next deliverable sequence rather than `0` (see [Sessions](Sessions.m The default smoke sequence opens a fresh stream (no cursor) and does not exercise the gap path; a resume-with-gap fixture case is tracked separately (TST-24). -The CLIs differ in how they *print* that library-level signal. Three of them consume -the typed gap and emit a dedicated row rather than a degenerate event row; the other -two hand the raw sentinel `MxEvent` straight to the formatter, so they print the -sentinel itself, whose `replayGap` field carries the same cursors: +All five CLIs consume the typed gap and emit a dedicated row rather than a +degenerate event row (the .NET and Java halves were the last to convert — NEXT-02): | CLI | Text mode | JSON mode | |-----|-----------|-----------| | `mxgw-rs` (Rust, canonical) | `REPLAY_GAP requested_after= oldest_available=` | `{"replayGap": {"requestedAfterSequence": , "oldestAvailableSequence": }}` as one entry of the `events` array | | `mxgw-go` (Go) | `REPLAY_GAP requested_after= oldest_available=` | one `{"replayGap": {"requestedAfterSequence": , "oldestAvailableSequence": }}` line, counted toward `-limit` like any other row | | `mxgw-py` (Python) | same JSON dump as `--json` | `{"replayGap": {"requestedAfterSequence": , "oldestAvailableSequence": }}` as one entry of the `events` array | -| `mxgw-dotnet` (.NET) | the raw sentinel `MxEvent` as protobuf JSON, including its `replayGap` field | same, as one entry of the `events` array | -| `mxgw-java` (Java) | the sentinel's `worker_sequence` and `family` (`0 MX_EVENT_FAMILY_UNSPECIFIED`) | the raw sentinel `MxEvent` as protobuf JSON, including its `replayGap` field | +| `mxgw-dotnet` (.NET) | one `{"replayGap": {...}}` line (its "text" mode is JSON-per-line) | the same row — per line with `--jsonl`, as one entry of the `events` array with `--json` | +| `mxgw-java` (Java) | `REPLAY_GAP requested_after= oldest_available=` | one `{"replayGap": {"requestedAfterSequence": , "oldestAvailableSequence": }}` line | -Rust, Go, and Python emit the same two key names and, deliberately, the same JSON -value **types**: the cursors are JSON numbers (`7`), not strings. That is why the -Go CLI types the row by hand instead of marshalling `ReplayGap` with `protojson` — -the proto3 JSON mapping renders 64-bit integers as strings (`"7"`), which is also -why the .NET and Java rows, which pass the sentinel through a protobuf JSON -formatter, carry **quoted** cursors. A matrix runner must therefore compare parsed -values, not raw bytes, and must not assume the same value type across all five -CLIs. +All five emit the same two key names and, deliberately, the same JSON value +**types**: the cursors are JSON numbers (`7`), not strings. That is why every +CLI types the row by hand instead of marshalling `ReplayGap` through its +protobuf JSON formatter — the proto3 JSON mapping renders 64-bit integers as +strings (`"7"`). Normal event rows still come from the protobuf formatters, so +a matrix runner must still compare parsed values, not raw bytes, when it mixes +gap rows with event rows. Two further formatting differences among the three canonical CLIs, none of them semantic: Python sorts object keys and uses `", "` / `": "` separators