@Test public void shouldInsertPaddingAndWriteToBuffer() { final int padding = 200; final int messageLength = 400; final int recordLength = messageLength + HEADER_LENGTH; final int alignedRecordLength = align(recordLength, ALIGNMENT); final long tail = 2 * CAPACITY - padding; final long head = tail; final long headCache = CAPACITY + 300; when(buffer.getLongVolatile(HEAD_COUNTER_INDEX)).thenReturn(head); when(buffer.getLongVolatile(TAIL_COUNTER_INDEX)).thenReturn(tail); when(buffer.getLongVolatile(HEAD_COUNTER_CACHE_INDEX)).thenReturn(headCache); when(buffer.compareAndSetLong(TAIL_COUNTER_INDEX, tail, tail + alignedRecordLength + padding)) .thenReturn(true); final UnsafeBuffer srcBuffer = new UnsafeBuffer(new byte[messageLength]); assertTrue(ringBuffer.write(MSG_TYPE_ID, srcBuffer, 0, messageLength)); }
public boolean write(final int msgTypeId, final DirectBuffer srcBuffer, final int offset, 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 false; } buffer.putIntOrdered(lengthOffset(recordIndex), -recordLength); UnsafeAccess.UNSAFE.storeFence(); buffer.putInt(typeOffset(recordIndex), msgTypeId); buffer.putBytes(encodedMsgOffset(recordIndex), srcBuffer, offset, length); buffer.putIntOrdered(lengthOffset(recordIndex), recordLength); return true; }
ManyToOneRingBuffer implements RingBuffer { public boolean write(final int msgTypeId, final DirectBuffer srcBuffer, final int offset, 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 false; } buffer.putIntOrdered(lengthOffset(recordIndex), -recordLength); UnsafeAccess.UNSAFE.storeFence(); buffer.putInt(typeOffset(recordIndex), msgTypeId); buffer.putBytes(encodedMsgOffset(recordIndex), srcBuffer, offset, length); buffer.putIntOrdered(lengthOffset(recordIndex), recordLength); return true; } }
ManyToOneRingBuffer implements RingBuffer { public boolean write(final int msgTypeId, final DirectBuffer srcBuffer, final int offset, 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 false; } buffer.putIntOrdered(lengthOffset(recordIndex), -recordLength); UnsafeAccess.UNSAFE.storeFence(); buffer.putInt(typeOffset(recordIndex), msgTypeId); buffer.putBytes(encodedMsgOffset(recordIndex), srcBuffer, offset, length); buffer.putIntOrdered(lengthOffset(recordIndex), recordLength); return true; } ManyToOneRingBuffer(final AtomicBuffer buffer); }
ManyToOneRingBuffer implements RingBuffer { public boolean write(final int msgTypeId, final DirectBuffer srcBuffer, final int offset, 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 false; } buffer.putIntOrdered(lengthOffset(recordIndex), -recordLength); UnsafeAccess.UNSAFE.storeFence(); buffer.putInt(typeOffset(recordIndex), msgTypeId); buffer.putBytes(encodedMsgOffset(recordIndex), srcBuffer, offset, length); buffer.putIntOrdered(lengthOffset(recordIndex), recordLength); return true; } 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 boolean write(final int msgTypeId, final DirectBuffer srcBuffer, final int offset, 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 false; } buffer.putIntOrdered(lengthOffset(recordIndex), -recordLength); UnsafeAccess.UNSAFE.storeFence(); buffer.putInt(typeOffset(recordIndex), msgTypeId); buffer.putBytes(encodedMsgOffset(recordIndex), srcBuffer, offset, length); buffer.putIntOrdered(lengthOffset(recordIndex), recordLength); return true; } 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);  }