@Test public void test() throws InterruptedException { 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); Assert.assertThat(ringBuffer.get(0).get(), is(1L)); Assert.assertThat(ringBuffer.get(1).get(), is(1L)); }
public RingBuffer<T> start(final Executor executor) { if (!started.compareAndSet(false, true)) { throw new IllegalStateException( "WorkerPool has already been started and cannot be restarted until halted."); } final long cursor = ringBuffer.getCursor(); workSequence.set(cursor); for (WorkProcessor<?> processor : workProcessors) { processor.getSequence().set(cursor); executor.execute(processor); } return ringBuffer; }
WorkerPool { public RingBuffer<T> start(final Executor executor) { if (!started.compareAndSet(false, true)) { throw new IllegalStateException( "WorkerPool has already been started and cannot be restarted until halted."); } final long cursor = ringBuffer.getCursor(); workSequence.set(cursor); for (WorkProcessor<?> processor : workProcessors) { processor.getSequence().set(cursor); executor.execute(processor); } return ringBuffer; } }
WorkerPool { public RingBuffer<T> start(final Executor executor) { if (!started.compareAndSet(false, true)) { throw new IllegalStateException( "WorkerPool has already been started and cannot be restarted until halted."); } final long cursor = ringBuffer.getCursor(); workSequence.set(cursor); for (WorkProcessor<?> processor : workProcessors) { processor.getSequence().set(cursor); executor.execute(processor); } return ringBuffer; } WorkerPool(final RingBuffer<T> ringBuffer, final SequenceBarrier sequenceBarrier,
                      final ExceptionHandler<? super T> exceptionHandler,
                      final WorkHandler<? super T>... workHandlers);  WorkerPool(final EventFactory<T> eventFactory,
                      final ExceptionHandler<? super T> exceptionHandler,
                      final WorkHandler<? super T>... workHandlers); }
WorkerPool { public RingBuffer<T> start(final Executor executor) { if (!started.compareAndSet(false, true)) { throw new IllegalStateException( "WorkerPool has already been started and cannot be restarted until halted."); } final long cursor = ringBuffer.getCursor(); workSequence.set(cursor); for (WorkProcessor<?> processor : workProcessors) { processor.getSequence().set(cursor); executor.execute(processor); } return ringBuffer; } WorkerPool(final RingBuffer<T> ringBuffer, final SequenceBarrier sequenceBarrier,
                      final ExceptionHandler<? super T> exceptionHandler,
                      final WorkHandler<? super T>... workHandlers);  WorkerPool(final EventFactory<T> eventFactory,
                      final ExceptionHandler<? super T> exceptionHandler,
                      final WorkHandler<? super T>... workHandlers); Sequence[] getWorkerSequences(); RingBuffer<T> start(final Executor executor); void drainAndHalt(); void halt(); boolean isRunning(); }
WorkerPool { public RingBuffer<T> start(final Executor executor) { if (!started.compareAndSet(false, true)) { throw new IllegalStateException( "WorkerPool has already been started and cannot be restarted until halted."); } final long cursor = ringBuffer.getCursor(); workSequence.set(cursor); for (WorkProcessor<?> processor : workProcessors) { processor.getSequence().set(cursor); executor.execute(processor); } return ringBuffer; } WorkerPool(final RingBuffer<T> ringBuffer, final SequenceBarrier sequenceBarrier,
                      final ExceptionHandler<? super T> exceptionHandler,
                      final WorkHandler<? super T>... workHandlers);  WorkerPool(final EventFactory<T> eventFactory,
                      final ExceptionHandler<? super T> exceptionHandler,
                      final WorkHandler<? super T>... workHandlers); Sequence[] getWorkerSequences(); RingBuffer<T> start(final Executor executor); void drainAndHalt(); void halt(); boolean isRunning();  }