- allows for local only codec lookup when in clustered mode (#4952)

* - allows for local only codec lookup when in clustered mode

* - is local decided by caller

Author: Reef3rm4n

src/main/java/io/vertx/core/eventbus/impl/EventBusImpl.java

@@ -119,14 +119,14 @@ public EventBus send(String address, Object message) {
 
   @Override
   public EventBus send(String address, Object message, DeliveryOptions options) {
-    MessageImpl msg = createMessage(true, address, options.getHeaders(), message, options.getCodecName());
+    MessageImpl msg = createMessage(true, isLocalOnly(options), address, options.getHeaders(), message, options.getCodecName());
     sendOrPubInternal(msg, options, null);
     return this;
   }
 
   @Override
   public <T> Future<Message<T>> request(String address, Object message, DeliveryOptions options) {
-    MessageImpl msg = createMessage(true, address, options.getHeaders(), message, options.getCodecName());
+    MessageImpl msg = createMessage(true, isLocalOnly(options), address, options.getHeaders(), message, options.getCodecName());
     ReplyHandler<T> handler = createReplyHandler(msg, true, options);
     sendOrPubInternal(msg, options, handler);
     return handler.result();
@@ -165,15 +165,15 @@ public EventBus publish(String address, Object message) {
 
   @Override
   public EventBus publish(String address, Object message, DeliveryOptions options) {
-    sendOrPubInternal(createMessage(false, address, options.getHeaders(), message, options.getCodecName()), options, null);
+    sendOrPubInternal(createMessage(false, isLocalOnly(options), address, options.getHeaders(), message, options.getCodecName()), options, null);
     return this;
   }
 
   @Override
   public <T> MessageConsumer<T> consumer(String address) {
     checkStarted();
     Objects.requireNonNull(address, "address");
-    return new MessageConsumerImpl<>(vertx.getOrCreateContext(), this, address,  false);
+    return new MessageConsumerImpl<>(vertx.getOrCreateContext(), this, address, false);
   }
 
   @Override
@@ -188,7 +188,7 @@ public <T> MessageConsumer<T> consumer(String address, Handler<Message<T>> handl
   public <T> MessageConsumer<T> localConsumer(String address) {
     checkStarted();
     Objects.requireNonNull(address, "address");
-    return new MessageConsumerImpl<>(vertx.getOrCreateContext(), this, address,  true);
+    return new MessageConsumerImpl<>(vertx.getOrCreateContext(), this, address, true);
   }
 
   @Override
@@ -253,9 +253,9 @@ public EventBusMetrics<?> getMetrics() {
     return metrics;
   }
 
-  public MessageImpl createMessage(boolean send, String address, MultiMap headers, Object body, String codecName) {
+  public MessageImpl createMessage(boolean send, boolean localOnly, String address, MultiMap headers, Object body, String codecName) {
     Objects.requireNonNull(address, "no null address accepted");
-    MessageCodec codec = codecManager.lookupCodec(body, codecName, true);
+    MessageCodec codec = codecManager.lookupCodec(body, codecName, localOnly);
     @SuppressWarnings("unchecked")
     MessageImpl msg = new MessageImpl(address, headers, body, codec, send, this);
     return msg;
@@ -382,7 +382,7 @@ protected ReplyException deliverMessageLocally(MessageImpl msg) {
         if (metrics != null) {
           metrics.messageReceived(msg.address(), !msg.isSend(), messageLocal, handlers.size());
         }
-        for (HandlerHolder holder: handlers) {
+        for (HandlerHolder holder : handlers) {
           if (messageLocal || !holder.isLocalOnly()) {
             holder.handler.receive(msg.copyBeforeReceive());
           }
@@ -412,8 +412,8 @@ protected String generateReplyAddress() {
   }
 
   <T> ReplyHandler<T> createReplyHandler(MessageImpl message,
-                                                 boolean src,
-                                                 DeliveryOptions options) {
+                                         boolean src,
+                                         DeliveryOptions options) {
     long timeout = options.getSendTimeout();
     String replyAddress = generateReplyAddress();
     message.setReplyAddress(replyAddress);
@@ -423,7 +423,7 @@ <T> ReplyHandler<T> createReplyHandler(MessageImpl message,
   }
 
   public <T> OutboundDeliveryContext<T> newSendContext(MessageImpl message, DeliveryOptions options,
-                                               ReplyHandler<T> handler) {
+                                                       ReplyHandler<T> handler) {
     return new OutboundDeliveryContext<>(vertx.getOrCreateContext(), message, options, handler);
   }
 
@@ -435,7 +435,7 @@ public <T> void sendOrPubInternal(OutboundDeliveryContext<T> senderCtx) {
   }
 
   public <T> Future<Void> sendOrPubInternal(MessageImpl message, DeliveryOptions options,
-                                    ReplyHandler<T> handler) {
+                                            ReplyHandler<T> handler) {
     checkStarted();
     OutboundDeliveryContext<T> ctx = newSendContext(message, options, handler);
     sendOrPubInternal(ctx);
@@ -478,7 +478,7 @@ private void removeInterceptor(AtomicReferenceFieldUpdater<EventBusImpl, Handler
     while (true) {
       Handler[] interceptors = updater.get(this);
       int idx = -1;
-      for (int i = 0;i < interceptors.length;i++) {
+      for (int i = 0; i < interceptors.length; i++) {
         if (interceptors[i].equals(interceptor)) {
           idx = i;
           break;
@@ -495,5 +495,12 @@ private void removeInterceptor(AtomicReferenceFieldUpdater<EventBusImpl, Handler
       }
     }
   }
+
+  private boolean isLocalOnly(DeliveryOptions options) {
+    if (vertx.isClustered()) {
+      return options.isLocalOnly();
+    }
+    return true;
+  }
 }
 

src/main/java/io/vertx/core/eventbus/impl/MessageImpl.java

@@ -119,7 +119,7 @@ public <R> Future<Message<R>> replyAndRequest(Object message, DeliveryOptions op
   }
 
   protected MessageImpl createReply(Object message, DeliveryOptions options) {
-    MessageImpl reply = bus.createMessage(true, replyAddress, options.getHeaders(), message, options.getCodecName());
+    MessageImpl reply = bus.createMessage(true, isLocal(), replyAddress, options.getHeaders(), message, options.getCodecName());
     reply.trace = trace;
     return reply;
   }

src/main/java/io/vertx/core/eventbus/impl/MessageProducerImpl.java

@@ -12,11 +12,9 @@
 package io.vertx.core.eventbus.impl;
 
 import io.vertx.core.Future;
-import io.vertx.core.Promise;
 import io.vertx.core.Vertx;
 import io.vertx.core.eventbus.*;
 import io.vertx.core.impl.ContextInternal;
-import io.vertx.core.impl.VertxInternal;
 
 /**
  * @author <a href="mailto:julien@julienviet.com">Julien Viet</a>
@@ -27,6 +25,7 @@ public class MessageProducerImpl<T> implements MessageProducer<T> {
   private final EventBusImpl bus;
   private final boolean send;
   private final String address;
+  private final boolean localOnly;
   private DeliveryOptions options;
 
   public MessageProducerImpl(Vertx vertx, String address, boolean send, DeliveryOptions options) {
@@ -35,6 +34,7 @@ public MessageProducerImpl(Vertx vertx, String address, boolean send, DeliveryOp
     this.address = address;
     this.send = send;
     this.options = options;
+    this.localOnly = vertx.isClustered() ? options.isLocalOnly() : true;
   }
 
   @Override
@@ -45,7 +45,7 @@ public synchronized MessageProducer<T> deliveryOptions(DeliveryOptions options)
 
   @Override
   public Future<Void> write(T body) {
-    MessageImpl msg = bus.createMessage(send, address, options.getHeaders(), body, options.getCodecName());
+    MessageImpl msg = bus.createMessage(send, localOnly, address, options.getHeaders(), body, options.getCodecName());
     return bus.sendOrPubInternal(msg, options, null);
   }
 

src/main/java/io/vertx/core/eventbus/impl/clustered/ClusteredEventBus.java

@@ -145,9 +145,9 @@ public void close(Promise<Void> promise) {
   }
 
   @Override
-  public MessageImpl createMessage(boolean send, String address, MultiMap headers, Object body, String codecName) {
+  public MessageImpl createMessage(boolean send, boolean local, String address, MultiMap headers, Object body, String codecName) {
     Objects.requireNonNull(address, "no null address accepted");
-    MessageCodec codec = codecManager.lookupCodec(body, codecName, false);
+    MessageCodec codec = codecManager.lookupCodec(body, codecName, local);
     @SuppressWarnings("unchecked")
     ClusteredMessage msg = new ClusteredMessage(nodeId, address, headers, body, codec, send, this);
     return msg;