Issue #9529 - Expose TCP connection establishment information.

Cleaned up contribution by @arsenalzp.
Added more test cases for blocking and non-blocking.
Added documentation.

Signed-off-by: Simone Bordet <[email protected]>

Author: sbordet

documentation/jetty/modules/code/examples/src/main/java/org/eclipse/jetty/docs/programming/client/http/HTTPClientDocs.java

@@ -20,10 +20,13 @@
 import java.net.SocketAddress;
 import java.net.URI;
 import java.nio.ByteBuffer;
+import java.nio.channels.SocketChannel;
 import java.nio.file.Path;
 import java.nio.file.Paths;
 import java.util.List;
 import java.util.concurrent.CompletableFuture;
+import java.util.concurrent.ConcurrentHashMap;
+import java.util.concurrent.ConcurrentMap;
 import java.util.concurrent.TimeUnit;
 import javax.net.ssl.SSLEngine;
 import javax.net.ssl.SSLException;
@@ -1233,4 +1236,38 @@ public void connectionInformation() throws Exception
             .send();
         // end::connectionInformation[]
     }
+
+    public void connectListener() throws Exception
+    {
+        // tag::connectListener[]
+        ClientConnector clientConnector = new ClientConnector();
+        clientConnector.addEventListener(new ClientConnector.ConnectListener()
+        {
+            private final ConcurrentMap<SocketChannel, Long> times = new ConcurrentHashMap<>();
+
+            @Override
+            public void onConnectBegin(SocketChannel socketChannel, SocketAddress socketAddress)
+            {
+                times.put(socketChannel, System.nanoTime());
+            }
+
+            @Override
+            public void onConnectSuccess(SocketChannel socketChannel)
+            {
+                Long begin = times.remove(socketChannel);
+                System.getLogger("connection").log(INFO, "established in %d ns", System.nanoTime() - begin);
+            }
+
+            @Override
+            public void onConnectFailure(SocketChannel socketChannel, SocketAddress socketAddress, Throwable failure)
+            {
+                Long begin = times.remove(socketChannel);
+                System.getLogger("connection").log(INFO, "failed in %d ns", System.nanoTime() - begin);
+            }
+        });
+
+        HttpClient httpClient = new HttpClient(new HttpClientTransportOverHTTP(clientConnector));
+        httpClient.start();
+        // end::connectListener[]
+    }
 }

documentation/jetty/modules/programming-guide/pages/client/http.adoc

@@ -541,7 +541,7 @@ include::code:example$src/main/java/org/eclipse/jetty/docs/programming/client/ht
 ----
 
 [[connection-information]]
-=== Request Connection Information
+=== Connection Information
 
 In order to send a request, it is necessary to obtain a connection, as explained in the xref:request-processing[request processing section].
 
@@ -562,6 +562,25 @@ This means that the connection is not available in the _request queued_ event, b
 For more information about request events, see xref:non-blocking[this section].
 ====
 
+[[connection-events]]
+=== Connection Events
+
+In order to send HTTP requests, a connection is necessary, as explained in the xref:request-processing[request processing section].
+
+HTTP/1.1 and HTTP/2 use `SocketChannel.connect(SocketAddress)` to establish a connection with the server, either via the TCP transport or via the Unix-Domain transport.
+
+You can listen to these `connect` events using a `ClientConnector.ConnectListener`, for example to record connection establishment times:
+
+[,java,indent=0]
+----
+include::code:example$src/main/java/org/eclipse/jetty/docs/programming/client/http/HTTPClientDocs.java[tags=connectListener]
+----
+
+This could be particularly useful when you notice that your client application seem "slow" to send requests.
+The `connect begin` and `connect success` events, along with the `request queued` and `request begin` event (detailed xref:non-blocking[here]), allow you to understand whether it is the server being slow at accepting connections, or it is the client being slow at processing queued requests.
+
+Once the low-level connection has been established, you can be notified of connection events using a `Connection.Listener`, or more concretely `ConnectionStatistics`, as described in xref:arch/io.adoc#connection-listener[this section].
+
 [[configuration]]
 == HttpClient Configuration
 

jetty-core/jetty-io/src/main/java/org/eclipse/jetty/io/ClientConnector.java

@@ -457,12 +457,11 @@ else if (networkChannel instanceof DatagramChannel)
             boolean connected = true;
             if (channel instanceof SocketChannel socketChannel)
             {
-                final SocketAddress socketAddress = address;
                 boolean blocking = isConnectBlocking() && address instanceof InetSocketAddress;
                 if (LOG.isDebugEnabled())
                     LOG.debug("Connecting {} to {}", blocking ? "blocking" : "non-blocking", address);
 
-                notifyConnectBegin(socketChannel, socketAddress);
+                notifyConnectBegin(socketChannel, address);
                 if (blocking)
                 {
                     socketChannel.socket().connect(address, (int)getConnectTimeout().toMillis());
@@ -473,6 +472,8 @@ else if (networkChannel instanceof DatagramChannel)
                 {
                     socketChannel.configureBlocking(false);
                     connected = socketChannel.connect(address);
+                    if (connected)
+                        notifyConnectSuccess(socketChannel);
                 }
             }
             else
@@ -494,7 +495,7 @@ else if (networkChannel instanceof DatagramChannel)
             if (x.getClass() == SocketException.class)
                 x = new SocketException("Could not connect to " + address).initCause(x);
             IO.close(channel);
-            connectFailed(channel, x, context);
+            connectFailed(channel, address, x, context);
         }
     }
 
@@ -580,16 +581,77 @@ protected void acceptFailed(Throwable failure, SelectableChannel channel, Map<St
             promise.failed(failure);
     }
 
-    protected void connectFailed(SelectableChannel channel, Throwable failure, Map<String, Object> context)
+    protected void connectFailed(SelectableChannel channel, SocketAddress address, Throwable failure, Map<String, Object> context)
     {
         if (LOG.isDebugEnabled())
-            LOG.debug("Could not connect to {}", context.get(REMOTE_SOCKET_ADDRESS_CONTEXT_KEY));
-        notifyConnectFailure((SocketChannel)channel, failure);
+            LOG.debug("Could not connect to {}", address);
+        notifyConnectFailure((SocketChannel)channel, address, failure);
         Promise<?> promise = (Promise<?>)context.get(CONNECTION_PROMISE_CONTEXT_KEY);
         if (promise != null)
             promise.failed(failure);
     }
 
+    @Override
+    public boolean addEventListener(EventListener listener)
+    {
+        if (listener instanceof ConnectListener connectListener)
+            return listeners.add(connectListener);
+        return super.addEventListener(listener);
+    }
+
+    @Override
+    public boolean removeEventListener(EventListener listener)
+    {
+        if (listener instanceof ConnectListener connectListener)
+            return listeners.remove(connectListener);
+        return super.removeEventListener(listener);
+    }
+
+    private void notifyConnectBegin(SocketChannel socketChannel, SocketAddress socketAddress)
+    {
+        for (ConnectListener listener : listeners)
+        {
+            try
+            {
+                listener.onConnectBegin(socketChannel, socketAddress);
+            }
+            catch (Throwable x)
+            {
+                LOG.info("failure notifying listener {}", listener, x);
+            }
+        }
+    }
+
+    private void notifyConnectSuccess(SocketChannel socketChannel)
+    {
+        for (ConnectListener listener : listeners)
+        {
+            try
+            {
+                listener.onConnectSuccess(socketChannel);
+            }
+            catch (Throwable x)
+            {
+                LOG.info("failure notifying listener {}", listener, x);
+            }
+        }
+    }
+
+    private void notifyConnectFailure(SocketChannel socketChannel, SocketAddress socketAddress, Throwable throwable)
+    {
+        for (ConnectListener listener : listeners)
+        {
+            try
+            {
+                listener.onConnectFailure(socketChannel, socketAddress, throwable);
+            }
+            catch (Throwable x)
+            {
+                LOG.info("failure notifying listener {}", listener, x);
+            }
+        }
+    }
+
     protected class ClientSelectorManager extends SelectorManager
     {
         public ClientSelectorManager(Executor executor, Scheduler scheduler, int selectors)
@@ -634,17 +696,20 @@ public void connectionOpened(Connection connection, Object context)
         }
 
         @Override
-        protected void connectionFailed(SelectableChannel channel, Throwable failure, Object attachment)
+        public void connectionSucceeded(SelectableChannel channel)
         {
-            @SuppressWarnings("unchecked")
-            Map<String, Object> context = (Map<String, Object>)attachment;
-            connectFailed(channel, failure, context);
+            super.connectionSucceeded(channel);
+            notifyConnectSuccess((SocketChannel)channel);
         }
 
         @Override
-        public void connectSuccess(SelectableChannel channel)
+        protected void connectionFailed(SelectableChannel channel, Throwable failure, Object attachment)
         {
-            listeners.forEach(listener -> listener.onConnectSuccess((SocketChannel)channel));
+            super.connectionFailed(channel, failure, attachment);
+            @SuppressWarnings("unchecked")
+            Map<String, Object> context = (Map<String, Object>)attachment;
+            SocketAddress address = (SocketAddress)context.get(REMOTE_SOCKET_ADDRESS_CONTEXT_KEY);
+            connectFailed(channel, address, failure, context);
         }
     }
 
@@ -770,119 +835,44 @@ public ChannelWithAddress newChannelWithAddress(ClientConnector clientConnector,
     }
 
     /**
-     * <p>A listener emits events for TCP connection establishment and failure</p>
-     * {@link #ClientConnector.ConnectListener} is notified of the events:
+     * <p>A listener for events about {@link SocketChannel#connect(SocketAddress)}.</p>
+     * <p>The events are:</p>
      * <ul>
-     * <li>Beginning of connect attempt</li>
-     * <li>Success of connect attempt</li>
-     * <li>Failed connect attempt</li>
+     * <li>{@link ConnectListener#onConnectBegin(SocketChannel, SocketAddress) begin}, just before the {@code connect()} call</li>
+     * <li>{@link ConnectListener#onConnectSuccess(SocketChannel) success}, when the {@code connect()} call succeeds</li>
+     * <li>{@link ConnectListener#onConnectFailure(SocketChannel, SocketAddress, Throwable) failure}, when the {@code connect()} call fails</li>
      * </ul>
-     * As an example {@link #ClientConnector.ConnectListener} can be used to trace attempts of connect for HttpClient
      */
     public interface ConnectListener extends EventListener
     {
         /**
-         * <p>Callback method is invoked for the very beginning of connect attempt</p>
-         * <p>The {@code socketChannel} parameter can be used to extract socket channel information</p>
-         * <p>The {@code socketAddress} parameter can be used to extract socket address information</p>
-         * 
-         * @param socketChannel the socket channel which is registered with connect attempt
-         * @param socketAddress the socket address which is registered with connect attempt
-         */
-        public void onConnectBegin(SocketChannel socketChannel, SocketAddress socketAddress);
-        
-        /**
-         * <p>Callback method is invoked for successfully connect attempt</p>
-         * <p>The {@code socketChannel} parameter can be used to extract socket channel information</p>
-         * 
-         * @param socketChannel the socket channel which is registered with connect attempt
+         * <p>Callback method invoked just before a {@link SocketChannel#connect(SocketAddress)} call.</p>
+         *
+         * @param socketChannel the local socket channel that is about to connect
+         * @param socketAddress the remote socket address to connect to
          */
-        public void onConnectSuccess(SocketChannel socketChannel);
-        
+        default void onConnectBegin(SocketChannel socketChannel, SocketAddress socketAddress)
+        {
+        }
+
         /**
-         * <p>Callback method is invoked for failed connect attempt</p>
-         * <p>The {@code socketChannel} parameter can be used to extract socket channel information</p>
-         * <p>The {@code throwable} parameter can be used to extract exception information of connect attempt</p>
-         * 
-         * @param socketChannel the socket channel which is registered with connect attempt
+         * <p>Callback method invoked when a {@link SocketChannel#connect(SocketAddress)} call completes successfully.</p>
+         *
+         * @param socketChannel the local socket channel that succeeded to connect to the remote socket address
          */
-        public void onConnectFailure(SocketChannel socketChannel, Throwable throwable);
-    }
-
-    @Override
-    public boolean addEventListener(EventListener listener)
-    {
-        if (listener instanceof ConnectListener connectListener)
-            return listeners.add(connectListener);
-        return super.addEventListener(listener);
-    }
-
-    @Override
-    public boolean removeEventListener(EventListener listener)
-    {
-        if (listener instanceof ConnectListener connectListener)
-            return listeners.remove(connectListener);
-        return super.removeEventListener(listener);
-    }
-    
-    /**
-     * <p>A wrapper for ConnectListener.notifyConnectBegin() call</p>
-     * @param socketChannel {@link #SocketChannel} registered for connect attempt
-     * @param socketAddress {@link #SocketAddress} registered for connect attempt
-     */
-    private void notifyConnectBegin(SocketChannel socketChannel, SocketAddress socketAddress)
-    {
-
-        for (ConnectListener listener : listeners)
+        default void onConnectSuccess(SocketChannel socketChannel)
         {
-            try
-            {
-                listener.onConnectBegin(socketChannel, socketAddress);
-            }
-            catch (Throwable x)
-            {
-                x.printStackTrace();
-            }
         }
-    }
-    
-    /**
-     * <p>Wrapper for ConnectListener.onConnectSuccess() call</p>
-     * @param socketChannel {@link #SocketChannel} for successfully connect attempt
-     */
-    private void notifyConnectSuccess(SocketChannel socketChannel)
-    {
 
-        for (ConnectListener listener : listeners)
-        {   
-            try
-            {
-                listener.onConnectSuccess(socketChannel);
-            }
-            catch (Throwable x)
-            {
-                x.printStackTrace();
-            }
-        }
-    }
-    
-    /**
-     * <p>Wrapper for ConnectListener.onConnectFailure() call</p>
-     * @param socketChannel {@link #SocketChannel} registered for failed connect attempt
-     * @param throwable {@link #Throwable} exception arisen with failed attempt
-     */
-    private void notifyConnectFailure(SocketChannel socketChannel, Throwable throwable)
-    {
-        for (ConnectListener listener : listeners)
+        /**
+         * <p>Callback method invoked when a {@link SocketChannel#connect(SocketAddress)} call completes with a failure.</p>
+         *
+         * @param socketChannel the local socket channel that failed to connect to the remote socket address
+         * @param socketAddress the remote socket address to connect to
+         * @param failure the failure cause
+         */
+        default void onConnectFailure(SocketChannel socketChannel, SocketAddress socketAddress, Throwable failure)
         {
-            try
-            {
-                listener.onConnectFailure(socketChannel, throwable);
-            }
-            catch (Throwable x)
-            {
-                x.printStackTrace();
-            }
         }
     }
 }

jetty-core/jetty-io/src/main/java/org/eclipse/jetty/io/ManagedSelector.java

@@ -350,7 +350,7 @@ private void processConnect(SelectionKey key, Connect connect)
             {
                 if (connect.timeout.cancel())
                 {
-                    _selectorManager.connectSuccess(channel);
+                    _selectorManager.connectionSucceeded(channel);
                     key.interestOps(0);
                     execute(new CreateEndPoint(connect, key));
                 }