Repository: eclipse-vertx/vert.x

Problem statement:
EventBus interceptor does not get message body when clustered
From [David Klotz](https://groups.google.com/d/msg/vertx/7otR0fsI6_8/E4ZdC0usCQAJ) on the forum

```
Hi, 

I'm currently a bit confused by the differing behavior of event bus interceptors in a non-clustered and a clustered vert.x environment (currently on 3.2.1, can test again in 3.4.x if something in that area has changed since then). When run in a non-clustered environment, the interceptor has access to both the message body and the message headers, but when run in a clustered environment, the message body is always null. Is this behavior by design or a bug?

Cheers, 
David
```




Hints:
Reply from @tsegismont

```
Can you put together a small reproducer on GitHub ? Thanks
```
Reply from David Klotz

```
It's actually quite simple to reproduce, just run this with clustered = false or true (depends just on vertx and vertx-hazelcast) and observe the different output (just tried it with 3.4.1, still reproducible): 

https://gist.github.com/dklotz/083b49e23266cda392c92c8ee5e0377a

Cheers, 
David
```

Patch:
diff --git a/src/main/java/io/vertx/core/eventbus/SendContext.java b/src/main/java/io/vertx/core/eventbus/SendContext.java
--- a/src/main/java/io/vertx/core/eventbus/SendContext.java
+++ b/src/main/java/io/vertx/core/eventbus/SendContext.java
@@ -22,8 +22,12 @@ public interface SendContext<T> {
   void next();
 
   /**
-   *
    * @return true if the message is being sent (point to point) or False if the message is being published
    */
   boolean send();
+
+  /**
+   * @return the value sent or published (before being processed by the codec)
+   */
+  Object sentBody();
 }
diff --git a/src/main/java/io/vertx/core/eventbus/impl/EventBusImpl.java b/src/main/java/io/vertx/core/eventbus/impl/EventBusImpl.java
--- a/src/main/java/io/vertx/core/eventbus/impl/EventBusImpl.java
+++ b/src/main/java/io/vertx/core/eventbus/impl/EventBusImpl.java
@@ -16,8 +16,22 @@
 
 package io.vertx.core.eventbus.impl;
 
-import io.vertx.core.*;
-import io.vertx.core.eventbus.*;
+import io.vertx.core.AsyncResult;
+import io.vertx.core.Closeable;
+import io.vertx.core.Context;
+import io.vertx.core.Future;
+import io.vertx.core.Handler;
+import io.vertx.core.MultiMap;
+import io.vertx.core.Vertx;
+import io.vertx.core.eventbus.DeliveryOptions;
+import io.vertx.core.eventbus.EventBus;
+import io.vertx.core.eventbus.Message;
+import io.vertx.core.eventbus.MessageCodec;
+import io.vertx.core.eventbus.MessageConsumer;
+import io.vertx.core.eventbus.MessageProducer;
+import io.vertx.core.eventbus.ReplyException;
+import io.vertx.core.eventbus.ReplyFailure;
+import io.vertx.core.eventbus.SendContext;
 import io.vertx.core.impl.VertxInternal;
 import io.vertx.core.logging.Logger;
 import io.vertx.core.logging.LoggerFactory;
@@ -455,6 +469,11 @@ public void next() {
     public boolean send() {
       return message.isSend();
     }
+
+    @Override
+    public Object sentBody() {
+      return message.sentBody;
+    }
   }
 
   protected class ReplySendContextImpl<T> extends SendContextImpl<T> {


Test patch:
diff --git a/src/test/java/io/vertx/test/core/EventBusTestBase.java b/src/test/java/io/vertx/test/core/EventBusTestBase.java
--- a/src/test/java/io/vertx/test/core/EventBusTestBase.java
+++ b/src/test/java/io/vertx/test/core/EventBusTestBase.java
@@ -542,6 +542,28 @@ public void start(Future<Void> startFuture) throws Exception {
     awaitLatch(closeLatch);
   }
 
+  @Test
+  public void testMessageBodyInterceptor() throws Exception {
+    String content = TestUtils.randomUnicodeString(13);
+    startNodes(2);
+    waitFor(2);
+    CountDownLatch latch = new CountDownLatch(1);
+    vertices[0].eventBus().registerCodec(new StringLengthCodec()).<Integer>consumer("whatever", msg -> {
+      assertEquals(content.length(), (int) msg.body());
+      complete();
+    }).completionHandler(ar -> latch.countDown());
+    awaitLatch(latch);
+    StringLengthCodec codec = new StringLengthCodec();
+    vertices[1].eventBus().registerCodec(codec).addInterceptor(sc -> {
+      if ("whatever".equals(sc.message().address())) {
+        assertEquals(content, sc.sentBody());
+        complete();
+      }
+      sc.next();
+    }).send("whatever", content, new DeliveryOptions().setCodecName(codec.name()));
+    await();
+  }
+
   protected <T> void testSend(T val) {
     testSend(val, null);
   }
@@ -770,4 +792,31 @@ public byte systemCodecID() {
     }
   }
 
+  public static class StringLengthCodec implements MessageCodec<String, Integer> {
+
+    @Override
+    public void encodeToWire(Buffer buffer, String s) {
+      buffer.appendInt(s.length());
+    }
+
+    @Override
+    public Integer decodeFromWire(int pos, Buffer buffer) {
+      return buffer.getInt(pos);
+    }
+
+    @Override
+    public Integer transform(String s) {
+      return s.length();
+    }
+
+    @Override
+    public String name() {
+      return getClass().getName();
+    }
+
+    @Override
+    public byte systemCodecID() {
+      return -1;
+    }
+  }
 }