Fix runtime review findings

clients/java/mxgateway-cli/src/test/java/com/dohertylan/mxgateway/cli/MxGatewayCliTests.java

@@ -32,7 +32,7 @@ final class MxGatewayCliTests {
         assertEquals(0, run.exitCode());
         assertEquals("", run.errors());
         assertTrue(run.output().contains("mxgateway-java 0.1.0"));
-        assertTrue(run.output().contains("gatewayProtocolVersion=1"));
+        assertTrue(run.output().contains("gatewayProtocolVersion=2"));
         assertTrue(run.output().contains("workerProtocolVersion=1"));
     }
 
@@ -42,7 +42,7 @@ final class MxGatewayCliTests {
 
         assertEquals(0, run.exitCode());
         assertTrue(run.output().contains("\"clientVersion\":\"0.1.0\""));
-        assertTrue(run.output().contains("\"gatewayProtocolVersion\":1"));
+        assertTrue(run.output().contains("\"gatewayProtocolVersion\":2"));
     }
 
     @Test

clients/java/mxgateway-client/src/main/java/com/dohertylan/mxgateway/client/DeployEventStream.java

@@ -11,6 +11,7 @@ import java.util.NoSuchElementException;
 import java.util.Objects;
 import java.util.concurrent.ArrayBlockingQueue;
 import java.util.concurrent.BlockingQueue;
+import java.util.concurrent.atomic.AtomicBoolean;
 
 /**
  * Iterator-style adaptor over the {@code WatchDeployEvents} server-streaming
@@ -22,8 +23,8 @@ public final class DeployEventStream implements Iterator<DeployEvent>, AutoClose
     private static final Object END = new Object();
 
     private final BlockingQueue<Object> queue;
+    private final AtomicBoolean closed = new AtomicBoolean();
     private volatile ClientCallStreamObserver<WatchDeployEventsRequest> requestStream;
-    private volatile boolean closed;
     private Object next;
 
     DeployEventStream(int capacity) {
@@ -35,6 +36,9 @@ public final class DeployEventStream implements Iterator<DeployEvent>, AutoClose
             @Override
             public void beforeStart(ClientCallStreamObserver<WatchDeployEventsRequest> requestStream) {
                 DeployEventStream.this.requestStream = requestStream;
+                if (closed.get()) {
+                    requestStream.cancel("client cancelled deploy event stream", null);
+                }
             }
 
             @Override
@@ -44,7 +48,7 @@ public final class DeployEventStream implements Iterator<DeployEvent>, AutoClose
 
             @Override
             public void onError(Throwable error) {
-                if (Status.fromThrowable(error).getCode() == Status.Code.CANCELLED && closed) {
+                if (Status.fromThrowable(error).getCode() == Status.Code.CANCELLED && closed.get()) {
                     offer(END);
                     return;
                 }
@@ -90,7 +94,7 @@ public final class DeployEventStream implements Iterator<DeployEvent>, AutoClose
 
     @Override
     public void close() {
-        closed = true;
+        closed.set(true);
         ClientCallStreamObserver<WatchDeployEventsRequest> stream = requestStream;
         if (stream != null) {
             stream.cancel("client cancelled deploy event stream", null);

clients/java/mxgateway-client/src/main/java/com/dohertylan/mxgateway/client/GalaxyRepositoryClient.java

@@ -36,6 +36,8 @@ import javax.net.ssl.SSLException;
  * {@link MxGatewayClient}.
  */
 public final class GalaxyRepositoryClient implements AutoCloseable {
+    private static final int DISCOVER_HIERARCHY_PAGE_SIZE = 5000;
+
     private final ManagedChannel ownedChannel;
     private final MxGatewayClientOptions options;
     private final GalaxyRepositoryGrpc.GalaxyRepositoryBlockingStub blockingStub;
@@ -130,9 +132,17 @@ public final class GalaxyRepositoryClient implements AutoCloseable {
      */
     public List<GalaxyObject> discoverHierarchy() {
         try {
-            DiscoverHierarchyReply reply =
-                    rawBlockingStub().discoverHierarchy(DiscoverHierarchyRequest.getDefaultInstance());
-            return reply.getObjectsList();
+            java.util.ArrayList<GalaxyObject> objects = new java.util.ArrayList<>();
+            String pageToken = "";
+            do {
+                DiscoverHierarchyReply reply = rawBlockingStub().discoverHierarchy(DiscoverHierarchyRequest.newBuilder()
+                        .setPageSize(DISCOVER_HIERARCHY_PAGE_SIZE)
+                        .setPageToken(pageToken)
+                        .build());
+                objects.addAll(reply.getObjectsList());
+                pageToken = reply.getNextPageToken();
+            } while (!pageToken.isBlank());
+            return objects;
         } catch (RuntimeException error) {
             if (error instanceof MxGatewayException) {
                 throw error;
@@ -142,8 +152,7 @@ public final class GalaxyRepositoryClient implements AutoCloseable {
     }
 
     public CompletableFuture<List<GalaxyObject>> discoverHierarchyAsync() {
-        return toCompletable(rawFutureStub().discoverHierarchy(DiscoverHierarchyRequest.getDefaultInstance()))
-                .thenApply(DiscoverHierarchyReply::getObjectsList);
+        return discoverHierarchyPageAsync("", new java.util.ArrayList<>());
     }
 
     /**
@@ -226,7 +235,7 @@ public final class GalaxyRepositoryClient implements AutoCloseable {
 
     private static ManagedChannel createChannel(MxGatewayClientOptions options) {
         NettyChannelBuilder builder = NettyChannelBuilder.forTarget(options.endpoint())
-                .maxInboundMessageSize(16 * 1024 * 1024);
+                .maxInboundMessageSize(options.maxGrpcMessageBytes());
         if (!options.connectTimeout().isNegative()) {
             builder.withOption(
                     io.grpc.netty.shaded.io.netty.channel.ChannelOption.CONNECT_TIMEOUT_MILLIS,
@@ -258,6 +267,21 @@ public final class GalaxyRepositoryClient implements AutoCloseable {
         return stub.withDeadlineAfter(options.callTimeout().toNanos(), TimeUnit.NANOSECONDS);
     }
 
+    private CompletableFuture<List<GalaxyObject>> discoverHierarchyPageAsync(
+            String pageToken, java.util.ArrayList<GalaxyObject> objects) {
+        DiscoverHierarchyRequest request = DiscoverHierarchyRequest.newBuilder()
+                .setPageSize(DISCOVER_HIERARCHY_PAGE_SIZE)
+                .setPageToken(pageToken)
+                .build();
+        return toCompletable(rawFutureStub().discoverHierarchy(request)).thenCompose(reply -> {
+            objects.addAll(reply.getObjectsList());
+            if (reply.getNextPageToken().isBlank()) {
+                return CompletableFuture.completedFuture(objects);
+            }
+            return discoverHierarchyPageAsync(reply.getNextPageToken(), objects);
+        });
+    }
+
     private static <T> CompletableFuture<T> toCompletable(com.google.common.util.concurrent.ListenableFuture<T> source) {
         CompletableFuture<T> target = new CompletableFuture<>();
         Futures.addCallback(

clients/java/mxgateway-client/src/main/java/com/dohertylan/mxgateway/client/MxGatewayClient.java

@@ -169,7 +169,7 @@ public final class MxGatewayClient implements AutoCloseable {
 
     private static ManagedChannel createChannel(MxGatewayClientOptions options) {
         NettyChannelBuilder builder = NettyChannelBuilder.forTarget(options.endpoint())
-                .maxInboundMessageSize(16 * 1024 * 1024);
+                .maxInboundMessageSize(options.maxGrpcMessageBytes());
         if (!options.connectTimeout().isNegative()) {
             builder.withOption(
                     io.grpc.netty.shaded.io.netty.channel.ChannelOption.CONNECT_TIMEOUT_MILLIS,

clients/java/mxgateway-client/src/main/java/com/dohertylan/mxgateway/client/MxGatewayClientOptions.java

@@ -7,6 +7,7 @@ import java.util.Objects;
 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;
 
     private final String endpoint;
     private final String apiKey;
@@ -16,6 +17,7 @@ public final class MxGatewayClientOptions {
     private final Duration connectTimeout;
     private final Duration callTimeout;
     private final Duration streamTimeout;
+    private final int maxGrpcMessageBytes;
 
     private MxGatewayClientOptions(Builder builder) {
         endpoint = requireText(builder.endpoint, "endpoint");
@@ -26,6 +28,9 @@ public final class MxGatewayClientOptions {
         connectTimeout = builder.connectTimeout == null ? DEFAULT_CONNECT_TIMEOUT : builder.connectTimeout;
         callTimeout = builder.callTimeout == null ? DEFAULT_CALL_TIMEOUT : builder.callTimeout;
         streamTimeout = builder.streamTimeout;
+        maxGrpcMessageBytes = builder.maxGrpcMessageBytes <= 0
+                ? DEFAULT_MAX_GRPC_MESSAGE_BYTES
+                : builder.maxGrpcMessageBytes;
     }
 
     public static Builder builder() {
@@ -68,6 +73,10 @@ public final class MxGatewayClientOptions {
         return streamTimeout;
     }
 
+    public int maxGrpcMessageBytes() {
+        return maxGrpcMessageBytes;
+    }
+
     @Override
     public String toString() {
         return "MxGatewayClientOptions{"
@@ -90,6 +99,8 @@ public final class MxGatewayClientOptions {
                 + callTimeout
                 + ", streamTimeout="
                 + streamTimeout
+                + ", maxGrpcMessageBytes="
+                + maxGrpcMessageBytes
                 + '}';
     }
 
@@ -109,6 +120,7 @@ public final class MxGatewayClientOptions {
         private Duration connectTimeout;
         private Duration callTimeout;
         private Duration streamTimeout;
+        private int maxGrpcMessageBytes;
 
         private Builder() {
         }
@@ -153,6 +165,11 @@ public final class MxGatewayClientOptions {
             return this;
         }
 
+        public Builder maxGrpcMessageBytes(int value) {
+            maxGrpcMessageBytes = value;
+            return this;
+        }
+
         public MxGatewayClientOptions build() {
             return new MxGatewayClientOptions(this);
         }

clients/java/mxgateway-client/src/main/java/com/dohertylan/mxgateway/client/MxGatewayClientVersion.java

@@ -1,7 +1,7 @@
 package com.dohertylan.mxgateway.client;
 
 public final class MxGatewayClientVersion {
-    private static final int GATEWAY_PROTOCOL_VERSION = 1;
+    private static final int GATEWAY_PROTOCOL_VERSION = 2;
     private static final int WORKER_PROTOCOL_VERSION = 1;
     private static final String CLIENT_VERSION = "0.1.0";
 

clients/java/mxgateway-client/src/test/java/com/dohertylan/mxgateway/client/GalaxyRepositoryClientTests.java

@@ -25,6 +25,8 @@ import io.grpc.ServerCallHandler;
 import io.grpc.ServerInterceptor;
 import io.grpc.inprocess.InProcessChannelBuilder;
 import io.grpc.inprocess.InProcessServerBuilder;
+import io.grpc.stub.ClientCallStreamObserver;
+import io.grpc.stub.ClientResponseObserver;
 import io.grpc.stub.StreamObserver;
 import java.time.Duration;
 import java.time.Instant;
@@ -100,31 +102,44 @@ final class GalaxyRepositoryClientTests {
 
     @Test
     void discoverHierarchyReturnsObjectsAndAttributes() throws Exception {
-        AtomicReference<DiscoverHierarchyRequest> seenRequest = new AtomicReference<>();
+        AtomicReference<DiscoverHierarchyRequest> firstRequest = new AtomicReference<>();
+        AtomicReference<DiscoverHierarchyRequest> secondRequest = new AtomicReference<>();
         TestService service = new TestService() {
             @Override
             public void discoverHierarchy(
                     DiscoverHierarchyRequest request, StreamObserver<DiscoverHierarchyReply> responseObserver) {
-                seenRequest.set(request);
-                responseObserver.onNext(DiscoverHierarchyReply.newBuilder()
-                        .addObjects(GalaxyObject.newBuilder()
-                                .setGobjectId(7)
-                                .setTagName("Pump_001")
-                                .setContainedName("Pump")
-                                .setBrowseName("Pump")
-                                .setParentGobjectId(1)
-                                .setIsArea(false)
-                                .setCategoryId(3)
-                                .setHostedByGobjectId(0)
-                                .addTemplateChain("$Pump")
-                                .addAttributes(GalaxyAttribute.newBuilder()
-                                        .setAttributeName("Speed")
-                                        .setFullTagReference("Pump_001.Speed")
-                                        .setMxDataType(5)
-                                        .setDataTypeName("MxFloat")
-                                        .setIsArray(false)
-                                        .setIsHistorized(true)))
-                        .build());
+                if (request.getPageToken().isEmpty()) {
+                    firstRequest.set(request);
+                    responseObserver.onNext(DiscoverHierarchyReply.newBuilder()
+                            .setNextPageToken("page-2")
+                            .setTotalObjectCount(2)
+                            .addObjects(GalaxyObject.newBuilder()
+                                    .setGobjectId(7)
+                                    .setTagName("Pump_001")
+                                    .setContainedName("Pump")
+                                    .setBrowseName("Pump")
+                                    .setParentGobjectId(1)
+                                    .setIsArea(false)
+                                    .setCategoryId(3)
+                                    .setHostedByGobjectId(0)
+                                    .addTemplateChain("$Pump")
+                                    .addAttributes(GalaxyAttribute.newBuilder()
+                                            .setAttributeName("Speed")
+                                            .setFullTagReference("Pump_001.Speed")
+                                            .setMxDataType(5)
+                                            .setDataTypeName("MxFloat")
+                                            .setIsArray(false)
+                                            .setIsHistorized(true)))
+                            .build());
+                } else {
+                    secondRequest.set(request);
+                    responseObserver.onNext(DiscoverHierarchyReply.newBuilder()
+                            .setTotalObjectCount(2)
+                            .addObjects(GalaxyObject.newBuilder()
+                                    .setGobjectId(8)
+                                    .setTagName("Pump_002"))
+                            .build());
+                }
                 responseObserver.onCompleted();
             }
         };
@@ -132,7 +147,10 @@ final class GalaxyRepositoryClientTests {
         try (InProcessGalaxy g = InProcessGalaxy.start(service, new AtomicReference<>());
                 GalaxyRepositoryClient client = g.client("")) {
             List<GalaxyObject> objects = client.discoverHierarchy();
-            assertEquals(1, objects.size());
+            assertEquals(2, objects.size());
+            assertEquals(5000, firstRequest.get().getPageSize());
+            assertEquals("", firstRequest.get().getPageToken());
+            assertEquals("page-2", secondRequest.get().getPageToken());
             GalaxyObject only = objects.get(0);
             assertEquals(7, only.getGobjectId());
             assertEquals("Pump_001", only.getTagName());
@@ -142,6 +160,20 @@ final class GalaxyRepositoryClientTests {
         }
     }
 
+    @Test
+    void deployEventStreamCloseBeforeBeforeStartCancelsStream() {
+        DeployEventStream stream = new DeployEventStream(4);
+        ClientResponseObserver<WatchDeployEventsRequest, DeployEvent> observer = stream.observer();
+        RecordingClientCallStreamObserver requestStream = new RecordingClientCallStreamObserver();
+
+        stream.close();
+        observer.beforeStart(requestStream);
+
+        assertTrue(requestStream.cancelled);
+        assertEquals("client cancelled deploy event stream", requestStream.cancelMessage);
+        assertFalse(stream.hasNext());
+    }
+
     @Test
     void watchDeployEventsReceivesEventsInOrder() throws Exception {
         DeployEvent first = DeployEvent.newBuilder()
@@ -281,6 +313,51 @@ final class GalaxyRepositoryClientTests {
         }
     }
 
+    private static final class RecordingClientCallStreamObserver
+            extends ClientCallStreamObserver<WatchDeployEventsRequest> {
+        private boolean cancelled;
+        private String cancelMessage;
+
+        @Override
+        public boolean isReady() {
+            return true;
+        }
+
+        @Override
+        public void setOnReadyHandler(Runnable onReadyHandler) {
+        }
+
+        @Override
+        public void disableAutoInboundFlowControl() {
+        }
+
+        @Override
+        public void request(int count) {
+        }
+
+        @Override
+        public void setMessageCompression(boolean enable) {
+        }
+
+        @Override
+        public void cancel(String message, Throwable cause) {
+            cancelled = true;
+            cancelMessage = message;
+        }
+
+        @Override
+        public void onNext(WatchDeployEventsRequest value) {
+        }
+
+        @Override
+        public void onError(Throwable error) {
+        }
+
+        @Override
+        public void onCompleted() {
+        }
+    }
+
     private record InProcessGalaxy(Server server, ManagedChannel channel) implements AutoCloseable {
         static InProcessGalaxy start(
                 GalaxyRepositoryGrpc.GalaxyRepositoryImplBase service, AtomicReference<String> authorization)