Improve internal insight of instances

Make toString() method of producer, consumer, and internal
components return a JSON document with information of their
state. Should be handy to understand the state of the overall
system (number of connections, etc).

Internal, no public contract.

References #50

Author: acogoluegnes

src/main/java/com/rabbitmq/stream/impl/Client.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").
@@ -96,6 +96,7 @@
 import java.io.IOException;
 import java.io.OutputStream;
 import java.lang.reflect.Field;
+import java.net.InetSocketAddress;
 import java.net.SocketAddress;
 import java.nio.charset.StandardCharsets;
 import java.time.Duration;
@@ -1237,6 +1238,29 @@ int getPort() {
     return port;
   }
 
+  String connectionName() {
+    StringBuilder builder = new StringBuilder();
+    SocketAddress localAddress = localAddress();
+    if (localAddress instanceof InetSocketAddress) {
+      InetSocketAddress address = (InetSocketAddress) localAddress;
+      builder.append(address.getHostString()).append(":").append(address.getPort());
+    } else {
+      builder.append("?");
+    }
+    builder.append(" -> ");
+    return builder.append(serverAddress()).toString();
+  }
+
+  private String serverAddress() {
+    SocketAddress remoteAddress = remoteAddress();
+    if (remoteAddress instanceof InetSocketAddress) {
+      InetSocketAddress address = (InetSocketAddress) remoteAddress;
+      return address.getHostString() + ":" + address.getPort();
+    } else {
+      return this.host + ":" + this.port;
+    }
+  }
+
   public List<String> route(String routingKey, String superStream) {
     if (routingKey == null || superStream == null) {
       throw new IllegalArgumentException("routing key and stream must not be null");

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

@@ -194,25 +194,24 @@ int poolSize() {
   @Override
   public String toString() {
     return ("[ \n"
-            + pools.entrySet().stream()
-                .map(
-                    poolEntry ->
-                        "  { 'broker' : '"
-                            + poolEntry.getKey()
-                            + "', 'clients' : [ "
-                            + poolEntry.getValue().managers.stream()
-                                .map(
-                                    manager ->
-                                        "{ 'consumer_count' : "
-                                            + manager.subscriptionTrackers.stream()
-                                                .filter(Objects::nonNull)
-                                                .count()
-                                            + " }")
-                                .collect(Collectors.joining(", "))
-                            + " ] }")
-                .collect(Collectors.joining(", \n"))
-            + "\n]")
-        .replace("'", "\"");
+        + pools.entrySet().stream()
+            .map(
+                poolEntry ->
+                    "  { \"broker\" : \""
+                        + poolEntry.getKey()
+                        + "\", \"clients\" : [ "
+                        + poolEntry.getValue().managers.stream()
+                            .map(
+                                manager ->
+                                    "{ \"consumer_count\" : "
+                                        + manager.subscriptionTrackers.stream()
+                                            .filter(Objects::nonNull)
+                                            .count()
+                                        + " }")
+                            .collect(Collectors.joining(", "))
+                        + " ] }")
+            .collect(Collectors.joining(", \n"))
+        + "\n]");
   }
 
   /**
@@ -267,10 +266,12 @@ boolean isClosing() {
     synchronized void assign(byte subscriptionIdInClient, ClientSubscriptionsManager manager) {
       this.subscriptionIdInClient = subscriptionIdInClient;
       this.manager = manager;
+      this.consumer.setSubscriptionClient(this.manager.client);
     }
 
     synchronized void detachFromManager() {
       this.manager = null;
+      this.consumer.setSubscriptionClient(null);
     }
   }
 

src/main/java/com/rabbitmq/stream/impl/OffsetTrackingCoordinator.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").
@@ -322,4 +322,9 @@ public void setTime(long time) {
       this.time = time;
     }
   }
+
+  @Override
+  public String toString() {
+    return "{ \"tracker_count\" : " + this.trackers.size() + " }";
+  }
 }

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

@@ -144,26 +144,25 @@ int clientCount() {
   @Override
   public String toString() {
     return ("[ \n"
-            + pools.entrySet().stream()
-                .map(
-                    poolEntry ->
-                        "  { 'broker' : '"
-                            + poolEntry.getKey()
-                            + "', 'clients' : [ "
-                            + poolEntry.getValue().managers.stream()
-                                .map(
-                                    manager ->
-                                        "{ 'producer_count' : "
-                                            + manager.producers.size()
-                                            + ", "
-                                            + "  'tracking_consumer_count' : "
-                                            + manager.trackingConsumerTrackers.size()
-                                            + " }")
-                                .collect(Collectors.joining(", "))
-                            + " ] }")
-                .collect(Collectors.joining(", \n"))
-            + "\n]")
-        .replace("'", "\"");
+        + pools.entrySet().stream()
+            .map(
+                poolEntry ->
+                    "  { \"broker\" : \""
+                        + poolEntry.getKey()
+                        + "\", \"clients\" : [ "
+                        + poolEntry.getValue().managers.stream()
+                            .map(
+                                manager ->
+                                    "{ \"producer_count\" : "
+                                        + manager.producers.size()
+                                        + ", "
+                                        + "  \"tracking_consumer_count\" : "
+                                        + manager.trackingConsumerTrackers.size()
+                                        + " }")
+                            .collect(Collectors.joining(", "))
+                        + " ] }")
+            .collect(Collectors.joining(", \n"))
+        + "\n]");
   }
 
   private interface AgentTracker {
@@ -280,7 +279,7 @@ public void assign(byte producerId, Client client, ClientProducersManager manage
       synchronized (TrackingConsumerTracker.this) {
         this.clientProducersManager = manager;
       }
-      this.consumer.setClient(client);
+      this.consumer.setTrackingClient(client);
     }
 
     @Override

src/main/java/com/rabbitmq/stream/impl/StreamConsumer.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").
@@ -44,6 +44,7 @@ class StreamConsumer implements Consumer {
   private final Runnable initCallback;
   private volatile Runnable closingCallback;
   private volatile Client trackingClient;
+  private volatile Client subscriptionClient;
   private volatile Status status;
   private volatile long lastRequestedStoredOffset = 0;
 
@@ -177,10 +178,14 @@ boolean isOpen() {
     return !this.closed.get();
   }
 
-  synchronized void setClient(Client client) {
+  synchronized void setTrackingClient(Client client) {
     this.trackingClient = client;
   }
 
+  void setSubscriptionClient(Client client) {
+    this.subscriptionClient = client;
+  }
+
   synchronized void unavailable() {
     this.status = Status.NOT_AVAILABLE;
     this.trackingClient = null;
@@ -253,6 +258,22 @@ public int hashCode() {
 
   @Override
   public String toString() {
-    return "StreamConsumer{" + "id=" + id + ", stream='" + stream + '\'' + '}';
+    Client subscriptionClient = this.subscriptionClient;
+    Client trackingClient = this.trackingClient;
+    return "{ "
+        + "\"id\" : "
+        + id
+        + ","
+        + "\"stream\" : \""
+        + stream
+        + "\","
+        + "\"subscription_client\" : "
+        + (subscriptionClient == null
+            ? "null"
+            : ("\"" + subscriptionClient.connectionName() + "\""))
+        + ", "
+        + "\"tracking_client\" : "
+        + (trackingClient == null ? "null" : ("\"" + trackingClient.connectionName() + "\""))
+        + "}";
   }
 }

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

@@ -604,13 +604,15 @@ TrackingConsumerRegistration registerTrackingConsumer(
   @Override
   public String toString() {
     Client locator = this.locator;
-    return "{ locator : "
-        + (locator == null ? "null" : ("'" + locator.getHost() + ":" + locator.getPort() + "'"))
+    return "{ \"locator\" : "
+        + (locator == null ? "null" : ("\"" + locator.connectionName() + "\""))
         + ", "
-        + "'producers' : "
+        + "\"producers\" : "
         + this.producersCoordinator
-        + ", 'consumers' : "
+        + ", \"consumers\" : "
         + this.consumersCoordinator
+        + ", \"offset_tracking\" : "
+        + this.offsetTrackingCoordinator
         + "}";
   }
 

src/main/java/com/rabbitmq/stream/impl/StreamProducer.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").
@@ -36,6 +36,7 @@
 import java.util.List;
 import java.util.Map;
 import java.util.Map.Entry;
+import java.util.Objects;
 import java.util.SortedMap;
 import java.util.TreeMap;
 import java.util.concurrent.ConcurrentHashMap;
@@ -52,8 +53,11 @@
 
 class StreamProducer implements Producer {
 
+  private static final AtomicLong ID_SEQUENCE = new AtomicLong(0);
+
   private static final Logger LOGGER = LoggerFactory.getLogger(StreamProducer.class);
   private static final ConfirmationHandler NO_OP_CONFIRMATION_HANDLER = confirmationStatus -> {};
+  private final long id;
   private final MessageAccumulator accumulator;
   // FIXME investigate a more optimized data structure to handle pending messages
   private final ConcurrentMap<Long, AccumulatedEntity> unconfirmedMessages;
@@ -87,6 +91,7 @@ class StreamProducer implements Producer {
       Duration confirmTimeout,
       Duration enqueueTimeout,
       StreamEnvironment environment) {
+    this.id = ID_SEQUENCE.getAndIncrement();
     this.environment = environment;
     this.name = name;
     this.stream = stream;
@@ -479,4 +484,36 @@ interface ConfirmationCallback {
 
     int handle(boolean confirmed, short code);
   }
+
+  @Override
+  public boolean equals(Object o) {
+    if (this == o) {
+      return true;
+    }
+    if (o == null || getClass() != o.getClass()) {
+      return false;
+    }
+    StreamProducer that = (StreamProducer) o;
+    return id == that.id && stream.equals(that.stream);
+  }
+
+  @Override
+  public int hashCode() {
+    return Objects.hash(id, stream);
+  }
+
+  @Override
+  public String toString() {
+    Client client = this.client;
+    return "{ "
+        + "\"id\" : "
+        + id
+        + ","
+        + "\"stream\" : \""
+        + stream
+        + "\","
+        + "\"publishing_client\" : "
+        + (client == null ? "null" : ("\"" + client.connectionName() + "\""))
+        + "}";
+  }
 }

src/main/java/com/rabbitmq/stream/perf/StreamPerfTest.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").
@@ -760,7 +760,6 @@ public Integer call() throws Exception {
                                 });
 
                         Consumer consumer = consumerBuilder.build();
-
                         return consumer;
                       })
                   .collect(Collectors.toList()));