Support StreamInfo command

References rabbitmq/rabbitmq-server#5412

Author: acogoluegnes

src/main/java/com/rabbitmq/stream/Constants.java

@@ -68,6 +68,7 @@ public final class Constants {
   public static final short COMMAND_ROUTE = 24;
   public static final short COMMAND_PARTITIONS = 25;
   public static final short COMMAND_EXCHANGE_COMMAND_VERSIONS = 27;
+  public static final short COMMAND_STREAM_INFO = 28;
 
   public static final short VERSION_1 = 1;
   public static final short VERSION_2 = 2;

src/main/java/com/rabbitmq/stream/Environment.java

@@ -1,4 +1,4 @@
-// Copyright (c) 2020-2021 VMware, Inc. or its affiliates.  All rights reserved.
+// Copyright (c) 2020-2022 VMware, Inc. or its affiliates.  All rights reserved.
 //
 // This software, the RabbitMQ Stream Java client library, is dual-licensed under the
 // Mozilla Public License 2.0 ("MPL"), and the Apache License version 2 ("ASL").
@@ -61,6 +61,8 @@ static EnvironmentBuilder builder() {
    */
   void deleteStream(String stream);
 
+  StreamInfo queryStreamInfo(String stream);
+
   /**
    * Create a {@link ProducerBuilder} to configure and create a {@link Producer}.
    *

src/main/java/com/rabbitmq/stream/MessageHandler.java

@@ -1,4 +1,4 @@
-// Copyright (c) 2020-2021 VMware, Inc. or its affiliates.  All rights reserved.
+// Copyright (c) 2020-2022 VMware, Inc. or its affiliates.  All rights reserved.
 //
 // This software, the RabbitMQ Stream Java client library, is dual-licensed under the
 // Mozilla Public License 2.0 ("MPL"), and the Apache License version 2 ("ASL").

src/main/java/com/rabbitmq/stream/NoOffsetException.java

@@ -0,0 +1,21 @@
+// Copyright (c) 2022 VMware, Inc. or its affiliates.  All rights reserved.
+//
+// This software, the RabbitMQ Stream Java client library, is dual-licensed under the
+// Mozilla Public License 2.0 ("MPL"), and the Apache License version 2 ("ASL").
+// For the MPL, please see LICENSE-MPL-RabbitMQ. For the ASL,
+// please see LICENSE-APACHE2.
+//
+// This software is distributed on an "AS IS" basis, WITHOUT WARRANTY OF ANY KIND,
+// either express or implied. See the LICENSE file for specific language governing
+// rights and limitations of this software.
+//
+// If you have any questions regarding licensing, please contact us at
+// info@rabbitmq.com.
+package com.rabbitmq.stream;
+
+public class NoOffsetException extends StreamException {
+
+  public NoOffsetException(String message) {
+    super(message, Constants.RESPONSE_CODE_NO_OFFSET);
+  }
+}

src/main/java/com/rabbitmq/stream/StreamDoesNotExistException.java

@@ -1,4 +1,4 @@
-// Copyright (c) 2020-2021 VMware, Inc. or its affiliates.  All rights reserved.
+// Copyright (c) 2020-2022 VMware, Inc. or its affiliates.  All rights reserved.
 //
 // This software, the RabbitMQ Stream Java client library, is dual-licensed under the
 // Mozilla Public License 2.0 ("MPL"), and the Apache License version 2 ("ASL").
@@ -19,7 +19,7 @@ public class StreamDoesNotExistException extends StreamException {
   private final String stream;
 
   public StreamDoesNotExistException(String stream) {
-    super("Stream " + stream + "  does not exist");
+    super("Stream " + stream + "  does not exist", Constants.RESPONSE_CODE_STREAM_DOES_NOT_EXIST);
     this.stream = stream;
   }
 

src/main/java/com/rabbitmq/stream/StreamInfo.java

@@ -0,0 +1,21 @@
+// Copyright (c) 2022 VMware, Inc. or its affiliates.  All rights reserved.
+//
+// This software, the RabbitMQ Stream Java client library, is dual-licensed under the
+// Mozilla Public License 2.0 ("MPL"), and the Apache License version 2 ("ASL").
+// For the MPL, please see LICENSE-MPL-RabbitMQ. For the ASL,
+// please see LICENSE-APACHE2.
+//
+// This software is distributed on an "AS IS" basis, WITHOUT WARRANTY OF ANY KIND,
+// either express or implied. See the LICENSE file for specific language governing
+// rights and limitations of this software.
+//
+// If you have any questions regarding licensing, please contact us at
+// info@rabbitmq.com.
+package com.rabbitmq.stream;
+
+public interface StreamInfo {
+
+  long firstOffset();
+
+  long committedOffset();
+}

src/main/java/com/rabbitmq/stream/StreamNotAvailableException.java

@@ -0,0 +1,24 @@
+// Copyright (c) 2022 VMware, Inc. or its affiliates.  All rights reserved.
+//
+// This software, the RabbitMQ Stream Java client library, is dual-licensed under the
+// Mozilla Public License 2.0 ("MPL"), and the Apache License version 2 ("ASL").
+// For the MPL, please see LICENSE-MPL-RabbitMQ. For the ASL,
+// please see LICENSE-APACHE2.
+//
+// This software is distributed on an "AS IS" basis, WITHOUT WARRANTY OF ANY KIND,
+// either express or implied. See the LICENSE file for specific language governing
+// rights and limitations of this software.
+//
+// If you have any questions regarding licensing, please contact us at
+// info@rabbitmq.com.
+package com.rabbitmq.stream;
+
+public class StreamNotAvailableException extends StreamException {
+
+  private final String stream;
+
+  public StreamNotAvailableException(String stream) {
+    super("Stream " + stream + " is not available", Constants.RESPONSE_CODE_STREAM_NOT_AVAILABLE);
+    this.stream = stream;
+  }
+}

src/main/java/com/rabbitmq/stream/impl/Client.java

@@ -32,6 +32,7 @@
 import static com.rabbitmq.stream.Constants.COMMAND_SASL_AUTHENTICATE;
 import static com.rabbitmq.stream.Constants.COMMAND_SASL_HANDSHAKE;
 import static com.rabbitmq.stream.Constants.COMMAND_STORE_OFFSET;
+import static com.rabbitmq.stream.Constants.COMMAND_STREAM_INFO;
 import static com.rabbitmq.stream.Constants.COMMAND_SUBSCRIBE;
 import static com.rabbitmq.stream.Constants.COMMAND_UNSUBSCRIBE;
 import static com.rabbitmq.stream.Constants.RESPONSE_CODE_AUTHENTICATION_FAILURE;
@@ -1354,6 +1355,31 @@ List<FrameHandlerInfo> exchangeCommandVersions() {
     }
   }
 
+  StreamInfoResponse streamInfo(String stream) {
+    if (stream == null) {
+      throw new IllegalArgumentException("stream must not be null");
+    }
+    int length = 2 + 2 + 4 + 2 + stream.length(); // API code, version, correlation ID, 1 string
+    int correlationId = correlationSequence.incrementAndGet();
+    try {
+      ByteBuf bb = allocate(length + 4);
+      bb.writeInt(length);
+      bb.writeShort(encodeRequestCode(COMMAND_STREAM_INFO));
+      bb.writeShort(VERSION_1);
+      bb.writeInt(correlationId);
+      bb.writeShort(stream.length());
+      bb.writeBytes(stream.getBytes(StandardCharsets.UTF_8));
+      OutstandingRequest<StreamInfoResponse> request = new OutstandingRequest<>(this.rpcTimeout);
+      outstandingRequests.put(correlationId, request);
+      channel.writeAndFlush(bb);
+      request.block();
+      return request.response.get();
+    } catch (RuntimeException e) {
+      outstandingRequests.remove(correlationId);
+      throw new StreamException(e);
+    }
+  }
+
   void shutdownReason(ShutdownReason reason) {
     this.shutdownReason = reason;
   }
@@ -1852,7 +1878,7 @@ static class QueryPublisherSequenceResponse extends Response {
 
     private final long sequence;
 
-    public QueryPublisherSequenceResponse(short responseCode, long sequence) {
+    QueryPublisherSequenceResponse(short responseCode, long sequence) {
       super(responseCode);
       this.sequence = sequence;
     }
@@ -1862,6 +1888,20 @@ public long getSequence() {
     }
   }
 
+  static class StreamInfoResponse extends Response {
+
+    private final Map<String, String> info;
+
+    StreamInfoResponse(short responseCode, Map<String, String> info) {
+      super(responseCode);
+      this.info = Collections.unmodifiableMap(new HashMap<>(info));
+    }
+
+    public Map<String, String> getInfo() {
+      return info;
+    }
+  }
+
   public static class StreamMetadata {
 
     private final String stream;

src/main/java/com/rabbitmq/stream/impl/ServerFrameHandler.java

@@ -34,6 +34,7 @@
 import static com.rabbitmq.stream.Constants.COMMAND_ROUTE;
 import static com.rabbitmq.stream.Constants.COMMAND_SASL_AUTHENTICATE;
 import static com.rabbitmq.stream.Constants.COMMAND_SASL_HANDSHAKE;
+import static com.rabbitmq.stream.Constants.COMMAND_STREAM_INFO;
 import static com.rabbitmq.stream.Constants.COMMAND_SUBSCRIBE;
 import static com.rabbitmq.stream.Constants.COMMAND_TUNE;
 import static com.rabbitmq.stream.Constants.COMMAND_UNSUBSCRIBE;
@@ -62,6 +63,7 @@
 import com.rabbitmq.stream.impl.Client.Response;
 import com.rabbitmq.stream.impl.Client.SaslAuthenticateResponse;
 import com.rabbitmq.stream.impl.Client.ShutdownContext.ShutdownReason;
+import com.rabbitmq.stream.impl.Client.StreamInfoResponse;
 import com.rabbitmq.stream.impl.Client.StreamMetadata;
 import com.rabbitmq.stream.impl.Client.SubscriptionOffset;
 import com.rabbitmq.stream.metrics.MetricsCollector;
@@ -132,6 +134,7 @@ class ServerFrameHandler {
     handlers.put(COMMAND_ROUTE, new RouteFrameHandler());
     handlers.put(COMMAND_PARTITIONS, new PartitionsFrameHandler());
     handlers.put(COMMAND_EXCHANGE_COMMAND_VERSIONS, new ExchangeCommandVersionsFrameHandler());
+    handlers.put(COMMAND_STREAM_INFO, new StreamInfoFrameHandler());
     HANDLERS = new FrameHandler[maxCommandKey + 1][];
     handlers
         .entrySet()
@@ -1146,4 +1149,38 @@ int doHandle(Client client, ChannelHandlerContext ctx, ByteBuf message) {
       return read;
     }
   }
+
+  private static class StreamInfoFrameHandler extends BaseFrameHandler {
+
+    @Override
+    int doHandle(Client client, ChannelHandlerContext ctx, ByteBuf message) {
+      int correlationId = message.readInt();
+      int read = 4;
+
+      short responseCode = message.readShort();
+      read += 2;
+
+      int infoCount = message.readInt();
+      read += 4;
+      Map<String, String> info = new LinkedHashMap<>(infoCount);
+
+      for (int i = 0; i < infoCount; i++) {
+        String key = readString(message);
+        read += 2 + key.length();
+        String value = readString(message);
+        read += 2 + value.length();
+        info.put(key, value);
+      }
+
+      OutstandingRequest<StreamInfoResponse> outstandingRequest =
+          remove(client.outstandingRequests, correlationId, StreamInfoResponse.class);
+      if (outstandingRequest == null) {
+        LOGGER.warn("Could not find outstanding request with correlation ID {}", correlationId);
+      } else {
+        outstandingRequest.response().set(new StreamInfoResponse(responseCode, info));
+        outstandingRequest.countDown();
+      }
+      return read;
+    }
+  }
 }

src/main/java/com/rabbitmq/stream/impl/StreamEnvironment.java

@@ -14,6 +14,7 @@
 package com.rabbitmq.stream.impl;
 
 import static com.rabbitmq.stream.impl.Utils.formatConstant;
+import static com.rabbitmq.stream.impl.Utils.propagateException;
 import static java.util.concurrent.TimeUnit.SECONDS;
 
 import com.rabbitmq.stream.Address;
@@ -24,13 +25,16 @@
 import com.rabbitmq.stream.Environment;
 import com.rabbitmq.stream.MessageHandler;
 import com.rabbitmq.stream.MessageHandler.Context;
+import com.rabbitmq.stream.NoOffsetException;
 import com.rabbitmq.stream.OffsetSpecification;
 import com.rabbitmq.stream.ProducerBuilder;
 import com.rabbitmq.stream.StreamCreator;
 import com.rabbitmq.stream.StreamException;
+import com.rabbitmq.stream.StreamInfo;
 import com.rabbitmq.stream.SubscriptionListener;
 import com.rabbitmq.stream.compression.CompressionCodecFactory;
 import com.rabbitmq.stream.impl.Client.ClientParameters;
+import com.rabbitmq.stream.impl.Client.StreamInfoResponse;
 import com.rabbitmq.stream.impl.OffsetTrackingCoordinator.Registration;
 import com.rabbitmq.stream.impl.StreamConsumerBuilder.TrackingConfiguration;
 import com.rabbitmq.stream.impl.StreamEnvironmentBuilder.DefaultTlsConfiguration;
@@ -45,6 +49,7 @@
 import java.net.URLDecoder;
 import java.util.Collections;
 import java.util.List;
+import java.util.Map;
 import java.util.Random;
 import java.util.concurrent.CopyOnWriteArrayList;
 import java.util.concurrent.ExecutionException;
@@ -54,9 +59,11 @@
 import java.util.concurrent.TimeoutException;
 import java.util.concurrent.atomic.AtomicBoolean;
 import java.util.concurrent.atomic.AtomicReference;
+import java.util.function.BiFunction;
 import java.util.function.Consumer;
 import java.util.function.Function;
 import java.util.function.LongConsumer;
+import java.util.function.LongSupplier;
 import java.util.function.Supplier;
 import java.util.stream.Collectors;
 import javax.net.ssl.SSLException;
@@ -382,6 +389,59 @@ public void deleteStream(String stream) {
     }
   }
 
+  @Override
+  public StreamInfo queryStreamInfo(String stream) {
+    StreamInfoResponse response = locatorOperation(client -> client.streamInfo(stream));
+    if (response.isOk()) {
+      Map<String, String> info = response.getInfo();
+      BiFunction<String, String, LongSupplier> offsetSupplierLogic =
+          (key, message) -> {
+            if (!info.containsKey(key) || "-1".equals(info.get(key))) {
+              return () -> {
+                throw new NoOffsetException(message);
+              };
+            } else {
+              try {
+                long firstOffset = Long.parseUnsignedLong(info.get(key));
+                return () -> firstOffset;
+              } catch (NumberFormatException e) {
+                return () -> {
+                  throw new NoOffsetException(message);
+                };
+              }
+            }
+          };
+      LongSupplier firstOffsetSupplier =
+          offsetSupplierLogic.apply("first_offset", "No first offset for stream " + stream);
+      LongSupplier committedOffsetSupplier =
+          offsetSupplierLogic.apply("committed_offset", "No committed offset for stream " + stream);
+      return new DefaultStreamInfo(firstOffsetSupplier, committedOffsetSupplier);
+    } else {
+      throw propagateException(response.getResponseCode(), stream);
+    }
+  }
+
+  private static class DefaultStreamInfo implements StreamInfo {
+
+    private final LongSupplier firstOffsetSupplier, committedOffsetSupplier;
+
+    private DefaultStreamInfo(
+        LongSupplier firstOffsetSupplier, LongSupplier committedOffsetSupplier) {
+      this.firstOffsetSupplier = firstOffsetSupplier;
+      this.committedOffsetSupplier = committedOffsetSupplier;
+    }
+
+    @Override
+    public long firstOffset() {
+      return firstOffsetSupplier.getAsLong();
+    }
+
+    @Override
+    public long committedOffset() {
+      return committedOffsetSupplier.getAsLong();
+    }
+  }
+
   @Override
   public ProducerBuilder producerBuilder() {
     return new StreamProducerBuilder(this);