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

fix(clients): render ReplayGap as the typed cross-CLI row in the .NET and Java CLIs (NEXT-02)

Browse Source 
The .NET and Java stream-events commands handed the raw ReplayGap sentinel
MxEvent to their protobuf JSON formatters (Java text mode printed
'0 MX_EVENT_FAMILY_UNSPECIFIED'), while the Go/Python/Rust CLIs already emit
the typed row (CLI-35/36). Both now branch on the sentinel: Java text mode
prints 'REPLAY_GAP requested_after=<n> oldest_available=<n>' and JSON mode a
hand-built {"replayGap":{...}} line via nextItem()/isReplayGap(); the .NET
CLI emits the same hand-built row in jsonl/text (its text mode is
JSON-per-line) and inside the --json events array. Rows are hand-built so
the cursors are JSON numbers like the other three CLIs, not the proto3 JSON
mapping's quoted uint64 strings — the CrossLanguageSmokeMatrix divergence
table collapses to a single converged contract.

Tests: .NET MxGatewayClientCliTests 35/35 (new RendersReplayGapAsTypedRow
covers jsonl + aggregate); Java gradle test 52/52 (new
streamEventsRendersReplayGapAsTypedRow covers --json + text over the
in-process harness). No generated-file churn.
This commit is contained in:
Joseph Doherty
2026-08-10 06:01:44 -04:00
parent 84dbf20a43
commit 8624e21372
5 changed files with 201 additions and 24 deletions
Download Patch File Download Diff File Expand all files Collapse all files
clients/dotnet/ZB.MOM.WW.MxGateway.Client.Cli/MxGatewayClientCli.cs
+32 -6
View File
@@ -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();
}
/// <summary>
/// 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.
/// </summary>
/// <param name="replayGap">Replay gap sentinel payload.</param>
/// <returns>A single-line JSON row describing the gap.</returns>
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)
clients/dotnet/ZB.MOM.WW.MxGateway.Client.Tests/MxGatewayClientCliTests.cs
+79
View File
@@ -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());
}
/// <summary>
/// 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).
/// </summary>
/// <returns>A task that represents the asynchronous operation.</returns>
[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());
}
/// <summary>Verifies that stream-alarms with --max-events stops output and distinguishes payload cases.</summary>
/// <returns>A task that represents the asynchronous operation.</returns>
clients/java/zb-mom-ww-mxgateway-cli/src/main/java/com/zb/mom/ww/mxgateway/cli/MxGatewayCli.java
+25 -4
View File
@@ -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<Integer> {
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) {
clients/java/zb-mom-ww-mxgateway-cli/src/test/java/com/zb/mom/ww/mxgateway/cli/MxGatewayCliTests.java
+54
View File
@@ -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