@Test public void shouldNotReadSingleMessagePartWayThroughWriting() { final long head = 0L; final int headIndex = (int)head; when(buffer.getLong(HEAD_COUNTER_INDEX)).thenReturn(head); when(buffer.getIntVolatile(lengthOffset(headIndex))).thenReturn(0); final int[] times = new int[1]; final MessageHandler handler = (msgTypeId, buffer, index, length) -> times[0]++; final int messagesRead = ringBuffer.read(handler); assertThat(messagesRead, is(0)); assertThat(times[0], is(0)); final InOrder inOrder = inOrder(buffer); inOrder.verify(buffer, times(1)).getIntVolatile(lengthOffset(headIndex)); inOrder.verify(buffer, times(0)).setMemory(anyInt(), anyInt(), anyByte()); inOrder.verify(buffer, times(0)).putLongOrdered(HEAD_COUNTER_INDEX, headIndex); }
public int read(final MessageHandler handler) { return read(handler, Integer.MAX_VALUE); }
OneToOneRingBuffer implements RingBuffer { public int read(final MessageHandler handler) { return read(handler, Integer.MAX_VALUE); } }
OneToOneRingBuffer implements RingBuffer { public int read(final MessageHandler handler) { return read(handler, Integer.MAX_VALUE); } OneToOneRingBuffer(final AtomicBuffer buffer); }
OneToOneRingBuffer implements RingBuffer { public int read(final MessageHandler handler) { return read(handler, Integer.MAX_VALUE); } OneToOneRingBuffer(final AtomicBuffer buffer); int capacity(); boolean write(final int msgTypeId, final DirectBuffer srcBuffer, final int offset, final int length); int read(final MessageHandler handler); int read(final MessageHandler handler, final int messageCountLimit); int maxMsgLength(); long nextCorrelationId(); AtomicBuffer buffer(); void consumerHeartbeatTime(final long time); long consumerHeartbeatTime(); long producerPosition(); long consumerPosition(); int size(); boolean unblock(); int tryClaim(final int msgTypeId, final int length); void commit(final int index); void abort(final int index); }
OneToOneRingBuffer implements RingBuffer { public int read(final MessageHandler handler) { return read(handler, Integer.MAX_VALUE); } OneToOneRingBuffer(final AtomicBuffer buffer); int capacity(); boolean write(final int msgTypeId, final DirectBuffer srcBuffer, final int offset, final int length); int read(final MessageHandler handler); int read(final MessageHandler handler, final int messageCountLimit); int maxMsgLength(); long nextCorrelationId(); AtomicBuffer buffer(); void consumerHeartbeatTime(final long time); long consumerHeartbeatTime(); long producerPosition(); long consumerPosition(); int size(); boolean unblock(); int tryClaim(final int msgTypeId, final int length); void commit(final int index); void abort(final int index);  }