Fixed bug in WorkerPool where it would consume messages before they were published in the multi-producer scenario

Author: mikeb01

src/main/java/com/lmax/disruptor/WorkProcessor.java

@@ -113,11 +113,12 @@ public void run()
                     sequence.set(nextSequence - 1L);
                 }
 
-                sequenceBarrier.waitFor(nextSequence);
-                event = ringBuffer.get(nextSequence);
-                workHandler.onEvent(event);
-
-                processedSequence = true;
+                if (sequenceBarrier.waitFor(nextSequence) >= nextSequence)
+                {
+                    event = ringBuffer.get(nextSequence);
+                    workHandler.onEvent(event);
+                    processedSequence = true;
+                }
             }
             catch (final AlertException ex)
             {
@@ -128,7 +129,7 @@ public void run()
             }
             catch (final Throwable ex)
             {
-                // handle, mark as procesed, unless the exception handler threw an exception
+                // handle, mark as processed, unless the exception handler threw an exception
                 exceptionHandler.handleEventException(ex, nextSequence, event);
                 processedSequence = true;
             }

src/test/java/com/lmax/disruptor/WorkerPoolTest.java

@@ -0,0 +1,75 @@
+package com.lmax.disruptor;
+
+import static org.hamcrest.CoreMatchers.is;
+import static org.junit.Assert.assertThat;
+
+import java.util.concurrent.Executor;
+import java.util.concurrent.Executors;
+import java.util.concurrent.atomic.AtomicLong;
+
+import org.junit.Test;
+
+import com.lmax.disruptor.util.DaemonThreadFactory;
+
+
+public class WorkerPoolTest
+{
+    @SuppressWarnings("unchecked")
+    @Test
+    public void shouldProcessEachMessageByOnlyOneWorker() throws Exception
+    {
+        Executor executor = Executors.newCachedThreadPool(DaemonThreadFactory.INSTANCE);
+        WorkerPool<AtomicLong> pool = new WorkerPool<AtomicLong>(new AtomicLongEventFactory(), new FatalExceptionHandler(),
+                new AtomicLongWorkHandler(), new AtomicLongWorkHandler());
+
+        RingBuffer<AtomicLong> ringBuffer = pool.start(executor);
+
+        ringBuffer.next();
+        ringBuffer.next();
+        ringBuffer.publish(0);
+        ringBuffer.publish(1);
+
+        Thread.sleep(500);
+
+        assertThat(ringBuffer.get(0).get(), is(1L));
+        assertThat(ringBuffer.get(1).get(), is(1L));
+    }
+
+    @SuppressWarnings("unchecked")
+    @Test
+    public void shouldProcessOnlyOnceItHasBeenPublished() throws Exception
+    {
+        Executor executor = Executors.newCachedThreadPool(DaemonThreadFactory.INSTANCE);
+        WorkerPool<AtomicLong> pool = new WorkerPool<AtomicLong>(new AtomicLongEventFactory(), new FatalExceptionHandler(),
+                new AtomicLongWorkHandler(), new AtomicLongWorkHandler());
+
+        RingBuffer<AtomicLong> ringBuffer = pool.start(executor);
+
+        ringBuffer.next();
+        ringBuffer.next();
+
+        Thread.sleep(1000);
+
+        assertThat(ringBuffer.get(0).get(), is(0L));
+        assertThat(ringBuffer.get(1).get(), is(0L));
+    }
+
+    private static class AtomicLongWorkHandler implements WorkHandler<AtomicLong>
+    {
+        @Override
+        public void onEvent(AtomicLong event) throws Exception
+        {
+            event.incrementAndGet();
+        }
+    }
+
+
+    private static class AtomicLongEventFactory implements EventFactory<AtomicLong>
+    {
+        @Override
+        public AtomicLong newInstance()
+        {
+            return new AtomicLong(0);
+        }
+    }
+}

src/test/java/com/lmax/disruptor/dsl/DisruptorTest.java

@@ -62,8 +62,8 @@ public class DisruptorTest
     private static final int TIMEOUT_IN_SECONDS = 2;
     private Disruptor<TestEvent> disruptor;
     private StubExecutor executor;
-    private Collection<DelayedEventHandler> delayedEventHandlers = new ArrayList<DelayedEventHandler>();
-    private Collection<TestWorkHandler> testWorkHandlers = new ArrayList<TestWorkHandler>();
+    private final Collection<DelayedEventHandler> delayedEventHandlers = new ArrayList<DelayedEventHandler>();
+    private final Collection<TestWorkHandler> testWorkHandlers = new ArrayList<TestWorkHandler>();
     private RingBuffer<TestEvent> ringBuffer;
     private TestEvent lastPublishedEvent;