More duplicated context unwrapping

Author: vietj

vertx-core/src/main/java/io/vertx/core/datagram/impl/DatagramSocketImpl.java

@@ -51,7 +51,7 @@
  */
 public class DatagramSocketImpl implements DatagramSocket, MetricsProvider, Closeable {
 
-  public static DatagramSocketImpl create(VertxInternal vertx, CloseFuture closeFuture, DatagramSocketOptions options) {
+  public static DatagramSocketImpl create(ContextInternal vertx, CloseFuture closeFuture, DatagramSocketOptions options) {
     DatagramSocketImpl socket = new DatagramSocketImpl(vertx, closeFuture, options);
     // Make sure object is fully initiliased to avoid race with async registration
     socket.init();
@@ -65,19 +65,18 @@ public static DatagramSocketImpl create(VertxInternal vertx, CloseFuture closeFu
   private Handler<Throwable> exceptionHandler;
   private final CloseFuture closeFuture;
 
-  private DatagramSocketImpl(VertxInternal vertx, CloseFuture closeFuture, DatagramSocketOptions options) {
-    Transport transport = vertx.transport();
+  private DatagramSocketImpl(ContextInternal context, CloseFuture closeFuture, DatagramSocketOptions options) {
+    Transport transport = context.owner().transport();
     DatagramChannel channel = transport.datagramChannel(options.isIpV6() ? InternetProtocolFamily.IPv6 : InternetProtocolFamily.IPv4);
     transport.configure(channel, new DatagramSocketOptions(options));
-    ContextInternal context = vertx.getOrCreateContext();
     channel.config().setOption(ChannelOption.DATAGRAM_CHANNEL_ACTIVE_ON_REGISTRATION, true);
     MaxMessagesRecvByteBufAllocator bufAllocator = channel.config().getRecvByteBufAllocator();
     bufAllocator.maxMessagesPerRead(1);
     context.nettyEventLoop().register(channel);
     if (options.getLogActivity()) {
       channel.pipeline().addLast("logging", new LoggingHandler(options.getActivityLogDataFormat()));
     }
-    VertxMetrics metrics = vertx.metrics();
+    VertxMetrics metrics = context.owner().metrics();
     this.metrics = metrics != null ? metrics.createDatagramSocketMetrics(options) : null;
     this.channel = channel;
     this.context = context;

vertx-core/src/main/java/io/vertx/core/dns/impl/DnsClientImpl.java

@@ -64,8 +64,7 @@ public DnsClientImpl(VertxInternal vertx, DnsClientOptions options) {
     this.vertx = vertx;
   }
 
-  private DnsNameResolver resolver(ContextInternal ctx, InternetProtocolFamily ipFamily) {
-    EventLoop el = ctx.nettyEventLoop();
+  private DnsNameResolver resolver(EventLoop el, InternetProtocolFamily ipFamily) {
     DnsNameResolver resolver;
     synchronized (this) {
       if (closed) {
@@ -169,7 +168,7 @@ public Future<List<SrvRecord>> resolveSRV(String name) {
   private Future<List<String>> resolveAll(String name, InternetProtocolFamily ipFamily) {
     Objects.requireNonNull(name);
     ContextInternal ctx = vertx.getOrCreateContext();
-    DnsNameResolver resolver = resolver(ctx, ipFamily);
+    DnsNameResolver resolver = resolver(ctx.nettyEventLoop(), ipFamily);
     if (resolver == null) {
       return ctx.failedFuture("DNS client is closed");
     }
@@ -200,7 +199,7 @@ private Future<List<String>> resolveAll(String name, InternetProtocolFamily ipFa
   private <T> Future<List<T>> queryAll(String name, DnsRecordType recordType, Function<DnsRecord, T> mapper) {
     Objects.requireNonNull(name);
     ContextInternal ctx = vertx.getOrCreateContext();
-    DnsNameResolver resolver = resolver(ctx, InternetProtocolFamily.IPv4);
+    DnsNameResolver resolver = resolver(ctx.nettyEventLoop(), InternetProtocolFamily.IPv4);
     if (resolver == null) {
       return ctx.failedFuture("DNS client is closed");
     }

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.

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

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.
    */

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;

vertx-core/src/main/java/io/vertx/core/http/impl/ClientWebSocketImpl.java

@@ -52,7 +52,7 @@ public class ClientWebSocketImpl implements ClientWebSocketInternal {
 
   @Override
   public Future<WebSocket> connect(WebSocketConnectOptions options) {
-    return connect(client.vertx().getOrCreateContext(), options);
+    return connect(client.vertx().getOrCreateContext().unwrap(), options);
   }
 
   @Override

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));
       }
     });
   }

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

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) {