@Test public void shouldTransmitIntoEmptyBuffer() { final long tail = 0L; final int recordOffset = (int)tail; final int length = 8; final int recordLength = length + HEADER_LENGTH; final int recordLengthAligned = align(recordLength, RECORD_ALIGNMENT); final UnsafeBuffer srcBuffer = new UnsafeBuffer(new byte[1024]); final int srcIndex = 0; broadcastTransmitter.transmit(MSG_TYPE_ID, srcBuffer, srcIndex, length); final InOrder inOrder = inOrder(buffer); inOrder.verify(buffer).getLong(TAIL_COUNTER_INDEX); inOrder.verify(buffer).putLongOrdered(TAIL_INTENT_COUNTER_OFFSET, tail + recordLengthAligned); inOrder.verify(buffer).putInt(lengthOffset(recordOffset), recordLength); inOrder.verify(buffer).putInt(typeOffset(recordOffset), MSG_TYPE_ID); inOrder.verify(buffer).putBytes(msgOffset(recordOffset), srcBuffer, srcIndex, length); inOrder.verify(buffer).putLongOrdered(LATEST_COUNTER_INDEX, tail); inOrder.verify(buffer).putLongOrdered(TAIL_COUNTER_INDEX, tail + recordLengthAligned); }
public void transmit(final int msgTypeId, final DirectBuffer srcBuffer, final int srcIndex, final int length) { checkTypeId(msgTypeId); checkMessageLength(length); final AtomicBuffer buffer = this.buffer; long currentTail = buffer.getLong(tailCounterIndex); int recordOffset = (int)currentTail & (capacity - 1); final int recordLength = HEADER_LENGTH + length; final int recordLengthAligned = BitUtil.align(recordLength, RECORD_ALIGNMENT); final long newTail = currentTail + recordLengthAligned; final int toEndOfBuffer = capacity - recordOffset; if (toEndOfBuffer < recordLengthAligned) { signalTailIntent(buffer, newTail + toEndOfBuffer); insertPaddingRecord(buffer, recordOffset, toEndOfBuffer); currentTail += toEndOfBuffer; recordOffset = 0; } else { signalTailIntent(buffer, newTail); } buffer.putInt(lengthOffset(recordOffset), recordLength); buffer.putInt(typeOffset(recordOffset), msgTypeId); buffer.putBytes(msgOffset(recordOffset), srcBuffer, srcIndex, length); buffer.putLongOrdered(latestCounterIndex, currentTail); buffer.putLongOrdered(tailCounterIndex, currentTail + recordLengthAligned); }
BroadcastTransmitter { public void transmit(final int msgTypeId, final DirectBuffer srcBuffer, final int srcIndex, final int length) { checkTypeId(msgTypeId); checkMessageLength(length); final AtomicBuffer buffer = this.buffer; long currentTail = buffer.getLong(tailCounterIndex); int recordOffset = (int)currentTail & (capacity - 1); final int recordLength = HEADER_LENGTH + length; final int recordLengthAligned = BitUtil.align(recordLength, RECORD_ALIGNMENT); final long newTail = currentTail + recordLengthAligned; final int toEndOfBuffer = capacity - recordOffset; if (toEndOfBuffer < recordLengthAligned) { signalTailIntent(buffer, newTail + toEndOfBuffer); insertPaddingRecord(buffer, recordOffset, toEndOfBuffer); currentTail += toEndOfBuffer; recordOffset = 0; } else { signalTailIntent(buffer, newTail); } buffer.putInt(lengthOffset(recordOffset), recordLength); buffer.putInt(typeOffset(recordOffset), msgTypeId); buffer.putBytes(msgOffset(recordOffset), srcBuffer, srcIndex, length); buffer.putLongOrdered(latestCounterIndex, currentTail); buffer.putLongOrdered(tailCounterIndex, currentTail + recordLengthAligned); } }
BroadcastTransmitter { public void transmit(final int msgTypeId, final DirectBuffer srcBuffer, final int srcIndex, final int length) { checkTypeId(msgTypeId); checkMessageLength(length); final AtomicBuffer buffer = this.buffer; long currentTail = buffer.getLong(tailCounterIndex); int recordOffset = (int)currentTail & (capacity - 1); final int recordLength = HEADER_LENGTH + length; final int recordLengthAligned = BitUtil.align(recordLength, RECORD_ALIGNMENT); final long newTail = currentTail + recordLengthAligned; final int toEndOfBuffer = capacity - recordOffset; if (toEndOfBuffer < recordLengthAligned) { signalTailIntent(buffer, newTail + toEndOfBuffer); insertPaddingRecord(buffer, recordOffset, toEndOfBuffer); currentTail += toEndOfBuffer; recordOffset = 0; } else { signalTailIntent(buffer, newTail); } buffer.putInt(lengthOffset(recordOffset), recordLength); buffer.putInt(typeOffset(recordOffset), msgTypeId); buffer.putBytes(msgOffset(recordOffset), srcBuffer, srcIndex, length); buffer.putLongOrdered(latestCounterIndex, currentTail); buffer.putLongOrdered(tailCounterIndex, currentTail + recordLengthAligned); } BroadcastTransmitter(final AtomicBuffer buffer); }
BroadcastTransmitter { public void transmit(final int msgTypeId, final DirectBuffer srcBuffer, final int srcIndex, final int length) { checkTypeId(msgTypeId); checkMessageLength(length); final AtomicBuffer buffer = this.buffer; long currentTail = buffer.getLong(tailCounterIndex); int recordOffset = (int)currentTail & (capacity - 1); final int recordLength = HEADER_LENGTH + length; final int recordLengthAligned = BitUtil.align(recordLength, RECORD_ALIGNMENT); final long newTail = currentTail + recordLengthAligned; final int toEndOfBuffer = capacity - recordOffset; if (toEndOfBuffer < recordLengthAligned) { signalTailIntent(buffer, newTail + toEndOfBuffer); insertPaddingRecord(buffer, recordOffset, toEndOfBuffer); currentTail += toEndOfBuffer; recordOffset = 0; } else { signalTailIntent(buffer, newTail); } buffer.putInt(lengthOffset(recordOffset), recordLength); buffer.putInt(typeOffset(recordOffset), msgTypeId); buffer.putBytes(msgOffset(recordOffset), srcBuffer, srcIndex, length); buffer.putLongOrdered(latestCounterIndex, currentTail); buffer.putLongOrdered(tailCounterIndex, currentTail + recordLengthAligned); } BroadcastTransmitter(final AtomicBuffer buffer); int capacity(); int maxMsgLength(); void transmit(final int msgTypeId, final DirectBuffer srcBuffer, final int srcIndex, final int length); }
BroadcastTransmitter { public void transmit(final int msgTypeId, final DirectBuffer srcBuffer, final int srcIndex, final int length) { checkTypeId(msgTypeId); checkMessageLength(length); final AtomicBuffer buffer = this.buffer; long currentTail = buffer.getLong(tailCounterIndex); int recordOffset = (int)currentTail & (capacity - 1); final int recordLength = HEADER_LENGTH + length; final int recordLengthAligned = BitUtil.align(recordLength, RECORD_ALIGNMENT); final long newTail = currentTail + recordLengthAligned; final int toEndOfBuffer = capacity - recordOffset; if (toEndOfBuffer < recordLengthAligned) { signalTailIntent(buffer, newTail + toEndOfBuffer); insertPaddingRecord(buffer, recordOffset, toEndOfBuffer); currentTail += toEndOfBuffer; recordOffset = 0; } else { signalTailIntent(buffer, newTail); } buffer.putInt(lengthOffset(recordOffset), recordLength); buffer.putInt(typeOffset(recordOffset), msgTypeId); buffer.putBytes(msgOffset(recordOffset), srcBuffer, srcIndex, length); buffer.putLongOrdered(latestCounterIndex, currentTail); buffer.putLongOrdered(tailCounterIndex, currentTail + recordLengthAligned); } BroadcastTransmitter(final AtomicBuffer buffer); int capacity(); int maxMsgLength(); void transmit(final int msgTypeId, final DirectBuffer srcBuffer, final int srcIndex, final int length);  }