@Test public void shouldThrowExceptionWhenMessageTypeIdInvalid() { final int invalidMsgId = -1; final UnsafeBuffer srcBuffer = new UnsafeBuffer(new byte[1024]); assertThrows(IllegalArgumentException.class, () -> broadcastTransmitter.transmit(invalidMsgId, srcBuffer, 0, 32)); }
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);  }