Merge pull request #5361 from tsegismont/jdk-domain-socket

Unix Domain Socket support with JDK16+

vertx-core/src/main/asciidoc/http.adoc

@@ -120,6 +120,15 @@ For example:
 {@link examples.HTTPExamples#example5}
 ----
 
+=== Listening to Unix domain sockets
+
+When running on JDK 16+, or using a <<_native_transports,native transport>>, a server can listen to Unix domain sockets:
+
+[source,$lang]
+----
+{@link examples.CoreExamples#httpServerWithDomainSockets}
+----
+
 === Getting notified of incoming requests
 
 To be notified when a request arrives you need to set a {@link io.vertx.core.http.HttpServer#requestHandler}:
@@ -872,21 +881,31 @@ the same preface from the server.
 The http server may not support HTTP/2, the actual version can be checked
 with {@link io.vertx.core.http.HttpClientResponse#version()} when the response arrives.
 
-When a clients connects to an HTTP/2 server, it sends to the server its {@link io.vertx.core.http.HttpClientOptions#getInitialSettings initial settings}.
+When a client connects to an HTTP/2 server, it sends to the server its {@link io.vertx.core.http.HttpClientOptions#getInitialSettings initial settings}.
 The settings define how the server can use the connection, the default initial settings for a client are the default
 values defined by the HTTP/2 RFC.
 
+=== Making connections to Unix domain sockets
+
+When running on JDK 16+, or using a <<_native_transports,native transport>>, a client can connect to Unix domain sockets:
+
+[source,$lang]
+----
+{@link examples.CoreExamples#httpClientWithDomainSockets}
+----
+
 === Pool configuration
 
- For performance purpose, the client uses connection pooling when interacting with HTTP/1.1 servers. The pool creates up
- to 5 connections per server. You can override the pool configuration like this:
+For performance purpose, the client uses connection pooling when interacting with HTTP/1.1 servers.
+The pool creates up to 5 connections per server.
+You can override the pool configuration like this:
 
- [source,$lang]
- ----
- {@link examples.HTTPExamples#examplePoolConfiguration}
- ----
+[source,$lang]
+----
+{@link examples.HTTPExamples#examplePoolConfiguration}
+----
 
- You can configure various pool {@link io.vertx.core.http.PoolOptions options} as follows
+You can configure various pool {@link io.vertx.core.http.PoolOptions options} as follows
 
 - {@link io.vertx.core.http.PoolOptions options#setHttp1MaxSize} the maximum number of opened per HTTP/1.x server (5 by default)
 - {@link io.vertx.core.http.PoolOptions options#setHttp2MaxSize} the maximum number of opened per HTTP/2 server (1 by default), you *should* not change this value since a single HTTP/2 connection is capable of delivering the same performance level than multiple HTTP/1.x connections

vertx-core/src/main/asciidoc/index.adoc

@@ -1013,6 +1013,7 @@ You can add it to your classpath to improve the integration and remove the warni
 </profile>
 ----
 
+[#_native_transports]
 == Native transports
 
 Vert.x can run with http://netty.io/wiki/native-transports.html[native transports] (when available) on BSD (OSX) and Linux:
@@ -1022,7 +1023,7 @@ Vert.x can run with http://netty.io/wiki/native-transports.html[native transport
 {@link examples.CoreExamples#configureNative()}
 ----
 
-NOTE: preferring native transport will not prevent the application to execute (for example a native dependency might be missing). If your application requires native transport, you need to check {@link io.vertx.core.Vertx#isNativeTransportEnabled()}.
+NOTE: Preferring native transport will not prevent the application to execute (for example a native dependency might be missing).If your application requires native transport, you need to check {@link io.vertx.core.Vertx#isNativeTransportEnabled()}.
 
 You can also explicitly configure the transport to use:
 
@@ -1104,36 +1105,6 @@ Native on BSD gives you extra networking options:
 {@link examples.CoreExamples#configureBSDOptions}
 ----
 
-=== Domain sockets
-
-Natives provide domain sockets support for servers:
-
-[source,$lang]
-----
-{@link examples.CoreExamples#tcpServerWithDomainSockets}
-----
-
-or for http:
-
-[source,$lang]
-----
-{@link examples.CoreExamples#httpServerWithDomainSockets}
-----
-
-As well as clients:
-
-[source,$lang]
-----
-{@link examples.CoreExamples#tcpClientWithDomainSockets}
-----
-
-or for http:
-
-[source,$lang]
-----
-{@link examples.CoreExamples#httpClientWithDomainSockets}
-----
-
 == Security notes
 
 Vert.x is a toolkit, not an opinionated framework where we force you to do things in a certain way. This gives you

vertx-core/src/main/asciidoc/net.adoc

@@ -1,6 +1,6 @@
 == Writing TCP servers and clients
 
-Vert.x allows you to easily write non blocking TCP clients and servers.
+Vert.x allows you to easily write non-blocking TCP clients and servers.
 
 === Creating a TCP server
 
@@ -65,6 +65,15 @@ To find out the real port the server is listening on you can call {@link io.vert
 {@link examples.NetExamples#example5_1}
 ----
 
+=== Listening to Unix domain sockets
+
+When running on JDK 16+, or using a <<_native_transports,native transport>>, a server can listen to Unix domain sockets:
+
+[source,$lang]
+----
+{@link examples.CoreExamples#tcpServerWithDomainSockets}
+----
+
 === Getting notified of incoming connections
 
 To be notified when a connection is made you need to set a {@link io.vertx.core.net.NetServer#connectHandler(io.vertx.core.Handler)}:
@@ -260,6 +269,15 @@ specifying the port and host of the server and a handler that will be called wit
 {@link examples.NetExamples#example15}
 ----
 
+=== Making connections to Unix domain sockets
+
+When running on JDK 16+, or using a <<_native_transports,native transport>>, a client can connect to Unix domain sockets:
+
+[source,$lang]
+----
+{@link examples.CoreExamples#tcpClientWithDomainSockets}
+----
+
 === Configuring connection attempts
 
 A client can be configured to automatically retry connecting to the server in the event that it cannot connect.

vertx-core/src/main/java/examples/CoreExamples.java

@@ -501,31 +501,49 @@ public void configureBSDOptions(Vertx vertx, boolean reusePort) {
   }
 
   public void tcpServerWithDomainSockets(Vertx vertx) {
-    // Only available on BSD and Linux
-    vertx.createNetServer().connectHandler(so -> {
+    NetServer netServer = vertx.createNetServer();
+
+    // Only available when running on JDK16+, or using a native transport
+    SocketAddress address = SocketAddress.domainSocketAddress("/var/tmp/myservice.sock");
+
+    netServer
+      .connectHandler(so -> {
       // Handle application
-    }).listen(SocketAddress.domainSocketAddress("/var/tmp/myservice.sock"));
+      })
+      .listen(address)
+      .onComplete(ar -> {
+        if (ar.succeeded()) {
+          // Bound to socket
+        } else {
+          // Handle failure
+        }
+      });
   }
 
   public void httpServerWithDomainSockets(Vertx vertx) {
-    vertx.createHttpServer()
+    HttpServer httpServer = vertx.createHttpServer();
+
+    // Only available when running on JDK16+, or using a native transport
+    SocketAddress address = SocketAddress.domainSocketAddress("/var/tmp/myservice.sock");
+
+    httpServer
       .requestHandler(req -> {
         // Handle application
       })
-      .listen(SocketAddress.domainSocketAddress("/var/tmp/myservice.sock"))
+      .listen(address)
       .onComplete(ar -> {
         if (ar.succeeded()) {
           // Bound to socket
         } else {
-          ar.cause().printStackTrace();
+          // Handle failure
         }
       });
   }
 
   public void tcpClientWithDomainSockets(Vertx vertx) {
     NetClient netClient = vertx.createNetClient();
 
-    // Only available on BSD and Linux
+    // Only available when running on JDK16+, or using a native transport
     SocketAddress addr = SocketAddress.domainSocketAddress("/var/tmp/myservice.sock");
 
     // Connect to the server
@@ -535,15 +553,15 @@ public void tcpClientWithDomainSockets(Vertx vertx) {
         if (ar.succeeded()) {
           // Connected
         } else {
-          ar.cause().printStackTrace();
+          // Handle failure
         }
       });
   }
 
   public void httpClientWithDomainSockets(Vertx vertx) {
     HttpClient httpClient = vertx.createHttpClient();
 
-    // Only available on BSD and Linux
+    // Only available when running on JDK16+, or using a native transport
     SocketAddress addr = SocketAddress.domainSocketAddress("/var/tmp/myservice.sock");
 
     // Send request to the server
@@ -552,10 +570,13 @@ public void httpClientWithDomainSockets(Vertx vertx) {
       .setHost("localhost")
       .setPort(8080)
       .setURI("/"))
-      .onSuccess(request -> {
-        request.send().onComplete(response -> {
+      .compose(request -> request.send().compose(HttpClientResponse::body))
+      .onComplete(ar -> {
+        if (ar.succeeded()) {
           // Process response
-        });
+        } else {
+          // Handle failure
+        }
       });
   }
 }

vertx-core/src/main/java/io/vertx/core/impl/transports/NioTransport.java

@@ -10,30 +10,59 @@
  */
 package io.vertx.core.impl.transports;
 
-import io.netty.channel.*;
+import io.netty.channel.Channel;
+import io.netty.channel.ChannelFactory;
+import io.netty.channel.IoHandlerFactory;
+import io.netty.channel.ServerChannel;
 import io.netty.channel.nio.NioIoHandler;
 import io.netty.channel.socket.DatagramChannel;
 import io.netty.channel.socket.InternetProtocolFamily;
-import io.netty.channel.socket.nio.NioDatagramChannel;
-import io.netty.channel.socket.nio.NioServerSocketChannel;
-import io.netty.channel.socket.nio.NioSocketChannel;
+import io.netty.channel.socket.nio.*;
 import io.vertx.core.spi.transport.Transport;
 
+import java.net.SocketAddress;
+
 public class NioTransport implements Transport {
   /**
    * The NIO transport, always there.
    */
   public static final Transport INSTANCE = new NioTransport();
 
+  private final UnixDomainSocketNioTransport unixDomainSocketNioTransport = UnixDomainSocketNioTransport.load();
+
+  @Override
+  public boolean supportsDomainSockets() {
+    return unixDomainSocketNioTransport != null;
+  }
+
+  @Override
+  public SocketAddress convert(io.vertx.core.net.SocketAddress address) {
+    if (address.isDomainSocket() && unixDomainSocketNioTransport != null) {
+      return unixDomainSocketNioTransport.convert(address);
+    } else {
+      return Transport.super.convert(address);
+    }
+  }
+
+  @Override
+  public io.vertx.core.net.SocketAddress convert(SocketAddress address) {
+    if (unixDomainSocketNioTransport != null && unixDomainSocketNioTransport.isUnixDomainSocketAddress(address)) {
+      return unixDomainSocketNioTransport.convert(address);
+    }
+    return Transport.super.convert(address);
+  }
+
   @Override
   public IoHandlerFactory ioHandlerFactory() {
     return NioIoHandler.newFactory();
   }
 
+  @Override
   public DatagramChannel datagramChannel() {
     return new NioDatagramChannel();
   }
 
+  @Override
   public DatagramChannel datagramChannel(InternetProtocolFamily family) {
     switch (family) {
       case IPv4:
@@ -45,16 +74,25 @@ public DatagramChannel datagramChannel(InternetProtocolFamily family) {
     }
   }
 
+  @Override
   public ChannelFactory<? extends Channel> channelFactory(boolean domainSocket) {
     if (domainSocket) {
-      throw new IllegalArgumentException("The Vertx instance must be created with the preferNativeTransport option set to true to create domain sockets");
+      if (unixDomainSocketNioTransport == null) {
+        throw new IllegalArgumentException("Domain sockets require JDK 16 and above, or the usage of a native transport");
+      }
+      return NioDomainSocketChannel::new;
+    } else {
+      return NioSocketChannel::new;
     }
-    return NioSocketChannel::new;
   }
 
+  @Override
   public ChannelFactory<? extends ServerChannel> serverChannelFactory(boolean domainSocket) {
     if (domainSocket) {
-      throw new IllegalArgumentException("The Vertx instance must be created with the preferNativeTransport option set to true to create domain sockets");
+      if (unixDomainSocketNioTransport == null) {
+        throw new IllegalArgumentException("Domain sockets require JDK 16 and above, or the usage of a native transport");
+      }
+      return NioServerDomainSocketChannel::new;
     }
     return NioServerSocketChannel::new;
   }