import io.aeron.Subscription; //导入依赖的package包/类
public static void main(final String... args) {
    final String aeronDirectoryName = args[0];
    final String channel = args[1];
    final int streamId = Integer.parseInt(args[2]);
    final long warmupCount = Long.parseLong(args[3]);
    final long measuredCount = Long.parseLong(args[4]);

    System.out.println("Started " + AeronSubscriber.class.getSimpleName() + ":");
    System.out.println("\twarmupCount       : " + warmupCount);
    System.out.println("\tmeasuredCount     : " + measuredCount);
    System.out.println("\tchannel           : " + channel);
    System.out.println("\tstreamId          : " + streamId);
    System.out.println();

    final Aeron aeron = aeron(aeronDirectoryName);
    final Subscription subscription = aeron.addSubscription(channel, streamId);
    try {
        run(subscription, warmupCount, measuredCount);
    } finally {
        subscription.close();
        aeron.close();
        System.out.println("Shutdown " + AeronSubscriber.class.getSimpleName() + "...");
    }
}
 

import io.aeron.Subscription; //导入依赖的package包/类
ClusterSubscription(
    final Subscription dataSubscription,
    final int clusterStreamId,
    final Subscription controlSubscription,
    final ArchiveReader archiveReader)
{
    this.controlSubscription = controlSubscription;
    this.archiveReader = archiveReader;
    // We use clusterStreamId as a reserved value filter
    if (clusterStreamId == NO_FILTER)
    {
        throw new IllegalArgumentException("ClusterStreamId must not be 0");
    }

    this.dataSubscription = dataSubscription;
    messageFilter = new MessageFilter(clusterStreamId);
    clusterHeader = new ClusterHeader(clusterStreamId);
}
 

import io.aeron.Subscription; //导入依赖的package包/类
void initialiseRoles(final Leader leader, final Candidate candidate, final Follower follower)
{
    final RaftPublication acknowledgementPublication = raftPublication(
        configuration.acknowledgementStream(), "acknowledgementPublication");
    final StreamIdentifier controlStream = configuration.controlStream();
    final RaftPublication controlPublication = raftPublication(controlStream, "controlStream");
    final Subscription controlSubscription = controlSubscription();

    leader
        .controlPublication(controlPublication)
        .controlSubscription(controlSubscription);

    candidate
        .controlPublication(controlPublication)
        .controlSubscription(controlSubscription);

    follower
        .controlPublication(controlPublication)
        .acknowledgementPublication(acknowledgementPublication)
        .controlSubscription(controlSubscription);
}
 

import io.aeron.Subscription; //导入依赖的package包/类
private Image replayImage(final String name, final int replaySessionId)
{
    final Subscription subscription = aeron.addSubscription(
        IPC_CHANNEL, OUTBOUND_REPLAY_STREAM);
    StreamInformation.print(name, subscription, configuration);

    // Await replay publication
    while (true)
    {
        final Image image = subscription.imageBySessionId(replaySessionId);
        if (image != null)
        {
            return image;
        }

        invokeAeronConductor();

        Thread.yield();
    }
}
 

import io.aeron.Subscription; //导入依赖的package包/类
static void awaitResponse(final Subscription controlResponse, final long expectedCorrelationId)
{
    final ControlResponseAdapter controlResponseAdapter = new ControlResponseAdapter(
        new FailControlResponseListener()
        {
            public void onResponse(
                final long controlSessionId,
                final long correlationId,
                final long relevantId,
                final ControlResponseCode code,
                final String errorMessage)
            {
                if (correlationId != expectedCorrelationId)
                {
                    throw new IllegalStateException(
                        "expected=" + expectedCorrelationId + " actual=" + correlationId);
                }
            }
        },
        controlResponse,
        1
    );

    await(() -> controlResponseAdapter.poll() != 0);
}
 

import io.aeron.Subscription; //导入依赖的package包/类
private Image mockImage(
    final int sessionId,
    final int initialTermId,
    final String sourceIdentity,
    final int termBufferLength,
    final Subscription subscription)
{
    final Image image = mock(Image.class);

    when(image.sessionId()).thenReturn(sessionId);
    when(image.initialTermId()).thenReturn(initialTermId);
    when(image.sourceIdentity()).thenReturn(sourceIdentity);
    when(image.termBufferLength()).thenReturn(termBufferLength);
    when(image.subscription()).thenReturn(subscription);
    when(image.mtuLength()).thenReturn(MTU_LENGTH);
    when(image.joinPosition()).thenReturn(START_POSITION);

    return image;
}
 

import io.aeron.Subscription; //导入依赖的package包/类
private void runRecordingEventPoller()
{
    try (Subscription subscription = aeron.addSubscription(
        AeronArchive.Configuration.recordingEventsChannel(),
        AeronArchive.Configuration.recordingEventsStreamId()))
    {
        final IdleStrategy idleStrategy = new BackoffIdleStrategy(10, 100, 1, 1);
        final RecordingEventsAdapter recordingEventsAdapter = new RecordingEventsAdapter(
            this, subscription, FRAGMENT_COUNT_LIMIT);

        while (isRunning)
        {
            idleStrategy.idle(recordingEventsAdapter.poll());
        }
    }
}
 

import io.aeron.Subscription; //导入依赖的package包/类
private static Thread startPong(final String embeddedDirName)
{
    return new Thread(() ->
    {
        System.out.println("Subscribing Ping at " + PING_CHANNEL + " on stream Id " + PING_STREAM_ID);
        System.out.println("Publishing Pong at " + PONG_CHANNEL + " on stream Id " + PONG_STREAM_ID);

        final Aeron.Context ctx = new Aeron.Context().aeronDirectoryName(embeddedDirName);

        try (Aeron aeron = Aeron.connect(ctx);
            Publication pongPublication = aeron.addPublication(PONG_CHANNEL, PONG_STREAM_ID);
            Subscription pingSubscription = aeron.addSubscription(PING_CHANNEL, PING_STREAM_ID))
        {
            final FragmentAssembler dataHandler = new FragmentAssembler(
                (buffer, offset, length, header) -> pingHandler(pongPublication, buffer, offset, length));

            while (RUNNING.get())
            {
                PING_HANDLER_IDLE_STRATEGY.idle(pingSubscription.poll(dataHandler, FRAME_COUNT_LIMIT));
            }

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

import io.aeron.Subscription; //导入依赖的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.Subscription; //导入依赖的package包/类
public static AeronReceiver<Command> commandReceiver(final DirectFactory directFactory,
                                                     final Subscription subscription) {
    return new AeronReceiver<>(subscription, new AeronHandler<Command>() {
        @Override
        protected Command message(DirectBuffer buffer, int offset, int length) {
            return directFactory.command();
        }
    });
}
 

import io.aeron.Subscription; //导入依赖的package包/类
public static AeronReceiver<Message> messageReceiver(final DirectFactory directFactory,
                                                     final Subscription subscription) {
    return new AeronReceiver<>(subscription, new AeronHandler<Message>() {
        @Override
        protected Message message(DirectBuffer buffer, int offset, int length) {
            final MessageType messageType = MessageType.readFrom(buffer, offset);
            return messageType.create(directFactory);
        }
    });
}
 

import io.aeron.Subscription; //导入依赖的package包/类
void injectLeaderSubscriptions(final Leader leader)
{
    final StreamIdentifier data = configuration.dataStream();
    final StreamIdentifier acknowledgement = configuration.acknowledgementStream();
    final Subscription leaderAcknowledgementSubscription = subscription(
        acknowledgement.channel(), acknowledgement.streamId(), "leaderAcknowledgementSubscription");

    final Subscription leaderDataSubscription = subscription(
        data.spyChannel(), data.streamId(), "leaderDataSubscription");

    leader
        .acknowledgementSubscription(leaderAcknowledgementSubscription)
        // The leader uses a spy subscription in order to avoid reading its own data
        .dataSubscription(leaderDataSubscription);
}
 

import io.aeron.Subscription; //导入依赖的package包/类
private void archiverSubscription(final Archiver archiver, final StreamIdentifier streamId)
{

    final Subscription subscription = aeron.addSubscription(streamId.channel(), streamId.streamId());
    StreamInformation.print("Archiver", subscription, configuration);
    archiver.subscription(subscription);
}
 

import io.aeron.Subscription; //导入依赖的package包/类
ClusterPositionSender(
    final Subscription outboundLibrarySubscription,
    final ClusterableSubscription outboundClusterSubscription,
    final GatewayPublication inboundLibraryPublication,
    final String agentNamePrefix)
{
    this.outboundLibrarySubscription = outboundLibrarySubscription;
    this.outboundClusterSubscription = outboundClusterSubscription;
    this.inboundLibraryPublication = inboundLibraryPublication;
    this.agentNamePrefix = agentNamePrefix;
}
 

import io.aeron.Subscription; //导入依赖的package包/类
public Subscription outboundLibrarySubscription(
    final String name, final UnavailableImageHandler unavailableImageHandler)
{
    final Subscription subscription = aeron.addSubscription(
        configuration.libraryAeronChannel(), OUTBOUND_LIBRARY_STREAM, null, unavailableImageHandler);
    StreamInformation.print(name, subscription, configuration);
    return subscription;
}
 

import io.aeron.Subscription; //导入依赖的package包/类
public static void print(
    final String name, final Subscription subscription, final boolean printAeronStreamIdentifiers)
{
    if (printAeronStreamIdentifiers)
    {
        System.out.printf(
            "%-40s - registrationId=%d,streamId=%d%n",
            name,
            subscription.registrationId(),
            subscription.streamId());
    }
}
 

import io.aeron.Subscription; //导入依赖的package包/类
protected Follower follower(
    final short id,
    final ClusterAgent clusterNode,
    final TermState termState)
{
    final ArchiveMetaData metaData = archiveMetaData(id);
    final Subscription subscription = dataSubscription();
    final StreamIdentifier streamId = new StreamIdentifier(subscription);
    final Archiver archiver = new Archiver(
        metaData,
        DEFAULT_LOGGER_CACHE_NUM_SETS,
        DEFAULT_LOGGER_CACHE_SET_SIZE,
        streamId,
        DEFAULT_NAME_PREFIX,
        mock(CompletionPosition.class))
        .subscription(subscription);

    return new Follower(
        id,
        clusterNode,
        0,
        TIMEOUT,
        termState,
        new RaftArchiver(termState.leaderSessionId(), archiver),
        NODE_STATE_BUFFER,
        nodeStateHandler)
        .controlSubscription(controlSubscription())
        .acknowledgementPublication(raftPublication(ClusterConfiguration.DEFAULT_ACKNOWLEDGEMENT_STREAM_ID))
        .controlPublication(raftPublication(ClusterConfiguration.DEFAULT_CONTROL_STREAM_ID))
        .follow(0);
}
 

import io.aeron.Subscription; //导入依赖的package包/类
/**
 * Create an adapter for a given subscription to an archive for control response messages.
 *
 * @param listener      to which responses are dispatched.
 * @param subscription  to poll for new events.
 * @param fragmentLimit to apply for each polling operation.
 */
public ControlResponseAdapter(
    final ControlResponseListener listener,
    final Subscription subscription,
    final int fragmentLimit)
{
    this.fragmentLimit = fragmentLimit;
    this.listener = listener;
    this.subscription = subscription;
}
 

import io.aeron.Subscription; //导入依赖的package包/类
/**
 * Create a poller for a given subscription to an archive for control response messages.
 *
 * @param subscription     to poll for new events.
 * @param fragmentLimit    to apply for each polling operation.
 * @param controlSessionId to filter the responses.
 */
public RecordingDescriptorPoller(
    final Subscription subscription, final int fragmentLimit, final long controlSessionId)
{
    this.subscription = subscription;
    this.fragmentLimit = fragmentLimit;
    this.controlSessionId = controlSessionId;
}
 

import io.aeron.Subscription; //导入依赖的package包/类
/**
 * Create a poller for a given subscription to an archive for recording events.
 *
 * @param listener      to which events are dispatched.
 * @param subscription  to poll for new events.
 * @param fragmentLimit to apply for each polling operation.
 */
public RecordingEventsAdapter(
    final RecordingEventsListener listener,
    final Subscription subscription,
    final int fragmentLimit)
{
    this.fragmentLimit = fragmentLimit;
    this.listener = listener;
    this.subscription = subscription;
}