feat(java-client): CLI-15 ReplayGap surface + CLI-04 typed-command parity (5/5 complete)
Completes both findings across all five clients — done locally on the Mac now that homebrew openjdk@17 is available (JAVA_HOME=/opt/homebrew/opt/openjdk@17). CLI-15: new MxEventStreamItem record + MxEventStream.nextItem() surfaces the gateway's ReplayGap sentinel as a typed, non-terminal signal (isReplayGap()/ replayGap()/event()); existing Iterator<MxEvent> path unchanged, sentinel never swallowed/synthesized. Javadoc covers gap semantics + resume contract. CLI-04: typed single-item helpers on MxGatewaySession — Phase 1 adviseSupervisory/writeSecured/writeSecured2/authenticateUser/archestrAUserToId, Phase 2 addBufferedItem/setBufferedUpdateInterval/suspend/activate (unregister already present). Each routes through invokeCommand -> ensureProtocolSuccess + ensureMxAccessSuccess (same validation as bulk). MXAccess parity preserved. Credentials flow only into the request proto; exceptions carry only the reply and gRPC status text is scrubbed via MxGatewaySecrets.redactCredentials — tests assert the password/secured value is absent from getMessage()/toString()/CLI output. New CLI subcommands write-secured/authenticate-user (credential via --password/--password-env, prints only the user id). gradle test: 106 tests, 0 failures (58 client + 48 cli); no generated churn. Shared docs: ClientLibrariesDesign + CLAUDE.md updated to "all five clients". Claude-Session: https://claude.ai/code/session_01DMXXvNuPekkkrTEyPNxEkW
This commit is contained in:
Joseph Doherty
@@ -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.
|
||||
|
||||
+73
-15
@@ -76,7 +76,40 @@ data-bearing MXAccess failure.
|
||||
`MxEventStream` implements `Iterator<MxEvent>` 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 <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 <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 <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 <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 <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 <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 <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"
|
||||
|
||||
+119
-7
@@ -155,6 +155,8 @@ public final class MxGatewayCli implements Callable<Integer> {
|
||||
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<Integer> {
|
||||
}
|
||||
}
|
||||
|
||||
@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<String, Object> 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<Integer> {
|
||||
|
||||
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<SubscribeResult> subscribeBulk(int serverHandle, List<String> items);
|
||||
|
||||
List<SubscribeResult> unsubscribeBulk(int serverHandle, List<Integer> itemHandles);
|
||||
@@ -1982,13 +2089,7 @@ public final class MxGatewayCli implements Callable<Integer> {
|
||||
|
||||
@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<Integer> {
|
||||
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<SubscribeResult> subscribeBulk(int serverHandle, List<String> items) {
|
||||
return session.subscribeBulk(serverHandle, items);
|
||||
|
||||
+67
@@ -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<String> 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<SubscribeResult> subscribeBulk(int serverHandle, List<String> items) {
|
||||
List<SubscribeResult> results = new ArrayList<>();
|
||||
|
||||
+36
@@ -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}.
|
||||
*
|
||||
* <p><strong>Single consumer.</strong> 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.
|
||||
*
|
||||
* <p><strong>Reconnect-replay gap.</strong> 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<MxEvent>, AutoCloseable {
|
||||
private static final Object END = new Object();
|
||||
@@ -91,6 +109,24 @@ public final class MxEventStream implements Iterator<MxEvent>, 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.
|
||||
*
|
||||
* <p>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;
|
||||
|
||||
+71
@@ -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}.
|
||||
*
|
||||
* <p>A {@code StreamEvents} stream resumed via {@code after_worker_sequence}
|
||||
* may begin with a gateway reconnect-replay <em>gap sentinel</em>: 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.
|
||||
*
|
||||
* <p>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.
|
||||
*
|
||||
* <p>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();
|
||||
}
|
||||
}
|
||||
+289
@@ -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.
|
||||
*
|
||||
* <p><strong>MXAccess parity:</strong> 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.
|
||||
*
|
||||
* <p><strong>Secret handling:</strong> {@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.
|
||||
*
|
||||
* <p><strong>Secret handling:</strong> {@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.
|
||||
|
||||
+230
@@ -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<MxEvent> 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<MxCommandRequest> 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<MxCommandReply> 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<MxCommandRequest> 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<MxCommandRequest> commandRequest = new AtomicReference<>();
|
||||
TestGatewayService service = new TestGatewayService() {
|
||||
@Override
|
||||
public void invoke(MxCommandRequest request, StreamObserver<MxCommandReply> 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<MxCommandReply> 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<MxCommandReply> 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<MxCommandRequest> commandRequest) {
|
||||
return new TestGatewayService() {
|
||||
@Override
|
||||
public void invoke(MxCommandRequest request, StreamObserver<MxCommandReply> 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)
|
||||
|
||||
@@ -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
|
||||
|
||||
|
||||
Reference in New Issue
Block a user