diff --git a/CLAUDE.md b/CLAUDE.md index a1ddf64..bf44902 100644 --- a/CLAUDE.md +++ b/CLAUDE.md @@ -76,7 +76,7 @@ powershell -ExecutionPolicy Bypass -File scripts/run-client-e2e-tests.ps1 - **Style guides** in `docs/style-guides/` are authoritative. Follow `CSharpStyleGuide.md` for gateway/worker/.NET-client code: file-scoped namespaces, `sealed` by default, `Async` suffix on Task-returning methods, MXAccess-aligned names (`MxStatusProxy`, `ServerHandle`, `ItemHandle`, `HResult`). - **MXAccess parity is the contract.** Don't "fix" surprising MXAccess behavior (e.g., `WriteSecured` failing before a value-bearing NMX body, distinct `OperationComplete` semantics, invalid-handle exceptions) unless the client explicitly opts into a non-parity mode. The installed MXAccess COM component is the baseline. - **Don't synthesize events.** The gateway forwards only events the worker emits; it never invents `OperationComplete` from write completion or command replies. -- **One worker per session** (invariant). Multi-subscriber event fan-out and reconnect-with-replay have shipped and are config-gated: `AllowMultipleEventSubscribers` (default `false`) enables fan-out up to `MaxEventSubscribersPerSession` (default `8`); `DetachGraceSeconds` (default `30`) retains a session after its last subscriber drops so clients can reconnect; `ReplayBufferCapacity` / `ReplayRetentionSeconds` control how much event history the replay ring keeps. Default config is single-subscriber (`AllowMultipleEventSubscribers` off), but detach-grace and replay retention are **on** by default (`DetachGraceSeconds=30`, `ReplayBufferCapacity=1024`, `ReplayRetentionSeconds=300`): a detached session is retained for 30 s and recent events are buffered for reconnect. The reconnect protocol is consumable end-to-end: a resuming `StreamEvents` (via `after_worker_sequence`) that predates the retained ring gets a `ReplayGap` sentinel, and the official clients surface it as a typed signal (shipped for .NET/Go/Rust/Python; the Java client is the remaining one, batched to windev). Orphan-worker reattach after a gateway restart is **deferred, not planned** — see `oldtasks.md` (session-resilience epic Phase 5); the invariant on the next line stands. See `docs/DesignDecisions.md` and `docs/Sessions.md`. +- **One worker per session** (invariant). Multi-subscriber event fan-out and reconnect-with-replay have shipped and are config-gated: `AllowMultipleEventSubscribers` (default `false`) enables fan-out up to `MaxEventSubscribersPerSession` (default `8`); `DetachGraceSeconds` (default `30`) retains a session after its last subscriber drops so clients can reconnect; `ReplayBufferCapacity` / `ReplayRetentionSeconds` control how much event history the replay ring keeps. Default config is single-subscriber (`AllowMultipleEventSubscribers` off), but detach-grace and replay retention are **on** by default (`DetachGraceSeconds=30`, `ReplayBufferCapacity=1024`, `ReplayRetentionSeconds=300`): a detached session is retained for 30 s and recent events are buffered for reconnect. The reconnect protocol is consumable end-to-end: a resuming `StreamEvents` (via `after_worker_sequence`) that predates the retained ring gets a `ReplayGap` sentinel, and all five official clients surface it as a typed signal. Orphan-worker reattach after a gateway restart is **deferred, not planned** — see `oldtasks.md` (session-resilience epic Phase 5); the invariant on the next line stands. See `docs/DesignDecisions.md` and `docs/Sessions.md`. - **Gateway restart does not reattach orphan workers.** The first version terminates orphaned workers on startup; do not design code paths that assume reattachment. - **No Blazor UI component libraries.** Dashboard uses local Bootstrap CSS/JS only — do not introduce MudBlazor, Radzen, FluentUI, etc. - **Don't log secrets or full tag values by default.** API keys, passwords, `WriteSecured` payloads, and `AuthenticateUser` credentials must never reach logs. Value logging is opt-in and redacted. diff --git a/clients/java/README.md b/clients/java/README.md index d7bead6..2e0b117 100644 --- a/clients/java/README.md +++ b/clients/java/README.md @@ -76,7 +76,40 @@ data-bearing MXAccess failure. `MxEventStream` implements `Iterator` and `AutoCloseable`. Closing it cancels the underlying gRPC stream. Canceling or timing out a Java client call only stops the client from waiting; it does not abort an in-flight MXAccess COM -call on the worker STA. +call on the worker STA. It is a **single-consumer** surface: drive +`hasNext()`/`next()` (or `nextItem()`) from one thread only. + +### Reconnect-replay gap signal + +When you resume a stream with `streamEventsAfter(afterWorkerSequence)` and the +requested cursor predates the oldest event the gateway still retains, the +gateway emits a single **replay-gap sentinel** at the head of the stream: an +`MxEvent` with its `replay_gap` field set, `family` unspecified, and the body +oneof unset. It means "you missed events — discard cached state and +re-snapshot": the events in the open interval `(requested_after_sequence, +oldest_available_sequence)` were evicted and cannot be replayed. The gateway +never synthesizes this signal from anything else, and the client never swallows +it. + +Use `nextItem()` to branch on it as a distinct typed item; `next()` still +returns the sentinel as a plain `MxEvent` (test `event.hasReplayGap()`). After a +gap, re-snapshot, then resume without another gap by requesting +`oldestAvailableSequence - 1` as the next `afterWorkerSequence`: + +```java +try (MxEventStream events = session.streamEventsAfter(lastSeenSequence)) { + while (events.hasNext()) { + MxEventStreamItem item = events.nextItem(); + if (item.isReplayGap()) { + long resumeFrom = item.replayGap().getOldestAvailableSequence() - 1; + // discard cached per-item state, re-snapshot, then resume from resumeFrom + continue; + } + MxEvent event = item.event(); + // normal event handling + } +} +``` For alarms, `MxGatewayClient` exposes `queryActiveAlarms` (one-shot snapshot), `streamAlarms` (returns an `MxGatewayAlarmFeedSubscription` whose iterator @@ -89,30 +122,52 @@ ack target). Close the subscription to cancel the underlying gRPC stream. These are MXAccess parity behaviors that surprise new callers. The gateway forwards them unchanged — it does not paper over them. +### Typed single-item command helpers + +`MxGatewaySession` exposes typed helpers for the parity-critical single-item +commands, so you do not need to build raw `MxCommand` messages: + +- `adviseSupervisory(serverHandle, itemHandle)` (and `adviseSupervisoryRaw`) +- `writeSecured(serverHandle, itemHandle, currentUserId, verifierUserId, value)` + and `writeSecured2(..., timestampValue)` (plus `*Raw` variants) +- `authenticateUser(serverHandle, verifyUser, verifyUserPassword)` → user id +- `archestrAUserToId(serverHandle, userIdGuid)` → user id +- `addBufferedItem(serverHandle, itemDefinition, itemContext)` → item handle +- `setBufferedUpdateInterval(serverHandle, updateIntervalMs)` +- `suspend(serverHandle, itemHandle)` / `activate(serverHandle, itemHandle)` → + the reply's `MxStatusProxy` + +All of them run the same MXAccess reply validation as the bulk helpers (protocol +status plus HRESULT/`MxStatusProxy` check) via the shared `invoke` path, so an +MXAccess COM-side failure surfaces as `MxAccessException`. + +**Secret redaction.** Credentials passed to `authenticateUser` (and the +credential-sensitive values passed to `writeSecured`/`writeSecured2`) travel +only in the request. They never appear in logs, exception messages, or +`toString()`: gateway status text is scrubbed through `MxGatewaySecrets`, and +MXAccess failures carry only the reply (never the request). Do not log the +credentials yourself. + ### Attributing a write to a user without `authenticateUser` MXAccess only stamps a plain `write`/`write2` with a Galaxy user id when the item carries an active *supervisory* advise. If you are **not** using the verified/secured path (`authenticateUser` → `writeSecured`/`writeSecured2`) but -still need the write attributed to a user id, you must first advise the item -supervisory and then pass that user id on the write. Without the supervisory -advise the `userId` on a plain write is ignored. - -The session exposes `advise`/`unAdvise` but not supervisory advise, so send it -through the generic command channel: +still need the write attributed to a user id, first advise the item supervisory +and then pass that user id on the write. Without the supervisory advise the +`userId` on a plain write is ignored. ```java -session.invokeCommand(MxCommand.newBuilder() - .setKind(MxCommandKind.MX_COMMAND_KIND_ADVISE_SUPERVISORY) - .setAdviseSupervisory(AdviseSupervisoryCommand.newBuilder() - .setServerHandle(serverHandle) - .setItemHandle(itemHandle)) - .build()); - +session.adviseSupervisory(serverHandle, itemHandle); session.write(serverHandle, itemHandle, value, userId); ``` -The CLI exposes the same command as `advise-supervisory`, and `write` / +**MXAccess parity:** `writeSecured` failing before a prior `authenticateUser` + +`adviseSupervisory`, or before a value-bearing body, is correct behavior — the +native failure is surfaced, not papered over. + +The CLI exposes `advise-supervisory`, `write-secured`, and `authenticate-user` +(credential via `--password` or `--password-env`, never echoed), and `write` / `write2` take `--user-id`. ### Array writes replace the whole array @@ -324,6 +379,9 @@ gradle :zb-mom-ww-mxgateway-cli:run --args="register --endpoint localhost:5000 - gradle :zb-mom-ww-mxgateway-cli:run --args="add-item --endpoint localhost:5000 --api-key-env MXGATEWAY_API_KEY --plaintext --session-id --server-handle 1 --item TestObject.TestInt --json" gradle :zb-mom-ww-mxgateway-cli:run --args="advise --endpoint localhost:5000 --api-key-env MXGATEWAY_API_KEY --plaintext --session-id --server-handle 1 --item-handle 1 --json" gradle :zb-mom-ww-mxgateway-cli:run --args="write --endpoint localhost:5000 --api-key-env MXGATEWAY_API_KEY --plaintext --session-id --server-handle 1 --item-handle 1 --type int32 --value 123 --json" +gradle :zb-mom-ww-mxgateway-cli:run --args="advise-supervisory --endpoint localhost:5000 --api-key-env MXGATEWAY_API_KEY --plaintext --session-id --server-handle 1 --item-handle 1 --json" +gradle :zb-mom-ww-mxgateway-cli:run --args="authenticate-user --endpoint localhost:5000 --api-key-env MXGATEWAY_API_KEY --plaintext --session-id --server-handle 1 --verify-user operator --password-env MXGATEWAY_VERIFY_PASSWORD --json" +gradle :zb-mom-ww-mxgateway-cli:run --args="write-secured --endpoint localhost:5000 --api-key-env MXGATEWAY_API_KEY --plaintext --session-id --server-handle 1 --item-handle 1 --type int32 --value 123 --current-user-id 100 --verifier-user-id 100 --json" gradle :zb-mom-ww-mxgateway-cli:run --args="stream-events --endpoint localhost:5000 --api-key-env MXGATEWAY_API_KEY --plaintext --session-id --limit 1 --json" gradle :zb-mom-ww-mxgateway-cli:run --args="stream-alarms --endpoint localhost:5000 --api-key-env MXGATEWAY_API_KEY --plaintext --filter-prefix Galaxy --limit 1 --json" gradle :zb-mom-ww-mxgateway-cli:run --args="acknowledge-alarm --endpoint localhost:5000 --api-key-env MXGATEWAY_API_KEY --plaintext --reference \"\\Galaxy\Area001.Pump001.PumpFault\" --json" 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 32ef7a5..181f14b 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 @@ -155,6 +155,8 @@ public final class MxGatewayCli implements Callable { commandLine.addSubcommand("advise", new AdviseCommand(clientFactory)); commandLine.addSubcommand( "advise-supervisory", new AdviseSupervisoryCommand(clientFactory)); + commandLine.addSubcommand("write-secured", new WriteSecuredCommand(clientFactory)); + commandLine.addSubcommand("authenticate-user", new AuthenticateUserCommand(clientFactory)); commandLine.addSubcommand("subscribe-bulk", new SubscribeBulkCommand(clientFactory)); commandLine.addSubcommand("unsubscribe-bulk", new UnsubscribeBulkCommand(clientFactory)); commandLine.addSubcommand("read-bulk", new ReadBulkCommand(clientFactory)); @@ -1074,6 +1076,106 @@ public final class MxGatewayCli implements Callable { } } + @Command( + name = "write-secured", + description = "Invokes MXAccess WriteSecured (verified single-item write).") + static final class WriteSecuredCommand extends GatewayCommand { + @Option(names = "--session-id", required = true, description = "Gateway session id.") + String sessionId; + + @Option(names = "--server-handle", required = true, description = "MXAccess server handle.") + int serverHandle; + + @Option(names = "--item-handle", required = true, description = "MXAccess item handle.") + int itemHandle; + + @Option(names = "--type", defaultValue = "string", description = "Value type.") + String type; + + @Option(names = "--value", required = true, description = "Value text (credential-sensitive; never echoed).") + String value; + + @Option(names = "--current-user-id", defaultValue = "0", description = "MXAccess current user id.") + int currentUserId; + + @Option(names = "--verifier-user-id", defaultValue = "0", description = "MXAccess verifier user id.") + int verifierUserId; + + WriteSecuredCommand(MxGatewayCliClientFactory clientFactory) { + super(clientFactory); + } + + @Override + public Integer call() { + try (MxGatewayCliClient client = clientFactory.connect(common.resolved())) { + // The secured write value is credential-sensitive: it goes only + // into the request. The reply printed below never carries it. + MxCommandReply reply = client.session(sessionId) + .writeSecuredRaw( + serverHandle, itemHandle, currentUserId, verifierUserId, parseValue(type, value)); + writeOutput("write-secured", common, json, reply, () -> reply.getKind().name()); + } + return 0; + } + } + + @Command( + name = "authenticate-user", + description = "Invokes MXAccess AuthenticateUser and prints the resolved user id.") + static final class AuthenticateUserCommand extends GatewayCommand { + @Option(names = "--session-id", required = true, description = "Gateway session id.") + String sessionId; + + @Option(names = "--server-handle", required = true, description = "MXAccess server handle.") + int serverHandle; + + @Option(names = "--verify-user", required = true, description = "Galaxy user name to authenticate.") + String verifyUser; + + @Option( + names = "--password", + description = "Galaxy user password (credential; prefer --password-env). Never echoed.") + String password = ""; + + @Option( + names = "--password-env", + defaultValue = "MXGATEWAY_VERIFY_PASSWORD", + description = "Environment variable holding the password when --password is omitted.") + String passwordEnv; + + AuthenticateUserCommand(MxGatewayCliClientFactory clientFactory) { + super(clientFactory); + } + + @Override + public Integer call() { + // Resolve the credential from the flag or environment. It flows only + // into the request; it is never written to output, logs, or errors. + String resolvedPassword = password == null || password.isBlank() + ? System.getenv(passwordEnv) + : password; + if (resolvedPassword == null) { + resolvedPassword = ""; + } + try (MxGatewayCliClient client = clientFactory.connect(common.resolved())) { + int userId = client.session(sessionId) + .authenticateUser(serverHandle, verifyUser, resolvedPassword); + PrintWriter out = common.spec.commandLine().getOut(); + if (json) { + Map output = new LinkedHashMap<>(); + output.put("command", "authenticate-user"); + output.put("options", common.redactedJsonMap()); + output.put("verifyUser", verifyUser); + output.put("userId", userId); + out.println(jsonObject(output)); + } else { + out.println(userId); + } + } + return 0; + } + } + @Command(name = "subscribe-bulk", description = "Invokes MXAccess SubscribeBulk.") static final class SubscribeBulkCommand extends GatewayCommand { @Option(names = "--session-id", required = true, description = "Gateway session id.") @@ -1864,6 +1966,11 @@ public final class MxGatewayCli implements Callable { MxCommandReply writeRaw(int serverHandle, int itemHandle, MxValue value, int userId); + MxCommandReply writeSecuredRaw( + int serverHandle, int itemHandle, int currentUserId, int verifierUserId, MxValue value); + + int authenticateUser(int serverHandle, String verifyUser, String verifyUserPassword); + List subscribeBulk(int serverHandle, List items); List unsubscribeBulk(int serverHandle, List itemHandles); @@ -1982,13 +2089,7 @@ public final class MxGatewayCli implements Callable { @Override public MxCommandReply adviseSupervisoryRaw(int serverHandle, int itemHandle) { - return session.invokeCommand(MxCommand.newBuilder() - .setKind(MxCommandKind.MX_COMMAND_KIND_ADVISE_SUPERVISORY) - .setAdviseSupervisory( - mxaccess_gateway.v1.MxaccessGateway.AdviseSupervisoryCommand.newBuilder() - .setServerHandle(serverHandle) - .setItemHandle(itemHandle)) - .build()); + return session.adviseSupervisoryRaw(serverHandle, itemHandle); } @Override @@ -1996,6 +2097,17 @@ public final class MxGatewayCli implements Callable { return session.writeRaw(serverHandle, itemHandle, value, userId); } + @Override + public MxCommandReply writeSecuredRaw( + int serverHandle, int itemHandle, int currentUserId, int verifierUserId, MxValue value) { + return session.writeSecuredRaw(serverHandle, itemHandle, currentUserId, verifierUserId, value); + } + + @Override + public int authenticateUser(int serverHandle, String verifyUser, String verifyUserPassword) { + return session.authenticateUser(serverHandle, verifyUser, verifyUserPassword); + } + @Override public List subscribeBulk(int serverHandle, List items) { return session.subscribeBulk(serverHandle, items); 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 2f26c63..9639cc2 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 @@ -168,6 +168,49 @@ final class MxGatewayCliTests { assertTrue(run.output().contains("\"kind\":\"MX_COMMAND_KIND_ADVISE_SUPERVISORY\"")); } + @Test + void writeSecuredCommandForwardsUserIdsAndValue() { + FakeClientFactory factory = new FakeClientFactory(); + CliRun run = execute( + factory, + "write-secured", + "--session-id", "session-cli", + "--server-handle", "12", + "--item-handle", "34", + "--type", "int32", + "--value", "77", + "--current-user-id", "100", + "--verifier-user-id", "200", + "--json"); + + assertEquals(0, run.exitCode()); + assertEquals(77, factory.client.session.lastWriteSecuredValue.getInt32Value()); + assertEquals(100, factory.client.session.lastWriteSecuredCurrentUserId); + assertEquals(200, factory.client.session.lastWriteSecuredVerifierUserId); + assertTrue(run.output().contains("\"kind\":\"MX_COMMAND_KIND_WRITE_SECURED\"")); + } + + @Test + void authenticateUserCommandForwardsCredentialAndPrintsUserIdWithoutEchoingPassword() { + FakeClientFactory factory = new FakeClientFactory(); + CliRun run = execute( + factory, + "authenticate-user", + "--session-id", "session-cli", + "--server-handle", "3", + "--verify-user", "operator", + "--password", "super-secret-pw", + "--json"); + + assertEquals(0, run.exitCode()); + // The credential reaches the session (request) but never the output. + assertEquals("operator", factory.client.session.lastAuthenticateUser); + assertEquals("super-secret-pw", factory.client.session.lastAuthenticatePassword); + assertTrue(run.output().contains("\"userId\":4242"), run.output()); + assertFalse(run.output().contains("super-secret-pw"), "password must never be echoed"); + assertFalse(run.errors().contains("super-secret-pw"), "password must never be echoed to stderr"); + } + // ---- ping subcommand (D4) ---- @Test @@ -1257,6 +1300,11 @@ final class MxGatewayCliTests { private boolean adviseCalled; private boolean adviseSupervisoryCalled; private MxValue lastWriteValue; + private MxValue lastWriteSecuredValue; + private int lastWriteSecuredCurrentUserId; + private int lastWriteSecuredVerifierUserId; + private String lastAuthenticateUser; + private String lastAuthenticatePassword; private String lastPingMessage; private long lastReadBulkTimeoutMs; private List lastReadBulkItems; @@ -1341,6 +1389,25 @@ final class MxGatewayCliTests { .build(); } + @Override + public MxCommandReply writeSecuredRaw( + int serverHandle, int itemHandle, int currentUserId, int verifierUserId, MxValue value) { + lastWriteSecuredValue = value; + lastWriteSecuredCurrentUserId = currentUserId; + lastWriteSecuredVerifierUserId = verifierUserId; + return MxCommandReply.newBuilder() + .setKind(MxCommandKind.MX_COMMAND_KIND_WRITE_SECURED) + .setProtocolStatus(ok()) + .build(); + } + + @Override + public int authenticateUser(int serverHandle, String verifyUser, String verifyUserPassword) { + lastAuthenticateUser = verifyUser; + lastAuthenticatePassword = verifyUserPassword; + return 4242; + } + @Override public List subscribeBulk(int serverHandle, List items) { List results = new ArrayList<>(); diff --git a/clients/java/zb-mom-ww-mxgateway-client/src/main/java/com/zb/mom/ww/mxgateway/client/MxEventStream.java b/clients/java/zb-mom-ww-mxgateway-client/src/main/java/com/zb/mom/ww/mxgateway/client/MxEventStream.java index 0fa27e4..e884c97 100644 --- a/clients/java/zb-mom-ww-mxgateway-client/src/main/java/com/zb/mom/ww/mxgateway/client/MxEventStream.java +++ b/clients/java/zb-mom-ww-mxgateway-client/src/main/java/com/zb/mom/ww/mxgateway/client/MxEventStream.java @@ -21,6 +21,24 @@ import mxaccess_gateway.v1.MxaccessGateway.StreamEventsRequest; * stream cancels the underlying gRPC call. If the queue overflows the call is * cancelled and a follow-up call to {@link #next()} throws * {@link MxGatewayException}. + * + *

Single consumer. This stream is not safe to drain from + * more than one thread. Interleave {@link #hasNext()}/{@link #next()} (or + * {@link #nextItem()}) from a single consumer only; concurrent drains race on + * the internal cursor. + * + *

Reconnect-replay gap. When the stream was resumed via + * {@code StreamEventsRequest.after_worker_sequence} and the requested cursor + * predates the oldest event the gateway still retains, the gateway emits a + * single gap sentinel {@link MxEvent} at the head of the stream with its + * {@code replay_gap} field set (family unspecified, body oneof unset). It is a + * distinct, non-terminal signal, forwarded verbatim — never synthesized and + * never swallowed. {@link #next()} returns it as a normal {@link MxEvent} + * (callers can test {@code event.hasReplayGap()}); {@link #nextItem()} wraps it + * in an {@link MxEventStreamItem} whose {@link MxEventStreamItem#isReplayGap()} + * is {@code true}. On a gap the consumer must discard cached per-item state and + * re-snapshot, then resume without a further gap by requesting + * {@code oldest_available_sequence - 1} as the next {@code after_worker_sequence}. */ public final class MxEventStream implements Iterator, AutoCloseable { private static final Object END = new Object(); @@ -91,6 +109,24 @@ public final class MxEventStream implements Iterator, AutoCloseable { return (MxEvent) value; } + /** + * Drains the next stream element as a typed {@link MxEventStreamItem} so a + * consumer can branch on the reconnect-replay gap sentinel via + * {@link MxEventStreamItem#isReplayGap()} without inspecting the raw event. + * + *

Equivalent to wrapping {@link #next()}; the gap sentinel is surfaced as + * a distinct typed item rather than being swallowed, and normal events are + * returned unchanged on {@link MxEventStreamItem#event()}. Do not mix + * {@link #next()} and {@code nextItem()} on the same element — each call + * advances the single shared cursor. + * + * @return the next stream item + * @throws NoSuchElementException if the stream is exhausted + */ + public MxEventStreamItem nextItem() { + return new MxEventStreamItem(next()); + } + @Override public void close() { closed = true; diff --git a/clients/java/zb-mom-ww-mxgateway-client/src/main/java/com/zb/mom/ww/mxgateway/client/MxEventStreamItem.java b/clients/java/zb-mom-ww-mxgateway-client/src/main/java/com/zb/mom/ww/mxgateway/client/MxEventStreamItem.java new file mode 100644 index 0000000..1b369fb --- /dev/null +++ b/clients/java/zb-mom-ww-mxgateway-client/src/main/java/com/zb/mom/ww/mxgateway/client/MxEventStreamItem.java @@ -0,0 +1,71 @@ +package com.zb.mom.ww.mxgateway.client; + +import java.util.Objects; +import mxaccess_gateway.v1.MxaccessGateway.MxEvent; +import mxaccess_gateway.v1.MxaccessGateway.ReplayGap; + +/** + * Typed view over a single item drained from an {@link MxEventStream}. + * + *

A {@code StreamEvents} stream resumed via {@code after_worker_sequence} + * may begin with a gateway reconnect-replay gap sentinel: a single + * {@link MxEvent} whose {@code replay_gap} field is set, whose + * {@code family} is {@code MX_EVENT_FAMILY_UNSPECIFIED}, and whose {@code body} + * oneof is unset. It means the requested resume cursor predates the oldest + * event the gateway still retains, so the events in the open interval + * {@code (requested_after_sequence, oldest_available_sequence)} were evicted and + * cannot be replayed. + * + *

This wrapper lets a consumer branch on that sentinel without inspecting + * the raw event: {@link #isReplayGap()} is {@code true} only for the sentinel, + * and {@link #replayGap()} exposes the typed {@link ReplayGap} cursors. Normal + * MXAccess events return {@code false} from {@link #isReplayGap()} and carry + * their payload on {@link #event()} exactly as before. + * + *

The gateway never synthesizes an {@code OperationComplete} or any other + * event from the gap; the sentinel is the gateway's own forwarded signal and is + * surfaced here untouched. On receiving a gap the consumer must discard any + * cached per-item state and re-snapshot, then resume without incurring another + * gap by requesting {@code oldest_available_sequence - 1} as the new + * {@code after_worker_sequence} cursor. + * + * @param event the raw event; for a gap sentinel this is the sentinel event + * itself (family unspecified, body unset, {@code replay_gap} set) + */ +public record MxEventStreamItem(MxEvent event) { + /** + * Creates a stream-item view over the supplied event. + * + * @param event the raw event; must not be {@code null} + */ + public MxEventStreamItem { + Objects.requireNonNull(event, "event"); + } + + /** + * Returns whether this item is the reconnect-replay gap sentinel rather + * than a normal MXAccess event. Detected via the generated + * {@code MxEvent.hasReplayGap()} presence flag. + * + * @return {@code true} for the gap sentinel, {@code false} for normal events + */ + public boolean isReplayGap() { + return event.hasReplayGap(); + } + + /** + * Returns the typed reconnect-replay gap descriptor when this item is the + * gap sentinel. + * + * @return the {@link ReplayGap} carrying {@code requested_after_sequence} + * and {@code oldest_available_sequence} + * @throws IllegalStateException if this item is a normal event (check + * {@link #isReplayGap()} first) + */ + public ReplayGap replayGap() { + if (!event.hasReplayGap()) { + throw new IllegalStateException("stream item is not a replay-gap sentinel"); + } + return event.getReplayGap(); + } +} diff --git a/clients/java/zb-mom-ww-mxgateway-client/src/main/java/com/zb/mom/ww/mxgateway/client/MxGatewaySession.java b/clients/java/zb-mom-ww-mxgateway-client/src/main/java/com/zb/mom/ww/mxgateway/client/MxGatewaySession.java index 8dd00e1..68c3843 100644 --- a/clients/java/zb-mom-ww-mxgateway-client/src/main/java/com/zb/mom/ww/mxgateway/client/MxGatewaySession.java +++ b/clients/java/zb-mom-ww-mxgateway-client/src/main/java/com/zb/mom/ww/mxgateway/client/MxGatewaySession.java @@ -7,11 +7,16 @@ import java.util.List; import java.util.Map; import java.util.Objects; import java.util.TreeMap; +import mxaccess_gateway.v1.MxaccessGateway.ActivateCommand; +import mxaccess_gateway.v1.MxaccessGateway.AddBufferedItemCommand; import mxaccess_gateway.v1.MxaccessGateway.AddItem2Command; import mxaccess_gateway.v1.MxaccessGateway.AddItemBulkCommand; import mxaccess_gateway.v1.MxaccessGateway.AddItemCommand; import mxaccess_gateway.v1.MxaccessGateway.AdviseItemBulkCommand; import mxaccess_gateway.v1.MxaccessGateway.AdviseCommand; +import mxaccess_gateway.v1.MxaccessGateway.AdviseSupervisoryCommand; +import mxaccess_gateway.v1.MxaccessGateway.ArchestrAUserToIdCommand; +import mxaccess_gateway.v1.MxaccessGateway.AuthenticateUserCommand; import mxaccess_gateway.v1.MxaccessGateway.BulkReadResult; import mxaccess_gateway.v1.MxaccessGateway.BulkWriteResult; import mxaccess_gateway.v1.MxaccessGateway.CloseSessionReply; @@ -23,15 +28,18 @@ import mxaccess_gateway.v1.MxaccessGateway.MxCommandRequest; import mxaccess_gateway.v1.MxaccessGateway.MxDataType; import mxaccess_gateway.v1.MxaccessGateway.MxSparseArray; import mxaccess_gateway.v1.MxaccessGateway.MxSparseElement; +import mxaccess_gateway.v1.MxaccessGateway.MxStatusProxy; import mxaccess_gateway.v1.MxaccessGateway.MxValue; import mxaccess_gateway.v1.MxaccessGateway.OpenSessionReply; import mxaccess_gateway.v1.MxaccessGateway.ReadBulkCommand; import mxaccess_gateway.v1.MxaccessGateway.RegisterCommand; import mxaccess_gateway.v1.MxaccessGateway.RemoveItemBulkCommand; import mxaccess_gateway.v1.MxaccessGateway.RemoveItemCommand; +import mxaccess_gateway.v1.MxaccessGateway.SetBufferedUpdateIntervalCommand; import mxaccess_gateway.v1.MxaccessGateway.StreamEventsRequest; import mxaccess_gateway.v1.MxaccessGateway.SubscribeBulkCommand; import mxaccess_gateway.v1.MxaccessGateway.SubscribeResult; +import mxaccess_gateway.v1.MxaccessGateway.SuspendCommand; import mxaccess_gateway.v1.MxaccessGateway.UnAdviseCommand; import mxaccess_gateway.v1.MxaccessGateway.UnAdviseItemBulkCommand; import mxaccess_gateway.v1.MxaccessGateway.UnsubscribeBulkCommand; @@ -42,6 +50,8 @@ import mxaccess_gateway.v1.MxaccessGateway.Write2Command; import mxaccess_gateway.v1.MxaccessGateway.WriteBulkCommand; import mxaccess_gateway.v1.MxaccessGateway.WriteBulkEntry; import mxaccess_gateway.v1.MxaccessGateway.WriteCommand; +import mxaccess_gateway.v1.MxaccessGateway.WriteSecured2Command; +import mxaccess_gateway.v1.MxaccessGateway.WriteSecuredCommand; import mxaccess_gateway.v1.MxaccessGateway.WriteSecured2BulkCommand; import mxaccess_gateway.v1.MxaccessGateway.WriteSecured2BulkEntry; import mxaccess_gateway.v1.MxaccessGateway.WriteSecuredBulkCommand; @@ -697,6 +707,285 @@ public final class MxGatewaySession implements AutoCloseable { .build()); } + /** + * Invokes MXAccess {@code AdviseSupervisory} so the item accepts + * supervisory-attributed writes. Required before a plain {@link #write} + * can stamp a Galaxy user id without the verified/secured path. + * + * @param serverHandle the {@code ServerHandle} owning the item + * @param itemHandle the {@code ItemHandle} to advise supervisory + * @throws MxGatewayException on transport or protocol failure + * @throws MxAccessException when MXAccess reports a COM-side failure + */ + public void adviseSupervisory(int serverHandle, int itemHandle) { + adviseSupervisoryRaw(serverHandle, itemHandle); + } + + /** + * Invokes MXAccess {@code AdviseSupervisory} and returns the raw reply. + * + * @param serverHandle the {@code ServerHandle} owning the item + * @param itemHandle the {@code ItemHandle} to advise supervisory + * @return the raw command reply + * @throws MxGatewayException on transport or protocol failure + * @throws MxAccessException when MXAccess reports a COM-side failure + */ + public MxCommandReply adviseSupervisoryRaw(int serverHandle, int itemHandle) { + return invokeCommand(MxCommand.newBuilder() + .setKind(MxCommandKind.MX_COMMAND_KIND_ADVISE_SUPERVISORY) + .setAdviseSupervisory(AdviseSupervisoryCommand.newBuilder() + .setServerHandle(serverHandle) + .setItemHandle(itemHandle)) + .build()); + } + + /** + * Invokes MXAccess {@code WriteSecured} — a verified write gated by a + * previously authenticated Galaxy user. + * + *

MXAccess parity: the native call fails if the caller + * has not first {@link #authenticateUser authenticated} and + * {@link #adviseSupervisory advised supervisory}, or if the value body is + * absent; that native failure is surfaced (as {@link MxAccessException}), + * not papered over. + * + *

Secret handling: {@code value} may carry a + * credential-sensitive payload. It is placed only in the request and never + * appears in any surfaced error (exceptions carry the reply, not the + * request), so callers must likewise avoid logging it. + * + * @param serverHandle the {@code ServerHandle} owning the item + * @param itemHandle the {@code ItemHandle} to write + * @param currentUserId the authenticated (current) Galaxy user id + * @param verifierUserId the verifier Galaxy user id (second-signature); use + * the same value as {@code currentUserId} when no separate verifier applies + * @param value the credential-sensitive value to write + * @throws MxGatewayException on transport or protocol failure + * @throws MxAccessException when MXAccess reports a COM-side failure + */ + public void writeSecured( + int serverHandle, int itemHandle, int currentUserId, int verifierUserId, MxValue value) { + writeSecuredRaw(serverHandle, itemHandle, currentUserId, verifierUserId, value); + } + + /** + * Invokes MXAccess {@code WriteSecured} and returns the raw reply. + * + * @param serverHandle the {@code ServerHandle} owning the item + * @param itemHandle the {@code ItemHandle} to write + * @param currentUserId the authenticated (current) Galaxy user id + * @param verifierUserId the verifier Galaxy user id + * @param value the credential-sensitive value to write + * @return the raw command reply + * @throws MxGatewayException on transport or protocol failure + * @throws MxAccessException when MXAccess reports a COM-side failure + */ + public MxCommandReply writeSecuredRaw( + int serverHandle, int itemHandle, int currentUserId, int verifierUserId, MxValue value) { + return invokeCommand(MxCommand.newBuilder() + .setKind(MxCommandKind.MX_COMMAND_KIND_WRITE_SECURED) + .setWriteSecured(WriteSecuredCommand.newBuilder() + .setServerHandle(serverHandle) + .setItemHandle(itemHandle) + .setCurrentUserId(currentUserId) + .setVerifierUserId(verifierUserId) + .setValue(value)) + .build()); + } + + /** + * Invokes MXAccess {@code WriteSecured2} — a verified, explicitly + * timestamped write. Parity and secret-handling notes mirror + * {@link #writeSecured}. + * + * @param serverHandle the {@code ServerHandle} owning the item + * @param itemHandle the {@code ItemHandle} to write + * @param currentUserId the authenticated (current) Galaxy user id + * @param verifierUserId the verifier Galaxy user id + * @param value the credential-sensitive value to write + * @param timestampValue the timestamp value to associate with the write + * @throws MxGatewayException on transport or protocol failure + * @throws MxAccessException when MXAccess reports a COM-side failure + */ + public void writeSecured2( + int serverHandle, + int itemHandle, + int currentUserId, + int verifierUserId, + MxValue value, + MxValue timestampValue) { + writeSecured2Raw(serverHandle, itemHandle, currentUserId, verifierUserId, value, timestampValue); + } + + /** + * Invokes MXAccess {@code WriteSecured2} and returns the raw reply. + * + * @param serverHandle the {@code ServerHandle} owning the item + * @param itemHandle the {@code ItemHandle} to write + * @param currentUserId the authenticated (current) Galaxy user id + * @param verifierUserId the verifier Galaxy user id + * @param value the credential-sensitive value to write + * @param timestampValue the timestamp value to associate with the write + * @return the raw command reply + * @throws MxGatewayException on transport or protocol failure + * @throws MxAccessException when MXAccess reports a COM-side failure + */ + public MxCommandReply writeSecured2Raw( + int serverHandle, + int itemHandle, + int currentUserId, + int verifierUserId, + MxValue value, + MxValue timestampValue) { + return invokeCommand(MxCommand.newBuilder() + .setKind(MxCommandKind.MX_COMMAND_KIND_WRITE_SECURED2) + .setWriteSecured2(WriteSecured2Command.newBuilder() + .setServerHandle(serverHandle) + .setItemHandle(itemHandle) + .setCurrentUserId(currentUserId) + .setVerifierUserId(verifierUserId) + .setValue(value) + .setTimestampValue(timestampValue)) + .build()); + } + + /** + * Invokes MXAccess {@code AuthenticateUser} and returns the resolved + * Galaxy user id used to attribute subsequent secured writes. + * + *

Secret handling: {@code verifyUserPassword} is a raw + * MXAccess credential. It is placed only in the request and never appears in + * any surfaced error, log, or {@code toString()} (exceptions carry the + * reply, not the request). Callers must not log it either. + * + * @param serverHandle the {@code ServerHandle} for the session + * @param verifyUser the Galaxy user name to authenticate + * @param verifyUserPassword the user's credential; never logged or surfaced + * @return the authenticated Galaxy user id + * @throws MxGatewayException on transport or protocol failure + * @throws MxAccessException when MXAccess rejects the credential + */ + public int authenticateUser(int serverHandle, String verifyUser, String verifyUserPassword) { + MxCommandReply reply = invokeCommand(MxCommand.newBuilder() + .setKind(MxCommandKind.MX_COMMAND_KIND_AUTHENTICATE_USER) + .setAuthenticateUser(AuthenticateUserCommand.newBuilder() + .setServerHandle(serverHandle) + .setVerifyUser(verifyUser) + .setVerifyUserPassword(verifyUserPassword)) + .build()); + if (reply.hasAuthenticateUser()) { + return reply.getAuthenticateUser().getUserId(); + } + return reply.getReturnValue().getInt32Value(); + } + + /** + * Invokes MXAccess {@code ArchestrAUserToId}, resolving a Galaxy user GUID + * to its integer user id. + * + * @param serverHandle the {@code ServerHandle} for the session + * @param userIdGuid the Galaxy user GUID string + * @return the resolved Galaxy user id + * @throws MxGatewayException on transport or protocol failure + * @throws MxAccessException when MXAccess reports a COM-side failure + */ + public int archestrAUserToId(int serverHandle, String userIdGuid) { + MxCommandReply reply = invokeCommand(MxCommand.newBuilder() + .setKind(MxCommandKind.MX_COMMAND_KIND_ARCHESTRA_USER_TO_ID) + .setArchestraUserToId(ArchestrAUserToIdCommand.newBuilder() + .setServerHandle(serverHandle) + .setUserIdGuid(userIdGuid)) + .build()); + if (reply.hasArchestraUserToId()) { + return reply.getArchestraUserToId().getUserId(); + } + return reply.getReturnValue().getInt32Value(); + } + + /** + * Invokes MXAccess {@code AddBufferedItem} and returns the new item handle. + * The buffered add family delivers coalesced {@code OnBufferedDataChange} + * updates on the interval set by {@link #setBufferedUpdateInterval}. + * + * @param serverHandle the {@code ServerHandle} owning the item + * @param itemDefinition the MXAccess item definition (tag reference) + * @param itemContext the MXAccess item context (e.g. galaxy/object scope) + * @return the {@code ItemHandle} assigned by MXAccess + * @throws MxGatewayException on transport or protocol failure + * @throws MxAccessException when MXAccess reports a COM-side failure + */ + public int addBufferedItem(int serverHandle, String itemDefinition, String itemContext) { + MxCommandReply reply = invokeCommand(MxCommand.newBuilder() + .setKind(MxCommandKind.MX_COMMAND_KIND_ADD_BUFFERED_ITEM) + .setAddBufferedItem(AddBufferedItemCommand.newBuilder() + .setServerHandle(serverHandle) + .setItemDefinition(itemDefinition) + .setItemContext(itemContext)) + .build()); + if (reply.hasAddBufferedItem()) { + return reply.getAddBufferedItem().getItemHandle(); + } + return reply.getReturnValue().getInt32Value(); + } + + /** + * Invokes MXAccess {@code SetBufferedUpdateInterval}, controlling how often + * the worker coalesces buffered updates for the given server handle. + * + * @param serverHandle the {@code ServerHandle} to configure + * @param updateIntervalMilliseconds the buffered update interval in milliseconds + * @throws MxGatewayException on transport or protocol failure + * @throws MxAccessException when MXAccess reports a COM-side failure + */ + public void setBufferedUpdateInterval(int serverHandle, int updateIntervalMilliseconds) { + invokeCommand(MxCommand.newBuilder() + .setKind(MxCommandKind.MX_COMMAND_KIND_SET_BUFFERED_UPDATE_INTERVAL) + .setSetBufferedUpdateInterval(SetBufferedUpdateIntervalCommand.newBuilder() + .setServerHandle(serverHandle) + .setUpdateIntervalMilliseconds(updateIntervalMilliseconds)) + .build()); + } + + /** + * Invokes MXAccess {@code Suspend} on an item and returns the reply's + * {@link MxStatusProxy}. + * + * @param serverHandle the {@code ServerHandle} owning the item + * @param itemHandle the {@code ItemHandle} to suspend + * @return the {@code MxStatusProxy} carried by the suspend reply + * @throws MxGatewayException on transport or protocol failure + * @throws MxAccessException when MXAccess reports a COM-side failure + */ + public MxStatusProxy suspend(int serverHandle, int itemHandle) { + MxCommandReply reply = invokeCommand(MxCommand.newBuilder() + .setKind(MxCommandKind.MX_COMMAND_KIND_SUSPEND) + .setSuspend(SuspendCommand.newBuilder() + .setServerHandle(serverHandle) + .setItemHandle(itemHandle)) + .build()); + return reply.getSuspend().getStatus(); + } + + /** + * Invokes MXAccess {@code Activate} on a previously suspended item and + * returns the reply's {@link MxStatusProxy}. + * + * @param serverHandle the {@code ServerHandle} owning the item + * @param itemHandle the {@code ItemHandle} to activate + * @return the {@code MxStatusProxy} carried by the activate reply + * @throws MxGatewayException on transport or protocol failure + * @throws MxAccessException when MXAccess reports a COM-side failure + */ + public MxStatusProxy activate(int serverHandle, int itemHandle) { + MxCommandReply reply = invokeCommand(MxCommand.newBuilder() + .setKind(MxCommandKind.MX_COMMAND_KIND_ACTIVATE) + .setActivate(ActivateCommand.newBuilder() + .setServerHandle(serverHandle) + .setItemHandle(itemHandle)) + .build()); + return reply.getActivate().getStatus(); + } + /** * Subscribes to gateway events for this session starting from the * beginning of the worker event log. diff --git a/clients/java/zb-mom-ww-mxgateway-client/src/test/java/com/zb/mom/ww/mxgateway/client/MxGatewayClientSessionTests.java b/clients/java/zb-mom-ww-mxgateway-client/src/test/java/com/zb/mom/ww/mxgateway/client/MxGatewayClientSessionTests.java index 16cc720..bfc3a2f 100644 --- a/clients/java/zb-mom-ww-mxgateway-client/src/test/java/com/zb/mom/ww/mxgateway/client/MxGatewayClientSessionTests.java +++ b/clients/java/zb-mom-ww-mxgateway-client/src/test/java/com/zb/mom/ww/mxgateway/client/MxGatewayClientSessionTests.java @@ -1,6 +1,7 @@ package com.zb.mom.ww.mxgateway.client; import static org.junit.jupiter.api.Assertions.assertEquals; +import static org.junit.jupiter.api.Assertions.assertFalse; import static org.junit.jupiter.api.Assertions.assertNotNull; import static org.junit.jupiter.api.Assertions.assertNull; import static org.junit.jupiter.api.Assertions.assertThrows; @@ -30,6 +31,7 @@ import mxaccess_gateway.v1.MxaccessGateway.AddItemReply; import mxaccess_gateway.v1.MxaccessGateway.AlarmConditionState; import mxaccess_gateway.v1.MxaccessGateway.AlarmFeedMessage; import mxaccess_gateway.v1.MxaccessGateway.AlarmTransitionKind; +import mxaccess_gateway.v1.MxaccessGateway.AuthenticateUserReply; import mxaccess_gateway.v1.MxaccessGateway.BulkSubscribeReply; import mxaccess_gateway.v1.MxaccessGateway.OnAlarmTransitionEvent; import mxaccess_gateway.v1.MxaccessGateway.CloseSessionReply; @@ -39,6 +41,7 @@ import mxaccess_gateway.v1.MxaccessGateway.MxCommandReply; import mxaccess_gateway.v1.MxaccessGateway.MxCommandRequest; import mxaccess_gateway.v1.MxaccessGateway.MxDataType; import mxaccess_gateway.v1.MxaccessGateway.MxEvent; +import mxaccess_gateway.v1.MxaccessGateway.MxEventFamily; import mxaccess_gateway.v1.MxaccessGateway.MxSparseElement; import mxaccess_gateway.v1.MxaccessGateway.MxValue; import mxaccess_gateway.v1.MxaccessGateway.OpenSessionReply; @@ -47,6 +50,7 @@ import mxaccess_gateway.v1.MxaccessGateway.ProtocolStatus; import mxaccess_gateway.v1.MxaccessGateway.ProtocolStatusCode; import mxaccess_gateway.v1.MxaccessGateway.QueryActiveAlarmsRequest; 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.StreamEventsRequest; @@ -509,6 +513,232 @@ final class MxGatewayClientSessionTests { } } + @Test + void streamEventsSurfacesReplayGapSentinelAsTypedItem() throws Exception { + // CLI-15: a resumed stream whose cursor predates the retained window + // begins with the gateway's replay-gap sentinel. It must surface as a + // distinct typed item, and normal events must be unaffected. + TestGatewayService service = new TestGatewayService() { + @Override + public void streamEvents(StreamEventsRequest request, StreamObserver responseObserver) { + responseObserver.onNext(MxEvent.newBuilder() + .setSessionId(request.getSessionId()) + .setReplayGap(ReplayGap.newBuilder() + .setRequestedAfterSequence(5) + .setOldestAvailableSequence(9)) + .build()); + responseObserver.onNext(MxEvent.newBuilder() + .setSessionId(request.getSessionId()) + .setFamily(MxEventFamily.MX_EVENT_FAMILY_ON_DATA_CHANGE) + .setWorkerSequence(9) + .build()); + responseObserver.onCompleted(); + } + }; + + try (InProcessGateway gateway = InProcessGateway.start(service, new AtomicReference<>()); + MxGatewayClient client = gateway.client("", Duration.ofSeconds(5)); + MxEventStream events = + MxGatewaySession.forSessionId(client, "resume-session").streamEventsAfter(5)) { + assertTrue(events.hasNext()); + MxEventStreamItem gap = events.nextItem(); + assertTrue(gap.isReplayGap(), "head sentinel must classify as a replay gap"); + assertEquals(5, gap.replayGap().getRequestedAfterSequence()); + assertEquals(9, gap.replayGap().getOldestAvailableSequence()); + // Sentinel carries no MXAccess payload. + assertEquals(MxEventFamily.MX_EVENT_FAMILY_UNSPECIFIED, gap.event().getFamily()); + + assertTrue(events.hasNext()); + MxEventStreamItem normal = events.nextItem(); + assertFalse(normal.isReplayGap(), "normal event must not classify as a replay gap"); + assertEquals(9, normal.event().getWorkerSequence()); + assertEquals(MxEventFamily.MX_EVENT_FAMILY_ON_DATA_CHANGE, normal.event().getFamily()); + } + } + + @Test + void adviseSupervisoryBuildsSupervisoryCommand() throws Exception { + AtomicReference commandRequest = new AtomicReference<>(); + TestGatewayService service = okInvokeService(commandRequest); + + try (InProcessGateway gateway = InProcessGateway.start(service, new AtomicReference<>()); + MxGatewayClient client = gateway.client("", Duration.ofSeconds(5))) { + MxGatewaySession session = MxGatewaySession.forSessionId(client, "advise-super-session"); + + session.adviseSupervisory(12, 34); + + assertEquals( + MxCommandKind.MX_COMMAND_KIND_ADVISE_SUPERVISORY, + commandRequest.get().getCommand().getKind()); + assertEquals(12, commandRequest.get().getCommand().getAdviseSupervisory().getServerHandle()); + assertEquals(34, commandRequest.get().getCommand().getAdviseSupervisory().getItemHandle()); + } + } + + @Test + void writeSecuredSurfacesNativeMxAccessFailure() throws Exception { + // CLI-04 parity: WriteSecured before authenticate/advise fails natively; + // the client surfaces the failure rather than papering over it. + TestGatewayService service = new TestGatewayService() { + @Override + public void invoke(MxCommandRequest request, StreamObserver responseObserver) { + responseObserver.onNext(MxCommandReply.newBuilder() + .setSessionId(request.getSessionId()) + .setKind(request.getCommand().getKind()) + .setProtocolStatus(ProtocolStatus.newBuilder() + .setCode(ProtocolStatusCode.PROTOCOL_STATUS_CODE_MXACCESS_FAILURE) + .setMessage("WriteSecured rejected: user not authenticated.")) + .setHresult(-2147220992) + .build()); + responseObserver.onCompleted(); + } + }; + + try (InProcessGateway gateway = InProcessGateway.start(service, new AtomicReference<>()); + MxGatewayClient client = gateway.client("", Duration.ofSeconds(5))) { + MxGatewaySession session = MxGatewaySession.forSessionId(client, "secured-session"); + + MxAccessException error = assertThrows( + MxAccessException.class, + () -> session.writeSecured(1, 2, 100, 100, MxValues.stringValue("secret-payload"))); + + assertEquals(-2147220992, error.reply().getHresult()); + // The credential-bearing value lives only in the request, so it must + // not appear in any surfaced error text. + assertFalse(String.valueOf(error.getMessage()).contains("secret-payload")); + } + } + + @Test + void writeSecuredScriptedSuccessSendsSecuredCommand() throws Exception { + AtomicReference commandRequest = new AtomicReference<>(); + TestGatewayService service = okInvokeService(commandRequest); + + try (InProcessGateway gateway = InProcessGateway.start(service, new AtomicReference<>()); + MxGatewayClient client = gateway.client("", Duration.ofSeconds(5))) { + MxGatewaySession session = MxGatewaySession.forSessionId(client, "secured-ok-session"); + + session.writeSecured(7, 8, 100, 200, MxValues.int32Value(42)); + + var command = commandRequest.get().getCommand(); + assertEquals(MxCommandKind.MX_COMMAND_KIND_WRITE_SECURED, command.getKind()); + assertEquals(7, command.getWriteSecured().getServerHandle()); + assertEquals(8, command.getWriteSecured().getItemHandle()); + assertEquals(100, command.getWriteSecured().getCurrentUserId()); + assertEquals(200, command.getWriteSecured().getVerifierUserId()); + assertEquals(42, command.getWriteSecured().getValue().getInt32Value()); + } + } + + @Test + void authenticateUserReturnsUserIdAndForwardsCredential() throws Exception { + AtomicReference commandRequest = new AtomicReference<>(); + TestGatewayService service = new TestGatewayService() { + @Override + public void invoke(MxCommandRequest request, StreamObserver responseObserver) { + commandRequest.set(request); + responseObserver.onNext(MxCommandReply.newBuilder() + .setSessionId(request.getSessionId()) + .setKind(request.getCommand().getKind()) + .setProtocolStatus(ok()) + .setAuthenticateUser(AuthenticateUserReply.newBuilder().setUserId(4242)) + .build()); + responseObserver.onCompleted(); + } + }; + + try (InProcessGateway gateway = InProcessGateway.start(service, new AtomicReference<>()); + MxGatewayClient client = gateway.client("", Duration.ofSeconds(5))) { + MxGatewaySession session = MxGatewaySession.forSessionId(client, "auth-session"); + + int userId = session.authenticateUser(3, "operator", "super-secret-pw"); + + assertEquals(4242, userId); + // The credential is forwarded in the request only. + assertEquals( + "super-secret-pw", + commandRequest.get().getCommand().getAuthenticateUser().getVerifyUserPassword()); + } + } + + @Test + void authenticateUserFailureKeepsCredentialOutOfSurfacedError() throws Exception { + TestGatewayService service = new TestGatewayService() { + @Override + public void invoke(MxCommandRequest request, StreamObserver responseObserver) { + responseObserver.onNext(MxCommandReply.newBuilder() + .setSessionId(request.getSessionId()) + .setKind(request.getCommand().getKind()) + .setProtocolStatus(ProtocolStatus.newBuilder() + .setCode(ProtocolStatusCode.PROTOCOL_STATUS_CODE_MXACCESS_FAILURE) + .setMessage("AuthenticateUser rejected the credential.")) + .setHresult(-2147220992) + .build()); + responseObserver.onCompleted(); + } + }; + + try (InProcessGateway gateway = InProcessGateway.start(service, new AtomicReference<>()); + MxGatewayClient client = gateway.client("", Duration.ofSeconds(5))) { + MxGatewaySession session = MxGatewaySession.forSessionId(client, "auth-fail-session"); + + MxAccessException error = assertThrows( + MxAccessException.class, + () -> session.authenticateUser(3, "operator", "super-secret-pw")); + + // The password must never reach the exception message or toString(). + assertFalse(String.valueOf(error.getMessage()).contains("super-secret-pw")); + assertFalse(error.toString().contains("super-secret-pw")); + } + } + + @Test + void suspendAndActivateReturnStatusProxy() throws Exception { + TestGatewayService service = new TestGatewayService() { + @Override + public void invoke(MxCommandRequest request, StreamObserver responseObserver) { + MxCommandReply.Builder reply = MxCommandReply.newBuilder() + .setSessionId(request.getSessionId()) + .setKind(request.getCommand().getKind()) + .setProtocolStatus(ok()); + if (request.getCommand().getKind() == MxCommandKind.MX_COMMAND_KIND_SUSPEND) { + reply.setSuspend(mxaccess_gateway.v1.MxaccessGateway.SuspendReply.newBuilder() + .setStatus(mxaccess_gateway.v1.MxaccessGateway.MxStatusProxy.newBuilder() + .setSuccess(1))); + } else if (request.getCommand().getKind() == MxCommandKind.MX_COMMAND_KIND_ACTIVATE) { + reply.setActivate(mxaccess_gateway.v1.MxaccessGateway.ActivateReply.newBuilder() + .setStatus(mxaccess_gateway.v1.MxaccessGateway.MxStatusProxy.newBuilder() + .setSuccess(1))); + } + responseObserver.onNext(reply.build()); + responseObserver.onCompleted(); + } + }; + + try (InProcessGateway gateway = InProcessGateway.start(service, new AtomicReference<>()); + MxGatewayClient client = gateway.client("", Duration.ofSeconds(5))) { + MxGatewaySession session = MxGatewaySession.forSessionId(client, "suspend-session"); + + assertTrue(MxStatuses.succeeded(session.suspend(1, 2))); + assertTrue(MxStatuses.succeeded(session.activate(1, 2))); + } + } + + private static TestGatewayService okInvokeService(AtomicReference commandRequest) { + return new TestGatewayService() { + @Override + public void invoke(MxCommandRequest request, StreamObserver responseObserver) { + commandRequest.set(request); + responseObserver.onNext(MxCommandReply.newBuilder() + .setSessionId(request.getSessionId()) + .setKind(request.getCommand().getKind()) + .setProtocolStatus(ok()) + .build()); + responseObserver.onCompleted(); + } + }; + } + private static ProtocolStatus ok() { return ProtocolStatus.newBuilder() .setCode(ProtocolStatusCode.PROTOCOL_STATUS_CODE_OK) diff --git a/docs/ClientLibrariesDesign.md b/docs/ClientLibrariesDesign.md index a616291..77356a0 100644 --- a/docs/ClientLibrariesDesign.md +++ b/docs/ClientLibrariesDesign.md @@ -83,8 +83,9 @@ discard cached state, re-snapshot, and resume with `after_worker_sequence = oldest_available_sequence - 1`. The per-language surface follows each idiom (.NET `MxEventStreamItem.IsReplayGap` via `StreamEventItemsAsync`; Go `EventResult.ReplayGap`/`IsReplayGap()`; Rust -`EventItem::ReplayGap`; Python yields a typed `ReplayGap` from -`stream_events`; Java — pending, tracked with the JDK-17/windows client pass). +`EventItem::ReplayGap`; Python yields a typed `ReplayGap` from `stream_events`; +Java `MxEventStreamItem.isReplayGap()` via `MxEventStream.nextItem()`). Shipped +in all five clients. ## Public Client Concepts @@ -124,8 +125,7 @@ reach logs, exception text, or `ToString`/`Debug`/`Display` — the value is car only on the wire. Each client's test suite asserts a distinctive credential is absent from any surfaced error. -Shipped for .NET / Go / Rust / Python; the Java client's typed parity helpers are -batched to the windows build host (no local JRE on the primary dev box). +Shipped in all five clients (.NET / Go / Rust / Python / Java). ## Shared API Shape