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);
    }
@Test
    void shouldTransmitIntoEndOfBuffer()
    {
        final int length = 8;
        final int recordLength = length + HEADER_LENGTH;
        final int recordLengthAligned = align(recordLength, RECORD_ALIGNMENT);
        final long tail = CAPACITY - recordLengthAligned;
        final int recordOffset = (int)tail;


        when(buffer.getLong(TAIL_COUNTER_INDEX)).thenReturn(tail);

        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);
    }