<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.archiver;

import io.aeron.*;
import io.aeron.archiver.client.ArchiveProxy;
import io.aeron.archiver.client.ControlResponsePoller;
import io.aeron.archiver.client.RecordingEventsPoller;
import io.aeron.driver.MediaDriver;
import io.aeron.driver.ThreadingMode;
import io.aeron.logbuffer.FrameDescriptor;
import io.aeron.logbuffer.Header;
import io.aeron.protocol.DataHeaderFlyweight;
import org.agrona.*;
import org.agrona.concurrent.UnsafeBuffer;
import org.junit.After;
import org.junit.Before;
import org.junit.Rule;
import org.junit.Test;
import org.junit.rules.TestWatcher;
import org.junit.runner.Description;

import java.io.File;
import java.io.IOException;
import java.util.Random;
import java.util.concurrent.CountDownLatch;
import java.util.concurrent.locks.LockSupport;

import static io.aeron.CommonContext.SPY_PREFIX;
import static io.aeron.archiver.TestUtil.*;
import static io.aeron.protocol.DataHeaderFlyweight.HEADER_LENGTH;
import static io.aeron.protocol.HeaderFlyweight.HDR_TYPE_PAD;
import static org.hamcrest.core.Is.is;
import static org.junit.Assert.*;

public class ArchiverSystemTest
{
    private static final double MEGABYTE = 1024.0d * 1024.0d;

    private static final String CONTROL_URI = "aeron:udp?endpoint=127.0.0.1:54327";
    private static final int CONTROL_STREAM_ID = 100;
    private static final String REPLAY_URI = "aeron:ipc";
    public static final int MESSAGE_COUNT = 5000;
    private String publishUri;
    private static final int PUBLISH_STREAM_ID = 1;
    private static final int MAX_FRAGMENT_SIZE = 1024;
    private static final int REPLAY_STREAM_ID = 101;

    private final MediaDriver.Context driverCtx = new MediaDriver.Context();
    private final Archiver.Context archiverCtx = new Archiver.Context();
    private Aeron publishingClient;
    private Archiver archiver;
    private MediaDriver driver;
    private final UnsafeBuffer buffer = new UnsafeBuffer(new byte[4096]);
    private File archiveDir = TestUtil.makeTempDir();
    private long recordingId;
    private long remaining;
    private int fragmentCount;
    private int[] fragmentLength;
    private long totalDataLength;
    private long totalRecordingLength;
    private long recorded;
    private long requestedJoinPosition;
    private volatile long endPosition = -1;
    private Throwable trackerError;
    private final Random rnd = new Random();
    private long seed = System.nanoTime();

    @Rule
    public TestWatcher testWatcher = new TestWatcher()
    {
        protected void failed(final Throwable t, final Description description)
        {
            System.err.println(ArchiverSystemTest.class.getName() + " failed with random seed: " + seed);
        }
    };

    private Subscription controlResponse;
    private long correlationId;
    private long joinPosition;
    private int requestedInitialTermId;

    @Before
    public void before() throws Exception
    {
        rnd.setSeed(seed);
        requestedInitialTermId = rnd.nextInt(1234);

        final int termLength = 1 << (16 + rnd.nextInt(10)); // 1M to 8M
        final int mtu = 1 << (10 + rnd.nextInt(3)); // 1024 to 8096
        final int termOffset = BitUtil.align(rnd.nextInt(termLength), FrameDescriptor.FRAME_ALIGNMENT);
        final int termId = requestedInitialTermId + rnd.nextInt(1000);

        final ChannelUriStringBuilder channelUriStringBuilder = new ChannelUriStringBuilder()
            .endpoint("1172.16.31.10:54325")
            .termLength(termLength)
            .mtu(mtu)
            .media("udp");

        channelUriStringBuilder
            .initialTermId(requestedInitialTermId)
            .termId(termId)
            .termOffset(termOffset);

        publishUri = channelUriStringBuilder.build();

        requestedJoinPosition = ((termId - requestedInitialTermId) * (long)termLength) + termOffset;

        driverCtx
            .termBufferSparseFile(true)
            .threadingMode(driverThreadingMode())
            .errorHandler(Throwable::printStackTrace)
            .dirsDeleteOnStart(true)
            .useConcurrentCounterManager(true);

        driver = MediaDriver.launch(driverCtx);

        archiverCtx
            .fileSyncLevel(2)
            .mediaDriverAgentInvoker(driver.sharedAgentInvoker())
            .archiveDir(archiveDir)
            .segmentFileLength(termLength << rnd.nextInt(4))
            .threadingMode(archiverThreadingMode())
            .countersManager(driverCtx.countersManager())
            .errorHandler(driverCtx.errorHandler());

        archiver = Archiver.launch(archiverCtx);

        println("Archiver started, dir: " + archiverCtx.archiveDir().getAbsolutePath());
        publishingClient = Aeron.connect();
    }

    ArchiverThreadingMode archiverThreadingMode()
    {
        return ArchiverThreadingMode.SHARED;
    }

    ThreadingMode driverThreadingMode()
    {<commit_msg>[Java] Set fileSyncLevel to 0 in ArchiverSystemTest to prevent timeout failures on Windows.<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.archiver;

import io.aeron.*;
import io.aeron.archiver.client.ArchiveProxy;
import io.aeron.archiver.client.ControlResponsePoller;
import io.aeron.archiver.client.RecordingEventsPoller;
import io.aeron.driver.MediaDriver;
import io.aeron.driver.ThreadingMode;
import io.aeron.logbuffer.FrameDescriptor;
import io.aeron.logbuffer.Header;
import io.aeron.protocol.DataHeaderFlyweight;
import org.agrona.*;
import org.agrona.concurrent.UnsafeBuffer;
import org.junit.After;
import org.junit.Before;
import org.junit.Rule;
import org.junit.Test;
import org.junit.rules.TestWatcher;
import org.junit.runner.Description;

import java.io.File;
import java.io.IOException;
import java.util.Random;
import java.util.concurrent.CountDownLatch;
import java.util.concurrent.locks.LockSupport;

import static io.aeron.CommonContext.SPY_PREFIX;
import static io.aeron.archiver.TestUtil.*;
import static io.aeron.protocol.DataHeaderFlyweight.HEADER_LENGTH;
import static io.aeron.protocol.HeaderFlyweight.HDR_TYPE_PAD;
import static org.hamcrest.core.Is.is;
import static org.junit.Assert.*;

public class ArchiverSystemTest
{
    private static final double MEGABYTE = 1024.0d * 1024.0d;

    private static final String CONTROL_URI = "aeron:udp?endpoint=127.0.0.1:54327";
    private static final int CONTROL_STREAM_ID = 100;
    private static final String REPLAY_URI = "aeron:ipc";
    public static final int MESSAGE_COUNT = 5000;
    private String publishUri;
    private static final int PUBLISH_STREAM_ID = 1;
    private static final int MAX_FRAGMENT_SIZE = 1024;
    private static final int REPLAY_STREAM_ID = 101;

    private final MediaDriver.Context driverCtx = new MediaDriver.Context();
    private final Archiver.Context archiverCtx = new Archiver.Context();
    private Aeron publishingClient;
    private Archiver archiver;
    private MediaDriver driver;
    private final UnsafeBuffer buffer = new UnsafeBuffer(new byte[4096]);
    private File archiveDir = TestUtil.makeTempDir();
    private long recordingId;
    private long remaining;
    private int fragmentCount;
    private int[] fragmentLength;
    private long totalDataLength;
    private long totalRecordingLength;
    private long recorded;
    private long requestedJoinPosition;
    private volatile long endPosition = -1;
    private Throwable trackerError;
    private final Random rnd = new Random();
    private long seed = System.nanoTime();

    @Rule
    public TestWatcher testWatcher = new TestWatcher()
    {
        protected void failed(final Throwable t, final Description description)
        {
            System.err.println(ArchiverSystemTest.class.getName() + " failed with random seed: " + seed);
        }
    };

    private Subscription controlResponse;
    private long correlationId;
    private long joinPosition;
    private int requestedInitialTermId;

    @Before
    public void before() throws Exception
    {
        rnd.setSeed(seed);
        requestedInitialTermId = rnd.nextInt(1234);

        final int termLength = 1 << (16 + rnd.nextInt(10)); // 1M to 8M
        final int mtu = 1 << (10 + rnd.nextInt(3)); // 1024 to 8096
        final int termOffset = BitUtil.align(rnd.nextInt(termLength), FrameDescriptor.FRAME_ALIGNMENT);
        final int termId = requestedInitialTermId + rnd.nextInt(1000);

        final ChannelUriStringBuilder channelUriStringBuilder = new ChannelUriStringBuilder()
            .endpoint("127.0.0.1:54325")
            .termLength(termLength)
            .mtu(mtu)
            .media("udp");

        channelUriStringBuilder
            .initialTermId(requestedInitialTermId)
            .termId(termId)
            .termOffset(termOffset);

        publishUri = channelUriStringBuilder.build();

        requestedJoinPosition = ((termId - requestedInitialTermId) * (long)termLength) + termOffset;

        driverCtx
            .termBufferSparseFile(true)
            .threadingMode(driverThreadingMode())
            .errorHandler(Throwable::printStackTrace)
            .dirsDeleteOnStart(true)
            .useConcurrentCounterManager(true);

        driver = MediaDriver.launch(driverCtx);

        archiverCtx
            .fileSyncLevel(0)
            .mediaDriverAgentInvoker(driver.sharedAgentInvoker())
            .archiveDir(archiveDir)
            .segmentFileLength(termLength << rnd.nextInt(4))
            .threadingMode(archiverThreadingMode())
            .countersManager(driverCtx.countersManager())
            .errorHandler(driverCtx.errorHandler());

        archiver = Archiver.launch(archiverCtx);

        println("Archiver started, dir: " + archiverCtx.archiveDir().getAbsolutePath());
        publishingClient = Aeron.connect();
    }

    ArchiverThreadingMode archiverThreadingMode()
    {
        return ArchiverThreadingMode.SHARED;
    }

    ThreadingMode driverThreadingMode()
    {
{'fraction_non_alphanumeric': 0.08622246298302637, 'fraction_numerical': 0.017244492596605272, 'mean_word_length': 3.4308, 'pattern_counts': {'":': 0, '<': 15, '<?xml version=': 0, '>': 3, 'https://': 0, 'lorem ipsum': 0, 'www.': 2, 'xml': 0}, 'pii_count': 4, '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': '4333959', 'n_tokens_mistral': 3215, 'n_tokens_neox': 3055, 'n_words': 806}