@SuppressWarnings("MethodLength")
    public void onFragment(final DirectBuffer buffer, final int offset, final int length, final Header header)
    {
        final MessageHeaderDecoder headerDecoder = decoders.header;
        headerDecoder.wrap(buffer, offset);

        final int schemaId = headerDecoder.schemaId();
        if (schemaId != MessageHeaderDecoder.SCHEMA_ID)
        {
            throw new ArchiveException("expected schemaId=" + MessageHeaderDecoder.SCHEMA_ID + ", actual=" + schemaId);
        }

        final int templateId = headerDecoder.templateId();
        switch (templateId)
        {
            case ConnectRequestDecoder.TEMPLATE_ID:
            {
                final ConnectRequestDecoder decoder = decoders.connectRequest;
                decoder.wrap(
                    buffer,
                    offset + MessageHeaderDecoder.ENCODED_LENGTH,
                    headerDecoder.blockLength(),
                    headerDecoder.version());

                final ControlSession session = conductor.newControlSession(
                    image.correlationId(),
                    decoder.correlationId(),
                    decoder.responseStreamId(),
                    decoder.version(),
                    decoder.responseChannel(),
                    ArrayUtil.EMPTY_BYTE_ARRAY,
                    this);
                controlSessionByIdMap.put(session.sessionId(), session);
                break;
            }

            case CloseSessionRequestDecoder.TEMPLATE_ID:
            {
                final CloseSessionRequestDecoder decoder = decoders.closeSessionRequest;
                decoder.wrap(
                    buffer,
                    offset + MessageHeaderDecoder.ENCODED_LENGTH,
                    headerDecoder.blockLength(),
                    headerDecoder.version());

                final long controlSessionId = decoder.controlSessionId();
                final ControlSession session = controlSessionByIdMap.get(controlSessionId);
                if (null != session)
                {
                    session.abort();
                }
                break;
            }

            case StartRecordingRequestDecoder.TEMPLATE_ID:
            {
                final StartRecordingRequestDecoder decoder = decoders.startRecordingRequest;
                decoder.wrap(
                    buffer,
                    offset + MessageHeaderDecoder.ENCODED_LENGTH,
                    headerDecoder.blockLength(),
                    headerDecoder.version());

                final long controlSessionId = decoder.controlSessionId();
                final long correlationId = decoder.correlationId();
                final ControlSession controlSession = getControlSession(correlationId, controlSessionId, templateId);

                if (null != controlSession)
                {
                    controlSession.onStartRecording(
                        correlationId,
                        decoder.streamId(),
                        decoder.sourceLocation(),
                        false,
                        decoder.channel());
                }
                break;
            }

            case StopRecordingRequestDecoder.TEMPLATE_ID:
            {
                final StopRecordingRequestDecoder decoder = decoders.stopRecordingRequest;
                decoder.wrap(
                    buffer,
                    offset + MessageHeaderDecoder.ENCODED_LENGTH,
                    headerDecoder.blockLength(),
                    headerDecoder.version());

                final long controlSessionId = decoder.controlSessionId();
                final long correlationId = decoder.correlationId();
                final ControlSession controlSession = getControlSession(correlationId, controlSessionId, templateId);

                if (null != controlSession)
                {
                    controlSession.onStopRecording(correlationId, decoder.streamId(), decoder.channel());
                }
                break;
            }

            case ReplayRequestDecoder.TEMPLATE_ID:
            {
                final ReplayRequestDecoder decoder = decoders.replayRequest;
                decoder.wrap(
                    buffer,
                    offset + MessageHeaderDecoder.ENCODED_LENGTH,
                    headerDecoder.blockLength(),
                    headerDecoder.version());

                final long controlSessionId = decoder.controlSessionId();
                final long correlationId = decoder.correlationId();
                final int fileIoMaxLength = FILE_IO_MAX_LENGTH_VERSION <= headerDecoder.version() ?
                    decoder.fileIoMaxLength() : Aeron.NULL_VALUE;
                final long recordingId = decoder.recordingId();
                final long position = decoder.position();
                final long replayLength = decoder.length();
                final int replayStreamId = decoder.replayStreamId();
                final long replayToken = REPLAY_TOKEN_VERSION <= headerDecoder.version() ?
                    decoder.replayToken() : Aeron.NULL_VALUE;

                final String replayChannel = decoder.replayChannel();
                final ChannelUri channelUri = ChannelUri.parse(replayChannel);
                final ControlSession controlSession = setupSessionAndChannelForReplay(
                    channelUri, replayToken, recordingId, correlationId, controlSessionId, templateId);

                if (null != controlSession)
                {
                    controlSession.onStartReplay(
                        correlationId,
                        recordingId,
                        position,
                        replayLength,
                        fileIoMaxLength,
                        replayStreamId,
                        channelUri.toString());
                }
                break;
            }

            case StopReplayRequestDecoder.TEMPLATE_ID:
            {
                final StopReplayRequestDecoder decoder = decoders.stopReplayRequest;
                decoder.wrap(
                    buffer,
                    offset + MessageHeaderDecoder.ENCODED_LENGTH,
                    headerDecoder.blockLength(),
                    headerDecoder.version());

                final long controlSessionId = decoder.controlSessionId();
                final long correlationId = decoder.correlationId();
                final ControlSession controlSession = getControlSession(correlationId, controlSessionId, templateId);

                if (null != controlSession)
                {
                    controlSession.onStopReplay(correlationId, decoder.replaySessionId());
                }
                break;
            }

            case ListRecordingsRequestDecoder.TEMPLATE_ID:
            {
                final ListRecordingsRequestDecoder decoder = decoders.listRecordingsRequest;
                decoder.wrap(
                    buffer,
                    offset + MessageHeaderDecoder.ENCODED_LENGTH,
                    headerDecoder.blockLength(),
                    headerDecoder.version());

                final long controlSessionId = decoder.controlSessionId();
                final long correlationId = decoder.correlationId();
                final ControlSession controlSession = getControlSession(correlationId, controlSessionId, templateId);

                if (null != controlSession)
                {
                    controlSession.onListRecordings(correlationId, decoder.fromRecordingId(), decoder.recordCount());
                }
                break;
            }

            case ListRecordingsForUriRequestDecoder.TEMPLATE_ID:
            {
                final ListRecordingsForUriRequestDecoder decoder = decoders.listRecordingsForUriRequest;
                decoder.wrap(
                    buffer,
                    offset + MessageHeaderDecoder.ENCODED_LENGTH,
                    headerDecoder.blockLength(),
                    headerDecoder.version());

                final long controlSessionId = decoder.controlSessionId();
                final long correlationId = decoder.correlationId();
                final ControlSession controlSession = getControlSession(correlationId, controlSessionId, templateId);

                if (null != controlSession)
                {
                    final int channelLength = decoder.channelLength();
                    final byte[] bytes = 0 == channelLength ? ArrayUtil.EMPTY_BYTE_ARRAY : new byte[channelLength];
                    decoder.getChannel(bytes, 0, channelLength);

                    controlSession.onListRecordingsForUri(
                        correlationId,
                        decoder.fromRecordingId(),
                        decoder.recordCount(),
                        decoder.streamId(),
                        bytes);
                }
                break;
            }

            case ListRecordingRequestDecoder.TEMPLATE_ID:
            {
                final ListRecordingRequestDecoder decoder = decoders.listRecordingRequest;
                decoder.wrap(
                    buffer,
                    offset + MessageHeaderDecoder.ENCODED_LENGTH,
                    headerDecoder.blockLength(),
                    headerDecoder.version());

                final long controlSessionId = decoder.controlSessionId();
                final long correlationId = decoder.correlationId();
                final ControlSession controlSession = getControlSession(correlationId, controlSessionId, templateId);

                if (null != controlSession)
                {
                    controlSession.onListRecording(correlationId, decoder.recordingId());
                }
                break;
            }

            case ExtendRecordingRequestDecoder.TEMPLATE_ID:
            {
                final ExtendRecordingRequestDecoder decoder = decoders.extendRecordingRequest;
                decoder.wrap(
                    buffer,
                    offset + MessageHeaderDecoder.ENCODED_LENGTH,
                    headerDecoder.blockLength(),
                    headerDecoder.version());

                final long controlSessionId = decoder.controlSessionId();
                final long correlationId = decoder.correlationId();
                final ControlSession controlSession = getControlSession(correlationId, controlSessionId, templateId);

                if (null != controlSession)
                {
                    controlSession.onExtendRecording(
                        correlationId,
                        decoder.recordingId(),
                        decoder.streamId(),
                        decoder.sourceLocation(),
                        false,
                        decoder.channel());
                }
                break;
            }

            case RecordingPositionRequestDecoder.TEMPLATE_ID:
            {
                final RecordingPositionRequestDecoder decoder = decoders.recordingPositionRequest;
                decoder.wrap(
                    buffer,
                    offset + MessageHeaderDecoder.ENCODED_LENGTH,
                    headerDecoder.blockLength(),
                    headerDecoder.version());

                final long controlSessionId = decoder.controlSessionId();
                final long correlationId = decoder.correlationId();
                final ControlSession controlSession = getControlSession(correlationId, controlSessionId, templateId);

                if (null != controlSession)
                {
                    controlSession.onGetRecordingPosition(correlationId, decoder.recordingId());
                }
                break;
            }

            case TruncateRecordingRequestDecoder.TEMPLATE_ID:
            {
                final TruncateRecordingRequestDecoder decoder = decoders.truncateRecordingRequest;
                decoder.wrap(
                    buffer,
                    offset + MessageHeaderDecoder.ENCODED_LENGTH,
                    headerDecoder.blockLength(),
                    headerDecoder.version());

                final long controlSessionId = decoder.controlSessionId();
                final long correlationId = decoder.correlationId();
                final ControlSession controlSession = getControlSession(correlationId, controlSessionId, templateId);

                if (null != controlSession)
                {
                    controlSession.onTruncateRecording(correlationId, decoder.recordingId(), decoder.position());
                }
                break;
            }

            case StopRecordingSubscriptionRequestDecoder.TEMPLATE_ID:
            {
                final StopRecordingSubscriptionRequestDecoder decoder = decoders.stopRecordingSubscriptionRequest;
                decoder.wrap(
                    buffer,
                    offset + MessageHeaderDecoder.ENCODED_LENGTH,
                    headerDecoder.blockLength(),
                    headerDecoder.version());

                final long controlSessionId = decoder.controlSessionId();
                final long correlationId = decoder.correlationId();
                final ControlSession controlSession = getControlSession(correlationId, controlSessionId, templateId);

                if (null != controlSession)
                {
                    controlSession.onStopRecordingSubscription(correlationId, decoder.subscriptionId());
                }
                break;
            }

            case StopPositionRequestDecoder.TEMPLATE_ID:
            {
                final StopPositionRequestDecoder decoder = decoders.stopPositionRequest;
                decoder.wrap(
                    buffer,
                    offset + MessageHeaderDecoder.ENCODED_LENGTH,
                    headerDecoder.blockLength(),
                    headerDecoder.version());

                final long controlSessionId = decoder.controlSessionId();
                final long correlationId = decoder.correlationId();
                final ControlSession controlSession = getControlSession(correlationId, controlSessionId, templateId);

                if (null != controlSession)
                {
                    controlSession.onGetStopPosition(correlationId, decoder.recordingId());
                }
                break;
            }

            case FindLastMatchingRecordingRequestDecoder.TEMPLATE_ID:
            {
                final FindLastMatchingRecordingRequestDecoder decoder = decoders.findLastMatchingRecordingRequest;
                decoder.wrap(
                    buffer,
                    offset + MessageHeaderDecoder.ENCODED_LENGTH,
                    headerDecoder.blockLength(),
                    headerDecoder.version());

                final long controlSessionId = decoder.controlSessionId();
                final long correlationId = decoder.correlationId();
                final ControlSession controlSession = getControlSession(correlationId, controlSessionId, templateId);

                if (null != controlSession)
                {
                    final int channelLength = decoder.channelLength();
                    final byte[] bytes = 0 == channelLength ? ArrayUtil.EMPTY_BYTE_ARRAY : new byte[channelLength];

                    decoder.getChannel(bytes, 0, channelLength);
                    controlSession.onFindLastMatchingRecording(
                        correlationId,
                        decoder.minRecordingId(),
                        decoder.sessionId(),
                        decoder.streamId(),
                        bytes);
                }
                break;
            }

            case ListRecordingSubscriptionsRequestDecoder.TEMPLATE_ID:
            {
                final ListRecordingSubscriptionsRequestDecoder decoder = decoders.listRecordingSubscriptionsRequest;
                decoder.wrap(
                    buffer,
                    offset + MessageHeaderDecoder.ENCODED_LENGTH,
                    headerDecoder.blockLength(),
                    headerDecoder.version());

                final long controlSessionId = decoder.controlSessionId();
                final long correlationId = decoder.correlationId();
                final ControlSession controlSession = getControlSession(correlationId, controlSessionId, templateId);

                if (null != controlSession)
                {
                    controlSession.onListRecordingSubscriptions(
                        correlationId,
                        decoder.pseudoIndex(),
                        decoder.subscriptionCount(),
                        decoder.applyStreamId() == BooleanType.TRUE,
                        decoder.streamId(),
                        decoder.channel());
                }
                break;
            }

            case BoundedReplayRequestDecoder.TEMPLATE_ID:
            {
                final BoundedReplayRequestDecoder decoder = decoders.boundedReplayRequest;
                decoder.wrap(
                    buffer,
                    offset + MessageHeaderDecoder.ENCODED_LENGTH,
                    headerDecoder.blockLength(),
                    headerDecoder.version());

                final long controlSessionId = decoder.controlSessionId();
                final long correlationId = decoder.correlationId();
                final long position = decoder.position();
                final long replayLength = decoder.length();
                final long recordingId = decoder.recordingId();
                final int limitCounterId = decoder.limitCounterId();
                final int replayStreamId = decoder.replayStreamId();
                final int fileIoMaxLength = FILE_IO_MAX_LENGTH_VERSION <= headerDecoder.version() ?
                    decoder.fileIoMaxLength() : Aeron.NULL_VALUE;
                final long replayToken = REPLAY_TOKEN_VERSION <= headerDecoder.version() ?
                    decoder.replayToken() : Aeron.NULL_VALUE;

                final String replayChannel = decoder.replayChannel();

                final ChannelUri channelUri = ChannelUri.parse(replayChannel);
                final ControlSession controlSession = setupSessionAndChannelForReplay(
                    channelUri, replayToken, recordingId, correlationId, controlSessionId, templateId);

                if (null != controlSession)
                {
                    controlSession.onStartBoundedReplay(
                        correlationId,
                        recordingId,
                        position,
                        replayLength,
                        limitCounterId,
                        fileIoMaxLength,
                        replayStreamId,
                        channelUri.toString());
                }
                break;
            }

            case StopAllReplaysRequestDecoder.TEMPLATE_ID:
            {
                final StopAllReplaysRequestDecoder decoder = decoders.stopAllReplaysRequest;
                decoder.wrap(
                    buffer,
                    offset + MessageHeaderDecoder.ENCODED_LENGTH,
                    headerDecoder.blockLength(),
                    headerDecoder.version());

                final long controlSessionId = decoder.controlSessionId();
                final long correlationId = decoder.correlationId();
                final ControlSession controlSession = getControlSession(correlationId, controlSessionId, templateId);

                if (null != controlSession)
                {
                    controlSession.onStopAllReplays(correlationId, decoder.recordingId());
                }
                break;
            }

            case ReplicateRequestDecoder.TEMPLATE_ID:
            {
                final ReplicateRequestDecoder decoder = decoders.replicateRequest;
                decoder.wrap(
                    buffer,
                    offset + MessageHeaderDecoder.ENCODED_LENGTH,
                    headerDecoder.blockLength(),
                    headerDecoder.version());

                final long controlSessionId = decoder.controlSessionId();
                final long correlationId = decoder.correlationId();
                final ControlSession controlSession = getControlSession(correlationId, controlSessionId, templateId);

                if (null != controlSession)
                {
                    controlSession.onReplicate(
                        correlationId,
                        decoder.srcRecordingId(),
                        decoder.dstRecordingId(),
                        AeronArchive.NULL_POSITION,
                        Aeron.NULL_VALUE,
                        Aeron.NULL_VALUE,
                        decoder.srcControlStreamId(),
                        Aeron.NULL_VALUE,
                        Aeron.NULL_VALUE,
                        decoder.srcControlChannel(),
                        decoder.liveDestination(),
                        "",
                        NullCredentialsSupplier.NULL_CREDENTIAL,
                        "");
                }
                break;
            }

            case StopReplicationRequestDecoder.TEMPLATE_ID:
            {
                final StopReplicationRequestDecoder decoder = decoders.stopReplicationRequest;
                decoder.wrap(
                    buffer,
                    offset + MessageHeaderDecoder.ENCODED_LENGTH,
                    headerDecoder.blockLength(),
                    headerDecoder.version());

                final long controlSessionId = decoder.controlSessionId();
                final long correlationId = decoder.correlationId();
                final ControlSession controlSession = getControlSession(correlationId, controlSessionId, templateId);

                if (null != controlSession)
                {
                    controlSession.onStopReplication(correlationId, decoder.replicationId());
                }
                break;
            }

            case StartPositionRequestDecoder.TEMPLATE_ID:
            {
                final StartPositionRequestDecoder decoder = decoders.startPositionRequest;
                decoder.wrap(
                    buffer,
                    offset + MessageHeaderDecoder.ENCODED_LENGTH,
                    headerDecoder.blockLength(),
                    headerDecoder.version());

                final long controlSessionId = decoder.controlSessionId();
                final long correlationId = decoder.correlationId();
                final ControlSession controlSession = getControlSession(correlationId, controlSessionId, templateId);

                if (null != controlSession)
                {
                    controlSession.onGetStartPosition(correlationId, decoder.recordingId());
                }
                break;
            }

            case DetachSegmentsRequestDecoder.TEMPLATE_ID:
            {
                final DetachSegmentsRequestDecoder decoder = decoders.detachSegmentsRequest;
                decoder.wrap(
                    buffer,
                    offset + MessageHeaderDecoder.ENCODED_LENGTH,
                    headerDecoder.blockLength(),
                    headerDecoder.version());

                final long controlSessionId = decoder.controlSessionId();
                final long correlationId = decoder.correlationId();
                final ControlSession controlSession = getControlSession(correlationId, controlSessionId, templateId);

                if (null != controlSession)
                {
                    controlSession.onDetachSegments(correlationId, decoder.recordingId(), decoder.newStartPosition());
                }
                break;
            }

            case DeleteDetachedSegmentsRequestDecoder.TEMPLATE_ID:
            {
                final DeleteDetachedSegmentsRequestDecoder decoder = decoders.deleteDetachedSegmentsRequest;
                decoder.wrap(
                    buffer,
                    offset + MessageHeaderDecoder.ENCODED_LENGTH,
                    headerDecoder.blockLength(),
                    headerDecoder.version());

                final long controlSessionId = decoder.controlSessionId();
                final long correlationId = decoder.correlationId();
                final ControlSession controlSession = getControlSession(correlationId, controlSessionId, templateId);

                if (null != controlSession)
                {
                    controlSession.onDeleteDetachedSegments(correlationId, decoder.recordingId());
                }
                break;
            }

            case PurgeSegmentsRequestDecoder.TEMPLATE_ID:
            {
                final PurgeSegmentsRequestDecoder decoder = decoders.purgeSegmentsRequest;
                decoder.wrap(
                    buffer,
                    offset + MessageHeaderDecoder.ENCODED_LENGTH,
                    headerDecoder.blockLength(),
                    headerDecoder.version());

                final long controlSessionId = decoder.controlSessionId();
                final long correlationId = decoder.correlationId();
                final ControlSession controlSession = getControlSession(correlationId, controlSessionId, templateId);

                if (null != controlSession)
                {
                    controlSession.onPurgeSegments(correlationId, decoder.recordingId(), decoder.newStartPosition());
                }
                break;
            }

            case AttachSegmentsRequestDecoder.TEMPLATE_ID:
            {
                final AttachSegmentsRequestDecoder decoder = decoders.attachSegmentsRequest;
                decoder.wrap(
                    buffer,
                    offset + MessageHeaderDecoder.ENCODED_LENGTH,
                    headerDecoder.blockLength(),
                    headerDecoder.version());

                final long controlSessionId = decoder.controlSessionId();
                final long correlationId = decoder.correlationId();
                final ControlSession controlSession = getControlSession(correlationId, controlSessionId, templateId);

                if (null != controlSession)
                {
                    controlSession.onAttachSegments(correlationId, decoder.recordingId());
                }
                break;
            }

            case MigrateSegmentsRequestDecoder.TEMPLATE_ID:
            {
                final MigrateSegmentsRequestDecoder decoder = decoders.migrateSegmentsRequest;
                decoder.wrap(
                    buffer,
                    offset + MessageHeaderDecoder.ENCODED_LENGTH,
                    headerDecoder.blockLength(),
                    headerDecoder.version());

                final long controlSessionId = decoder.controlSessionId();
                final long correlationId = decoder.correlationId();
                final ControlSession controlSession = getControlSession(correlationId, controlSessionId, templateId);

                if (null != controlSession)
                {
                    controlSession.onMigrateSegments(correlationId, decoder.srcRecordingId(), decoder.dstRecordingId());
                }
                break;
            }

            case AuthConnectRequestDecoder.TEMPLATE_ID:
            {
                final AuthConnectRequestDecoder decoder = decoders.authConnectRequest;
                decoder.wrap(
                    buffer,
                    offset + MessageHeaderDecoder.ENCODED_LENGTH,
                    headerDecoder.blockLength(),
                    headerDecoder.version());

                final String responseChannel = decoder.responseChannel();
                final int credentialsLength = decoder.encodedCredentialsLength();
                final byte[] credentials;

                if (credentialsLength > 0)
                {
                    credentials = new byte[credentialsLength];
                    decoder.getEncodedCredentials(credentials, 0, credentialsLength);
                }
                else
                {
                    credentials = ArrayUtil.EMPTY_BYTE_ARRAY;
                }

                final ControlSession session = conductor.newControlSession(
                    image.correlationId(),
                    decoder.correlationId(),
                    decoder.responseStreamId(),
                    decoder.version(),
                    responseChannel,
                    credentials,
                    this);
                controlSessionByIdMap.put(session.sessionId(), session);
                break;
            }

            case ChallengeResponseDecoder.TEMPLATE_ID:
            {
                final ChallengeResponseDecoder decoder = decoders.challengeResponse;
                decoder.wrap(
                    buffer,
                    offset + MessageHeaderDecoder.ENCODED_LENGTH,
                    headerDecoder.blockLength(),
                    headerDecoder.version());

                final long controlSessionId = decoder.controlSessionId();
                final ControlSession session = controlSessionByIdMap.get(controlSessionId);
                if (null != session)
                {
                    final int credentialsLength = decoder.encodedCredentialsLength();
                    final byte[] credentials;

                    if (credentialsLength > 0)
                    {
                        credentials = new byte[credentialsLength];
                        decoder.getEncodedCredentials(credentials, 0, credentialsLength);
                    }
                    else
                    {
                        credentials = ArrayUtil.EMPTY_BYTE_ARRAY;
                    }

                    session.onChallengeResponse(decoder.correlationId(), credentials);
                }
                break;
            }

            case KeepAliveRequestDecoder.TEMPLATE_ID:
            {
                final KeepAliveRequestDecoder decoder = decoders.keepAliveRequest;
                decoder.wrap(
                    buffer,
                    offset + MessageHeaderDecoder.ENCODED_LENGTH,
                    headerDecoder.blockLength(),
                    headerDecoder.version());

                final long controlSessionId = decoder.controlSessionId();
                final long correlationId = decoder.correlationId();
                final ControlSession controlSession = getControlSession(correlationId, controlSessionId, templateId);

                if (null != controlSession)
                {
                    controlSession.onKeepAlive(correlationId);
                }
                break;
            }

            case TaggedReplicateRequestDecoder.TEMPLATE_ID:
            {
                final TaggedReplicateRequestDecoder decoder = decoders.taggedReplicateRequest;
                decoder.wrap(
                    buffer,
                    offset + MessageHeaderDecoder.ENCODED_LENGTH,
                    headerDecoder.blockLength(),
                    headerDecoder.version());

                final long controlSessionId = decoder.controlSessionId();
                final long correlationId = decoder.correlationId();
                final ControlSession controlSession = getControlSession(correlationId, controlSessionId, templateId);

                if (null != controlSession)
                {
                    controlSession.onReplicate(
                        correlationId,
                        decoder.srcRecordingId(),
                        decoder.dstRecordingId(),
                        AeronArchive.NULL_POSITION,
                        decoder.channelTagId(),
                        decoder.subscriptionTagId(),
                        decoder.srcControlStreamId(),
                        Aeron.NULL_VALUE,
                        Aeron.NULL_VALUE,
                        decoder.srcControlChannel(),
                        decoder.liveDestination(),
                        "",
                        NullCredentialsSupplier.NULL_CREDENTIAL,
                        "");
                }
                break;
            }

            case StartRecordingRequest2Decoder.TEMPLATE_ID:
            {
                final StartRecordingRequest2Decoder decoder = decoders.startRecordingRequest2;
                decoder.wrap(
                    buffer,
                    offset + MessageHeaderDecoder.ENCODED_LENGTH,
                    headerDecoder.blockLength(),
                    headerDecoder.version());

                final long controlSessionId = decoder.controlSessionId();
                final long correlationId = decoder.correlationId();
                final ControlSession controlSession = getControlSession(correlationId, controlSessionId, templateId);

                if (null != controlSession)
                {
                    controlSession.onStartRecording(
                        correlationId,
                        decoder.streamId(),
                        decoder.sourceLocation(),
                        decoder.autoStop() == BooleanType.TRUE,
                        decoder.channel());
                }
                break;
            }

            case ExtendRecordingRequest2Decoder.TEMPLATE_ID:
            {
                final ExtendRecordingRequest2Decoder decoder = decoders.extendRecordingRequest2;
                decoder.wrap(
                    buffer,
                    offset + MessageHeaderDecoder.ENCODED_LENGTH,
                    headerDecoder.blockLength(),
                    headerDecoder.version());

                final long controlSessionId = decoder.controlSessionId();
                final long correlationId = decoder.correlationId();
                final ControlSession controlSession = getControlSession(correlationId, controlSessionId, templateId);

                if (null != controlSession)
                {
                    controlSession.onExtendRecording(
                        correlationId,
                        decoder.recordingId(),
                        decoder.streamId(),
                        decoder.sourceLocation(),
                        decoder.autoStop() == BooleanType.TRUE,
                        decoder.channel());
                }
                break;
            }

            case StopRecordingByIdentityRequestDecoder.TEMPLATE_ID:
            {
                final StopRecordingByIdentityRequestDecoder decoder = decoders.stopRecordingByIdentityRequest;
                decoder.wrap(
                    buffer,
                    offset + MessageHeaderDecoder.ENCODED_LENGTH,
                    headerDecoder.blockLength(),
                    headerDecoder.version());

                final long controlSessionId = decoder.controlSessionId();
                final long correlationId = decoder.correlationId();
                final ControlSession controlSession = getControlSession(correlationId, controlSessionId, templateId);

                if (null != controlSession)
                {
                    controlSession.onStopRecordingByIdentity(correlationId, decoder.recordingId());
                }
                break;
            }

            case ReplicateRequest2Decoder.TEMPLATE_ID:
            {
                final ReplicateRequest2Decoder decoder = decoders.replicateRequest2;
                decoder.wrap(
                    buffer,
                    offset + MessageHeaderDecoder.ENCODED_LENGTH,
                    headerDecoder.blockLength(),
                    headerDecoder.version());

                final long controlSessionId = decoder.controlSessionId();
                final long correlationId = decoder.correlationId();
                final ControlSession controlSession = getControlSession(correlationId, controlSessionId, templateId);

                final int fileIoMaxLength = FILE_IO_MAX_LENGTH_VERSION <= headerDecoder.version() ?
                    decoder.fileIoMaxLength() : Aeron.NULL_VALUE;

                final int sessionId = SESSION_ID_VERSION <= headerDecoder.version() ?
                    decoder.replicationSessionId() : Aeron.NULL_VALUE;

                final String srcControlChannel = decoder.srcControlChannel();
                final String liveDestination = decoder.liveDestination();
                final String replicationChannel = decoder.replicationChannel();
                final byte[] encodedCredentials;
                if (ENCODED_CREDENTIALS_VERSION <= headerDecoder.version())
                {
                    encodedCredentials = new byte[decoder.encodedCredentialsLength()];
                    decoder.getEncodedCredentials(encodedCredentials, 0, decoder.encodedCredentialsLength());
                }
                else
                {
                    encodedCredentials = NullCredentialsSupplier.NULL_CREDENTIAL;
                }
                final String srcResponseChannel = decoder.srcResponseChannel();

                if (null != controlSession)
                {
                    controlSession.onReplicate(
                        correlationId,
                        decoder.srcRecordingId(),
                        decoder.dstRecordingId(),
                        decoder.stopPosition(),
                        decoder.channelTagId(),
                        decoder.subscriptionTagId(),
                        decoder.srcControlStreamId(),
                        fileIoMaxLength,
                        sessionId,
                        srcControlChannel,
                        liveDestination,
                        replicationChannel,
                        encodedCredentials,
                        srcResponseChannel
                    );
                }
                break;
            }

            case PurgeRecordingRequestDecoder.TEMPLATE_ID:
            {
                final PurgeRecordingRequestDecoder decoder = decoders.purgeRecordingRequest;
                decoder.wrap(
                    buffer,
                    offset + MessageHeaderDecoder.ENCODED_LENGTH,
                    headerDecoder.blockLength(),
                    headerDecoder.version());

                final long controlSessionId = decoder.controlSessionId();
                final long correlationId = decoder.correlationId();
                final ControlSession controlSession = getControlSession(correlationId, controlSessionId, templateId);

                if (null != controlSession)
                {
                    controlSession.onPurgeRecording(correlationId, decoder.recordingId());
                }
                break;
            }

            case MaxRecordedPositionRequestDecoder.TEMPLATE_ID:
            {
                final MaxRecordedPositionRequestDecoder decoder = decoders.maxRecordedPositionRequest;
                decoder.wrap(
                    buffer,
                    offset + MessageHeaderDecoder.ENCODED_LENGTH,
                    headerDecoder.blockLength(),
                    headerDecoder.version());

                final long controlSessionId = decoder.controlSessionId();
                final long correlationId = decoder.correlationId();
                final ControlSession controlSession = getControlSession(correlationId, controlSessionId, templateId);

                if (null != controlSession)
                {
                    controlSession.onGetMaxRecordedPosition(correlationId, decoder.recordingId());
                }
                break;
            }

            case ArchiveIdRequestDecoder.TEMPLATE_ID:
            {
                final ArchiveIdRequestDecoder decoder = decoders.archiveIdRequestDecoder;
                decoder.wrap(
                    buffer,
                    offset + MessageHeaderDecoder.ENCODED_LENGTH,
                    headerDecoder.blockLength(),
                    headerDecoder.version());

                final long controlSessionId = decoder.controlSessionId();
                final long correlationId = decoder.correlationId();
                final ControlSession controlSession = getControlSession(correlationId, controlSessionId, templateId);

                if (null != controlSession)
                {
                    controlSession.onArchiveId(correlationId);
                }
                break;
            }

            case ReplayTokenRequestDecoder.TEMPLATE_ID:
            {
                final ReplayTokenRequestDecoder decoder = decoders.replayTokenRequestDecoder;
                decoder.wrap(
                    buffer,
                    offset + MessageHeaderDecoder.ENCODED_LENGTH,
                    headerDecoder.blockLength(),
                    headerDecoder.version());

                final long controlSessionId = decoder.controlSessionId();
                final long correlationId = decoder.correlationId();
                final long recordingId = decoder.recordingId();
                final ControlSession controlSession = getControlSession(correlationId, controlSessionId, templateId);
                if (null != controlSession)
                {
                    final long replayToken = conductor.generateReplayToken(controlSession, recordingId);
                    controlSession.sendResponse(
                        correlationId, replayToken, ControlResponseCode.OK, "", conductor.controlResponseProxy());
                }
            }
        }
    }
@Test
    void shouldHandleReplayRequest()
    {
        final ControlSessionDemuxer controlSessionDemuxer = new ControlSessionDemuxer(
            new ControlRequestDecoders(), mockImage, mockConductor, mockAuthorisationService);
        setupControlSession(controlSessionDemuxer, CONTROL_SESSION_ID);

        final ExpandableArrayBuffer buffer = new ExpandableArrayBuffer();
        final MessageHeaderEncoder headerEncoder = new MessageHeaderEncoder();
        final ReplayRequestEncoder replayRequestEncoder = new ReplayRequestEncoder();

        replayRequestEncoder.wrapAndApplyHeader(buffer, 0, headerEncoder);

        replayRequestEncoder
            .controlSessionId(928374L)
            .correlationId(9382475L)
            .recordingId(9827345897L)
            .position(982374L)
            .fileIoMaxLength(4096)
            .replayStreamId(9832475)
            .replayChannel("aeron:ipc");

        final int replicateRequestLength = replayRequestEncoder.encodedLength();

        controlSessionDemuxer.onFragment(buffer, 0, replicateRequestLength, mockHeader);

        final ReplayRequestDecoder expected = new ReplayRequestDecoder()
            .wrapAndApplyHeader(buffer, 0, new MessageHeaderDecoder());

        verify(mockSession).onStartReplay(
            expected.correlationId(),
            expected.recordingId(),
            expected.position(),
            expected.length(),
            expected.fileIoMaxLength(),
            expected.replayStreamId(),
            expected.replayChannel());
    }