From 48c9323d4ad5030f6d74a45352ca4e34e766065c Mon Sep 17 00:00:00 2001 From: Joseph Doherty Date: Thu, 22 Oct 2026 06:09:25 -0400 Subject: [PATCH] fix(java-client): configurable event queue, keep events on overflow, cancel on interrupt, typed diagnostics - eventQueueCapacity option (default 1024, was hard-coded 16) - overflow retains buffered events and delivers the error after them - interrupt cancels the gRPC call and restores the interrupt flag - ping/getSessionState/getWorkerInfo/drainEvents on MxGatewaySession --- clients/java/README.md | 24 +++++ .../ww/mxgateway/client/MxEventStream.java | 40 ++++++-- .../ww/mxgateway/client/MxGatewayClient.java | 2 +- .../client/MxGatewayClientOptions.java | 30 ++++++ .../ww/mxgateway/client/MxGatewaySession.java | 68 ++++++++++++++ .../mxgateway/client/MxEventStreamTests.java | 93 +++++++++++++++++++ .../client/MxGatewayClientSessionTests.java | 60 ++++++++++++ 7 files changed, 310 insertions(+), 7 deletions(-) create mode 100644 clients/java/zb-mom-ww-mxgateway-client/src/test/java/com/zb/mom/ww/mxgateway/client/MxEventStreamTests.java diff --git a/clients/java/README.md b/clients/java/README.md index 05e6c08..8c591cf 100644 --- a/clients/java/README.md +++ b/clients/java/README.md @@ -79,6 +79,30 @@ only stops the client from waiting; it does not abort an in-flight MXAccess COM call on the worker STA. It is a **single-consumer** surface: drive `hasNext()`/`next()` (or `nextItem()`) from one thread only. +### Event buffer, overflow, and interrupts + +The iterator buffers events in a bounded queue whose size is +`MxGatewayClientOptions.builder().eventQueueCapacity(n)` (default 1024; it was a +hard-coded 16, which a short burst could overflow). If the consumer falls more +than `n` events behind, the client cancels the call, keeps the events already +buffered, and throws `MxGatewayException` only after you have drained them. Resume +with `streamEventsAfter(lastSeenWorkerSequence)`. If the consuming thread is +interrupted, the call is cancelled, the interrupt flag is restored, and the +iterator throws a `CANCELLED` `StatusRuntimeException`. + +`streamEventsAsync(request, StreamObserver)` is the non-dropping +alternative: it delivers each event to your observer on the gRPC thread with no +client-side queue, so nothing is discarded by the client (a slow observer +instead applies normal gRPC flow control). + +### Session diagnostics + +`MxGatewaySession` exposes typed helpers for the worker diagnostic commands: +`ping(message)` (returns the raw `MxCommandReply`), `getSessionState()` +(`SessionState`), `getWorkerInfo()` (`WorkerInfoReply`: process id, version, +ProgID, CLSID) and `drainEvents(maxEvents)` (`List`; replies are +bounded server-side, so loop until an empty list). + ### Reconnect-replay gap signal When you resume a stream with `streamEventsAfter(afterWorkerSequence)` and the 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 e884c97..26d1a4f 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 @@ -19,8 +19,12 @@ import mxaccess_gateway.v1.MxaccessGateway.StreamEventsRequest; *

Events arrive on a background gRPC thread and are buffered in a bounded * blocking queue; the iterator drains them on the calling thread. Closing the * 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}. + * cancelled, the events already buffered remain readable, and the + * {@link MxGatewayException} is thrown only after they are drained (the consumer + * can then resume with {@code after_worker_sequence}). The capacity comes from + * {@link MxGatewayClientOptions#eventQueueCapacity()}. Callers that must never be + * dropped should use {@link MxGatewayClient#streamEventsAsync}, which delivers on + * the gRPC thread with the caller's own flow control. * *

Single consumer. This stream is not safe to drain from * more than one thread. Interleave {@link #hasNext()}/{@link #next()} (or @@ -46,10 +50,15 @@ public final class MxEventStream implements Iterator, AutoCloseable { private final BlockingQueue queue; private volatile ClientCallStreamObserver requestStream; private volatile boolean closed; + private final int capacity; + private boolean overflowed; private Object next; MxEventStream(int capacity) { - queue = new ArrayBlockingQueue<>(capacity); + this.capacity = capacity; + // Two extra slots are reserved so the overflow error and END sentinel + // can always be enqueued behind a full buffer of received events. + queue = new ArrayBlockingQueue<>(capacity + 2); } ClientResponseObserver observer() { @@ -143,12 +152,16 @@ public final class MxEventStream implements Iterator, AutoCloseable { return queue.take(); } catch (InterruptedException error) { Thread.currentThread().interrupt(); + ClientCallStreamObserver stream = requestStream; + if (stream != null) { + stream.cancel("interrupted while reading events", error); + } return new StatusRuntimeException(Status.CANCELLED.withDescription("interrupted while reading events")); } } } - private void offer(Object value) { + private synchronized void offer(Object value) { Objects.requireNonNull(value, "value"); if (value == END) { if (!queue.offer(value)) { @@ -157,14 +170,29 @@ public final class MxEventStream implements Iterator, AutoCloseable { } return; } - if (!queue.offer(value)) { + if (overflowed) { + return; + } + if (value instanceof MxEvent) { + if (queue.size() < capacity) { + queue.offer(value); + return; + } + // Keep the events already received so the consumer can drain them + // and resume from the last one via after_worker_sequence; the + // overflow error is delivered behind them. + overflowed = true; ClientCallStreamObserver stream = requestStream; if (stream != null) { stream.cancel("client event stream queue overflowed", null); } - queue.clear(); queue.offer(new MxGatewayException("gateway stream events queue overflowed")); queue.offer(END); + return; + } + if (!queue.offer(value)) { + queue.clear(); + queue.offer(value); } } } diff --git a/clients/java/zb-mom-ww-mxgateway-client/src/main/java/com/zb/mom/ww/mxgateway/client/MxGatewayClient.java b/clients/java/zb-mom-ww-mxgateway-client/src/main/java/com/zb/mom/ww/mxgateway/client/MxGatewayClient.java index 59798cb..024fead 100644 --- a/clients/java/zb-mom-ww-mxgateway-client/src/main/java/com/zb/mom/ww/mxgateway/client/MxGatewayClient.java +++ b/clients/java/zb-mom-ww-mxgateway-client/src/main/java/com/zb/mom/ww/mxgateway/client/MxGatewayClient.java @@ -245,7 +245,7 @@ public final class MxGatewayClient implements AutoCloseable { * @return an iterator-style stream of events */ public MxEventStream streamEvents(StreamEventsRequest request) { - MxEventStream stream = new MxEventStream(16); + MxEventStream stream = new MxEventStream(options.eventQueueCapacity()); withStreamDeadline(rawAsyncStub()).streamEvents(request, stream.observer()); return stream; } diff --git a/clients/java/zb-mom-ww-mxgateway-client/src/main/java/com/zb/mom/ww/mxgateway/client/MxGatewayClientOptions.java b/clients/java/zb-mom-ww-mxgateway-client/src/main/java/com/zb/mom/ww/mxgateway/client/MxGatewayClientOptions.java index 2f5642f..e5a8998 100644 --- a/clients/java/zb-mom-ww-mxgateway-client/src/main/java/com/zb/mom/ww/mxgateway/client/MxGatewayClientOptions.java +++ b/clients/java/zb-mom-ww-mxgateway-client/src/main/java/com/zb/mom/ww/mxgateway/client/MxGatewayClientOptions.java @@ -15,6 +15,8 @@ public final class MxGatewayClientOptions { private static final Duration DEFAULT_CONNECT_TIMEOUT = Duration.ofSeconds(10); private static final Duration DEFAULT_CALL_TIMEOUT = Duration.ofSeconds(30); private static final int DEFAULT_MAX_GRPC_MESSAGE_BYTES = 16 * 1024 * 1024; + /** Default {@code StreamEvents} iterator buffer; sized to absorb bursts without dropping a slow consumer. */ + static final int DEFAULT_EVENT_QUEUE_CAPACITY = 1024; private final String endpoint; private final String apiKey; @@ -26,6 +28,7 @@ public final class MxGatewayClientOptions { private final Duration callTimeout; private final Duration streamTimeout; private final int maxGrpcMessageBytes; + private final int eventQueueCapacity; private MxGatewayClientOptions(Builder builder) { endpoint = requireText(builder.endpoint, "endpoint"); @@ -40,6 +43,10 @@ public final class MxGatewayClientOptions { maxGrpcMessageBytes = builder.maxGrpcMessageBytes <= 0 ? DEFAULT_MAX_GRPC_MESSAGE_BYTES : builder.maxGrpcMessageBytes; + if (builder.eventQueueCapacity <= 0) { + throw new IllegalArgumentException("eventQueueCapacity must be positive"); + } + eventQueueCapacity = builder.eventQueueCapacity; } /** @@ -149,6 +156,17 @@ public final class MxGatewayClientOptions { return maxGrpcMessageBytes; } + /** + * Returns the capacity of the buffer behind {@link MxGatewayClient#streamEvents}. + * A consumer that falls more than this many events behind has its stream + * cancelled with an overflow error after the buffered events are drained. + * + * @return the event buffer capacity (default 1024) + */ + public int eventQueueCapacity() { + return eventQueueCapacity; + } + @Override public String toString() { return "MxGatewayClientOptions{" @@ -199,6 +217,7 @@ public final class MxGatewayClientOptions { private Duration callTimeout; private Duration streamTimeout; private int maxGrpcMessageBytes; + private int eventQueueCapacity = DEFAULT_EVENT_QUEUE_CAPACITY; private Builder() { } @@ -314,6 +333,17 @@ public final class MxGatewayClientOptions { return this; } + /** + * Sets the capacity of the iterator-style event stream buffer. Defaults to 1024. + * + * @param value the buffer capacity, must be positive + * @return this builder + */ + public Builder eventQueueCapacity(int value) { + eventQueueCapacity = value; + return this; + } + /** * Builds an immutable {@link MxGatewayClientOptions} from the current state. * 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 485ba5b..e0c803e 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 @@ -37,6 +37,13 @@ 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.DrainEventsCommand; +import mxaccess_gateway.v1.MxaccessGateway.GetSessionStateCommand; +import mxaccess_gateway.v1.MxaccessGateway.GetWorkerInfoCommand; +import mxaccess_gateway.v1.MxaccessGateway.MxEvent; +import mxaccess_gateway.v1.MxaccessGateway.PingCommand; +import mxaccess_gateway.v1.MxaccessGateway.SessionState; +import mxaccess_gateway.v1.MxaccessGateway.WorkerInfoReply; import mxaccess_gateway.v1.MxaccessGateway.StreamEventsRequest; import mxaccess_gateway.v1.MxaccessGateway.SubscribeBulkCommand; import mxaccess_gateway.v1.MxaccessGateway.SubscribeResult; @@ -1033,6 +1040,67 @@ public final class MxGatewaySession implements AutoCloseable { .build()); } + /** + * Sends a {@code Ping} diagnostic command to the session's worker. + * + * @param message free-form text carried in the command + * @return the raw command reply + * @throws MxGatewayException on transport or protocol failure + */ + public MxCommandReply ping(String message) { + return invokeCommand(MxCommand.newBuilder() + .setKind(MxCommandKind.MX_COMMAND_KIND_PING) + .setPing(PingCommand.newBuilder().setMessage(message == null ? "" : message)) + .build()); + } + + /** + * Queries the worker-side state of this session. + * + * @return the reported {@link SessionState} + * @throws MxGatewayException on transport or protocol failure + */ + public SessionState getSessionState() { + return invokeCommand(MxCommand.newBuilder() + .setKind(MxCommandKind.MX_COMMAND_KIND_GET_SESSION_STATE) + .setGetSessionState(GetSessionStateCommand.getDefaultInstance()) + .build()) + .getSessionState() + .getState(); + } + + /** + * Queries the worker process and MXAccess COM identity. + * + * @return the {@link WorkerInfoReply} (process id, version, ProgID, CLSID) + * @throws MxGatewayException on transport or protocol failure + */ + public WorkerInfoReply getWorkerInfo() { + return invokeCommand(MxCommand.newBuilder() + .setKind(MxCommandKind.MX_COMMAND_KIND_GET_WORKER_INFO) + .setGetWorkerInfo(GetWorkerInfoCommand.getDefaultInstance()) + .build()) + .getWorkerInfo(); + } + + /** + * Pulls queued events from the worker without opening a stream. The reply + * is bounded server-side and may carry fewer than {@code maxEvents}; drain + * iteratively until an empty list is returned. + * + * @param maxEvents the maximum number of events to return + * @return the drained events, possibly empty + * @throws MxGatewayException on transport or protocol failure + */ + public List drainEvents(int maxEvents) { + return invokeCommand(MxCommand.newBuilder() + .setKind(MxCommandKind.MX_COMMAND_KIND_DRAIN_EVENTS) + .setDrainEvents(DrainEventsCommand.newBuilder().setMaxEvents(maxEvents)) + .build()) + .getDrainEvents() + .getEventsList(); + } + /** * Sends a pre-built {@link MxCommand} for this session and returns the raw * reply, attaching a freshly generated client correlation id. diff --git a/clients/java/zb-mom-ww-mxgateway-client/src/test/java/com/zb/mom/ww/mxgateway/client/MxEventStreamTests.java b/clients/java/zb-mom-ww-mxgateway-client/src/test/java/com/zb/mom/ww/mxgateway/client/MxEventStreamTests.java new file mode 100644 index 0000000..b2ef375 --- /dev/null +++ b/clients/java/zb-mom-ww-mxgateway-client/src/test/java/com/zb/mom/ww/mxgateway/client/MxEventStreamTests.java @@ -0,0 +1,93 @@ +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.assertThrows; +import static org.junit.jupiter.api.Assertions.assertTrue; + +import io.grpc.stub.ClientCallStreamObserver; +import io.grpc.stub.ClientResponseObserver; +import java.util.concurrent.atomic.AtomicInteger; +import mxaccess_gateway.v1.MxaccessGateway.MxEvent; +import mxaccess_gateway.v1.MxaccessGateway.StreamEventsRequest; +import org.junit.jupiter.api.Test; + +final class MxEventStreamTests { + @Test + void overflowKeepsRetainedEventsThenSurfacesError() { + MxEventStream stream = new MxEventStream(3); + RecordingCall call = new RecordingCall(); + ClientResponseObserver observer = stream.observer(); + observer.beforeStart(call); + + for (long i = 1; i <= 5; i++) { + observer.onNext(MxEvent.newBuilder().setWorkerSequence(i).build()); + } + + assertTrue(call.cancelCount.get() >= 1, "overflow cancels the call"); + assertEquals(1L, stream.next().getWorkerSequence()); + assertEquals(2L, stream.next().getWorkerSequence()); + assertEquals(3L, stream.next().getWorkerSequence()); + MxGatewayException error = assertThrows(MxGatewayException.class, stream::hasNext); + assertTrue(error.getMessage().contains("overflowed")); + assertFalse(stream.hasNext()); + } + + @Test + void interruptCancelsCallAndRestoresInterruptFlag() { + MxEventStream stream = new MxEventStream(4); + RecordingCall call = new RecordingCall(); + stream.observer().beforeStart(call); + + Thread.currentThread().interrupt(); + try { + assertThrows(io.grpc.StatusRuntimeException.class, stream::hasNext); + assertTrue(Thread.currentThread().isInterrupted(), "interrupt flag restored"); + assertEquals(1, call.cancelCount.get(), "call cancelled on interrupt"); + } finally { + Thread.interrupted(); + } + } + + private static final class RecordingCall extends ClientCallStreamObserver { + final AtomicInteger cancelCount = new AtomicInteger(); + + @Override + public void cancel(String message, Throwable cause) { + cancelCount.incrementAndGet(); + } + + @Override + public boolean isReady() { + return true; + } + + @Override + public void setOnReadyHandler(Runnable onReadyHandler) { + } + + @Override + public void request(int count) { + } + + @Override + public void setMessageCompression(boolean enable) { + } + + @Override + public void disableAutoInboundFlowControl() { + } + + @Override + public void onNext(StreamEventsRequest value) { + } + + @Override + public void onError(Throwable t) { + } + + @Override + public void onCompleted() { + } + } +} 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 b6c04bf..b0cb356 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 @@ -51,7 +51,10 @@ 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.DrainEventsReply; import mxaccess_gateway.v1.MxaccessGateway.SessionState; +import mxaccess_gateway.v1.MxaccessGateway.SessionStateReply; +import mxaccess_gateway.v1.MxaccessGateway.WorkerInfoReply; import mxaccess_gateway.v1.MxaccessGateway.StreamAlarmsRequest; import mxaccess_gateway.v1.MxaccessGateway.StreamEventsRequest; import mxaccess_gateway.v1.MxaccessGateway.SubscribeResult; @@ -727,6 +730,63 @@ final class MxGatewayClientSessionTests { } } + @Test + void diagnosticsHelpersBuildTypedCommandsAndDecodeReplies() throws Exception { + AtomicReference last = new AtomicReference<>(); + TestGatewayService service = new TestGatewayService() { + @Override + public void invoke(MxCommandRequest request, StreamObserver responseObserver) { + last.set(request); + MxCommandReply.Builder reply = MxCommandReply.newBuilder() + .setSessionId(request.getSessionId()) + .setKind(request.getCommand().getKind()) + .setProtocolStatus(ok()); + switch (request.getCommand().getKind()) { + case MX_COMMAND_KIND_GET_SESSION_STATE -> reply.setSessionState( + SessionStateReply.newBuilder().setState(SessionState.SESSION_STATE_READY)); + case MX_COMMAND_KIND_GET_WORKER_INFO -> reply.setWorkerInfo( + WorkerInfoReply.newBuilder().setWorkerProcessId(99).setWorkerVersion("1.2.3")); + case MX_COMMAND_KIND_DRAIN_EVENTS -> reply.setDrainEvents(DrainEventsReply.newBuilder() + .addEvents(MxEvent.newBuilder().setWorkerSequence(5))); + default -> { } + } + 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, "existing-session"); + + MxCommandReply ping = session.ping("hello"); + assertEquals(MxCommandKind.MX_COMMAND_KIND_PING, ping.getKind()); + assertEquals("hello", last.get().getCommand().getPing().getMessage()); + + assertEquals(SessionState.SESSION_STATE_READY, session.getSessionState()); + assertEquals(MxCommandKind.MX_COMMAND_KIND_GET_SESSION_STATE, last.get().getCommand().getKind()); + + WorkerInfoReply info = session.getWorkerInfo(); + assertEquals(99, info.getWorkerProcessId()); + assertEquals("1.2.3", info.getWorkerVersion()); + + List events = session.drainEvents(10); + assertEquals(1, events.size()); + assertEquals(5L, events.get(0).getWorkerSequence()); + assertEquals(10, last.get().getCommand().getDrainEvents().getMaxEvents()); + } + } + + @Test + void eventQueueCapacityOptionDefaultsAndValidates() { + assertEquals(1024, MxGatewayClientOptions.builder().endpoint("x").build().eventQueueCapacity()); + assertEquals( + 7, MxGatewayClientOptions.builder().endpoint("x").eventQueueCapacity(7).build().eventQueueCapacity()); + assertThrows( + IllegalArgumentException.class, + () -> MxGatewayClientOptions.builder().endpoint("x").eventQueueCapacity(0).build()); + } + private static TestGatewayService okInvokeService(AtomicReference commandRequest) { return new TestGatewayService() { @Override