import org.agrona.BitUtil; //导入依赖的package包/类
@Test
public void shouldUpdatePositionInWrite() throws Exception
{
    int totalWrittenBytes = 0;

    for (int i = 0; i < MESSAGE_COUNT; i++)
    {
        MESSAGE_BUFFER.putInt(i);
        MESSAGE_BUFFER.flip();
        final int writtenBytes = journalling.write(MESSAGE_BUFFER);
        totalWrittenBytes += writtenBytes;
        MESSAGE_BUFFER.clear();

        assertTrue(writtenBytes == BitUtil.SIZE_OF_INT);
        assertTrue(journalling.position() == totalWrittenBytes);
    }
}
 

import org.agrona.BitUtil; //导入依赖的package包/类
@Test
public void shouldTriggerSnapshotDepletionHandlerOnJournalDepletion() throws Exception
{
    final boolean[] lastJournalReached = { false };

    journalling.depletionHandler((journalling) -> lastJournalReached[0] = true);

    int totalWrittenBytes = 0;
    while (!lastJournalReached[0])
    {
        MESSAGE_BUFFER.putInt(0);
        MESSAGE_BUFFER.flip();
        final int writtenBytes = journalling.write(MESSAGE_BUFFER);
        totalWrittenBytes += writtenBytes;
        MESSAGE_BUFFER.clear();

        assertTrue(writtenBytes == BitUtil.SIZE_OF_INT);
    }

    assertTrue(totalWrittenBytes == (JOURNAL_FILE_SIZE+BitUtil.SIZE_OF_INT));

    journalling.depletionHandler(null);
}
 

import org.agrona.BitUtil; //导入依赖的package包/类
public EncodedData2ObjectHashMap(
        final int initialCapacity,
        final float loadFactor,
        final int keyLengthInBytes,
        final BiConsumer<K, ByteBuffer> keyEncoder,
        final ToIntFunction<ByteBuffer> hashFunction,
        final K nullKey)
{
    this.capacity = BitUtil.findNextPositivePowerOfTwo(initialCapacity);
    this.loadFactor = loadFactor;
    this.keyLengthInBytes = keyLengthInBytes;
    this.keyBuffer = ByteBuffer.allocate(keyLengthInBytes);
    this.nullKeyBuffer = ByteBuffer.allocate(keyLengthInBytes);
    this.keyEncoder = keyEncoder;
    this.keySpace = ByteBuffer.allocate(keyLengthInBytes * capacity);
    this.values = new Object[capacity];
    this.hashFunction = hashFunction;

    keyEncoder.accept(nullKey, nullKeyBuffer);
    fillKeySpaceWithNullKey(keySpace, capacity);
    if(loadFactor >= 1f)
    {
        throw new IllegalArgumentException("loadFactor must be < 1");
    }
    resizeThreshold = (int) (capacity * loadFactor);
}
 

import org.agrona.BitUtil; //导入依赖的package包/类
@SuppressWarnings("SameReturnValue")
private boolean validateFragment1(final UnsafeBuffer buffer, final int offset, final int length)
{
    final DataHeaderFlyweight headerFlyweight = new DataHeaderFlyweight();
    headerFlyweight.wrap(buffer.addressOffset() + offset - HEADER_LENGTH, HEADER_LENGTH);
    if (headerFlyweight.headerType() == HDR_TYPE_PAD)
    {
        return true;
    }

    final int expectedLength = messageLengths[messageCount] - HEADER_LENGTH;
    assertThat("on fragment[" + messageCount + "]", length, is(expectedLength));
    assertThat(buffer.getInt(offset), is(messageCount));
    assertThat(buffer.getByte(offset + 4), is((byte)'z'));

    remaining -= BitUtil.align(messageLengths[messageCount], FrameDescriptor.FRAME_ALIGNMENT);
    messageCount++;

    return true;
}
 

import org.agrona.BitUtil; //导入依赖的package包/类
/**
 * Check that term length is valid and alignment is valid.
 *
 * @param termLength to be checked.
 * @throws IllegalStateException if the length is not as expected.
 */
public static void checkTermLength(final int termLength)
{
    if (termLength < TERM_MIN_LENGTH)
    {
        throw new IllegalStateException(
            "Term length less than min length of " + TERM_MIN_LENGTH + ": length=" + termLength);
    }

    if (termLength > TERM_MAX_LENGTH)
    {
        throw new IllegalStateException(
            "Term length more than max length of " + TERM_MAX_LENGTH + ": length=" + termLength);
    }

    if (!BitUtil.isPowerOfTwo(termLength))
    {
        throw new IllegalStateException("Term length not a power of 2: length=" + termLength);
    }
}
 

import org.agrona.BitUtil; //导入依赖的package包/类
/**
 * Check that page size is valid and alignment is valid.
 *
 * @param pageSize to be checked.
 * @throws IllegalStateException if the size is not as expected.
 */
public static void checkPageSize(final int pageSize)
{
    if (pageSize < PAGE_MIN_SIZE)
    {
        throw new IllegalStateException(
            "Page size less than min size of " + PAGE_MIN_SIZE + ": page size=" + pageSize);
    }

    if (pageSize > PAGE_MAX_SIZE)
    {
        throw new IllegalStateException(
            "Page size more than max size of " + PAGE_MAX_SIZE + ": page size=" + pageSize);
    }

    if (!BitUtil.isPowerOfTwo(pageSize))
    {
        throw new IllegalStateException("Page size not a power of 2: page size=" + pageSize);
    }
}
 

import org.agrona.BitUtil; //导入依赖的package包/类
@Test
public void shouldInsertLastFrameIntoBuffer()
{
    final int frameLength = BitUtil.align(256, FRAME_ALIGNMENT);
    final int srcOffset = 0;
    final int tail = TERM_BUFFER_CAPACITY - frameLength;
    final int termOffset = tail;
    final UnsafeBuffer packet = new UnsafeBuffer(ByteBuffer.allocateDirect(frameLength));
    packet.putShort(typeOffset(srcOffset), (short)PADDING_FRAME_TYPE, LITTLE_ENDIAN);
    packet.putInt(srcOffset, frameLength, LITTLE_ENDIAN);

    TermRebuilder.insert(termBuffer, termOffset, packet, frameLength);

    verify(termBuffer).putBytes(
        tail + HEADER_LENGTH, packet, srcOffset + HEADER_LENGTH, frameLength - HEADER_LENGTH);
}
 

import org.agrona.BitUtil; //导入依赖的package包/类
@Test
public void shouldFillAfterAGap()
{
    final int frameLength = 50;
    final int alignedFrameLength = BitUtil.align(frameLength, FRAME_ALIGNMENT);
    final int srcOffset = 0;
    final UnsafeBuffer packet = new UnsafeBuffer(ByteBuffer.allocateDirect(alignedFrameLength));
    final int termOffset = alignedFrameLength * 2;

    TermRebuilder.insert(termBuffer, termOffset, packet, alignedFrameLength);

    verify(termBuffer).putBytes(
        (alignedFrameLength * 2) + HEADER_LENGTH,
        packet,
        srcOffset + HEADER_LENGTH,
        alignedFrameLength - HEADER_LENGTH);
}
 

import org.agrona.BitUtil; //导入依赖的package包/类
@Test
public void shouldFillGapButNotMoveTailOrHwm()
{
    final int frameLength = 50;
    final int alignedFrameLength = BitUtil.align(frameLength, FRAME_ALIGNMENT);
    final int srcOffset = 0;
    final UnsafeBuffer packet = new UnsafeBuffer(ByteBuffer.allocateDirect(alignedFrameLength));
    final int termOffset = alignedFrameLength * 2;

    TermRebuilder.insert(termBuffer, termOffset, packet, alignedFrameLength);

    verify(termBuffer).putBytes(
        (alignedFrameLength * 2) + HEADER_LENGTH,
        packet,
        srcOffset + HEADER_LENGTH,
        alignedFrameLength - HEADER_LENGTH);
}
 

import org.agrona.BitUtil; //导入依赖的package包/类
@Test
public void shouldReadBlockOfThreeMessagesThatFillBuffer()
{
    final int offset = 0;
    final int limit = termBuffer.capacity();
    final int messageLength = 50;
    final int alignedMessageLength = BitUtil.align(messageLength, FRAME_ALIGNMENT);
    final int thirdMessageLength = limit - (alignedMessageLength * 2);

    when(termBuffer.getIntVolatile(lengthOffset(offset))).thenReturn(messageLength);
    when(termBuffer.getIntVolatile(lengthOffset(alignedMessageLength))).thenReturn(messageLength);
    when(termBuffer.getIntVolatile(lengthOffset(alignedMessageLength * 2))).thenReturn(thirdMessageLength);

    final int newOffset = TermBlockScanner.scan(termBuffer, offset, limit);

    assertThat(newOffset, is(limit));
}
 

import org.agrona.BitUtil; //导入依赖的package包/类
@Test
public void shouldReadBlockOfTwoMessagesBecauseOfLimit()
{
    final int offset = 0;
    final int messageLength = 50;
    final int alignedMessageLength = BitUtil.align(messageLength, FRAME_ALIGNMENT);
    final int limit = (alignedMessageLength * 2) + 1;

    when(termBuffer.getIntVolatile(lengthOffset(offset))).thenReturn(messageLength);
    when(termBuffer.getIntVolatile(lengthOffset(alignedMessageLength))).thenReturn(messageLength);
    when(termBuffer.getIntVolatile(lengthOffset(alignedMessageLength * 2))).thenReturn(messageLength);

    final int newOffset = TermBlockScanner.scan(termBuffer, offset, limit);

    assertThat(newOffset, is(alignedMessageLength * 2));
}
 

import org.agrona.BitUtil; //导入依赖的package包/类
/**
 * Validate that page size is valid and alignment is valid.
 *
 * @param pageSize to be checked.
 * @throws ConfigurationException if the size is not as expected.
 */
static void validatePageSize(final int pageSize)
{
    if (pageSize < PAGE_MIN_SIZE)
    {
        throw new ConfigurationException(
            "Page size less than min size of " + PAGE_MIN_SIZE + ": " + pageSize);
    }

    if (pageSize > PAGE_MAX_SIZE)
    {
        throw new ConfigurationException(
            "Page size greater than max size of " + PAGE_MAX_SIZE + ": " + pageSize);
    }

    if (!BitUtil.isPowerOfTwo(pageSize))
    {
        throw new ConfigurationException("Page size not a power of 2: " + pageSize);
    }
}
 

import org.agrona.BitUtil; //导入依赖的package包/类
@Theory
public void shouldGetByteArrayFromBuffer(final AtomicBuffer buffer)
{
    final byte[] testArray = { 'H', 'e', 'l', 'l', 'o' };

    int i = INDEX;
    for (final byte v : testArray)
    {
        buffer.putByte(i, v);
        i += BitUtil.SIZE_OF_BYTE;
    }

    final byte[] result = new byte[testArray.length];
    buffer.getBytes(INDEX, result);

    assertThat(result, is(testArray));
}
 

import org.agrona.BitUtil; //导入依赖的package包/类
public void run()
{
    try
    {
        barrier.await();
    }
    catch (final Exception ignore)
    {
    }

    final int length = BitUtil.SIZE_OF_INT * 2;
    final int repsValueOffset = BitUtil.SIZE_OF_INT;
    final UnsafeBuffer srcBuffer = new UnsafeBuffer(new byte[1024]);

    srcBuffer.putInt(0, producerId);

    for (int i = 0; i < reps; i++)
    {
        srcBuffer.putInt(repsValueOffset, i);

        while (!ringBuffer.write(MSG_TYPE_ID, srcBuffer, 0, length))
        {
            Thread.yield();
        }
    }
}
 

import org.agrona.BitUtil; //导入依赖的package包/类
/**
 * Get a media driver context
 * for sending ndarrays
 * based on a given length
 * where length is the length (number of elements)
 * in the ndarrays hat are being sent
 * @param length the length to based the ipc length
 * @return the media driver context based on the given length
 */
public static MediaDriver.Context getMediaDriverContext(int length) {
    //length of array * sizeof(float)
    int ipcLength = length * 16;
    //padding for NDArrayMessage
    ipcLength += 64;
    //must be a power of 2
    ipcLength *= 2;
    //ipc length must be positive power of 2
    while (!BitUtil.isPowerOfTwo(ipcLength))
        ipcLength += 2;
    // System.setProperty("aeron.term.buffer.size",String.valueOf(ipcLength));
    final MediaDriver.Context ctx =
                    new MediaDriver.Context().threadingMode(ThreadingMode.SHARED).dirsDeleteOnStart(true)
                                    /*  .ipcTermBufferLength(ipcLength)
                                      .publicationTermBufferLength(ipcLength)
                                      .maxTermBufferLength(ipcLength)*/
                                    .conductorIdleStrategy(new BusySpinIdleStrategy())
                                    .receiverIdleStrategy(new BusySpinIdleStrategy())
                                    .senderIdleStrategy(new BusySpinIdleStrategy());
    return ctx;
}
 

import org.agrona.BitUtil; //导入依赖的package包/类
protected AbstractJournalling(final long fileSize, final int pageSize, final JournalAllocator<T> journalAllocator)
{
    Check.enforce(fileSize > 0, "Journal file size must be positive");
    Check.enforce(BitUtil.isPowerOfTwo(pageSize), "Journal page size must be power of 2");
    Check.enforce(fileSize % pageSize == 0, "Journal file size must be multiple of journal page size");

    this.fileSize = fileSize;
    this.pageSize = pageSize;
    this.journalAllocator = journalAllocator;
}
 

import org.agrona.BitUtil; //导入依赖的package包/类
public static void checkZeroedAllocation(final Path journalPath, final long journalSize)
{
    try
    {
        FileChannel journal = FileChannel.open(journalPath, StandardOpenOption.READ);
        assertThat(journal.size(), is(journalSize));

        final ByteBuffer readBuffer = ByteBuffer.allocate(JournalAllocator.BLOCK_SIZE);

        while (journal.read(readBuffer) != -1)
        {
            readBuffer.flip();

            while (readBuffer.remaining() >= BitUtil.SIZE_OF_INT)
            {
                int readInt = readBuffer.getInt();
                assertThat(readInt, anyOf(is(0xDEADCAFE), is(0x00000000)));
            }

            readBuffer.clear();
        }

        journal.close();
    }
    catch (IOException e)
    {
        throw new UncheckedIOException(e);
    }
}
 

import org.agrona.BitUtil; //导入依赖的package包/类
@Test
public void shouldSupportNullSnapshotDepletionHandlerOnJournalDepletion() throws Exception
{
    int totalWrittenBytes = 0;
    while (totalWrittenBytes < (JOURNAL_FILE_SIZE + BitUtil.SIZE_OF_INT))
    {
        MESSAGE_BUFFER.putInt(0);
        MESSAGE_BUFFER.flip();
        final int writtenBytes = journalling.write(MESSAGE_BUFFER);
        totalWrittenBytes += writtenBytes;
        MESSAGE_BUFFER.clear();

        assertTrue(writtenBytes == BitUtil.SIZE_OF_INT);
    }
}
 

import org.agrona.BitUtil; //导入依赖的package包/类
@Test
public void shouldAllocateCacheAlignedDirectBufferWithSpecifiedCapacity()
{
    final ByteBuffer buffer = DirectBufferAllocator.allocateCacheAligned(CAPACITY);

    assertTrue(buffer.isDirect());
    assertThat(buffer.capacity(), is(CAPACITY));

    final long address = ((sun.nio.ch.DirectBuffer)buffer).address();
    assertTrue(BitUtil.isAligned(address, BitUtil.CACHE_LINE_LENGTH));
}
 

import org.agrona.BitUtil; //导入依赖的package包/类
static void checkIndexFileSize(final int indexFileSize)
{
    if (!BitUtil.isPowerOfTwo(recordCapacity(indexFileSize)))
    {
        throw new IllegalStateException(
            "IndexFileSize must be a positive power of 2 + INITIAL_RECORD_OFFSET: indexFileSize=" + indexFileSize);
    }
}