<commit_before>/*
 * Copyright 2014-2017 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.archive;

import io.aeron.ExclusivePublication;
import io.aeron.Publication;
import io.aeron.archive.ArchiveConductor.ReplayPublicationSupplier;
import io.aeron.archive.codecs.ControlResponseCode;
import io.aeron.archive.codecs.RecordingDescriptorDecoder;
import io.aeron.logbuffer.ExclusiveBufferClaim;
import io.aeron.logbuffer.FrameDescriptor;
import io.aeron.protocol.DataHeaderFlyweight;
import org.agrona.CloseHelper;
import org.agrona.LangUtil;
import org.agrona.concurrent.EpochClock;
import org.agrona.concurrent.UnsafeBuffer;
import org.agrona.concurrent.status.AtomicCounter;

import java.io.File;

import static io.aeron.archive.Catalog.wrapDescriptorDecoder;
import static io.aeron.logbuffer.FrameDescriptor.frameFlags;
import static io.aeron.logbuffer.FrameDescriptor.frameType;
import static io.aeron.protocol.DataHeaderFlyweight.RESERVED_VALUE_OFFSET;
import static java.nio.ByteOrder.LITTLE_ENDIAN;

/**
 * A replay session with a client which works through the required request response flow and streaming of recorded data.
 * The {@link ArchiveConductor} will initiate a session on receiving a ReplayRequest
 * (see {@link io.aeron.archive.codecs.ReplayRequestDecoder}). The session will:
 * <ul>
 * <li>Validate request parameters and respond with appropriate error if unable to replay </li>
 * <li>Wait for replay subscription to connect to the requested replay publication. If no subscription appears within
 * {@link #CONNECT_TIMEOUT_MS} the session will terminate and respond will error.</li>
 * <li>Once the replay publication is connected send an OK response to control client</li>
 * <li>Stream recorded data into the replayPublication {@link ExclusivePublication}</li>
 * <li>If the replay is aborted part way through, send a ReplayAborted message and terminate.</li>
 * </ul>
 */
class ReplaySession implements Session, SimplifiedControlledFragmentHandler
{
    enum State
    {
        INIT, REPLAY, INACTIVE, CLOSED
    }

    /**
     * Timeout within which a replay connection needs to be established.
     */
    static final long CONNECT_TIMEOUT_MS = 5000;

    private static final int REPLAY_FRAGMENT_LIMIT = Archive.Configuration.replayFragmentLimit();

    private long connectDeadlineMs;
    private final long replaySessionId;
    private final long correlationId;
    private final ExclusiveBufferClaim bufferClaim = new ExclusiveBufferClaim();
    private final ExclusivePublication replayPublication;
    private final RecordingFragmentReader cursor;
    private ControlResponseProxy threadLocalControlResponseProxy;
    private final ControlSession controlSession;
    private final EpochClock epochClock;
    private State state = State.INIT;

    @SuppressWarnings("ConstantConditions")
    ReplaySession(
        final long replayPosition,
        final long replayLength,
        final ReplayPublicationSupplier supplier,
        final ControlSession controlSession,
        final File archiveDir,
        final ControlResponseProxy threadLocalControlResponseProxy,
        final long replaySessionId,
        final long correlationId,
        final EpochClock epochClock,
        final String replayChannel,
        final int replayStreamId,
        final UnsafeBuffer descriptorBuffer,
        final AtomicCounter recordingPosition)
    {
        this.controlSession = controlSession;
        this.threadLocalControlResponseProxy = threadLocalControlResponseProxy;
        this.replaySessionId = replaySessionId;
        this.correlationId = correlationId;
        this.epochClock = epochClock;
        this.connectDeadlineMs = epochClock.time() + CONNECT_TIMEOUT_MS;

        final RecordingDescriptorDecoder descriptorDecoder = new RecordingDescriptorDecoder();
        wrapDescriptorDecoder(descriptorDecoder, descriptorBuffer);

        final long startPosition = descriptorDecoder.startPosition();
        final int mtuLength = descriptorDecoder.mtuLength();
        final int termBufferLength = descriptorDecoder.termBufferLength();
        final int initialTermId = descriptorDecoder.initialTermId();

        if (replayPosition - startPosition < 0)
        {
            final String errorMessage = "requested replay start position(=" + replayPosition +
                ") is before recording start position(=" + startPosition + ")";
            closeOnError(new IllegalArgumentException(errorMessage), errorMessage);
            cursor = null;
            replayPublication = null;
            return;
        }

        final long stopPosition = descriptorDecoder.stopPosition();
        if (replayPosition - stopPosition >= 0)
        {
            final String errorMessage = "requested replay start position(=" + replayPosition +
                ") must be before current highest recorded position(=" + stopPosition + ")";
            closeOnError(new IllegalArgumentException(errorMessage), errorMessage);
            cursor = null;
            replayPublication = null;

            return;
        }

        RecordingFragmentReader cursor = null;
        try
        {
            cursor = new RecordingFragmentReader(
                descriptorDecoder,
                archiveDir,
                replayPosition,
                replayLength,
                recordingPosition);
        }
        catch (final Exception ex)
        {
            closeOnError(ex, "failed to open cursor on a recording because: " + ex.getMessage());
        }

        this.cursor = cursor;

        ExclusivePublication replayPublication = null;
        try
        {
            replayPublication = supplier.newReplayPublication(
                replayChannel,
                replayStreamId,
                cursor.fromPosition(),
                mtuLength,
                initialTermId,
                termBufferLength);
        }
        catch (final Exception ex)
        {
            closeOnError(ex, "failed to create replay publication because: " + ex.getMessage());
        }

        this.replayPublication = replayPublication;
        controlSession.sendOkResponse(correlationId, threadLocalControlResponseProxy);
    }

    public void close()
    {
        state = State.CLOSED;

        CloseHelper.quietClose(cursor);
        CloseHelper.quietClose(replayPublication);
    }

    public long sessionId()
    {
        return replaySessionId;
    }

    public int doWork()
    {
        int workCount = 0;

        switch (state)
        {
            case INIT:<commit_msg>[Java] Set connect timeout for after things have been setup.<commit_after>/*
 * Copyright 2014-2017 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.archive;

import io.aeron.ExclusivePublication;
import io.aeron.Publication;
import io.aeron.archive.ArchiveConductor.ReplayPublicationSupplier;
import io.aeron.archive.codecs.ControlResponseCode;
import io.aeron.archive.codecs.RecordingDescriptorDecoder;
import io.aeron.logbuffer.ExclusiveBufferClaim;
import io.aeron.logbuffer.FrameDescriptor;
import io.aeron.protocol.DataHeaderFlyweight;
import org.agrona.CloseHelper;
import org.agrona.LangUtil;
import org.agrona.concurrent.EpochClock;
import org.agrona.concurrent.UnsafeBuffer;
import org.agrona.concurrent.status.AtomicCounter;

import java.io.File;

import static io.aeron.archive.Catalog.wrapDescriptorDecoder;
import static io.aeron.logbuffer.FrameDescriptor.frameFlags;
import static io.aeron.logbuffer.FrameDescriptor.frameType;
import static io.aeron.protocol.DataHeaderFlyweight.RESERVED_VALUE_OFFSET;
import static java.nio.ByteOrder.LITTLE_ENDIAN;

/**
 * A replay session with a client which works through the required request response flow and streaming of recorded data.
 * The {@link ArchiveConductor} will initiate a session on receiving a ReplayRequest
 * (see {@link io.aeron.archive.codecs.ReplayRequestDecoder}). The session will:
 * <ul>
 * <li>Validate request parameters and respond with appropriate error if unable to replay </li>
 * <li>Wait for replay subscription to connect to the requested replay publication. If no subscription appears within
 * {@link #CONNECT_TIMEOUT_MS} the session will terminate and respond will error.</li>
 * <li>Once the replay publication is connected send an OK response to control client</li>
 * <li>Stream recorded data into the replayPublication {@link ExclusivePublication}</li>
 * <li>If the replay is aborted part way through, send a ReplayAborted message and terminate.</li>
 * </ul>
 */
class ReplaySession implements Session, SimplifiedControlledFragmentHandler
{
    enum State
    {
        INIT, REPLAY, INACTIVE, CLOSED
    }

    /**
     * Timeout within which a replay connection needs to be established.
     */
    static final long CONNECT_TIMEOUT_MS = 5000;

    private static final int REPLAY_FRAGMENT_LIMIT = Archive.Configuration.replayFragmentLimit();

    private long connectDeadlineMs;
    private final long replaySessionId;
    private final long correlationId;
    private final ExclusiveBufferClaim bufferClaim = new ExclusiveBufferClaim();
    private final ExclusivePublication replayPublication;
    private final RecordingFragmentReader cursor;
    private ControlResponseProxy threadLocalControlResponseProxy;
    private final ControlSession controlSession;
    private final EpochClock epochClock;
    private State state = State.INIT;

    @SuppressWarnings("ConstantConditions")
    ReplaySession(
        final long replayPosition,
        final long replayLength,
        final ReplayPublicationSupplier supplier,
        final ControlSession controlSession,
        final File archiveDir,
        final ControlResponseProxy threadLocalControlResponseProxy,
        final long replaySessionId,
        final long correlationId,
        final EpochClock epochClock,
        final String replayChannel,
        final int replayStreamId,
        final UnsafeBuffer descriptorBuffer,
        final AtomicCounter recordingPosition)
    {
        this.controlSession = controlSession;
        this.threadLocalControlResponseProxy = threadLocalControlResponseProxy;
        this.replaySessionId = replaySessionId;
        this.correlationId = correlationId;
        this.epochClock = epochClock;

        final RecordingDescriptorDecoder descriptorDecoder = new RecordingDescriptorDecoder();
        wrapDescriptorDecoder(descriptorDecoder, descriptorBuffer);

        final long startPosition = descriptorDecoder.startPosition();
        final int mtuLength = descriptorDecoder.mtuLength();
        final int termBufferLength = descriptorDecoder.termBufferLength();
        final int initialTermId = descriptorDecoder.initialTermId();

        if (replayPosition - startPosition < 0)
        {
            final String errorMessage = "requested replay start position(=" + replayPosition +
                ") is before recording start position(=" + startPosition + ")";
            closeOnError(new IllegalArgumentException(errorMessage), errorMessage);
            cursor = null;
            replayPublication = null;
            return;
        }

        final long stopPosition = descriptorDecoder.stopPosition();
        if (replayPosition - stopPosition >= 0)
        {
            final String errorMessage = "requested replay start position(=" + replayPosition +
                ") must be before current highest recorded position(=" + stopPosition + ")";
            closeOnError(new IllegalArgumentException(errorMessage), errorMessage);
            cursor = null;
            replayPublication = null;

            return;
        }

        RecordingFragmentReader cursor = null;
        try
        {
            cursor = new RecordingFragmentReader(
                descriptorDecoder,
                archiveDir,
                replayPosition,
                replayLength,
                recordingPosition);
        }
        catch (final Exception ex)
        {
            closeOnError(ex, "failed to open cursor on a recording because: " + ex.getMessage());
        }

        this.cursor = cursor;

        ExclusivePublication replayPublication = null;
        try
        {
            replayPublication = supplier.newReplayPublication(
                replayChannel,
                replayStreamId,
                cursor.fromPosition(),
                mtuLength,
                initialTermId,
                termBufferLength);
        }
        catch (final Exception ex)
        {
            closeOnError(ex, "failed to create replay publication because: " + ex.getMessage());
        }

        this.replayPublication = replayPublication;
        controlSession.sendOkResponse(correlationId, threadLocalControlResponseProxy);

        connectDeadlineMs = epochClock.time() + CONNECT_TIMEOUT_MS;
    }

    public void close()
    {
        state = State.CLOSED;

        CloseHelper.quietClose(cursor);
        CloseHelper.quietClose(replayPublication);
    }

    public long sessionId()
    {
        return replaySessionId;
    }

    public int doWork()
    {
        int workCount = 0;

        switch (state)
        {
            case INIT:
{'fraction_non_alphanumeric': 0.0681834190321899, 'fraction_numerical': 0.0026766218215115867, 'mean_word_length': 3.0005635390250776, 'pattern_counts': {'":': 0, '<': 29, '<?xml version=': 0, '>': 29, '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': '677387', 'n_tokens_mistral': 3583, 'n_tokens_neox': 3430, 'n_words': 1060}