Do consume messages on the correct thread after resume

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

@@ -152,7 +152,7 @@ protected void dispatch(Message<T> msg, ContextInternal context, Handler<Message
     if (handler == null) {
       throw new NullPointerException();
     }
-    context.dispatch(msg, handler);
+    context.emit(msg, handler);
   }
 
   private void deliver(Handler<Message<T>> theHandler, Message<T> message) {

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

@@ -0,0 +1,62 @@
+package io.vertx.tests.eventbus;
+
+import io.vertx.core.*;
+import io.vertx.core.eventbus.MessageConsumer;
+import io.vertx.test.core.VertxTestBase;
+import org.junit.Test;
+
+import java.util.HashMap;
+import java.util.Map;
+import java.util.concurrent.CountDownLatch;
+
+public class MessageConsumerTest extends VertxTestBase {
+
+
+  @Test
+  public void testMessageConsumptionStayOnWorkerThreadAfterResume() throws Exception {
+    TestVerticle verticle = new TestVerticle(2);
+    Future<String> deployVerticle = vertx.deployVerticle(verticle, new DeploymentOptions().setThreadingModel(ThreadingModel.WORKER));
+
+    CountDownLatch startLatch = new CountDownLatch(1);
+    deployVerticle.onComplete(onSuccess(cf -> startLatch.countDown()));
+    awaitLatch(startLatch);
+
+    vertx.eventBus().send("testAddress", "message1");
+    vertx.eventBus().send("testAddress", "message2");
+
+    awaitLatch(verticle.msgLatch);
+
+    assertEquals(2, verticle.messageArrivedOnWorkerThread.size());
+    assertTrue("message1 should be processed on worker thread", verticle.messageArrivedOnWorkerThread.get("message1"));
+    assertTrue("message2 should be processed on worker thread", verticle.messageArrivedOnWorkerThread.get("message2"));
+  }
+
+
+  private static class TestVerticle extends AbstractVerticle {
+
+    private final CountDownLatch msgLatch;
+
+    private final Map<String, Boolean> messageArrivedOnWorkerThread = new HashMap<>();
+
+    private TestVerticle(Integer numberOfExpectedMessages) {
+      this.msgLatch = new CountDownLatch(numberOfExpectedMessages);
+    }
+
+    @Override
+    public void start() {
+      MessageConsumer<String> consumer = vertx.eventBus().localConsumer("testAddress");
+      handleMessages(consumer);
+    }
+
+    private void handleMessages(MessageConsumer<String> consumer) {
+      consumer.handler(msg -> {
+        consumer.pause();
+        messageArrivedOnWorkerThread.putIfAbsent(msg.body(), Context.isOnWorkerThread());
+        msgLatch.countDown();
+        vertx.setTimer(20, id -> {
+          consumer.resume();
+        });
+      });
+    }
+  }
+}