Let the DeliveryContext next implementation work correctly when used by different threads.

Motivation:

The current implementation of DeliveryContext#next assumes the same thread makes interceptor delivery progress, this API however assumes any thread can make interceptor progress.

Changes:

Rewrite the implementation of DeliveryContext#next with an atomic interceptor index that guarantees that progress is visible accross threads until the last interceptor is invoked.

Author: vietj

vertx-core/src/main/java/io/vertx/core/eventbus/impl/DeliveryContextBase.java

@@ -15,16 +15,17 @@
 import io.vertx.core.eventbus.Message;
 import io.vertx.core.internal.ContextInternal;
 
+import java.util.concurrent.atomic.AtomicIntegerFieldUpdater;
+
 abstract class DeliveryContextBase<T> implements DeliveryContext<T> {
 
+  private static final AtomicIntegerFieldUpdater<DeliveryContextBase> UPDATER = AtomicIntegerFieldUpdater.newUpdater(DeliveryContextBase.class, "interceptorIdx");
+
   public final MessageImpl<?, T> message;
   public final ContextInternal context;
 
   private final Handler<DeliveryContext>[] interceptors;
-
-  private int interceptorIdx;
-  private boolean invoking;
-  private boolean invokeNext;
+  private volatile int interceptorIdx;
 
   protected DeliveryContextBase(MessageImpl<?, T> message, Handler<DeliveryContext>[] interceptors, ContextInternal context) {
     this.message = message;
@@ -34,12 +35,7 @@ protected DeliveryContextBase(MessageImpl<?, T> message, Handler<DeliveryContext
   }
 
   void dispatch() {
-    this.interceptorIdx = 0;
-    if (invoking) {
-      this.invokeNext = true;
-    } else {
-      next();
-    }
+    next();
   }
 
   @Override
@@ -49,33 +45,24 @@ public Message<T> message() {
 
   protected abstract void execute();
 
-
   @Override
   public void next() {
-    if (invoking) {
-      invokeNext = true;
-    } else {
-      while (interceptorIdx < interceptors.length) {
-        Handler<DeliveryContext> interceptor = interceptors[interceptorIdx];
-        invoking = true;
-        interceptorIdx++;
-        if (context.inThread()) {
-          context.dispatch(this, interceptor);
-        } else {
-          try {
-            interceptor.handle(this);
-          } catch (Throwable t) {
-            context.reportException(t);
-          }
+    int idx = UPDATER.getAndIncrement(this);
+    if (idx < interceptors.length) {
+      Handler<DeliveryContext> interceptor = interceptors[idx];
+      if (context.inThread()) {
+        context.dispatch(this, interceptor);
+      } else {
+        try {
+          interceptor.handle(this);
+        } catch (Throwable t) {
+          context.reportException(t);
         }
-        invoking = false;
-        if (!invokeNext) {
-          return;
-        }
-        invokeNext = false;
       }
-      interceptorIdx = 0;
+    } else if (idx == interceptors.length) {
       execute();
+    } else {
+      throw new IllegalStateException();
     }
   }
 }

vertx-core/src/test/java/io/vertx/tests/eventbus/EventBusInterceptorTest.java

@@ -385,6 +385,7 @@ public void testOutboundInterceptorFromNonVertxThreadFailure() {
 
   @Test
   public void testInboundInterceptorFromNonVertxThreadDispatch() {
+    disableThreadChecks();
     AtomicReference<Thread> interceptorThread = new AtomicReference<>();
     AtomicReference<Thread> th = new AtomicReference<>();
     eb.addInboundInterceptor(sc -> {
@@ -394,12 +395,13 @@ public void testInboundInterceptorFromNonVertxThreadDispatch() {
       }).start();
     });
     eb.addInboundInterceptor(sc -> {
+      assertTrue(!Context.isOnEventLoopThread());
       interceptorThread.set(Thread.currentThread());
     });
     eb.consumer("some-address", msg -> {
     });
     eb.send("some-address", "armadillo");
-    waitUntil(() -> interceptorThread.get() != null);
+    assertWaitUntil(() -> interceptorThread.get() != null);
     assertSame(th.get(), interceptorThread.get());
   }