Add ClientFactory.numConnections() (#3613)

Motivation:

It's be nice if `ClientFactory` has `numConnections()` like `Server`,
so that a user can easily figure out how many connections it is managing.

Modifications:

- Add `Clientfactory.numConnections()`, get connection count from allChannels directly.
- Add test for multiple h1c request & multie servers

Result:

- Closes #3596

core/src/main/java/com/linecorp/armeria/client/ClientFactory.java

@@ -196,6 +196,11 @@ ReleasableHolder<EventLoop> acquireEventLoop(SessionProtocol sessionProtocol,
      */
     Object newClient(ClientBuilderParams params);
 
+    /**
+     * Returns the number of open connections managed by this {@link ClientFactory}.
+     */
+    int numConnections();
+
     /**
      * Returns the {@link ClientBuilderParams} held in {@code client}. This is used when creating a new derived
      * {@link Client} which inherits {@link ClientBuilderParams} from {@code client}. If this

core/src/main/java/com/linecorp/armeria/client/DecoratingClientFactory.java

@@ -142,4 +142,9 @@ public CompletableFuture<?> closeAsync() {
     public void close() {
         unwrap().close();
     }
+
+    @Override
+    public int numConnections() {
+        return unwrap().numConnections();
+    }
 }

core/src/main/java/com/linecorp/armeria/client/DefaultClientFactory.java

@@ -148,6 +148,11 @@ public void setMeterRegistry(MeterRegistry meterRegistry) {
         httpClientFactory.setMeterRegistry(meterRegistry);
     }
 
+    @Override
+    public int numConnections() {
+        return httpClientFactory.numConnections();
+    }
+
     @Override
     public ClientFactoryOptions options() {
         return httpClientFactory.options();

core/src/main/java/com/linecorp/armeria/client/HttpChannelPool.java

@@ -418,6 +418,13 @@ void connect(SocketAddress remoteAddress, SessionProtocol desiredProtocol,
         });
     }
 
+    /**
+     * Returns the number of open connections on this {@link HttpChannelPool}.
+     */
+    int numConnections() {
+        return allChannels.size();
+    }
+
     void invokeProxyConnectFailed(SessionProtocol protocol, PoolKey poolKey, Throwable cause) {
         try {
             final ProxyConfig proxyConfig = poolKey.proxyConfig;

core/src/main/java/com/linecorp/armeria/client/HttpClientFactory.java

@@ -368,6 +368,11 @@ public void close() {
         }
     }
 
+    @Override
+    public int numConnections() {
+        return pools.values().stream().mapToInt(HttpChannelPool::numConnections).sum();
+    }
+
     HttpChannelPool pool(EventLoop eventLoop) {
         final HttpChannelPool pool = pools.get(eventLoop);
         if (pool != null) {

core/src/test/java/com/linecorp/armeria/client/HttpClientFactoryTest.java

@@ -0,0 +1,106 @@
+/*
+ * Copyright 2021 LINE Corporation
+ *
+ * LINE Corporation licenses this file to you under the Apache License,
+ * version 2.0 (the "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at:
+ *
+ *   https://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS, WITHOUT
+ * WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the
+ * License for the specific language governing permissions and limitations
+ * under the Licenses
+ */
+
+package com.linecorp.armeria.client;
+
+import static org.assertj.core.api.Assertions.assertThat;
+import static org.awaitility.Awaitility.await;
+
+import org.junit.jupiter.api.Test;
+import org.junit.jupiter.api.extension.RegisterExtension;
+
+import com.linecorp.armeria.common.HttpRequest;
+import com.linecorp.armeria.common.HttpResponse;
+import com.linecorp.armeria.common.SessionProtocol;
+import com.linecorp.armeria.server.AbstractHttpService;
+import com.linecorp.armeria.server.Server;
+import com.linecorp.armeria.server.ServerBuilder;
+import com.linecorp.armeria.server.ServiceRequestContext;
+import com.linecorp.armeria.testing.junit5.server.ServerExtension;
+
+class HttpClientFactoryTest {
+    @RegisterExtension
+    public static final ServerExtension server = new ServerExtension() {
+        @Override
+        protected void configure(ServerBuilder sb) throws Exception {
+            sb.service("/", new AbstractHttpService() {
+                @Override
+                protected HttpResponse doGet(ServiceRequestContext ctx, HttpRequest req) {
+                    return HttpResponse.streaming();
+                }
+            });
+        }
+    };
+
+    @Test
+    void numConnections() {
+        final ClientFactory clientFactory = ClientFactory.builder().build();
+        assertThat(clientFactory.numConnections()).isZero();
+
+        final WebClient client = WebClient.builder(server.httpUri()).factory(clientFactory).build();
+        final HttpResponse response = client.get("/");
+
+        await().untilAsserted(() -> {
+            assertThat(response.isOpen()).isTrue();
+            assertThat(clientFactory.numConnections()).isOne();
+        });
+
+        clientFactory.close();
+        await().untilAsserted(() -> assertThat(clientFactory.numConnections()).isZero());
+    }
+
+    @Test
+    void numConnections_multipleH1Client() {
+        try (ClientFactory clientFactory = ClientFactory.builder().build()) {
+            for (int i = 0; i < 15; i++) {
+                final WebClient client = WebClient.builder(server.httpEndpoint()
+                                                                 .toUri(SessionProtocol.H1C))
+                                                  .factory(clientFactory).build();
+                final HttpResponse response = client.get("/");
+                await().untilAsserted(() -> assertThat(response.isOpen()).isTrue());
+            }
+            await().untilAsserted(() -> assertThat(clientFactory.numConnections()).isEqualTo(15));
+        }
+    }
+
+    @Test
+    void numConnections_multipleServers() {
+        try (Server server2 = Server.builder()
+                                    .service("/", new AbstractHttpService() {
+                                        @Override
+                                        protected HttpResponse doGet(ServiceRequestContext ctx,
+                                                                     HttpRequest req) {
+                                            return HttpResponse.streaming();
+                                        }
+                                    }).build();
+             ClientFactory clientFactory = ClientFactory.builder().build()) {
+            server2.start().join();
+
+            final WebClient client1 = WebClient.builder(server.httpUri())
+                                               .factory(clientFactory).build();
+            final HttpResponse response1 = client1.get("/");
+            final WebClient client2 = WebClient.builder("http://127.0.0.1:" + server2.activeLocalPort())
+                                               .factory(clientFactory).build();
+            final HttpResponse response2 = client2.get("/");
+
+            await().untilAsserted(() -> {
+                assertThat(response1.isOpen()).isTrue();
+                assertThat(response2.isOpen()).isTrue();
+                assertThat(clientFactory.numConnections()).isEqualTo(2);
+            });
+        }
+    }
+}