swallow error handler exceptions

src/main/java/net/engio/mbassy/bus/AbstractPubSubSupport.java

@@ -101,7 +101,14 @@ protected Collection<Subscription> getSubscriptionsByMessageType(Class messageTy
 
     protected void handlePublicationError(PublicationError error) {
         for (IPublicationErrorHandler errorHandler : errorHandlers) {
-            errorHandler.handleError(error);
+            try
+            {
+                errorHandler.handleError(error);
+            }
+            catch (Throwable ex)
+            {
+                ex.printStackTrace();
+            }
         }
     }
 

src/test/java/net/engio/mbassy/MethodDispatchTest.java

@@ -1,11 +1,14 @@
 package net.engio.mbassy;
 
 import net.engio.mbassy.bus.common.IMessageBus;
-import net.engio.mbassy.bus.config.BusConfiguration;
+import net.engio.mbassy.bus.config.Feature;
 import net.engio.mbassy.common.MessageBusTest;
 import net.engio.mbassy.listener.Handler;
+import org.junit.Assert;
 import org.junit.Test;
 
+import java.util.concurrent.atomic.AtomicInteger;
+
 /**
  * Very simple test to verify dispatch to correct message handler
  *
@@ -17,8 +20,6 @@ public class MethodDispatchTest extends MessageBusTest{
    private boolean listener1Called = false;
    private boolean listener2Called = false;
 
-
-
     // a simple event listener
     public class EventListener1 {
 
@@ -54,4 +55,24 @@ public void testDispatch1(){
         assertTrue(listener1Called);
     }
 
+    @Test
+    public void testAsyncDispatchAfterExceptionInErrorHandler() throws InterruptedException
+    {
+        IMessageBus bus = createBus(SyncAsync(true /*configures an error handler that throws exception*/).addFeature(Feature.AsynchronousMessageDispatch.Default().setNumberOfMessageDispatchers(1)));
+        final AtomicInteger msgHandlerCounter=new AtomicInteger(0);
+        bus.subscribe(new Object()
+        {
+            @Handler
+            public void handleAndThrowException(String s) throws Exception
+            {
+                msgHandlerCounter.incrementAndGet();
+                throw new Exception("error in msg handler on call no. " + msgHandlerCounter.get());
+            }
+        });
+        bus.post("first event - event handler will raise exception followed by another exception in the error handler").asynchronously();
+        bus.post("second event - expecting that msg dispatcher will still dispatch this after encountering exception in error handler").asynchronously();
+        pause(200);
+        Assert.assertEquals("msg handler is called also on the 2nd event after an exception in error handler following 1st event error", 2, msgHandlerCounter.get());
+        Assert.assertFalse("no more messages left to process", bus.hasPendingMessages());
+    }
 }

src/test/java/net/engio/mbassy/bus/AbstractPubSubSupportTest.java

@@ -1,6 +1,7 @@
 package net.engio.mbassy.bus;
 
 import com.mycila.testing.junit.MycilaJunitRunner;
+import junit.framework.Assert;
 import net.engio.mbassy.bus.config.IBusConfiguration;
 import net.engio.mbassy.bus.error.IPublicationErrorHandler;
 import net.engio.mbassy.bus.error.PublicationError;
@@ -12,6 +13,7 @@
 
 import java.io.ByteArrayOutputStream;
 import java.io.PrintStream;
+import java.util.concurrent.atomic.AtomicInteger;
 
 import static org.assertj.core.api.Assertions.assertThat;
 import static org.mockito.Mockito.verify;
@@ -183,5 +185,17 @@ public void testHandlePublicationError_default_construct_async() {
 
     }
 
-
+    @Test
+    public void testHandlePublicationError_raises_exception() {
+        final AtomicInteger invocationCounter = new AtomicInteger(0);
+        SyncMessageBus<String> bus = new SyncMessageBus<String>(new IPublicationErrorHandler() {
+            @Override
+            public void handleError(PublicationError error) {
+                invocationCounter.incrementAndGet();
+                throw new RuntimeException("exception encountered in error handler");
+            }
+        });
+        bus.handlePublicationError(publicationError);
+        Assert.assertEquals(1, invocationCounter.get());
+    }
 }