@Test public void tryClaimReturnsIndexAtWhichEncodedMessageStartsAfterPadding() { final int length = 10; final int recordLength = length + HEADER_LENGTH; final int alignedRecordLength = align(recordLength, ALIGNMENT); final long headPosition = 248L; final int padding = 22; final long tailPosition = CAPACITY - padding; final int paddingIndex = (int)tailPosition; final int recordIndex = 0; when(buffer.getLongVolatile(HEAD_COUNTER_CACHE_INDEX)).thenReturn(headPosition); when(buffer.getLongVolatile(TAIL_COUNTER_INDEX)).thenReturn(tailPosition); when(buffer.compareAndSetLong(TAIL_COUNTER_INDEX, tailPosition, tailPosition + alignedRecordLength + padding)) .thenReturn(TRUE); final int index = ringBuffer.tryClaim(MSG_TYPE_ID, length); assertEquals(recordIndex + HEADER_LENGTH, index); final InOrder inOrder = inOrder(buffer); inOrder.verify(buffer).getLongVolatile(HEAD_COUNTER_CACHE_INDEX); inOrder.verify(buffer).getLongVolatile(TAIL_COUNTER_INDEX); inOrder.verify(buffer) .compareAndSetLong(TAIL_COUNTER_INDEX, tailPosition, tailPosition + alignedRecordLength + padding); inOrder.verify(buffer).putIntOrdered(lengthOffset(paddingIndex), -padding); inOrder.verify(buffer).putInt(typeOffset(paddingIndex), PADDING_MSG_TYPE_ID); inOrder.verify(buffer).putIntOrdered(lengthOffset(paddingIndex), padding); inOrder.verify(buffer).putIntOrdered(lengthOffset(recordIndex), -recordLength); inOrder.verify(buffer).putInt(typeOffset(recordIndex), MSG_TYPE_ID); inOrder.verifyNoMoreInteractions(); }
public int tryClaim(final int msgTypeId, final int length) { checkTypeId(msgTypeId); checkMsgLength(length); final AtomicBuffer buffer = this.buffer; final int recordLength = length + HEADER_LENGTH; final int recordIndex = claimCapacity(buffer, recordLength); if (INSUFFICIENT_CAPACITY == recordIndex) { return recordIndex; } buffer.putIntOrdered(lengthOffset(recordIndex), -recordLength); UnsafeAccess.UNSAFE.storeFence(); buffer.putInt(typeOffset(recordIndex), msgTypeId); return encodedMsgOffset(recordIndex); }
ManyToOneRingBuffer implements RingBuffer { public int tryClaim(final int msgTypeId, final int length) { checkTypeId(msgTypeId); checkMsgLength(length); final AtomicBuffer buffer = this.buffer; final int recordLength = length + HEADER_LENGTH; final int recordIndex = claimCapacity(buffer, recordLength); if (INSUFFICIENT_CAPACITY == recordIndex) { return recordIndex; } buffer.putIntOrdered(lengthOffset(recordIndex), -recordLength); UnsafeAccess.UNSAFE.storeFence(); buffer.putInt(typeOffset(recordIndex), msgTypeId); return encodedMsgOffset(recordIndex); } }
ManyToOneRingBuffer implements RingBuffer { public int tryClaim(final int msgTypeId, final int length) { checkTypeId(msgTypeId); checkMsgLength(length); final AtomicBuffer buffer = this.buffer; final int recordLength = length + HEADER_LENGTH; final int recordIndex = claimCapacity(buffer, recordLength); if (INSUFFICIENT_CAPACITY == recordIndex) { return recordIndex; } buffer.putIntOrdered(lengthOffset(recordIndex), -recordLength); UnsafeAccess.UNSAFE.storeFence(); buffer.putInt(typeOffset(recordIndex), msgTypeId); return encodedMsgOffset(recordIndex); } ManyToOneRingBuffer(final AtomicBuffer buffer); }
ManyToOneRingBuffer implements RingBuffer { public int tryClaim(final int msgTypeId, final int length) { checkTypeId(msgTypeId); checkMsgLength(length); final AtomicBuffer buffer = this.buffer; final int recordLength = length + HEADER_LENGTH; final int recordIndex = claimCapacity(buffer, recordLength); if (INSUFFICIENT_CAPACITY == recordIndex) { return recordIndex; } buffer.putIntOrdered(lengthOffset(recordIndex), -recordLength); UnsafeAccess.UNSAFE.storeFence(); buffer.putInt(typeOffset(recordIndex), msgTypeId); return encodedMsgOffset(recordIndex); } 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 tryClaim(final int msgTypeId, final int length) { checkTypeId(msgTypeId); checkMsgLength(length); final AtomicBuffer buffer = this.buffer; final int recordLength = length + HEADER_LENGTH; final int recordIndex = claimCapacity(buffer, recordLength); if (INSUFFICIENT_CAPACITY == recordIndex) { return recordIndex; } buffer.putIntOrdered(lengthOffset(recordIndex), -recordLength); UnsafeAccess.UNSAFE.storeFence(); buffer.putInt(typeOffset(recordIndex), msgTypeId); return encodedMsgOffset(recordIndex); } 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);  }