@Test @SuppressWarnings("unchecked") public void shouldPollForEvents() throws Exception { final Sequence gatingSequence = new Sequence(); final SingleProducerSequencer sequencer = new SingleProducerSequencer(16, new BusySpinWaitStrategy()); final EventPoller.Handler<Object> handler = new EventPoller.Handler<Object>() { public boolean onEvent(Object event, long sequence, boolean endOfBatch) throws Exception { return false; } }; final Object[] data = new Object[16]; final DataProvider<Object> provider = new DataProvider<Object>() { public Object get(long sequence) { return data[(int) sequence]; } }; final EventPoller<Object> poller = sequencer.newPoller(provider, gatingSequence); final Object event = new Object(); data[0] = event; assertThat(poller.poll(handler), is(EventPoller.PollState.IDLE)); sequencer.publish(sequencer.next()); assertThat(poller.poll(handler), is(EventPoller.PollState.GATING)); gatingSequence.incrementAndGet(); assertThat(poller.poll(handler), is(EventPoller.PollState.PROCESSING)); }
public PollState poll(final Handler<T> eventHandler) throws Exception { final long currentSequence = sequence.get(); long nextSequence = currentSequence + 1; final long availableSequence = sequencer.getHighestPublishedSequence(nextSequence, gatingSequence.get()); if (nextSequence <= availableSequence) { boolean processNextEvent; long processedSequence = currentSequence; try { do { final T event = dataProvider.get(nextSequence); processNextEvent = eventHandler.onEvent(event, nextSequence, nextSequence == availableSequence); processedSequence = nextSequence; nextSequence++; } while (nextSequence <= availableSequence & processNextEvent); } finally { sequence.set(processedSequence); } return PollState.PROCESSING; } else if (sequencer.getCursor() >= nextSequence) { return PollState.GATING; } else { return PollState.IDLE; } }
EventPoller { public PollState poll(final Handler<T> eventHandler) throws Exception { final long currentSequence = sequence.get(); long nextSequence = currentSequence + 1; final long availableSequence = sequencer.getHighestPublishedSequence(nextSequence, gatingSequence.get()); if (nextSequence <= availableSequence) { boolean processNextEvent; long processedSequence = currentSequence; try { do { final T event = dataProvider.get(nextSequence); processNextEvent = eventHandler.onEvent(event, nextSequence, nextSequence == availableSequence); processedSequence = nextSequence; nextSequence++; } while (nextSequence <= availableSequence & processNextEvent); } finally { sequence.set(processedSequence); } return PollState.PROCESSING; } else if (sequencer.getCursor() >= nextSequence) { return PollState.GATING; } else { return PollState.IDLE; } } }
EventPoller { public PollState poll(final Handler<T> eventHandler) throws Exception { final long currentSequence = sequence.get(); long nextSequence = currentSequence + 1; final long availableSequence = sequencer.getHighestPublishedSequence(nextSequence, gatingSequence.get()); if (nextSequence <= availableSequence) { boolean processNextEvent; long processedSequence = currentSequence; try { do { final T event = dataProvider.get(nextSequence); processNextEvent = eventHandler.onEvent(event, nextSequence, nextSequence == availableSequence); processedSequence = nextSequence; nextSequence++; } while (nextSequence <= availableSequence & processNextEvent); } finally { sequence.set(processedSequence); } return PollState.PROCESSING; } else if (sequencer.getCursor() >= nextSequence) { return PollState.GATING; } else { return PollState.IDLE; } } EventPoller(final DataProvider<T> dataProvider, final Sequencer sequencer,
                       final Sequence sequence, final Sequence gatingSequence); }
EventPoller { public PollState poll(final Handler<T> eventHandler) throws Exception { final long currentSequence = sequence.get(); long nextSequence = currentSequence + 1; final long availableSequence = sequencer.getHighestPublishedSequence(nextSequence, gatingSequence.get()); if (nextSequence <= availableSequence) { boolean processNextEvent; long processedSequence = currentSequence; try { do { final T event = dataProvider.get(nextSequence); processNextEvent = eventHandler.onEvent(event, nextSequence, nextSequence == availableSequence); processedSequence = nextSequence; nextSequence++; } while (nextSequence <= availableSequence & processNextEvent); } finally { sequence.set(processedSequence); } return PollState.PROCESSING; } else if (sequencer.getCursor() >= nextSequence) { return PollState.GATING; } else { return PollState.IDLE; } } EventPoller(final DataProvider<T> dataProvider, final Sequencer sequencer,
                       final Sequence sequence, final Sequence gatingSequence); PollState poll(final Handler<T> eventHandler); static EventPoller<T> newInstance(final DataProvider<T> dataProvider,
                                                 final Sequencer sequencer,
                                                 final Sequence sequence,
                                                 final Sequence cursorSequence,
                                                 final Sequence... gatingSequences); Sequence getSequence(); }
EventPoller { public PollState poll(final Handler<T> eventHandler) throws Exception { final long currentSequence = sequence.get(); long nextSequence = currentSequence + 1; final long availableSequence = sequencer.getHighestPublishedSequence(nextSequence, gatingSequence.get()); if (nextSequence <= availableSequence) { boolean processNextEvent; long processedSequence = currentSequence; try { do { final T event = dataProvider.get(nextSequence); processNextEvent = eventHandler.onEvent(event, nextSequence, nextSequence == availableSequence); processedSequence = nextSequence; nextSequence++; } while (nextSequence <= availableSequence & processNextEvent); } finally { sequence.set(processedSequence); } return PollState.PROCESSING; } else if (sequencer.getCursor() >= nextSequence) { return PollState.GATING; } else { return PollState.IDLE; } } EventPoller(final DataProvider<T> dataProvider, final Sequencer sequencer,
                       final Sequence sequence, final Sequence gatingSequence); PollState poll(final Handler<T> eventHandler); static EventPoller<T> newInstance(final DataProvider<T> dataProvider,
                                                 final Sequencer sequencer,
                                                 final Sequence sequence,
                                                 final Sequence cursorSequence,
                                                 final Sequence... gatingSequences); Sequence getSequence();  }