import io.aeron.logbuffer.FragmentHandler; //导入依赖的package包/类
public void forEachFragment(
    final StreamIdentifier streamId,
    final FragmentHandler handler,
    final ErrorHandler errorHandler)
{
    final UnsafeBuffer termBuffer = new UnsafeBuffer(0, 0);
    for (final File logFile : directoryDescriptor.listLogFiles(streamId))
    {
        final ByteBuffer byteBuffer = LoggerUtil.mapExistingFile(logFile);
        if (byteBuffer.capacity() > 0)
        {
            termBuffer.wrap(byteBuffer);
            final int initialTermId = LogBufferDescriptor.initialTermId(termBuffer);
            final Header header = new Header(initialTermId, termBuffer.capacity());
            TermReader.read(
                termBuffer,
                0,
                handler,
                Integer.MAX_VALUE,
                header,
                errorHandler);
        }
    }
}
 

import io.aeron.logbuffer.FragmentHandler; //导入依赖的package包/类
private static void consume(final Subscription subscription, final int count, final String prefix)
{
    final MutableInteger received = new MutableInteger(0);

    final FragmentHandler fragmentHandler = new FragmentAssembler(
        (buffer, offset, length, header) ->
        {
            final String expected = prefix + received.value;
            final String actual = buffer.getStringWithoutLengthAscii(offset, length);

            assertEquals(expected, actual);

            received.value++;
        });

    while (received.value < count)
    {
        if (0 == subscription.poll(fragmentHandler, FRAGMENT_LIMIT))
        {
            Thread.yield();
        }
    }

    assertThat(received.get(), is(count));
}
 

import io.aeron.logbuffer.FragmentHandler; //导入依赖的package包/类
private static void consume(
    final Subscription subscription, final int startIndex, final int count, final String prefix)
{
    final MutableInteger received = new MutableInteger(startIndex);

    final FragmentHandler fragmentHandler = new FragmentAssembler(
        (buffer, offset, length, header) ->
        {
            final String expected = prefix + received.value;
            final String actual = buffer.getStringWithoutLengthAscii(offset, length);

            assertEquals(expected, actual);

            received.value++;
        });

    while (received.value < (startIndex + count))
    {
        if (0 == subscription.poll(fragmentHandler, FRAGMENT_LIMIT))
        {
            Thread.yield();
        }
    }

    assertThat(received.get(), is(startIndex + count));
}
 

import io.aeron.logbuffer.FragmentHandler; //导入依赖的package包/类
private void verifyData(final UnsafeBuffer srcBuffer, final FragmentHandler mockFragmentHandler)
{
    final ArgumentCaptor<DirectBuffer> bufferArg = ArgumentCaptor.forClass(DirectBuffer.class);
    final ArgumentCaptor<Integer> offsetArg = ArgumentCaptor.forClass(Integer.class);

    verify(mockFragmentHandler, times(1)).onFragment(
        bufferArg.capture(), offsetArg.capture(), eq(srcBuffer.capacity()), any(Header.class));

    final DirectBuffer capturedBuffer = bufferArg.getValue();
    final int offset = offsetArg.getValue();
    for (int i = 0; i < srcBuffer.capacity(); i++)
    {
        final int index = offset + i;
        assertThat("same at " + index, capturedBuffer.getByte(index), is(srcBuffer.getByte(i)));
    }
}
 

import io.aeron.logbuffer.FragmentHandler; //导入依赖的package包/类
@Test
public void shouldReadData()
{
    subscription.addImage(imageOneMock);

    when(imageOneMock.poll(any(FragmentHandler.class), anyInt())).then(
        (invocation) ->
        {
            final FragmentHandler handler = (FragmentHandler)invocation.getArguments()[0];
            handler.onFragment(atomicReadBuffer, HEADER_LENGTH, READ_BUFFER_CAPACITY - HEADER_LENGTH, header);

            return 1;
        });

    assertThat(subscription.poll(fragmentHandler, FRAGMENT_COUNT_LIMIT), is(1));
    verify(fragmentHandler).onFragment(
        eq(atomicReadBuffer),
        eq(HEADER_LENGTH),
        eq(READ_BUFFER_CAPACITY - HEADER_LENGTH),
        any(Header.class));
}
 

import io.aeron.logbuffer.FragmentHandler; //导入依赖的package包/类
/**
 * Return a reusable, parameterized {@link FragmentHandler} that prints to stdout for the first stream(STREAM)
 *
 * @param streamId to show when printing
 * @return subscription data handler function that prints the message contents
 */
public static FragmentHandler reassembledStringMessage1(final int streamId)
{
    return (buffer, offset, length, header) ->
    {
        final byte[] data = new byte[length];
        buffer.getBytes(offset, data);

        System.out.format(
            "message to stream %d from session %x term id %x term offset %d (%[email protected]%d)%n",
            streamId, header.sessionId(), header.termId(), header.termOffset(), length, offset);

        if (length != 10000)
        {
            System.out.format(
                "Received message was not assembled properly;" +
                " received length was %d, but was expecting 10000%n",
                length);
        }
    };
}
 

import io.aeron.logbuffer.FragmentHandler; //导入依赖的package包/类
/**
 * Return a reusable, parameterized {@link FragmentHandler} that prints to stdout for the second stream (STREAM + 1)
 *
 * @param streamId to show when printing
 * @return subscription data handler function that prints the message contents
 */
public static FragmentHandler reassembledStringMessage2(final int streamId)
{
    return (buffer, offset, length, header) ->
    {
        final byte[] data = new byte[length];
        buffer.getBytes(offset, data);

        System.out.format(
            "message to stream %d from session %x term id %x term offset %d (%[email protected]%d)%n",
            streamId, header.sessionId(), header.termId(), header.termOffset(), length, offset);

        if (length != 9000)
        {
            System.out.format(
                "Received message was not assembled properly; received length was %d, but was expecting 9000%n",
                length);
        }
    };
}
 

import io.aeron.logbuffer.FragmentHandler; //导入依赖的package包/类
private static void run(final Subscription subscription, final long warmupCount, final long measuredCount) {
    final NanoClock clock = new SystemNanoClock();
    final Histogram histogram = new Histogram(1, 1000000000, 3);
    final MutableMarketDataSnapshot snapshot = new MutableMarketDataSnapshot();
    final UnsafeBuffer unsafeBuffer = new UnsafeBuffer(0, 0);
    final AtomicLong t0 = new AtomicLong();
    final AtomicLong t1 = new AtomicLong();
    final AtomicLong t2 = new AtomicLong();
    final long n = warmupCount + measuredCount;
    final AtomicLong count = new AtomicLong();
    final FragmentHandler fh = (buf, offset, len, header) -> {
        if (count.get() == 0) t0.set(clock.nanoTime());
        else if (count.get() == warmupCount-1) t1.set(clock.nanoTime());
        else if (count.get() == n-1) t2.set(clock.nanoTime());
        unsafeBuffer.wrap(buf, offset, len);
        final MarketDataSnapshot decoded = SerializerHelper.decode(unsafeBuffer, snapshot.builder());
        final long time = clock.nanoTime();
        if (count.incrementAndGet() <= n) {
            histogram.recordValue(time - decoded.getEventTimestamp());
        }
        if (count.get() == warmupCount) {
            histogram.reset();
        }
    };
    while (count.get() < n) {
        subscription.poll(fh, 256);
    }
    final long c = count.get();
    System.out.println((t2.get() - t0.get())/1000.0 + " us total receiving time (" + (t2.get() - t0.get())/(1000f*c) + " us/message, " + c/((t2.get()-t0.get())/1000000000f) + " messages/second)");
    System.out.println();
    HistogramPrinter.printHistogram(histogram);
}
 

import io.aeron.logbuffer.FragmentHandler; //导入依赖的package包/类
private void pollForFragment(
    final Subscription subscription, final FragmentHandler handler, final MutableInteger fragmentsRead)
{
    SystemTestHelper.executeUntil(
        () -> fragmentsRead.get() > 0,
        (j) ->
        {
            fragmentsRead.value += subscription.poll(handler, 10);
            Thread.yield();
        },
        Integer.MAX_VALUE,
        TimeUnit.MILLISECONDS.toNanos(500));
}
 

import io.aeron.logbuffer.FragmentHandler; //导入依赖的package包/类
private void verifyFragments(final FragmentHandler fragmentHandler, final int numMessagesToSend)
{
    verify(fragmentHandler, times(numMessagesToSend)).onFragment(
        any(DirectBuffer.class),
        anyInt(),
        eq(MESSAGE_LENGTH),
        any(Header.class));
}
 

import io.aeron.logbuffer.FragmentHandler; //导入依赖的package包/类
/**
 * Construct an adapter to reassemble message fragments and delegate on whole messages.
 *
 * @param delegate            onto which whole messages are forwarded.
 * @param initialBufferLength to be used for each session.
 * @param isDirectByteBuffer  is the underlying buffer to be a direct {@link java.nio.ByteBuffer}?
 */
public FragmentAssembler(
    final FragmentHandler delegate, final int initialBufferLength, final boolean isDirectByteBuffer)
{
    this.initialBufferLength = initialBufferLength;
    this.delegate = delegate;
    this.isDirectByteBuffer = isDirectByteBuffer;
}
 

import io.aeron.logbuffer.FragmentHandler; //导入依赖的package包/类
public static void main(final String[] args)
{
    System.out.println("Subscribing to " + CHANNEL + " on stream Id " + STREAM_ID);

    final FragmentHandler fragmentHandler = SamplesUtil.printStringMessage(STREAM_ID);
    final AtomicBoolean running = new AtomicBoolean(true);

    // Register a SIGINT handler for graceful shutdown.
    SigInt.register(() -> running.set(false));

    // Create an Aeron instance using the configured Context and create a
    // Subscription on that instance that subscribes to the configured
    // channel and stream ID.
    // The Aeron and Subscription classes implement "AutoCloseable" and will automatically
    // clean up resources when this try block is finished
    try (AeronArchive archive = AeronArchive.connect())
    {
        final long recordingId = findLatestRecording(archive, CHANNEL, STREAM_ID);
        final long position = 0L;
        final long length = Long.MAX_VALUE;

        try (Subscription subscription = archive.replay(recordingId, position, length, CHANNEL, REPLAY_STREAM_ID))
        {
            SamplesUtil.subscriberLoop(fragmentHandler, FRAGMENT_COUNT_LIMIT, running).accept(subscription);

            System.out.println("Shutting down...");
        }
    }
}
 

import io.aeron.logbuffer.FragmentHandler; //导入依赖的package包/类
private static void roundTripMessages(
    final FragmentHandler fragmentHandler,
    final Publication publication,
    final Subscription subscription,
    final long count)
{
    while (!subscription.isConnected())
    {
        Thread.yield();
    }

    final Image image = subscription.imageAtIndex(0);

    for (long i = 0; i < count; i++)
    {
        long offeredPosition;

        do
        {
            ATOMIC_BUFFER.putLong(0, System.nanoTime());
        }
        while ((offeredPosition = publication.offer(ATOMIC_BUFFER, 0, MESSAGE_LENGTH)) < 0L);

        POLLING_IDLE_STRATEGY.reset();

        do
        {
            while (image.poll(fragmentHandler, FRAGMENT_COUNT_LIMIT) <= 0)
            {
                POLLING_IDLE_STRATEGY.idle();
            }
        }
        while (image.position() < offeredPosition);
    }
}
 

import io.aeron.logbuffer.FragmentHandler; //导入依赖的package包/类
public static void main(final String[] args)
{
    System.out.println("Subscribing to " + CHANNEL + " on stream Id " + STREAM_ID);

    final MediaDriver driver = EMBEDDED_MEDIA_DRIVER ? MediaDriver.launchEmbedded() : null;
    final Aeron.Context ctx = new Aeron.Context()
        .availableImageHandler(SamplesUtil::printAvailableImage)
        .unavailableImageHandler(SamplesUtil::printUnavailableImage);

    if (EMBEDDED_MEDIA_DRIVER)
    {
        ctx.aeronDirectoryName(driver.aeronDirectoryName());
    }

    final FragmentHandler fragmentHandler = SamplesUtil.printStringMessage(STREAM_ID);
    final AtomicBoolean running = new AtomicBoolean(true);

    // Register a SIGINT handler for graceful shutdown.
    SigInt.register(() -> running.set(false));

    // Create an Aeron instance using the configured Context and create a
    // Subscription on that instance that subscribes to the configured
    // channel and stream ID.
    // The Aeron and Subscription classes implement "AutoCloseable" and will automatically
    // clean up resources when this try block is finished
    try (Aeron aeron = Aeron.connect(ctx);
        Subscription subscription = aeron.addSubscription(CHANNEL, STREAM_ID))
    {
        SamplesUtil.subscriberLoop(fragmentHandler, FRAGMENT_COUNT_LIMIT, running).accept(subscription);

        System.out.println("Shutting down...");
    }

    CloseHelper.quietClose(driver);
}
 

import io.aeron.logbuffer.FragmentHandler; //导入依赖的package包/类
private static void roundTripMessages(
    final FragmentHandler fragmentHandler,
    final Publication pingPublication,
    final Subscription pongSubscription,
    final long numMessages)
{
    while (!pongSubscription.isConnected())
    {
        Thread.yield();
    }

    final Image image = pongSubscription.imageAtIndex(0);

    for (long i = 0; i < numMessages; i++)
    {
        long offeredPosition;

        do
        {
            ATOMIC_BUFFER.putLong(0, System.nanoTime());
        }
        while ((offeredPosition = pingPublication.offer(ATOMIC_BUFFER, 0, MESSAGE_LENGTH)) < 0L);

        PONG_HANDLER_IDLE_STRATEGY.reset();
        do
        {
            while (image.poll(fragmentHandler, FRAGMENT_COUNT_LIMIT) <= 0)
            {
                PONG_HANDLER_IDLE_STRATEGY.idle();
            }
        }
        while (image.position() < offeredPosition);
    }
}
 

import io.aeron.logbuffer.FragmentHandler; //导入依赖的package包/类
/**
 * Return a reusable, parametrised event loop that calls and idler when no messages are received
 *
 * @param fragmentHandler    to be called back for each message.
 * @param endOfStreamHandler to be called back when the stream has been completely received.
 * @param limit              passed to {@link Subscription#poll(FragmentHandler, int)}
 * @param running            indication for loop
 * @param idleStrategy       to use for loop
 * @return loop function
 */
public static Consumer<Subscription> subscriberLoop(
    final FragmentHandler fragmentHandler,
    final EndOfStreamHandler endOfStreamHandler,
    final int limit,
    final AtomicBoolean running,
    final IdleStrategy idleStrategy)
{
    return
        (subscription) ->
        {
            try
            {
                boolean reachedEos = false;

                while (running.get())
                {
                    final int fragmentsRead = subscription.poll(fragmentHandler, limit);

                    if (0 == fragmentsRead)
                    {
                        if (!reachedEos && subscription.pollEndOfStreams(endOfStreamHandler) > 0)
                        {
                            reachedEos = true;
                        }
                    }

                    idleStrategy.idle(fragmentsRead);
                }
            }
            catch (final Exception ex)
            {
                LangUtil.rethrowUnchecked(ex);
            }
        };
}
 

import io.aeron.logbuffer.FragmentHandler; //导入依赖的package包/类
/**
 * Return a reusable, parametrised {@link FragmentHandler} that prints to stdout
 *
 * @param streamId to show when printing
 * @return subscription data handler function that prints the message contents
 */
public static FragmentHandler printStringMessage(final int streamId)
{
    return (buffer, offset, length, header) ->
    {
        final byte[] data = new byte[length];
        buffer.getBytes(offset, data);

        System.out.println(String.format(
            "Message to stream %d from session %d (%[email protected]%d) <<%s>>",
            streamId, header.sessionId(), length, offset, new String(data)));
    };
}
 

import io.aeron.logbuffer.FragmentHandler; //导入依赖的package包/类
/**
 * Return a reusable, parameterized event
 * loop that calls and idler
 * when no messages are received
 *
 * @param fragmentHandler to be called back for each message.
 * @param limit           passed to {@link Subscription#poll(FragmentHandler, int)}
 * @param running         indication for loop
 * @param idleStrategy    to use for loop
 * @return loop function
 */
public static Consumer<Subscription> subscriberLoop(final FragmentHandler fragmentHandler, final int limit,
                final AtomicBoolean running, final IdleStrategy idleStrategy, final AtomicBoolean launched) {
    return (subscription) -> {
        try {
            while (running.get()) {
                idleStrategy.idle(subscription.poll(fragmentHandler, limit));
                launched.set(true);
            }
        } catch (final Exception ex) {
            LangUtil.rethrowUnchecked(ex);
        }
    };
}
 

import io.aeron.logbuffer.FragmentHandler; //导入依赖的package包/类
/**
 * Return a reusable, parameterized {@link FragmentHandler} that prints to stdout
 *
 * @param streamId to show when printing
 * @return subscription data handler function that prints the message contents
 */
public static FragmentHandler printStringMessage(final int streamId) {
    return (buffer, offset, length, header) -> {
        final byte[] data = new byte[length];
        buffer.getBytes(offset, data);

        System.out.println(String.format("Message to stream %d from session %d (%[email protected]%d) <<%s>>", streamId,
                        header.sessionId(), length, offset, new String(data)));
    };
}
 

import io.aeron.logbuffer.FragmentHandler; //导入依赖的package包/类
static void poll(final Subscription subscription, final FragmentHandler handler)
{
    await(() -> subscription.poll(handler, 1) > 0);
}