More duplicated context unwrapping

vietj · vietj · commit 99d9583c15b7 · 2025-06-12T20:08:24.000+02:00
diff --git a/vertx-core/src/main/java/io/vertx/core/eventbus/EventBus.java b/vertx-core/src/main/java/io/vertx/core/eventbus/EventBus.java
@@ -21,6 +21,7 @@
 import io.vertx.core.eventbus.impl.DefaultSerializableChecker;
 import io.vertx.core.metrics.Measured;
 
+import java.util.Objects;
 import java.util.function.Function;
 
 import static io.vertx.codegen.annotations.GenIgnore.PERMITTED_TYPE;
@@ -175,7 +176,12 @@ default <T> Future<Message<T>> request(String address, @Nullable Object message)
    * @param handler  the handler that will process the received messages
    * @return the event bus message consumer
    */
-  <T> MessageConsumer<T> localConsumer(String address, Handler<Message<T>> handler);
+  default <T> MessageConsumer<T> localConsumer(String address, Handler<Message<T>> handler) {
+    Objects.requireNonNull(handler, "handler");
+    MessageConsumer<T> consumer = localConsumer(address);
+    consumer.handler(handler);
+    return consumer;
+  }
 
   /**
    * Create a message sender against the specified address.
diff --git a/vertx-core/src/main/java/io/vertx/core/eventbus/impl/EventBusImpl.java b/vertx-core/src/main/java/io/vertx/core/eventbus/impl/EventBusImpl.java
@@ -187,7 +187,7 @@ public <T> MessageConsumer<T> consumer(MessageConsumerOptions options, Handler<M
   public <T> MessageConsumer<T> consumer(String address) {
     checkStarted();
     Objects.requireNonNull(address, "address");
-    return new MessageConsumerImpl<>(vertx.getOrCreateContext(), this, address, false, MessageConsumerOptions.DEFAULT_MAX_BUFFERED_MESSAGES);
+    return new MessageConsumerImpl<>(vertx.getOrCreateContext().unwrap(), this, address, false, MessageConsumerOptions.DEFAULT_MAX_BUFFERED_MESSAGES);
   }
 
   @Override
@@ -200,17 +200,18 @@ public <T> MessageConsumer<T> consumer(String address, Handler<Message<T>> handl
 
   @Override
   public <T> MessageConsumer<T> localConsumer(String address) {
-    checkStarted();
-    Objects.requireNonNull(address, "address");
-    return new MessageConsumerImpl<>(vertx.getOrCreateContext(), this, address, true, MessageConsumerOptions.DEFAULT_MAX_BUFFERED_MESSAGES);
+    return localConsumer(vertx.getOrCreateContext(), address);
   }
 
   @Override
-  public <T> MessageConsumer<T> localConsumer(String address, Handler<Message<T>> handler) {
-    Objects.requireNonNull(handler, "handler");
-    MessageConsumer<T> consumer = localConsumer(address);
-    consumer.handler(handler);
-    return consumer;
+  public <T> MessageConsumer<T> localConsumer(Context context, String address) {
+    checkStarted();
+    Objects.requireNonNull(context, "context");
+    Objects.requireNonNull(address, "address");
+    if (context.owner() != vertx) {
+      throw new IllegalArgumentException("Invalid context instance");
+    }
+    return new MessageConsumerImpl<>((ContextInternal) context, this, address, true, MessageConsumerOptions.DEFAULT_MAX_BUFFERED_MESSAGES);
   }
 
   @Override
diff --git a/vertx-core/src/main/java/io/vertx/core/eventbus/impl/EventBusInternal.java b/vertx-core/src/main/java/io/vertx/core/eventbus/impl/EventBusInternal.java
@@ -11,8 +11,13 @@
 
 package io.vertx.core.eventbus.impl;
 
+import io.vertx.core.Context;
 import io.vertx.core.Promise;
 import io.vertx.core.eventbus.EventBus;
+import io.vertx.core.eventbus.MessageConsumer;
+import io.vertx.core.eventbus.MessageConsumerOptions;
+
+import java.util.Objects;
 
 public interface EventBusInternal extends EventBus {
 
@@ -21,6 +26,8 @@ public interface EventBusInternal extends EventBus {
    */
   void start(Promise<Void> promise);
 
+  <T> MessageConsumer<T> localConsumer(Context context, String address);
+
   /**
    * Close the event bus and release any resources held.
    */
diff --git a/vertx-core/src/main/java/io/vertx/core/eventbus/impl/MessageConsumerImpl.java b/vertx-core/src/main/java/io/vertx/core/eventbus/impl/MessageConsumerImpl.java
@@ -38,6 +38,11 @@ public class MessageConsumerImpl<T> extends HandlerRegistration<T> implements Me
 
   MessageConsumerImpl(ContextInternal context, EventBusImpl eventBus, String address, boolean localOnly, int maxBufferedMessages) {
     super(context, eventBus, address, false);
+
+    if (context.isDuplicate()) {
+      throw new IllegalArgumentException("Duplicated context are not allowed");
+    }
+
     this.localOnly = localOnly;
     this.result = context.promise();
     this.maxBufferedMessages = maxBufferedMessages;
diff --git a/vertx-core/src/main/java/io/vertx/core/http/impl/Http1xServerConnection.java b/vertx-core/src/main/java/io/vertx/core/http/impl/Http1xServerConnection.java
@@ -297,7 +297,7 @@ void createWebSocket(Http1xServerRequest request, PromiseInternal<ServerWebSocke
           promise.fail(e);
           return;
         }
-        promise.complete(new ServerWebSocketHandshaker(request, handshaker, options));
+        promise.complete(new ServerWebSocketHandshaker(request, promise.context(), handshaker, options));
       }
     });
   }
diff --git a/vertx-core/src/main/java/io/vertx/core/http/impl/Http1xServerRequest.java b/vertx-core/src/main/java/io/vertx/core/http/impl/Http1xServerRequest.java
@@ -436,22 +436,14 @@ public String getFormAttribute(String attributeName) {
 
   @Override
   public Future<ServerWebSocket> toWebSocket() {
-    return webSocketHandshake().compose(handshake -> handshake.accept());
-  }
-
-  /**
-   * @return a future of the un-accepted WebSocket
-   */
-  Future<ServerWebSocketHandshake> webSocketHandshake() {
-    PromiseInternal<ServerWebSocketHandshake> promise = context.promise();
-    webSocketHandshake(promise);
-    return promise.future();
+    return webSocketHandshake(context.unwrap()).compose(handshake -> handshake.accept());
   }
 
   /**
    * Handle the request when a WebSocket upgrade header is present.
    */
-  private void webSocketHandshake(PromiseInternal<ServerWebSocketHandshake> promise) {
+  Future<ServerWebSocketHandshake> webSocketHandshake(ContextInternal context) {
+    PromiseInternal<ServerWebSocketHandshake> promise = context.promise();
     BufferInternal body = BufferInternal.buffer();
     boolean[] failed = new boolean[1];
     handler(buff -> {
@@ -482,6 +474,7 @@ private void webSocketHandshake(PromiseInternal<ServerWebSocketHandshake> promis
     });
     // In case we were paused
     resume();
+    return promise.future();
   }
 
   @Override
diff --git a/vertx-core/src/main/java/io/vertx/core/http/impl/Http1xServerRequestHandler.java b/vertx-core/src/main/java/io/vertx/core/http/impl/Http1xServerRequestHandler.java
@@ -44,7 +44,7 @@ public void handle(HttpServerRequest req) {
     if (wsHandler != null || wsHandshakeHandler != null) {
       if (req.headers().contains(UPGRADE, WEBSOCKET, true) && handlers.server.wsAccept()) {
         // Missing upgrade header + null request handler will be handled when creating the handshake by sending a 400 error
-        ((Http1xServerRequest)req).webSocketHandshake().onComplete(ar -> {
+        ((Http1xServerRequest)req).webSocketHandshake(((Http1xServerRequest)req).context.unwrap()).onComplete(ar -> {
           if (ar.succeeded()) {
             ServerWebSocketHandshaker handshake = (ServerWebSocketHandshaker) ar.result();
             if (wsHandshakeHandler == null) {
diff --git a/vertx-core/src/main/java/io/vertx/core/http/impl/HttpChannelConnector.java b/vertx-core/src/main/java/io/vertx/core/http/impl/HttpChannelConnector.java
@@ -166,6 +166,9 @@ public Future<HttpClientConnectionInternal> wrap(ContextInternal context, NetSoc
   }
 
   public Future<HttpClientConnectionInternal> httpConnect(ContextInternal context) {
+    if (context.isDuplicate()) {
+      throw new IllegalArgumentException("Cannot accept duplicate contexts");
+    }
     Promise<NetSocket> promise = context.promise();
     Future<NetSocket> future = promise.future();
     // We perform the compose operation before calling connect to be sure that the composition happens
diff --git a/vertx-core/src/main/java/io/vertx/core/http/impl/HttpClientImpl.java b/vertx-core/src/main/java/io/vertx/core/http/impl/HttpClientImpl.java
@@ -242,7 +242,7 @@ public Future<HttpClientConnection> connect(HttpConnectOptions connect) {
       server,
       false,
       0);
-    return (Future) connector.httpConnect(vertx.getOrCreateContext()).map(conn -> new UnpooledHttpClientConnection(conn).init());
+    return (Future) connector.httpConnect(vertx.getOrCreateContext().unwrap()).map(conn -> new UnpooledHttpClientConnection(conn).init());
   }
 
   @Override
diff --git a/vertx-core/src/main/java/io/vertx/core/http/impl/HttpServerImpl.java b/vertx-core/src/main/java/io/vertx/core/http/impl/HttpServerImpl.java
@@ -169,7 +169,27 @@ public Future<HttpServer> listen() {
   }
 
   @Override
-  public synchronized Future<HttpServer> listen(SocketAddress address) {
+  public Future<HttpServer> listen(SocketAddress address) {
+    ContextInternal context = vertx.getOrCreateContext();
+    ContextInternal listenContext;
+    if (context.isEventLoopContext()) {
+      listenContext = context.unwrap();
+    } else {
+      listenContext = context.toBuilder()
+        .withThreadingModel(ThreadingModel.EVENT_LOOP)
+        .build();
+    }
+    Promise<HttpServer> promise = context.promise();
+    Supplier<ContextInternal> streamContextSupplier = context.unwrap()::duplicate;
+    listen(listenContext, context.threadingModel(), streamContextSupplier, address, promise);
+    return promise.future();
+  }
+
+  private synchronized void listen(ContextInternal listenContext,
+                                   ThreadingModel threadingModel,
+                                   Supplier<ContextInternal> streamContextSupplier,
+                                   SocketAddress address,
+                                   Promise<HttpServer> promise) {
     if (requestHandler == null && webSocketHandler == null && webSocketHandhakeHandler == null) {
       throw new IllegalStateException("Set request or WebSocket handler first");
     }
@@ -181,23 +201,12 @@ public synchronized Future<HttpServer> listen(SocketAddress address) {
     if (tcpOptions.getSslOptions() != null) {
       configureApplicationLayerProtocols(tcpOptions.getSslOptions());
     }
-    ContextInternal context = vertx.getOrCreateContext();
-    ContextInternal listenContext;
-    // Not sure of this
-    if (context.isEventLoopContext()) {
-      listenContext = context.unwrap();
-    } else {
-      listenContext = context.toBuilder()
-        .withThreadingModel(ThreadingModel.EVENT_LOOP)
-        .build();
-    }
     NetServerInternal server = vertx.createNetServer(tcpOptions);
     Handler<Throwable> h = exceptionHandler;
     Handler<Throwable> exceptionHandler = h != null ? h : DEFAULT_EXCEPTION_HANDLER;
     server.exceptionHandler(exceptionHandler);
     server.connectHandler(so -> {
       NetSocketImpl soi = (NetSocketImpl) so;
-      Supplier<ContextInternal> streamContextSupplier = context.unwrap()::duplicate;
       String host = address.isInetSocket() ? address.host() : "localhost";
       int port = address.port();
       String serverOrigin = (tcpOptions.isSsl() ? "https" : "http") + "://" + host + ":" + port;
@@ -211,7 +220,7 @@ public synchronized Future<HttpServer> listen(SocketAddress address) {
         exceptionHandler);
       HttpServerConnectionInitializer initializer = new HttpServerConnectionInitializer(
         listenContext,
-        context.threadingModel(),
+        threadingModel,
         streamContextSupplier,
         this,
         vertx,
@@ -224,15 +233,7 @@ public synchronized Future<HttpServer> listen(SocketAddress address) {
     });
     tcpServer = server;
     closeSequence = new CloseSequence(p -> doClose(server, p), p -> doShutdown(server, p ));
-    Promise<HttpServer> result = context.promise();
-    tcpServer.listen(listenContext, address).onComplete(ar -> {
-      if (ar.succeeded()) {
-        result.complete(this);
-      } else {
-        result.fail(ar.cause());
-      }
-    });
-    return result.future();
+    tcpServer.listen(listenContext, address).map(this).onComplete(promise);
   }
 
   private void doShutdown(NetServer netServer, Completable<Void> p) {
diff --git a/vertx-core/src/main/java/io/vertx/core/http/impl/ServerWebSocketHandshaker.java b/vertx-core/src/main/java/io/vertx/core/http/impl/ServerWebSocketHandshaker.java
@@ -52,8 +52,8 @@ public class ServerWebSocketHandshaker extends FutureImpl<ServerWebSocket> imple
   private final WebSocketServerHandshaker handshaker;
   private boolean done;
 
-  public ServerWebSocketHandshaker(Http1xServerRequest request, WebSocketServerHandshaker handshaker, HttpServerOptions options) {
-    super(request.context);
+  public ServerWebSocketHandshaker(Http1xServerRequest request, ContextInternal context, WebSocketServerHandshaker handshaker, HttpServerOptions options) {
+    super(context);
     this.request = request;
     this.handshaker = handshaker;
     this.options = options;
@@ -94,7 +94,7 @@ public Future<ServerWebSocket> accept() {
     }
     ServerWebSocket ws;
     try {
-      ws = acceptHandshake();
+      ws = acceptHandshake(context);
     } catch (Exception e) {
       return rejectHandshake(BAD_REQUEST.code())
         .transform(ar -> {
@@ -159,7 +159,7 @@ private Future<Void> rejectHandshake(int sc) {
     return response.setStatusCode(sc).end(status.reasonPhrase());
   }
 
-  private ServerWebSocket acceptHandshake() {
+  private ServerWebSocket acceptHandshake(ContextInternal context) {
     Http1xServerConnection httpConn = (Http1xServerConnection) request.connection();
     ChannelHandlerContext chctx = httpConn.channelHandlerContext();
     Channel channel = chctx.channel();
@@ -175,9 +175,9 @@ private ServerWebSocket acceptHandshake() {
     }
     VertxHandler<WebSocketConnectionImpl> handler = VertxHandler.create(ctx -> {
       long closingTimeoutMS = options.getWebSocketClosingTimeout() >= 0 ? options.getWebSocketClosingTimeout() * 1000L : 0L;
-      WebSocketConnectionImpl webSocketConn = new WebSocketConnectionImpl(request.context, ctx, true, closingTimeoutMS,httpConn.metrics);
+      WebSocketConnectionImpl webSocketConn = new WebSocketConnectionImpl(context, ctx, true, closingTimeoutMS,httpConn.metrics);
       ServerWebSocketImpl webSocket = new ServerWebSocketImpl(
-        request.context(),
+        context,
         webSocketConn,
         handshaker.version() != WebSocketVersion.V00,
         request,
diff --git a/vertx-core/src/main/java/io/vertx/core/http/impl/WebSocketImplBase.java b/vertx-core/src/main/java/io/vertx/core/http/impl/WebSocketImplBase.java
@@ -25,6 +25,7 @@
 import io.vertx.core.MultiMap;
 import io.vertx.core.Promise;
 import io.vertx.core.buffer.Buffer;
+import io.vertx.core.eventbus.impl.EventBusInternal;
 import io.vertx.core.internal.buffer.BufferInternal;
 import io.vertx.core.eventbus.EventBus;
 import io.vertx.core.eventbus.Message;
@@ -88,6 +89,9 @@ public abstract class WebSocketImplBase<S extends WebSocket> implements WebSocke
                     int maxWebSocketFrameSize,
                     int maxWebSocketMessageSize,
                     boolean registerWebSocketWriteHandlers) {
+    if (context.isDuplicate()) {
+      throw new IllegalArgumentException();
+    }
     this.supportsContinuation = supportsContinuation;
     if (registerWebSocketWriteHandlers) {
       textHandlerID = "__vertx.ws." + UUID.randomUUID();
@@ -117,12 +121,12 @@ protected void handleMessage(WebSocketFrameInternal msg) {
     this.headers = headers;
   }
 
-  void registerHandler(EventBus eventBus) {
+  void registerHandler(EventBusInternal eventBus) {
     if (binaryHandlerID != null) {
       Handler<Message<Buffer>> binaryHandler = msg -> writeBinaryFrameInternal(msg.body());
       Handler<Message<String>> textHandler = msg -> writeTextFrameInternal(msg.body());
-      binaryHandlerRegistration = eventBus.<Buffer>localConsumer(binaryHandlerID).handler(binaryHandler);
-      textHandlerRegistration = eventBus.<String>localConsumer(textHandlerID).handler(textHandler);
+      binaryHandlerRegistration = eventBus.<Buffer>localConsumer(context, binaryHandlerID).handler(binaryHandler);
+      textHandlerRegistration = eventBus.<String>localConsumer(context, textHandlerID).handler(textHandler);
     }
   }
 
diff --git a/vertx-core/src/main/java/io/vertx/core/impl/VertxImpl.java b/vertx-core/src/main/java/io/vertx/core/impl/VertxImpl.java
@@ -426,7 +426,7 @@ public HttpClientBuilder httpClientBuilder() {
     return new HttpClientBuilderInternal(this);
   }
 
-  public EventBus eventBus() {
+  public EventBusInternal eventBus() {
     return eventBus;
   }
 
diff --git a/vertx-core/src/main/java/io/vertx/core/internal/VertxInternal.java b/vertx-core/src/main/java/io/vertx/core/internal/VertxInternal.java
@@ -14,6 +14,8 @@
 
 import io.netty.channel.EventLoopGroup;
 import io.vertx.core.*;
+import io.vertx.core.eventbus.EventBus;
+import io.vertx.core.eventbus.impl.EventBusInternal;
 import io.vertx.core.impl.*;
 import io.vertx.core.internal.deployment.DeploymentManager;
 import io.vertx.core.internal.resolver.NameResolver;
@@ -102,6 +104,9 @@ default NetServerInternal createNetServer() {
   @Override
   ContextInternal getOrCreateContext();
 
+  @Override
+  EventBusInternal eventBus();
+
   EventLoopGroup eventLoopGroup();
 
   EventLoopGroup acceptorEventLoopGroup();
diff --git a/vertx-core/src/main/java/io/vertx/core/internal/VertxWrapper.java b/vertx-core/src/main/java/io/vertx/core/internal/VertxWrapper.java
@@ -17,6 +17,7 @@
 import io.vertx.core.dns.DnsClient;
 import io.vertx.core.dns.DnsClientOptions;
 import io.vertx.core.eventbus.EventBus;
+import io.vertx.core.eventbus.impl.EventBusInternal;
 import io.vertx.core.file.FileSystem;
 import io.vertx.core.http.*;
 import io.vertx.core.internal.deployment.DeploymentManager;
@@ -106,7 +107,7 @@ public FileSystem fileSystem() {
   }
 
   @Override
-  public EventBus eventBus() {
+  public EventBusInternal eventBus() {
     return delegate.eventBus();
   }
 
diff --git a/vertx-core/src/main/java/io/vertx/core/net/impl/ConnectionBase.java b/vertx-core/src/main/java/io/vertx/core/net/impl/ConnectionBase.java
@@ -81,6 +81,10 @@ public abstract class ConnectionBase {
 
   protected ConnectionBase(ContextInternal context, ChannelHandlerContext chctx) {
 
+    if (context.isDuplicate()) {
+      throw new IllegalArgumentException("Cannot accept duplicate contexts");
+    }
+
     PromiseInternal<Void> f = context.promise();
     chctx
       .channel()
diff --git a/vertx-core/src/main/java/io/vertx/core/net/impl/NetClientImpl.java b/vertx-core/src/main/java/io/vertx/core/net/impl/NetClientImpl.java
@@ -139,7 +139,7 @@ public Future<NetSocket> connect(SocketAddress remoteAddress, String serverName)
 
   @Override
   public Future<NetSocket> connect(ConnectOptions connectOptions) {
-    ContextInternal context = vertx.getOrCreateContext();
+    ContextInternal context = vertx.getOrCreateContext().unwrap();
     Promise<NetSocket> promise = context.promise();
     connectInternal(connectOptions, options.isRegisterWriteHandler(), promise, context, options.getReconnectAttempts());
     return promise.future();
@@ -237,6 +237,9 @@ private void connectInternal(ConnectOptions connectOptions,
                                Promise<NetSocket> connectHandler,
                                ContextInternal context,
                                int remainingAttempts) {
+    if (context.isDuplicate()) {
+      throw new IllegalArgumentException("Cannot accept duplicate contexts");
+    }
     if (closeSequence.started()) {
       connectHandler.fail(new IllegalStateException("Client is closed"));
     } else {
diff --git a/vertx-core/src/main/java/io/vertx/core/net/impl/NetServerImpl.java b/vertx-core/src/main/java/io/vertx/core/net/impl/NetServerImpl.java
diff --git a/vertx-core/src/test/java/io/vertx/tests/eventbus/EventBusTestBase.java b/vertx-core/src/test/java/io/vertx/tests/eventbus/EventBusTestBase.java
diff --git a/vertx-core/src/test/java/io/vertx/tests/http/HttpTest.java b/vertx-core/src/test/java/io/vertx/tests/http/HttpTest.java
diff --git a/vertx-core/src/test/java/io/vertx/tests/net/NetTest.java b/vertx-core/src/test/java/io/vertx/tests/net/NetTest.java
diff --git a/vertx-core/src/test/java/io/vertx/tests/timer/TimerTest.java b/vertx-core/src/test/java/io/vertx/tests/timer/TimerTest.java

Original file line number	Diff line number	Diff line change
`@@ -297,7 +297,7 @@ void createWebSocket(Http1xServerRequest request, PromiseInternal<ServerWebSocke`
`297`	`297`	`promise.fail(e);`
`298`	`298`	`return;`
`299`	`299`	`}`
`300`		`- promise.complete(new ServerWebSocketHandshaker(request, handshaker, options));`
	`300`	`+ promise.complete(new ServerWebSocketHandshaker(request, promise.context(), handshaker, options));`
`301`	`301`	`}`
`302`	`302`	`});`
`303`	`303`	`}`
Original file line number	Diff line number	Diff line change
`@@ -166,6 +166,9 @@ public Future<HttpClientConnectionInternal> wrap(ContextInternal context, NetSoc`
`166`	`166`	`}`
`167`	`167`
`168`	`168`	`public Future<HttpClientConnectionInternal> httpConnect(ContextInternal context) {`
	`169`	`+ if (context.isDuplicate()) {`
	`170`	`+ throw new IllegalArgumentException("Cannot accept duplicate contexts");`
	`171`	`+ }`
`169`	`172`	`Promise<NetSocket> promise = context.promise();`
`170`	`173`	`Future<NetSocket> future = promise.future();`
`171`	`174`	`// We perform the compose operation before calling connect to be sure that the composition happens`
Original file line number	Diff line number	Diff line change
`@@ -242,7 +242,7 @@ public Future<HttpClientConnection> connect(HttpConnectOptions connect) {`
`242`	`242`	`server,`
`243`	`243`	`false,`
`244`	`244`	`0);`
`245`		`- return (Future) connector.httpConnect(vertx.getOrCreateContext()).map(conn -> new UnpooledHttpClientConnection(conn).init());`
	`245`	`+ return (Future) connector.httpConnect(vertx.getOrCreateContext().unwrap()).map(conn -> new UnpooledHttpClientConnection(conn).init());`
`246`	`246`	`}`
`247`	`247`
`248`	`248`	`@Override`