<commit_before>/*
 * Copyright 2014-2018 Real Logic Ltd.
 *
 * Licensed under the Apache License, Version 2.0 (the "License");
 * you may not use this file except in compliance with the License.
 * You may obtain a copy of the License at
 *
 * http://www.apache.org/licenses/LICENSE-2.0
 *
 * Unless required by applicable law or agreed to in writing, software
 * distributed under the License is distributed on an "AS IS" BASIS,
 * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
 * See the License for the specific language governing permissions and
 * limitations under the License.
 */
package io.aeron.cluster;

import io.aeron.*;
import io.aeron.archive.client.AeronArchive;
import io.aeron.archive.client.ControlResponsePoller;
import io.aeron.archive.codecs.ControlResponseCode;
import io.aeron.archive.codecs.SourceLocation;
import io.aeron.archive.status.RecordingPos;
import io.aeron.cluster.client.ClusterException;
import io.aeron.cluster.codecs.MessageHeaderDecoder;
import io.aeron.cluster.codecs.SnapshotMarkerDecoder;
import io.aeron.cluster.codecs.SnapshotRecordingsDecoder;
import io.aeron.logbuffer.ControlledFragmentHandler;
import io.aeron.logbuffer.Header;
import org.agrona.CloseHelper;
import org.agrona.DirectBuffer;
import org.agrona.concurrent.status.CountersReader;

import java.util.ArrayList;
import java.util.concurrent.TimeUnit;

import static io.aeron.Aeron.NULL_VALUE;
import static io.aeron.CommonContext.ENDPOINT_PARAM_NAME;
import static io.aeron.archive.client.AeronArchive.NULL_LENGTH;
import static io.aeron.cluster.ConsensusModule.Configuration.SNAPSHOT_TYPE_ID;

public class DynamicJoin implements AutoCloseable
{
    enum State
    {
        INIT,
        PASSIVE_FOLLOWER,
        SNAPSHOT_RETRIEVE,
        SNAPSHOT_LOAD,
        JOIN_CLUSTER,
        DONE
    }

    private final AeronArchive localArchive;
    private final MemberStatusAdapter memberStatusAdapter;
    private final MemberStatusPublisher memberStatusPublisher;
    private final ConsensusModule.Context ctx;
    private final ConsensusModuleAgent consensusModuleAgent;
    private final String[] clusterMembersStatusEndpoints;
    private final String memberEndpoints;
    private final String memberStatusEndpoint;
    private final String transferEndpoint;
    private final ArrayList<RecordingLog.Snapshot> leaderSnapshots = new ArrayList<>();
    private final long intervalMs;

    private ExclusivePublication clusterPublication;
    private State state = State.INIT;
    private ClusterMember[] clusterMembers;
    private ClusterMember leaderMember;
    private AeronArchive.AsyncConnect leaderArchiveAsyncConnect;
    private AeronArchive leaderArchive;
    private Subscription snapshotRetrieveSubscription;
    private Image snapshotRetrieveImage;
    private SnapshotReader snapshotReader;
    private Counter recoveryStateCounter;
    private long timeOfLastActivity = 0;
    private long correlationId = NULL_VALUE;
    private long snapshotRetrieveSubscriptionId = NULL_VALUE;
    private int memberId = NULL_VALUE;
    private int highMemberId = NULL_VALUE;
    private int clusterMembersSattusEndpointsCursor = 0;
    private int recordingIdCursor = 0;
    private int snapshotReplaySessionId = NULL_VALUE;

    public DynamicJoin(
        final String clusterMembersStatusEndpoints,
        final AeronArchive localArchive,
        final MemberStatusAdapter memberStatusAdapter,
        final MemberStatusPublisher memberStatusPublisher,
        final ConsensusModule.Context ctx,
        final ConsensusModuleAgent consensusModuleAgent)
    {
        final ClusterMember thisMember = ClusterMember.parseEndpoints(-1, ctx.memberEndpoints());

        this.localArchive = localArchive;
        this.memberStatusAdapter = memberStatusAdapter;
        this.memberStatusPublisher = memberStatusPublisher;
        this.ctx = ctx;
        this.consensusModuleAgent = consensusModuleAgent;
        this.intervalMs = TimeUnit.NANOSECONDS.toMillis(ctx.dynamicJoinIntervalNs());
        this.memberEndpoints = ctx.memberEndpoints();
        this.memberStatusEndpoint = thisMember.memberFacingEndpoint();
        this.transferEndpoint = thisMember.transferEndpoint();
        this.clusterMembersStatusEndpoints = clusterMembersStatusEndpoints.split(",");

        final ChannelUri memberStatusUri = ChannelUri.parse(ctx.memberStatusChannel());
        memberStatusUri.put(
            ENDPOINT_PARAM_NAME, this.clusterMembersStatusEndpoints[clusterMembersSattusEndpointsCursor]);
        clusterPublication = ctx.aeron().addExclusivePublication(
            memberStatusUri.toString(), ctx.memberStatusStreamId());
    }

    public void close()
    {
        CloseHelper.close(clusterPublication);
        CloseHelper.close(snapshotRetrieveSubscription);
    }

    boolean isDone()
    {
        return State.DONE == state;
    }

    int doWork(final long nowMs)
    {
        int workCount = 0;
        workCount += memberStatusAdapter.poll();

        switch (state)
        {
            case INIT:
                workCount += init(nowMs);
                break;

            case PASSIVE_FOLLOWER:
                workCount += passiveFollower(nowMs);
                break;

            case SNAPSHOT_RETRIEVE:
                workCount += snapshotRetrieve(nowMs);
                break;

            case SNAPSHOT_LOAD:
                workCount += snapshotLoad(nowMs);
                break;

            case JOIN_CLUSTER:
                workCount += joinCluster(nowMs);
                break;
        }

        return workCount;
    }

    public void onClusterMembersChange(
        final long correlationId, final int leaderMemberId, final String activeMembers, final String passiveMembers)
    {
        if (State.INIT == state && correlationId == this.correlationId)
        {
            final ClusterMember[] passiveFollowers = ClusterMember.parse(passiveMembers);

            for (final ClusterMember follower : passiveFollowers)
            {
                if (memberStatusEndpoint.equals(follower.memberFacingEndpoint()))
                {
                    memberId = follower.id();

                    clusterMembers = ClusterMember.parse(activeMembers);
                    leaderMember = ClusterMember.findMember(clusterMembers, leaderMemberId);

                    if (null != leaderMember)
                    {
                        if (!leaderMember.memberFacingEndpoint().equals(
                            clusterMembersStatusEndpoints[clusterMembersSattusEndpointsCursor]))
                        {
                            clusterPublication.close();

                            final ChannelUri memberStatusUri = ChannelUri.parse(ctx.memberStatusChannel());
                            memberStatusUri.put(ENDPOINT_PARAM_NAME, leaderMember.memberFacingEndpoint());
                            clusterPublication = ctx.aeron().addExclusivePublication(
                                memberStatusUri.toString(), ctx.memberStatusStreamId());
                        }

                        final ChannelUri leaderArchiveUri = ChannelUri.parse(
                            ctx.archiveContext().controlRequestChannel());
                        final ChannelUri localArchiveUri = ChannelUri.parse(
                            ctx.archiveContext().controlResponseChannel());
                        leaderArchiveUri.put(ENDPOINT_PARAM_NAME, leaderMember.archiveEndpoint());

                        final AeronArchive.Context leaderArchiveCtx = new AeronArchive.Context()
                            .aeron(ctx.aeron())
                            .controlRequestChannel(leaderArchiveUri.toString())
                            .controlRequestStreamId(ctx.archiveContext().controlResponseStreamId())
                            .controlResponseChannel(localArchiveUri.toString())
                            .controlResponseStreamId(ctx.archiveContext().controlResponseStreamId());

                        leaderArchiveAsyncConnect = AeronArchive.asyncConnect(leaderArchiveCtx);

                        timeOfLastActivity = 0;
                        state = State.PASSIVE_FOLLOWER;
                    }
                    break;
                }
            }
        }
    }

    public void onSnapshotRecordings(
        final long correlationId, final SnapshotRecordingsDecoder snapshotRecordingsDecoder)
    {
        if (State.PASSIVE_FOLLOWER == state && correlationId == this.correlationId)
        {
            final SnapshotRecordingsDecoder.SnapshotsDecoder snapshotsDecoder = snapshotRecordingsDecoder.snapshots();

            if (snapshotsDecoder.count() > 0)
            {
                for (final SnapshotRecordingsDecoder.SnapshotsDecoder snapshot : snapshotsDecoder)
                {
                    if (snapshot.serviceId() <= ctx.serviceCount())
                    {
                        leaderSnapshots.add(new RecordingLog.Snapshot(
                            snapshot.recordingId(),
                            snapshot.leadershipTermId(),
                            snapshot.termBaseLogPosition(),
                            snapshot.logPosition(),
                            snapshot.timestamp(),
                            snapshot.serviceId()));
                    }
                }
            }

            timeOfLastActivity = 0;
            recordingIdCursor = 0;
            this.correlationId = NULL_VALUE;
            state = leaderSnapshots.isEmpty() ? State.JOIN_CLUSTER : State.SNAPSHOT_RETRIEVE;
        }
    }

    private int init(final long nowMs)
    {
        if (nowMs > (timeOfLastActivity + intervalMs))
        {
            correlationId = ctx.aeron().nextCorrelationId();

            if (memberStatusPublisher.addClusterMember(clusterPublication, correlationId, memberEndpoints))
            {
                timeOfLastActivity = nowMs;
                return 1;
            }
        }
        return 0;
    }

    private int passiveFollower(final long nowms)
    {
        if (nowms > (timeOfLastActivity + intervalMs))
        {
            correlationId = ctx.aeron().nextCorrelationId();

            if (memberStatusPublisher.snapshotRecordingQuery(clusterPublication, correlationId, memberId))
            {
                timeOfLastActivity = nowms;
                return 1;
            }
        }
        return 0;
    }

    private int snapshotRetrieve(final long nowMs)
    {
        int workCount = 0;

        if (null == leaderArchive)
        {
            leaderArchive = leaderArchiveAsyncConnect.poll();
            return (null == leaderArchive) ? 0 : 1;
        }

        if (null != snapshotReader)
        {<commit_msg>[Java] Naming.<commit_after>/*
 * Copyright 2014-2018 Real Logic Ltd.
 *
 * Licensed under the Apache License, Version 2.0 (the "License");
 * you may not use this file except in compliance with the License.
 * You may obtain a copy of the License at
 *
 * http://www.apache.org/licenses/LICENSE-2.0
 *
 * Unless required by applicable law or agreed to in writing, software
 * distributed under the License is distributed on an "AS IS" BASIS,
 * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
 * See the License for the specific language governing permissions and
 * limitations under the License.
 */
package io.aeron.cluster;

import io.aeron.*;
import io.aeron.archive.client.AeronArchive;
import io.aeron.archive.client.ControlResponsePoller;
import io.aeron.archive.codecs.ControlResponseCode;
import io.aeron.archive.codecs.SourceLocation;
import io.aeron.archive.status.RecordingPos;
import io.aeron.cluster.client.ClusterException;
import io.aeron.cluster.codecs.MessageHeaderDecoder;
import io.aeron.cluster.codecs.SnapshotMarkerDecoder;
import io.aeron.cluster.codecs.SnapshotRecordingsDecoder;
import io.aeron.logbuffer.ControlledFragmentHandler;
import io.aeron.logbuffer.Header;
import org.agrona.CloseHelper;
import org.agrona.DirectBuffer;
import org.agrona.concurrent.status.CountersReader;

import java.util.ArrayList;
import java.util.concurrent.TimeUnit;

import static io.aeron.Aeron.NULL_VALUE;
import static io.aeron.CommonContext.ENDPOINT_PARAM_NAME;
import static io.aeron.archive.client.AeronArchive.NULL_LENGTH;
import static io.aeron.cluster.ConsensusModule.Configuration.SNAPSHOT_TYPE_ID;

public class DynamicJoin implements AutoCloseable
{
    enum State
    {
        INIT,
        PASSIVE_FOLLOWER,
        SNAPSHOT_RETRIEVE,
        SNAPSHOT_LOAD,
        JOIN_CLUSTER,
        DONE
    }

    private final AeronArchive localArchive;
    private final MemberStatusAdapter memberStatusAdapter;
    private final MemberStatusPublisher memberStatusPublisher;
    private final ConsensusModule.Context ctx;
    private final ConsensusModuleAgent consensusModuleAgent;
    private final String[] clusterMemberStatusEndpoints;
    private final String memberEndpoints;
    private final String memberStatusEndpoint;
    private final String transferEndpoint;
    private final ArrayList<RecordingLog.Snapshot> leaderSnapshots = new ArrayList<>();
    private final long intervalMs;

    private ExclusivePublication clusterPublication;
    private State state = State.INIT;
    private ClusterMember[] clusterMembers;
    private ClusterMember leaderMember;
    private AeronArchive.AsyncConnect leaderArchiveAsyncConnect;
    private AeronArchive leaderArchive;
    private Subscription snapshotRetrieveSubscription;
    private Image snapshotRetrieveImage;
    private SnapshotReader snapshotReader;
    private Counter recoveryStateCounter;
    private long timeOfLastActivityMs = 0;
    private long correlationId = NULL_VALUE;
    private long snapshotRetrieveSubscriptionId = NULL_VALUE;
    private int memberId = NULL_VALUE;
    private int highMemberId = NULL_VALUE;
    private int clusterMembersStatusEndpointsCursor = 0;
    private int recordingIdCursor = 0;
    private int snapshotReplaySessionId = NULL_VALUE;

    public DynamicJoin(
        final String clusterMemberStatusEndpoints,
        final AeronArchive localArchive,
        final MemberStatusAdapter memberStatusAdapter,
        final MemberStatusPublisher memberStatusPublisher,
        final ConsensusModule.Context ctx,
        final ConsensusModuleAgent consensusModuleAgent)
    {
        final ClusterMember thisMember = ClusterMember.parseEndpoints(-1, ctx.memberEndpoints());

        this.localArchive = localArchive;
        this.memberStatusAdapter = memberStatusAdapter;
        this.memberStatusPublisher = memberStatusPublisher;
        this.ctx = ctx;
        this.consensusModuleAgent = consensusModuleAgent;
        this.intervalMs = TimeUnit.NANOSECONDS.toMillis(ctx.dynamicJoinIntervalNs());
        this.memberEndpoints = ctx.memberEndpoints();
        this.memberStatusEndpoint = thisMember.memberFacingEndpoint();
        this.transferEndpoint = thisMember.transferEndpoint();
        this.clusterMemberStatusEndpoints = clusterMemberStatusEndpoints.split(",");

        final ChannelUri memberStatusUri = ChannelUri.parse(ctx.memberStatusChannel());
        memberStatusUri.put(
            ENDPOINT_PARAM_NAME, this.clusterMemberStatusEndpoints[clusterMembersStatusEndpointsCursor]);
        clusterPublication = ctx.aeron().addExclusivePublication(
            memberStatusUri.toString(), ctx.memberStatusStreamId());
    }

    public void close()
    {
        CloseHelper.close(clusterPublication);
        CloseHelper.close(snapshotRetrieveSubscription);
    }

    boolean isDone()
    {
        return State.DONE == state;
    }

    int doWork(final long nowMs)
    {
        int workCount = 0;
        workCount += memberStatusAdapter.poll();

        switch (state)
        {
            case INIT:
                workCount += init(nowMs);
                break;

            case PASSIVE_FOLLOWER:
                workCount += passiveFollower(nowMs);
                break;

            case SNAPSHOT_RETRIEVE:
                workCount += snapshotRetrieve(nowMs);
                break;

            case SNAPSHOT_LOAD:
                workCount += snapshotLoad(nowMs);
                break;

            case JOIN_CLUSTER:
                workCount += joinCluster(nowMs);
                break;
        }

        return workCount;
    }

    public void onClusterMembersChange(
        final long correlationId, final int leaderMemberId, final String activeMembers, final String passiveMembers)
    {
        if (State.INIT == state && correlationId == this.correlationId)
        {
            final ClusterMember[] passiveFollowers = ClusterMember.parse(passiveMembers);

            for (final ClusterMember follower : passiveFollowers)
            {
                if (memberStatusEndpoint.equals(follower.memberFacingEndpoint()))
                {
                    memberId = follower.id();

                    clusterMembers = ClusterMember.parse(activeMembers);
                    leaderMember = ClusterMember.findMember(clusterMembers, leaderMemberId);

                    if (null != leaderMember)
                    {
                        if (!leaderMember.memberFacingEndpoint().equals(
                            clusterMemberStatusEndpoints[clusterMembersStatusEndpointsCursor]))
                        {
                            clusterPublication.close();

                            final ChannelUri memberStatusUri = ChannelUri.parse(ctx.memberStatusChannel());
                            memberStatusUri.put(ENDPOINT_PARAM_NAME, leaderMember.memberFacingEndpoint());
                            clusterPublication = ctx.aeron().addExclusivePublication(
                                memberStatusUri.toString(), ctx.memberStatusStreamId());
                        }

                        final ChannelUri leaderArchiveUri = ChannelUri.parse(
                            ctx.archiveContext().controlRequestChannel());
                        final ChannelUri localArchiveUri = ChannelUri.parse(
                            ctx.archiveContext().controlResponseChannel());
                        leaderArchiveUri.put(ENDPOINT_PARAM_NAME, leaderMember.archiveEndpoint());

                        final AeronArchive.Context leaderArchiveCtx = new AeronArchive.Context()
                            .aeron(ctx.aeron())
                            .controlRequestChannel(leaderArchiveUri.toString())
                            .controlRequestStreamId(ctx.archiveContext().controlResponseStreamId())
                            .controlResponseChannel(localArchiveUri.toString())
                            .controlResponseStreamId(ctx.archiveContext().controlResponseStreamId());

                        leaderArchiveAsyncConnect = AeronArchive.asyncConnect(leaderArchiveCtx);

                        timeOfLastActivityMs = 0;
                        state = State.PASSIVE_FOLLOWER;
                    }
                    break;
                }
            }
        }
    }

    public void onSnapshotRecordings(
        final long correlationId, final SnapshotRecordingsDecoder snapshotRecordingsDecoder)
    {
        if (State.PASSIVE_FOLLOWER == state && correlationId == this.correlationId)
        {
            final SnapshotRecordingsDecoder.SnapshotsDecoder snapshotsDecoder = snapshotRecordingsDecoder.snapshots();

            if (snapshotsDecoder.count() > 0)
            {
                for (final SnapshotRecordingsDecoder.SnapshotsDecoder snapshot : snapshotsDecoder)
                {
                    if (snapshot.serviceId() <= ctx.serviceCount())
                    {
                        leaderSnapshots.add(new RecordingLog.Snapshot(
                            snapshot.recordingId(),
                            snapshot.leadershipTermId(),
                            snapshot.termBaseLogPosition(),
                            snapshot.logPosition(),
                            snapshot.timestamp(),
                            snapshot.serviceId()));
                    }
                }
            }

            timeOfLastActivityMs = 0;
            recordingIdCursor = 0;
            this.correlationId = NULL_VALUE;
            state = leaderSnapshots.isEmpty() ? State.JOIN_CLUSTER : State.SNAPSHOT_RETRIEVE;
        }
    }

    private int init(final long nowMs)
    {
        if (nowMs > (timeOfLastActivityMs + intervalMs))
        {
            correlationId = ctx.aeron().nextCorrelationId();

            if (memberStatusPublisher.addClusterMember(clusterPublication, correlationId, memberEndpoints))
            {
                timeOfLastActivityMs = nowMs;
                return 1;
            }
        }
        return 0;
    }

    private int passiveFollower(final long nowMs)
    {
        if (nowMs > (timeOfLastActivityMs + intervalMs))
        {
            correlationId = ctx.aeron().nextCorrelationId();

            if (memberStatusPublisher.snapshotRecordingQuery(clusterPublication, correlationId, memberId))
            {
                timeOfLastActivityMs = nowMs;
                return 1;
            }
        }
        return 0;
    }

    private int snapshotRetrieve(final long nowMs)
    {
        int workCount = 0;

        if (null == leaderArchive)
        {
            leaderArchive = leaderArchiveAsyncConnect.poll();
            return (null == leaderArchive) ? 0 : 1;
        }

        if (null != snapshotReader)
        {
{'fraction_non_alphanumeric': 0.07126543929154043, 'fraction_numerical': 0.0026101141924959217, 'mean_word_length': 2.252387448840382, 'pattern_counts': {'":': 0, '<': 9, '<?xml version=': 0, '>': 13, 'https://': 0, 'lorem ipsum': 0, 'www.': 2, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 0, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '3313846', 'n_tokens_mistral': 5368, 'n_tokens_neox': 5064, 'n_words': 1105}