@Test public void shouldReadTwoMessages() { final int msgLength = 16; final int recordLength = HEADER_LENGTH + msgLength; final int alignedRecordLength = align(recordLength, ALIGNMENT); final long tail = alignedRecordLength * 2; final long head = 0L; final int headIndex = (int)head; when(buffer.getLong(HEAD_COUNTER_INDEX)).thenReturn(head); when(buffer.getInt(typeOffset(headIndex))).thenReturn(MSG_TYPE_ID); when(buffer.getIntVolatile(lengthOffset(headIndex))).thenReturn(recordLength); when(buffer.getInt(typeOffset(headIndex + alignedRecordLength))).thenReturn(MSG_TYPE_ID); when(buffer.getIntVolatile(lengthOffset(headIndex + alignedRecordLength))).thenReturn(recordLength); final int[] times = new int[1]; final MessageHandler handler = (msgTypeId, buffer, index, length) -> times[0]++; final int messagesRead = ringBuffer.read(handler); assertThat(messagesRead, is(2)); assertThat(times[0], is(2)); final InOrder inOrder = inOrder(buffer); inOrder.verify(buffer, times(1)).setMemory(headIndex, alignedRecordLength * 2, (byte)0); inOrder.verify(buffer, times(1)).putLongOrdered(HEAD_COUNTER_INDEX, tail); }
public int read(final MessageHandler handler) { return read(handler, Integer.MAX_VALUE); }
ManyToOneRingBuffer implements RingBuffer { public int read(final MessageHandler handler) { return read(handler, Integer.MAX_VALUE); } }
ManyToOneRingBuffer implements RingBuffer { public int read(final MessageHandler handler) { return read(handler, Integer.MAX_VALUE); } ManyToOneRingBuffer(final AtomicBuffer buffer); }
ManyToOneRingBuffer implements RingBuffer { public int read(final MessageHandler handler) { return read(handler, Integer.MAX_VALUE); } ManyToOneRingBuffer(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); }
ManyToOneRingBuffer implements RingBuffer { public int read(final MessageHandler handler) { return read(handler, Integer.MAX_VALUE); } ManyToOneRingBuffer(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);  }