Merge stream G (Java client: configurable event queue, overflow retention, interrupt cancel, typed diagnostics)

2026-10-22 06:09:55 -04:00
parent 7636573ebb 48c9323d4a
commit c18566bcf2
7 changed files with 310 additions and 7 deletions
@@ -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<MxEvent>)` 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<MxEvent>`; 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
@@ -19,8 +19,12 @@ import mxaccess_gateway.v1.MxaccessGateway.StreamEventsRequest;
 * <p>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.
 *
 * <p><strong>Single consumer.</strong> 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<MxEvent>, AutoCloseable {
    private final BlockingQueue<Object> queue;
    private volatile ClientCallStreamObserver<StreamEventsRequest> 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<StreamEventsRequest, MxEvent> observer() {
@@ -143,12 +152,16 @@ public final class MxEventStream implements Iterator<MxEvent>, AutoCloseable {
                return queue.take();
            } catch (InterruptedException error) {
                Thread.currentThread().interrupt();
+                ClientCallStreamObserver<StreamEventsRequest> 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<MxEvent>, 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<StreamEventsRequest> 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);
        }
    }
 }
@@ -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;
    }
@@ -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.
         *
@@ -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<MxEvent> 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.
@@ -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<StreamEventsRequest, MxEvent> 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<StreamEventsRequest> {
+        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() {
+        }
+    }
+}
@@ -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<MxCommandRequest> last = new AtomicReference<>();
+        TestGatewayService service = new TestGatewayService() {
+            @Override
+            public void invoke(MxCommandRequest request, StreamObserver<MxCommandReply> 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<MxEvent> 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<MxCommandRequest> commandRequest) {
        return new TestGatewayService() {
            @Override