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

import io.aeron.*;
import io.aeron.archive.Archive;
import io.aeron.archive.ArchiveThreadingMode;
import io.aeron.archive.ArchivingMediaDriver;
import io.aeron.archive.client.AeronArchive;
import io.aeron.archive.codecs.SourceLocation;
import io.aeron.archive.status.RecordingPos;
import io.aeron.driver.MediaDriver;
import io.aeron.driver.ThreadingMode;
import io.aeron.logbuffer.ControlledFragmentHandler;
import io.aeron.logbuffer.Header;
import io.aeron.logbuffer.LogBufferDescriptor;
import org.agrona.CloseHelper;
import org.agrona.DirectBuffer;
import org.agrona.SystemUtil;
import org.agrona.collections.LongArrayList;
import org.agrona.concurrent.IdleStrategy;
import org.agrona.concurrent.UnsafeBuffer;
import org.agrona.concurrent.YieldingIdleStrategy;
import org.agrona.concurrent.status.CountersReader;

import java.io.File;
import java.nio.ByteOrder;
import java.util.Random;

import static io.aeron.Aeron.NULL_VALUE;
import static io.aeron.CommonContext.getAeronDirectoryName;
import static org.agrona.BitUtil.SIZE_OF_LONG;

/**
 * Example of how to create a recorded stream that is replicated and indexed. Indexers run in parallel to the recording
 * for greater throughput and lower latency.
 * <p>
 * The index allows for the lookup of message start position in a recording based on message index plus a basic
 * time series for when a message was published.
 * <p>
 * The secondary (destination) archive launches after the primary (source) archive and replicates from the sources
 * the history of the stream it missed then when it catches up to live it will merge with the live stream and stop
 * the replay replication from the source.
 */
public class IndexedReplicatedRecording implements AutoCloseable
{
    static final int MESSAGE_INDEX_OFFSET = 0;
    static final int TIMESTAMP_OFFSET = SIZE_OF_LONG;
    static final int HEADER_LENGTH = TIMESTAMP_OFFSET + SIZE_OF_LONG;
    static final int MESSAGE_BURST_COUNT = 10_000;

    private static final int TERM_LENGTH = LogBufferDescriptor.TERM_MIN_LENGTH;
    private static final long CATALOG_CAPACITY = 64 * 1024;
    private static final int SRC_CONTROL_STREAM_ID = AeronArchive.Configuration.CONTROL_STREAM_ID_DEFAULT;
    private static final String SRC_CONTROL_REQUEST_CHANNEL = "aeron:udp?endpoint=localhost:8090";
    private static final String SRC_CONTROL_RESPONSE_CHANNEL = "aeron:udp?endpoint=localhost:0";
    private static final String DST_CONTROL_REQUEST_CHANNEL = "aeron:udp?endpoint=localhost:8095";
    private static final String DST_CONTROL_RESPONSE_CHANNEL = "aeron:udp?endpoint=localhost:0";
    private static final String SRC_REPLICATION_CHANNEL = "aeron:udp?endpoint=localhost:0";
    private static final String DST_REPLICATION_CHANNEL = "aeron:udp?endpoint=localhost:0";

    private static final int LIVE_STREAM_ID = 1033;
    private static final String LIVE_CHANNEL = new ChannelUriStringBuilder()
        .media("udp")
        .controlEndpoint("localhost:8100")
        .termLength(TERM_LENGTH)
        .build();

    private static final int INDEX_STREAM_ID = 1097;
    private static final String INDEX_CHANNEL = new ChannelUriStringBuilder()
        .media("ipc")
        .termLength(TERM_LENGTH)
        .build();

    private final ArchivingMediaDriver srcArchivingMediaDriver;
    private final ArchivingMediaDriver dstArchivingMediaDriver;
    private final Aeron srcAeron;
    private final Aeron dstAeron;
    private final AeronArchive srcAeronArchive;
    private final AeronArchive dstAeronArchive;

    IndexedReplicatedRecording()
    {
        final String srcAeronDirectoryName = getAeronDirectoryName() + "-src";
        System.out.println("srcAeronDirectoryName=" + srcAeronDirectoryName);
        final String dstAeronDirectoryName = getAeronDirectoryName() + "-dst";
        System.out.println("dstAeronDirectoryName=" + dstAeronDirectoryName);

        final File srcArchiveDir = new File(SystemUtil.tmpDirName(), "src-archive");
        System.out.println("srcArchiveDir=" + srcArchiveDir);
        srcArchivingMediaDriver = ArchivingMediaDriver.launch(
            new MediaDriver.Context()
                .aeronDirectoryName(srcAeronDirectoryName)
                .termBufferSparseFile(true)
                .threadingMode(ThreadingMode.SHARED)
                .errorHandler(Throwable::printStackTrace)
                .spiesSimulateConnection(true)
                .dirDeleteOnShutdown(true)
                .dirDeleteOnStart(true),
            new Archive.Context()
                .catalogCapacity(CATALOG_CAPACITY)
                .controlChannel(SRC_CONTROL_REQUEST_CHANNEL)
                .archiveClientContext(new AeronArchive.Context().controlResponseChannel(SRC_CONTROL_RESPONSE_CHANNEL))
                .recordingEventsEnabled(false)
                .replicationChannel(SRC_REPLICATION_CHANNEL)
                .deleteArchiveOnStart(true)
                .archiveDir(srcArchiveDir)
                .fileSyncLevel(0)
                .threadingMode(ArchiveThreadingMode.SHARED));

        final File dstArchiveDir = new File(SystemUtil.tmpDirName(), "dst-archive");
        System.out.println("dstArchiveDir=" + dstArchiveDir);
        dstArchivingMediaDriver = ArchivingMediaDriver.launch(
            new MediaDriver.Context()
                .aeronDirectoryName(dstAeronDirectoryName)
                .termBufferSparseFile(true)
                .threadingMode(ThreadingMode.SHARED)
                .errorHandler(Throwable::printStackTrace)
                .spiesSimulateConnection(true)
                .dirDeleteOnShutdown(true)
                .dirDeleteOnStart(true),
            new Archive.Context()
                .catalogCapacity(CATALOG_CAPACITY)
                .controlChannel(DST_CONTROL_REQUEST_CHANNEL)
                .archiveClientContext(new AeronArchive.Context().controlResponseChannel(DST_CONTROL_RESPONSE_CHANNEL))
                .recordingEventsEnabled(false)
                .replicationChannel(DST_REPLICATION_CHANNEL)
                .deleteArchiveOnStart(true)
                .archiveDir(dstArchiveDir)
                .fileSyncLevel(0)
                .threadingMode(ArchiveThreadingMode.SHARED));

        srcAeron = Aeron.connect(
            new Aeron.Context()
                .aeronDirectoryName(srcAeronDirectoryName));

        dstAeron = Aeron.connect(
            new Aeron.Context()<commit_msg>[Java] Remove properties that are now defaulted for IndexedReplicatedRecording.<commit_after>/*
 * Copyright 2014-2022 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
 *
 * 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.samples.archive;

import io.aeron.*;
import io.aeron.archive.Archive;
import io.aeron.archive.ArchiveThreadingMode;
import io.aeron.archive.ArchivingMediaDriver;
import io.aeron.archive.client.AeronArchive;
import io.aeron.archive.codecs.SourceLocation;
import io.aeron.archive.status.RecordingPos;
import io.aeron.driver.MediaDriver;
import io.aeron.driver.ThreadingMode;
import io.aeron.logbuffer.ControlledFragmentHandler;
import io.aeron.logbuffer.Header;
import io.aeron.logbuffer.LogBufferDescriptor;
import org.agrona.CloseHelper;
import org.agrona.DirectBuffer;
import org.agrona.SystemUtil;
import org.agrona.collections.LongArrayList;
import org.agrona.concurrent.IdleStrategy;
import org.agrona.concurrent.UnsafeBuffer;
import org.agrona.concurrent.YieldingIdleStrategy;
import org.agrona.concurrent.status.CountersReader;

import java.io.File;
import java.nio.ByteOrder;
import java.util.Random;

import static io.aeron.Aeron.NULL_VALUE;
import static io.aeron.CommonContext.getAeronDirectoryName;
import static org.agrona.BitUtil.SIZE_OF_LONG;

/**
 * Example of how to create a recorded stream that is replicated and indexed. Indexers run in parallel to the recording
 * for greater throughput and lower latency.
 * <p>
 * The index allows for the lookup of message start position in a recording based on message index plus a basic
 * time series for when a message was published.
 * <p>
 * The secondary (destination) archive launches after the primary (source) archive and replicates from the sources
 * the history of the stream it missed then when it catches up to live it will merge with the live stream and stop
 * the replay replication from the source.
 */
public class IndexedReplicatedRecording implements AutoCloseable
{
    static final int MESSAGE_INDEX_OFFSET = 0;
    static final int TIMESTAMP_OFFSET = SIZE_OF_LONG;
    static final int HEADER_LENGTH = TIMESTAMP_OFFSET + SIZE_OF_LONG;
    static final int MESSAGE_BURST_COUNT = 10_000;

    private static final int TERM_LENGTH = LogBufferDescriptor.TERM_MIN_LENGTH;
    private static final long CATALOG_CAPACITY = 64 * 1024;
    private static final int SRC_CONTROL_STREAM_ID = AeronArchive.Configuration.CONTROL_STREAM_ID_DEFAULT;
    private static final String SRC_CONTROL_REQUEST_CHANNEL = "aeron:udp?endpoint=localhost:8090";
    private static final String SRC_CONTROL_RESPONSE_CHANNEL = "aeron:udp?endpoint=localhost:0";
    private static final String DST_CONTROL_REQUEST_CHANNEL = "aeron:udp?endpoint=localhost:8095";
    private static final String DST_CONTROL_RESPONSE_CHANNEL = "aeron:udp?endpoint=localhost:0";
    private static final String SRC_REPLICATION_CHANNEL = "aeron:udp?endpoint=localhost:0";
    private static final String DST_REPLICATION_CHANNEL = "aeron:udp?endpoint=localhost:0";

    private static final int LIVE_STREAM_ID = 1033;
    private static final String LIVE_CHANNEL = new ChannelUriStringBuilder()
        .media("udp")
        .controlEndpoint("localhost:8100")
        .termLength(TERM_LENGTH)
        .build();

    private static final int INDEX_STREAM_ID = 1097;
    private static final String INDEX_CHANNEL = new ChannelUriStringBuilder()
        .media("ipc")
        .termLength(TERM_LENGTH)
        .build();

    private final ArchivingMediaDriver srcArchivingMediaDriver;
    private final ArchivingMediaDriver dstArchivingMediaDriver;
    private final Aeron srcAeron;
    private final Aeron dstAeron;
    private final AeronArchive srcAeronArchive;
    private final AeronArchive dstAeronArchive;

    IndexedReplicatedRecording()
    {
        final String srcAeronDirectoryName = getAeronDirectoryName() + "-src";
        System.out.println("srcAeronDirectoryName=" + srcAeronDirectoryName);
        final String dstAeronDirectoryName = getAeronDirectoryName() + "-dst";
        System.out.println("dstAeronDirectoryName=" + dstAeronDirectoryName);

        final File srcArchiveDir = new File(SystemUtil.tmpDirName(), "src-archive");
        System.out.println("srcArchiveDir=" + srcArchiveDir);
        srcArchivingMediaDriver = ArchivingMediaDriver.launch(
            new MediaDriver.Context()
                .aeronDirectoryName(srcAeronDirectoryName)
                .termBufferSparseFile(true)
                .threadingMode(ThreadingMode.SHARED)
                .errorHandler(Throwable::printStackTrace)
                .spiesSimulateConnection(true)
                .dirDeleteOnShutdown(true)
                .dirDeleteOnStart(true),
            new Archive.Context()
                .catalogCapacity(CATALOG_CAPACITY)
                .controlChannel(SRC_CONTROL_REQUEST_CHANNEL)
                .archiveClientContext(new AeronArchive.Context().controlResponseChannel(SRC_CONTROL_RESPONSE_CHANNEL))
                .replicationChannel(SRC_REPLICATION_CHANNEL)
                .deleteArchiveOnStart(true)
                .archiveDir(srcArchiveDir)
                .fileSyncLevel(0)
                .threadingMode(ArchiveThreadingMode.SHARED));

        final File dstArchiveDir = new File(SystemUtil.tmpDirName(), "dst-archive");
        System.out.println("dstArchiveDir=" + dstArchiveDir);
        dstArchivingMediaDriver = ArchivingMediaDriver.launch(
            new MediaDriver.Context()
                .aeronDirectoryName(dstAeronDirectoryName)
                .termBufferSparseFile(true)
                .threadingMode(ThreadingMode.SHARED)
                .errorHandler(Throwable::printStackTrace)
                .spiesSimulateConnection(true)
                .dirDeleteOnShutdown(true)
                .dirDeleteOnStart(true),
            new Archive.Context()
                .catalogCapacity(CATALOG_CAPACITY)
                .controlChannel(DST_CONTROL_REQUEST_CHANNEL)
                .archiveClientContext(new AeronArchive.Context().controlResponseChannel(DST_CONTROL_RESPONSE_CHANNEL))
                .replicationChannel(DST_REPLICATION_CHANNEL)
                .deleteArchiveOnStart(true)
                .archiveDir(dstArchiveDir)
                .fileSyncLevel(0)
                .threadingMode(ArchiveThreadingMode.SHARED));

        srcAeron = Aeron.connect(
            new Aeron.Context()
                .aeronDirectoryName(srcAeronDirectoryName));

        dstAeron = Aeron.connect(
            new Aeron.Context()
{'fraction_non_alphanumeric': 0.07473639213451126, 'fraction_numerical': 0.007124536905101168, 'mean_word_length': 3.4561904761904763, 'pattern_counts': {'":': 0, '<': 7, '<?xml version=': 0, '>': 7, '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': '3440429', 'n_tokens_mistral': 3975, 'n_tokens_neox': 3747, 'n_words': 935}