/*
 * Copyright 2014-2020 Real Logic Limited.
 *
 * 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
 *
 * https://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 org.agrona.CloseHelper;

final class LogReplay
{
    private final long recordingId;
    private final long startPosition;
    private final long stopPosition;
    private final long leadershipTermId;
    private final int logSessionId;
    private final AeronArchive archive;
    private final ConsensusModuleAgent consensusModuleAgent;
    private final ConsensusModule.Context ctx;
    private final LogAdapter logAdapter;
    private final Subscription logSubscription;

    private long replaySessionId = Aeron.NULL_VALUE;
    private boolean isDone = false;

    LogReplay(
        final AeronArchive archive,
        final long recordingId,
        final long startPosition,
        final long stopPosition,
        final long leadershipTermId,
        final int logSessionId,
        final LogAdapter logAdapter,
        final ConsensusModule.Context ctx)
    {
        this.archive = archive;
        this.recordingId = recordingId;
        this.startPosition = startPosition;
        this.stopPosition = stopPosition;
        this.leadershipTermId = leadershipTermId;
        this.logSessionId = logSessionId;
        this.logAdapter = logAdapter;
        this.consensusModuleAgent = logAdapter.consensusModuleAgent();
        this.ctx = ctx;

        final ChannelUri channelUri = ChannelUri.parse(ctx.replayChannel());
        channelUri.put(CommonContext.SESSION_ID_PARAM_NAME, Integer.toString(logSessionId));
        logSubscription = ctx.aeron().addSubscription(channelUri.toString(), ctx.replayStreamId());
    }

    void close()
    {
        logAdapter.image(null);
        CloseHelper.close(ctx.countedErrorHandler(), logSubscription);
    }

    int doWork()
    {
        int workCount = 0;

        if (Aeron.NULL_VALUE == replaySessionId)
        {
            final String channel = logSubscription.channel();
            final int streamId = logSubscription.streamId();
            consensusModuleAgent.awaitServicesReady(
                channel, streamId, logSessionId, leadershipTermId, startPosition, stopPosition, true);

            final long length = stopPosition - startPosition;
            replaySessionId = archive.startReplay(recordingId, startPosition, length, channel, streamId);
            workCount += 1;
        }

        if (null == logAdapter.image())
        {
            final Image image = logSubscription.imageBySessionId((int)replaySessionId);
            if (null != image)
            {
                logAdapter.image(image);
                workCount += 1;
            }
        }
        else
        {
            workCount += consensusModuleAgent.replayLogPoll(logAdapter, stopPosition);
            if (logAdapter.position() >= stopPosition)
            {
                isDone = true;
                workCount += 1;
            }
        }

        return workCount;
    }

    boolean isDone()
    {
        return isDone;
    }
}

@@ -749,44 +749,8 @@
 y%0A%7B%0A
-    private final long recordingId;%0A
     
@@ -775,32 +775,32 @@
  startPosition;%0A
+
     private fina
@@ -901,48 +901,8 @@
 Id;%0A
-    private final AeronArchive archive;%0A
     
@@ -1111,47 +1111,36 @@
 ate 
-long replaySessionId = Aeron.NULL_VALUE
+boolean awaitServices = true
 ;%0A  
@@ -1483,80 +1483,8 @@
   %7B%0A
-        this.archive = archive;%0A        this.recordingId = recordingId;%0A
     
@@ -1973,93 +1973,449 @@
     
-logSubscription = ctx.aeron().addSubscription(channelUri.toString(), ctx.replayStream
+final String channel = channelUri.toString();%0A        final int streamId = ctx.replayStreamId();%0A        logSubscription = ctx.aeron().addSubscription(channel, streamId);%0A%0A        final long length = stopPosition - startPosition;%0A        final long correlationId = ctx.aeron().nextCorrelationId();%0A        archive.archiveProxy().replay(%0A            recordingId, startPosition, length, channel, streamId, correlationId, archive.controlSession
 Id()
@@ -2624,43 +2624,21 @@
 if (
-Aeron.NULL_VALUE == replaySessionId
+awaitServices
 )%0A  
@@ -2941,162 +2941,29 @@
     
-final long length = stopPosition - startPosition;%0A            replaySessionId = archive.startReplay(recordingId, startPosition, length, channel, streamId)
+awaitServices = false
 ;%0A  
@@ -3118,19 +3118,11 @@
 nId(
-(int)replay
+log
 Sess

Use an async replay command to the archive for LogReplay.