Use StreamStats lingo, not StreamInfo

References #165

Author: acogoluegnes

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

@@ -62,15 +62,15 @@ static EnvironmentBuilder builder() {
   void deleteStream(String stream);
 
   /**
-   * Query information on a stream.
+   * Query statistics on a stream.
    *
    * <p>Requires RabbitMQ 3.11 or more.
    *
    * @param stream
-   * @return information on the stream
+   * @return statistics on the stream
    * @throws UnsupportedOperationException if the broker does not support this command
    */
-  StreamStats queryStreamInfo(String stream);
+  StreamStats queryStreamStats(String stream);
 
   /**
    * Create a {@link ProducerBuilder} to configure and create a {@link Producer}.

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

@@ -16,9 +16,9 @@
 import com.rabbitmq.stream.MessageHandler.Context;
 
 /**
- * Information on a stream.
+ * Statistics on a stream.
  *
- * @see Environment#queryStreamInfo(String)
+ * @see Environment#queryStreamStats(String)
  */
 public interface StreamStats {
 

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

@@ -1360,7 +1360,7 @@ List<FrameHandlerInfo> exchangeCommandVersions() {
     }
   }
 
-  StreamInfoResponse streamStats(String stream) {
+  StreamStatsResponse streamStats(String stream) {
     if (stream == null) {
       throw new IllegalArgumentException("stream must not be null");
     }
@@ -1374,7 +1374,7 @@ StreamInfoResponse streamStats(String stream) {
       bb.writeInt(correlationId);
       bb.writeShort(stream.length());
       bb.writeBytes(stream.getBytes(StandardCharsets.UTF_8));
-      OutstandingRequest<StreamInfoResponse> request = new OutstandingRequest<>(this.rpcTimeout);
+      OutstandingRequest<StreamStatsResponse> request = new OutstandingRequest<>(this.rpcTimeout);
       outstandingRequests.put(correlationId, request);
       channel.writeAndFlush(bb);
       request.block();
@@ -1921,11 +1921,11 @@ public long getSequence() {
     }
   }
 
-  static class StreamInfoResponse extends Response {
+  static class StreamStatsResponse extends Response {
 
     private final Map<String, Long> info;
 
-    StreamInfoResponse(short responseCode, Map<String, Long> info) {
+    StreamStatsResponse(short responseCode, Map<String, Long> info) {
       super(responseCode);
       this.info = Collections.unmodifiableMap(new HashMap<>(info));
     }

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

@@ -65,7 +65,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.StreamStatsResponse;
 import com.rabbitmq.stream.impl.Client.StreamMetadata;
 import com.rabbitmq.stream.impl.Client.SubscriptionOffset;
 import com.rabbitmq.stream.metrics.MetricsCollector;
@@ -1196,12 +1196,12 @@ int doHandle(Client client, ChannelHandlerContext ctx, ByteBuf message) {
         read += 8;
       }
 
-      OutstandingRequest<StreamInfoResponse> outstandingRequest =
-          remove(client.outstandingRequests, correlationId, StreamInfoResponse.class);
+      OutstandingRequest<StreamStatsResponse> outstandingRequest =
+          remove(client.outstandingRequests, correlationId, StreamStatsResponse.class);
       if (outstandingRequest == null) {
         LOGGER.warn("Could not find outstanding request with correlation ID {}", correlationId);
       } else {
-        outstandingRequest.response().set(new StreamInfoResponse(responseCode, info));
+        outstandingRequest.response().set(new StreamStatsResponse(responseCode, info));
         outstandingRequest.countDown();
       }
       return read;

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

@@ -34,7 +34,7 @@
 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.Client.StreamStatsResponse;
 import com.rabbitmq.stream.impl.OffsetTrackingCoordinator.Registration;
 import com.rabbitmq.stream.impl.StreamConsumerBuilder.TrackingConfiguration;
 import com.rabbitmq.stream.impl.StreamEnvironmentBuilder.DefaultTlsConfiguration;
@@ -390,8 +390,8 @@ public void deleteStream(String stream) {
   }
 
   @Override
-  public StreamStats queryStreamInfo(String stream) {
-    StreamInfoResponse response =
+  public StreamStats queryStreamStats(String stream) {
+    StreamStatsResponse response =
         locatorOperation(
             client -> {
               if (Utils.is3_11_OrMore(client.brokerVersion())) {

src/test/java/com/rabbitmq/stream/impl/ClientTest.java

@@ -38,7 +38,7 @@
 import com.rabbitmq.stream.codec.SwiftMqCodec;
 import com.rabbitmq.stream.impl.Client.ClientParameters;
 import com.rabbitmq.stream.impl.Client.Response;
-import com.rabbitmq.stream.impl.Client.StreamInfoResponse;
+import com.rabbitmq.stream.impl.Client.StreamStatsResponse;
 import com.rabbitmq.stream.impl.Client.StreamParametersBuilder;
 import com.rabbitmq.stream.impl.ServerFrameHandler.FrameHandlerInfo;
 import com.rabbitmq.stream.impl.TestUtils.BrokerVersion;
@@ -883,7 +883,7 @@ void deliverVersion2LastCommittedOffsetShouldBeSet() throws Exception {
 
   @Test
   @BrokerVersionAtLeast(BrokerVersion.RABBITMQ_3_11)
-  void streamInfoShouldReturnFirstOffsetAndCommittedOffset() throws Exception {
+  void streamStatsShouldReturnFirstOffsetAndCommittedOffset() throws Exception {
     int publishCount = 20_000;
     CountDownLatch latch = new CountDownLatch(publishCount);
 
@@ -903,7 +903,7 @@ void streamInfoShouldReturnFirstOffsetAndCommittedOffset() throws Exception {
             new Client.ClientParameters()
                 .chunkListener(chunkListener)
                 .messageListener(messageListener));
-    StreamInfoResponse response = client.streamStats(stream);
+    StreamStatsResponse response = client.streamStats(stream);
     assertThat(response.getInfo()).containsEntry("first_chunk_id", -1L);
     assertThat(response.getInfo()).containsEntry("committed_chunk_id", -1L);
     TestUtils.publishAndWaitForConfirms(cf, publishCount, stream);
@@ -924,14 +924,14 @@ void streamInfoShouldReturnFirstOffsetAndCommittedOffset() throws Exception {
 
   @Test
   @BrokerVersionAtLeast(BrokerVersion.RABBITMQ_3_11)
-  void streamInfoShouldReturnErrorWhenStreamDoesNotExist() {
+  void streamStatsShouldReturnErrorWhenStreamDoesNotExist() {
     assertThat(cf.get().streamStats("does not exist").getResponseCode())
         .isEqualTo(Constants.RESPONSE_CODE_STREAM_DOES_NOT_EXIST);
   }
 
   @Test
   @BrokerVersionAtLeast(BrokerVersion.RABBITMQ_3_11)
-  void streamInfoFirstOffsetShouldChangeAfterRetentionKickedIn(TestInfo info) {
+  void streamStatsFirstOffsetShouldChangeAfterRetentionKickedIn(TestInfo info) {
     int messageCount = 1000;
     int payloadSize = 1000;
     String s = TestUtils.streamName(info);
@@ -948,7 +948,7 @@ void streamInfoFirstOffsetShouldChangeAfterRetentionKickedIn(TestInfo info) {
                   .isOk())
           .isTrue();
 
-      StreamInfoResponse response = client.streamStats(s);
+      StreamStatsResponse response = client.streamStats(s);
       assertThat(response.getInfo()).containsEntry("first_chunk_id", -1L);
       assertThat(response.getInfo()).containsEntry("committed_chunk_id", -1L);
 

src/test/java/com/rabbitmq/stream/impl/StreamEnvironmentTest.java

@@ -506,18 +506,18 @@ void createPublishConsumeDelete(boolean lazyInit, TestInfo info) {
 
   @Test
   @BrokerVersionAtLeast(BrokerVersion.RABBITMQ_3_11)
-  void queryStreamInfoShouldReturnFirstOffsetAndCommittedOffset() throws Exception {
+  void queryStreamStatsShouldReturnFirstOffsetAndCommittedOffset() throws Exception {
     try (Environment env = environmentBuilder.build()) {
-      StreamStats info = env.queryStreamInfo(stream);
-      assertThatThrownBy(() -> info.firstOffset()).isInstanceOf(NoOffsetException.class);
-      assertThatThrownBy(() -> info.committedChunkId()).isInstanceOf(NoOffsetException.class);
+      StreamStats stats = env.queryStreamStats(stream);
+      assertThatThrownBy(() -> stats.firstOffset()).isInstanceOf(NoOffsetException.class);
+      assertThatThrownBy(() -> stats.committedChunkId()).isInstanceOf(NoOffsetException.class);
 
       int publishCount = 20_000;
       TestUtils.publishAndWaitForConfirms(cf, publishCount, stream);
 
-      StreamStats info2 = env.queryStreamInfo(stream);
-      assertThat(info2.firstOffset()).isZero();
-      assertThat(info2.committedChunkId()).isPositive();
+      StreamStats stats2 = env.queryStreamStats(stream);
+      assertThat(stats2.firstOffset()).isZero();
+      assertThat(stats2.committedChunkId()).isPositive();
 
       CountDownLatch latch = new CountDownLatch(publishCount);
       AtomicLong committedChunkId = new AtomicLong();
@@ -532,15 +532,15 @@ void queryStreamInfoShouldReturnFirstOffsetAndCommittedOffset() throws Exception
 
       assertThat(latch.await(10, SECONDS)).isTrue();
       assertThat(committedChunkId.get()).isPositive();
-      assertThat(committedChunkId).hasValue(info2.committedChunkId());
+      assertThat(committedChunkId).hasValue(stats2.committedChunkId());
     }
   }
 
   @Test
   @BrokerVersionAtLeast(BrokerVersion.RABBITMQ_3_11)
-  void queryStreamInfoShouldThrowExceptionWhenStreamDoesNotExist() {
+  void queryStreamStatsShouldThrowExceptionWhenStreamDoesNotExist() {
     try (Environment env = environmentBuilder.build()) {
-      assertThatThrownBy(() -> env.queryStreamInfo("does not exist"))
+      assertThatThrownBy(() -> env.queryStreamStats("does not exist"))
           .isInstanceOf(StreamDoesNotExistException.class);
     }
   }