Adding per-host statistics

Author: Diagoras

client/src/main/java/org/asynchttpclient/ClientStats.java

@@ -13,63 +13,80 @@
  */
 package org.asynchttpclient;
 
+import java.util.Collections;
+import java.util.Map;
 import java.util.Objects;
 
 /**
- * A record class representing the state of an (@link org.asynchttpclient.AsyncHttpClient)
+ * A record class representing the state of an (@link org.asynchttpclient.AsyncHttpClient).
  */
 public class ClientStats {
 
-    private final long activeConnectionCount;
-    private final long idleConnectionCount;
+    private final Map<String, HostStats> statsPerHost;
 
-    public ClientStats(long activeConnectionCount,
-                       long idleConnectionCount) {
-        this.activeConnectionCount = activeConnectionCount;
-        this.idleConnectionCount = idleConnectionCount;
+    public ClientStats(Map<String, HostStats> statsPerHost) {
+        this.statsPerHost = Collections.unmodifiableMap(statsPerHost);
     }
 
     /**
-     * @return The sum of {@link #getActiveConnectionCount()} and {@link #getIdleConnectionCount()},
+     * @return A map from hostname to statistics on that host's connections.
+     * The returned map is an {@link java.util.Collections.UnmodifiableMap}.
+     */
+    public Map<String, HostStats> getStatsPerHost() {
+        return statsPerHost;
+    }
+
+    /**
+     * @return The sum of {@link #getTotalActiveConnectionCount()} and {@link #getTotalIdleConnectionCount()},
      * a long representing the total number of connections in the connection pool.
      */
     public long getTotalConnectionCount() {
-        return activeConnectionCount + idleConnectionCount;
+        return statsPerHost
+                .values()
+                .stream()
+                .mapToLong(HostStats::getHostConnectionCount)
+                .sum();
     }
 
     /**
-     * @return A long representing the number of active connection in the connection pool.
+     * @return A long representing the number of active connections in the connection pool.
      */
-    public long getActiveConnectionCount() {
-        return activeConnectionCount;
+    public long getTotalActiveConnectionCount() {
+        return statsPerHost
+                .values()
+                .stream()
+                .mapToLong(HostStats::getHostActiveConnectionCount)
+                .sum();
     }
 
     /**
-     *
      * @return A long representing the number of idle connections in the connection pool.
      */
-    public long getIdleConnectionCount() {
-        return idleConnectionCount;
+    public long getTotalIdleConnectionCount() {
+        return statsPerHost
+                .values()
+                .stream()
+                .mapToLong(HostStats::getHostIdleConnectionCount)
+                .sum();
     }
 
     @Override
     public String toString() {
         return "There are " + getTotalConnectionCount() +
-                " total connections, " + getActiveConnectionCount() +
-                " are active and " + getIdleConnectionCount() + " are idle.";
+                " total connections, " + getTotalActiveConnectionCount() +
+                " are active and " + getTotalIdleConnectionCount() + " are idle.";
     }
 
     @Override
     public boolean equals(final Object o) {
         if (this == o) return true;
         if (o == null || getClass() != o.getClass()) return false;
         final ClientStats that = (ClientStats) o;
-        return activeConnectionCount == that.activeConnectionCount &&
-                idleConnectionCount == that.idleConnectionCount;
+        return Objects.equals(statsPerHost, that.statsPerHost);
     }
 
     @Override
     public int hashCode() {
-        return Objects.hash(activeConnectionCount, idleConnectionCount);
+        return Objects.hashCode(statsPerHost);
     }
 }

client/src/main/java/org/asynchttpclient/HostStats.java

@@ -0,0 +1,74 @@
+/*
+ * Copyright (c) 2014 AsyncHttpClient Project. All rights reserved.
+ *
+ * This program is licensed to you under the Apache License Version 2.0,
+ * and you may not use this file except in compliance with the Apache License Version 2.0.
+ * You may obtain a copy of the Apache License Version 2.0 at
+ *     http://www.apache.org/licenses/LICENSE-2.0.
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the Apache License Version 2.0 is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the Apache License Version 2.0 for the specific language governing permissions and limitations there under.
+ */
+package org.asynchttpclient;
+
+import java.util.Objects;
+
+/**
+ * A record class representing the status of connections to some host.
+ */
+public class HostStats {
+
+    private final long activeConnectionCount;
+    private final long idleConnectionCount;
+
+    public HostStats(long activeConnectionCount,
+                     long idleConnectionCount) {
+        this.activeConnectionCount = activeConnectionCount;
+        this.idleConnectionCount = idleConnectionCount;
+    }
+
+    /**
+     * @return The sum of {@link #getHostActiveConnectionCount()} and {@link #getHostIdleConnectionCount()},
+     * a long representing the total number of connections to this host.
+     */
+    public long getHostConnectionCount() {
+        return activeConnectionCount + idleConnectionCount;
+    }
+
+    /**
+     * @return A long representing the number of active connections to the host.
+     */
+    public long getHostActiveConnectionCount() {
+        return activeConnectionCount;
+    }
+
+    /**
+     * @return A long representing the number of idle connections in the connection pool.
+     */
+    public long getHostIdleConnectionCount() {
+        return idleConnectionCount;
+    }
+
+    @Override
+    public String toString() {
+        return "There are " + getHostConnectionCount() +
+                " total connections, " + getHostActiveConnectionCount() +
+                " are active and " + getHostIdleConnectionCount() + " are idle.";
+    }
+
+    @Override
+    public boolean equals(final Object o) {
+        if (this == o) return true;
+        if (o == null || getClass() != o.getClass()) return false;
+        final HostStats hostStats = (HostStats) o;
+        return activeConnectionCount == hostStats.activeConnectionCount &&
+                idleConnectionCount == hostStats.idleConnectionCount;
+    }
+
+    @Override
+    public int hashCode() {
+        return Objects.hash(activeConnectionCount, idleConnectionCount);
+    }
+}

client/src/main/java/org/asynchttpclient/channel/ChannelPool.java

@@ -13,6 +13,8 @@
  */
 package org.asynchttpclient.channel;
 
+import java.util.Map;
+
 import io.netty.channel.Channel;
 
 public interface ChannelPool {
@@ -72,7 +74,7 @@ public interface ChannelPool {
     void flushPartitions(ChannelPoolPartitionSelector selector);
 
     /**
-     * @return The number of idle channels.
+     * @return The number of idle channels per host.
      */
-    long getIdleChannelCount();
+    Map<String, Long> getIdleChannelCountPerHost();
 }

client/src/main/java/org/asynchttpclient/channel/NoopChannelPool.java

@@ -13,6 +13,9 @@
  */
 package org.asynchttpclient.channel;
 
+import java.util.Collections;
+import java.util.Map;
+
 import io.netty.channel.Channel;
 
 public enum NoopChannelPool implements ChannelPool {
@@ -52,7 +55,7 @@ public void flushPartitions(ChannelPoolPartitionSelector selector) {
     }
 
     @Override
-    public long getIdleChannelCount() {
-        return 0;
+    public Map<String, Long> getIdleChannelCountPerHost() {
+        return Collections.emptyMap();
     }
 }

client/src/main/java/org/asynchttpclient/netty/channel/ChannelManager.java

@@ -40,19 +40,23 @@
 import io.netty.util.concurrent.GlobalEventExecutor;
 
 import java.io.IOException;
+import java.net.InetSocketAddress;
+import java.util.Map;
 import java.util.Map.Entry;
+import java.util.Optional;
+import java.util.OptionalLong;
 import java.util.concurrent.ConcurrentHashMap;
 import java.util.concurrent.Semaphore;
 import java.util.concurrent.ThreadFactory;
 import java.util.concurrent.TimeUnit;
+import java.util.function.Function;
+import java.util.stream.Collectors;
+import java.util.stream.Stream;
 
 import javax.net.ssl.SSLEngine;
 import javax.net.ssl.SSLException;
 
-import org.asynchttpclient.AsyncHandler;
-import org.asynchttpclient.AsyncHttpClientConfig;
-import org.asynchttpclient.ClientStats;
-import org.asynchttpclient.SslEngineFactory;
+import org.asynchttpclient.*;
 import org.asynchttpclient.channel.ChannelPool;
 import org.asynchttpclient.channel.ChannelPoolPartitioning;
 import org.asynchttpclient.channel.NoopChannelPool;
@@ -491,9 +495,26 @@ public EventLoopGroup getEventLoopGroup() {
     }
 
     public ClientStats getClientStats() {
-        final long totalConnectionCount = openChannels.size();
-        final long idleConnectionCount = channelPool.getIdleChannelCount();
-        final long activeConnectionCount = totalConnectionCount - idleConnectionCount;
-        return new ClientStats(activeConnectionCount, idleConnectionCount);
+        final Map<String, Long> totalConnectionsPerHost = openChannels
+                    .stream()
+                    .map(Channel::remoteAddress)
+                    .filter(a -> a.getClass() == InetSocketAddress.class)
+                    .map(a -> (InetSocketAddress) a)
+                    .map(InetSocketAddress::getHostName)
+                    .collect(Collectors.groupingBy(Function.identity(), Collectors.counting()));
+        final Map<String, Long> idleConnectionsPerHost = channelPool.getIdleChannelCountPerHost();
+        final Map<String, HostStats> statsPerHost = totalConnectionsPerHost
+                .entrySet()
+                .stream()
+                .collect(Collectors.toMap(
+                        Entry::getKey,
+                        entry -> {
+                            final long totalConnectionCount = entry.getValue();
+                            final long idleConnectionCount = idleConnectionsPerHost.getOrDefault(entry.getKey(), 0L);
+                            final long activeConnectionCount = totalConnectionCount - idleConnectionCount;
+                            return new HostStats(activeConnectionCount, idleConnectionCount);
+                        }
+                ));
+        return new ClientStats(statsPerHost);
     }
 }

client/src/main/java/org/asynchttpclient/netty/channel/DefaultChannelPool.java

@@ -21,11 +21,14 @@
 import io.netty.util.Timer;
 import io.netty.util.TimerTask;
 
+import java.net.InetSocketAddress;
 import java.util.*;
 import java.util.concurrent.ConcurrentHashMap;
 import java.util.concurrent.ConcurrentLinkedDeque;
 import java.util.concurrent.TimeUnit;
 import java.util.concurrent.atomic.AtomicBoolean;
+import java.util.function.Function;
+import java.util.stream.Collectors;
 
 import org.asynchttpclient.AsyncHttpClientConfig;
 import org.asynchttpclient.channel.ChannelPool;
@@ -120,6 +123,10 @@ public boolean takeOwnership() {
             return owned.compareAndSet(false, true);
         }
 
+        public Channel getChannel() {
+            return channel;
+        }
+
         @Override
         // only depends on channel
         public boolean equals(Object o) {
@@ -357,8 +364,16 @@ public void flushPartitions(ChannelPoolPartitionSelector selector) {
     }
 
     @Override
-    public long getIdleChannelCount() {
-        return partitions.values().stream().mapToLong(ConcurrentLinkedDeque::size).sum();
+    public Map<String, Long> getIdleChannelCountPerHost() {
+        return partitions
+                .values()
+                .stream()
+                .flatMap(ConcurrentLinkedDeque::stream)
+                .map(idle -> idle.getChannel().remoteAddress())
+                .filter(a -> a.getClass() == InetSocketAddress.class)
+                .map(a -> (InetSocketAddress) a)
+                .map(InetSocketAddress::getHostName)
+                .collect(Collectors.groupingBy(Function.identity(), Collectors.counting()));
     }
 
     public enum PoolLeaseStrategy {