<commit_before>/*
 * Copyright 2014-2019 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.driver;

import io.aeron.CommonContext;
import io.aeron.DriverProxy;
import io.aeron.ErrorCode;
import io.aeron.driver.buffer.RawLog;
import io.aeron.driver.buffer.RawLogFactory;
import io.aeron.driver.media.ReceiveChannelEndpoint;
import io.aeron.driver.media.ReceiveChannelEndpointThreadLocals;
import io.aeron.driver.media.UdpChannel;
import io.aeron.driver.status.SystemCounters;
import io.aeron.logbuffer.HeaderWriter;
import io.aeron.logbuffer.LogBufferDescriptor;
import io.aeron.logbuffer.TermAppender;
import io.aeron.protocol.StatusMessageFlyweight;
import org.agrona.ErrorHandler;
import org.agrona.concurrent.*;
import org.agrona.concurrent.ringbuffer.ManyToOneRingBuffer;
import org.agrona.concurrent.ringbuffer.RingBuffer;
import org.agrona.concurrent.status.AtomicCounter;
import org.agrona.concurrent.status.CountersManager;
import org.junit.After;
import org.junit.Before;
import org.junit.Test;
import org.mockito.ArgumentCaptor;
import org.mockito.InOrder;
import org.mockito.stubbing.Answer;

import java.net.InetSocketAddress;
import java.nio.ByteBuffer;
import java.nio.charset.StandardCharsets;
import java.util.concurrent.TimeUnit;
import java.util.function.BooleanSupplier;
import java.util.function.LongConsumer;

import static io.aeron.ErrorCode.*;
import static io.aeron.driver.Configuration.*;
import static io.aeron.protocol.DataHeaderFlyweight.createDefaultHeader;
import static org.agrona.concurrent.status.CountersReader.METADATA_LENGTH;
import static org.hamcrest.Matchers.anyOf;
import static org.hamcrest.Matchers.is;
import static org.junit.Assert.*;
import static org.mockito.Mockito.*;

public class DriverConductorTest
{
    private static final String CHANNEL_4000 = "aeron:udp?endpoint=localhost:4000";
    private static final String CHANNEL_4001 = "aeron:udp?endpoint=localhost:4001";
    private static final String CHANNEL_4002 = "aeron:udp?endpoint=localhost:4002";
    private static final String CHANNEL_4003 = "aeron:udp?endpoint=localhost:4003";
    private static final String CHANNEL_4004 = "aeron:udp?endpoint=localhost:4004";
    private static final String CHANNEL_4000_TAG_ID_1 = "aeron:udp?endpoint=localhost:4000|tags=1001";
    private static final String CHANNEL_TAG_ID_1 = "aeron:udp?tags=1001";
    private static final String CHANNEL_SUB_CONTROL_MODE_MANUAL = "aeron:udp?control-mode=manual";
    private static final String CHANNEL_IPC = "aeron:ipc";
    private static final String INVALID_URI = "aeron:udp://";
    private static final String COUNTER_LABEL = "counter label";
    private static final int SESSION_ID = 100;
    private static final int STREAM_ID_1 = 10;
    private static final int STREAM_ID_2 = 20;
    private static final int STREAM_ID_3 = 30;
    private static final int STREAM_ID_4 = 40;
    private static final int TERM_BUFFER_LENGTH = Configuration.TERM_BUFFER_LENGTH_DEFAULT;
    private static final int BUFFER_LENGTH = 16 * 1024;
    private static final int COUNTER_TYPE_ID = 101;
    private static final int COUNTER_KEY_OFFSET = 0;
    private static final int COUNTER_KEY_LENGTH = 12;
    private static final int COUNTER_LABEL_OFFSET = COUNTER_KEY_OFFSET + COUNTER_KEY_LENGTH;
    private static final int COUNTER_LABEL_LENGTH = COUNTER_LABEL.length();
    private static final long CLIENT_LIVENESS_TIMEOUT_NS = clientLivenessTimeoutNs();
    private static final long PUBLICATION_LINGER_TIMEOUT_NS = publicationLingerTimeoutNs();
    private static final int MTU_LENGTH = Configuration.mtuLength();

    private final ByteBuffer conductorBuffer = ByteBuffer.allocateDirect(CONDUCTOR_BUFFER_LENGTH_DEFAULT);
    private final UnsafeBuffer counterKeyAndLabel = new UnsafeBuffer(new byte[BUFFER_LENGTH]);

    private final RawLogFactory mockRawLogFactory = mock(RawLogFactory.class);

    private final RingBuffer toDriverCommands = new ManyToOneRingBuffer(new UnsafeBuffer(conductorBuffer));
    private final ClientProxy mockClientProxy = mock(ClientProxy.class);

    private final ErrorHandler mockErrorHandler = mock(ErrorHandler.class);
    private final AtomicCounter mockErrorCounter = mock(AtomicCounter.class);

    private final SenderProxy senderProxy = mock(SenderProxy.class);
    private final ReceiverProxy receiverProxy = mock(ReceiverProxy.class);
    private final DriverConductorProxy driverConductorProxy = mock(DriverConductorProxy.class);

    private long currentTimeMs;
    private final EpochClock epochClock = () -> currentTimeMs;
    private long currentTimeNs;
    private final NanoClock nanoClock = () -> currentTimeNs;

    private CountersManager spyCountersManager;

    private DriverProxy driverProxy;

    private DriverConductor driverConductor;

    private final Answer<Void> closeChannelEndpointAnswer = (invocation) ->
    {
        final Object[] args = invocation.getArguments();
        final ReceiveChannelEndpoint channelEndpoint = (ReceiveChannelEndpoint)args[0];
        channelEndpoint.close();

        return null;
    };

    @Before
    public void setUp()
    {
        // System GC required in order to ensure that the direct byte buffers get cleaned and avoid OOM.
        System.gc();

        when(mockRawLogFactory.newNetworkPublication(
            anyString(), anyInt(), anyInt(), anyLong(), anyInt(), anyBoolean()))
            .thenReturn(LogBufferHelper.newTestLogBuffers(TERM_BUFFER_LENGTH));

        when(mockRawLogFactory.newNetworkedImage(
            anyString(), anyInt(), anyInt(), anyLong(), eq(TERM_BUFFER_LENGTH), anyBoolean()))
            .thenReturn(LogBufferHelper.newTestLogBuffers(TERM_BUFFER_LENGTH));

        when(mockRawLogFactory.newIpcPublication(anyInt(), anyInt(), anyLong(), anyInt(), anyBoolean()))
            .thenReturn(LogBufferHelper.newTestLogBuffers(TERM_BUFFER_LENGTH));

        currentTimeNs = 0;

        counterKeyAndLabel.putInt(COUNTER_KEY_OFFSET, 42);
        counterKeyAndLabel.putStringAscii(COUNTER_LABEL_OFFSET, COUNTER_LABEL);

        final UnsafeBuffer counterBuffer = new UnsafeBuffer(ByteBuffer.allocateDirect(BUFFER_LENGTH));
        spyCountersManager = spy(new CountersManager(
            new UnsafeBuffer(ByteBuffer.allocateDirect(BUFFER_LENGTH * 2)), counterBuffer, StandardCharsets.US_ASCII));

        final MediaDriver.Context ctx = new MediaDriver.Context()
            .tempBuffer(new UnsafeBuffer(new byte[METADATA_LENGTH]))
            .publicationTermBufferLength(TERM_BUFFER_LENGTH)
            .ipcTermBufferLength(TERM_BUFFER_LENGTH)
            .applicationSpecificFeedback(Configuration.applicationSpecificFeedback())
            .unicastFlowControlSupplier(Configuration.unicastFlowControlSupplier())
            .multicastFlowControlSupplier(Configuration.multicastFlowControlSupplier())
            .driverCommandQueue(new ManyToOneConcurrentArrayQueue<>(Configuration.CMD_QUEUE_CAPACITY))
            .errorHandler(mockErrorHandler)
            .rawLogBuffersFactory(mockRawLogFactory)
            .countersManager(spyCountersManager)
            .epochClock(epochClock)
            .nanoClock(nanoClock)
            .cachedEpochClock(new CachedEpochClock())
            .cachedNanoClock(new CachedNanoClock())
            .sendChannelEndpointSupplier(Configuration.sendChannelEndpointSupplier())
            .receiveChannelEndpointSupplier(Configuration.receiveChannelEndpointSupplier())
            .congestControlSupplier(Configuration.congestionControlSupplier());

        ctx.toDriverCommands(toDriverCommands)
            .clientProxy(mockClientProxy)
            .countersValuesBuffer(counterBuffer);

        final SystemCounters mockSystemCounters = mock(SystemCounters.class);
        when(mockSystemCounters.get(any())).thenReturn(mockErrorCounter);

        ctx.systemCounters(mockSystemCounters)
            .receiverProxy(receiverProxy)<commit_msg>[Java] Move some test buffer from off to on heap in the DriverConductorTest.<commit_after>/*
 * Copyright 2014-2019 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.driver;

import io.aeron.CommonContext;
import io.aeron.DriverProxy;
import io.aeron.ErrorCode;
import io.aeron.driver.buffer.RawLog;
import io.aeron.driver.buffer.RawLogFactory;
import io.aeron.driver.media.ReceiveChannelEndpoint;
import io.aeron.driver.media.ReceiveChannelEndpointThreadLocals;
import io.aeron.driver.media.UdpChannel;
import io.aeron.driver.status.SystemCounters;
import io.aeron.logbuffer.HeaderWriter;
import io.aeron.logbuffer.LogBufferDescriptor;
import io.aeron.logbuffer.TermAppender;
import io.aeron.protocol.StatusMessageFlyweight;
import org.agrona.ErrorHandler;
import org.agrona.concurrent.*;
import org.agrona.concurrent.ringbuffer.ManyToOneRingBuffer;
import org.agrona.concurrent.ringbuffer.RingBuffer;
import org.agrona.concurrent.status.AtomicCounter;
import org.agrona.concurrent.status.CountersManager;
import org.junit.After;
import org.junit.Before;
import org.junit.Test;
import org.mockito.ArgumentCaptor;
import org.mockito.InOrder;
import org.mockito.stubbing.Answer;

import java.net.InetSocketAddress;
import java.nio.ByteBuffer;
import java.nio.charset.StandardCharsets;
import java.util.concurrent.TimeUnit;
import java.util.function.BooleanSupplier;
import java.util.function.LongConsumer;

import static io.aeron.ErrorCode.*;
import static io.aeron.driver.Configuration.*;
import static io.aeron.protocol.DataHeaderFlyweight.createDefaultHeader;
import static org.agrona.concurrent.status.CountersReader.METADATA_LENGTH;
import static org.hamcrest.Matchers.anyOf;
import static org.hamcrest.Matchers.is;
import static org.junit.Assert.*;
import static org.mockito.Mockito.*;

public class DriverConductorTest
{
    private static final String CHANNEL_4000 = "aeron:udp?endpoint=localhost:4000";
    private static final String CHANNEL_4001 = "aeron:udp?endpoint=localhost:4001";
    private static final String CHANNEL_4002 = "aeron:udp?endpoint=localhost:4002";
    private static final String CHANNEL_4003 = "aeron:udp?endpoint=localhost:4003";
    private static final String CHANNEL_4004 = "aeron:udp?endpoint=localhost:4004";
    private static final String CHANNEL_4000_TAG_ID_1 = "aeron:udp?endpoint=localhost:4000|tags=1001";
    private static final String CHANNEL_TAG_ID_1 = "aeron:udp?tags=1001";
    private static final String CHANNEL_SUB_CONTROL_MODE_MANUAL = "aeron:udp?control-mode=manual";
    private static final String CHANNEL_IPC = "aeron:ipc";
    private static final String INVALID_URI = "aeron:udp://";
    private static final String COUNTER_LABEL = "counter label";
    private static final int SESSION_ID = 100;
    private static final int STREAM_ID_1 = 10;
    private static final int STREAM_ID_2 = 20;
    private static final int STREAM_ID_3 = 30;
    private static final int STREAM_ID_4 = 40;
    private static final int TERM_BUFFER_LENGTH = Configuration.TERM_BUFFER_LENGTH_DEFAULT;
    private static final int BUFFER_LENGTH = 16 * 1024;
    private static final int COUNTER_TYPE_ID = 101;
    private static final int COUNTER_KEY_OFFSET = 0;
    private static final int COUNTER_KEY_LENGTH = 12;
    private static final int COUNTER_LABEL_OFFSET = COUNTER_KEY_OFFSET + COUNTER_KEY_LENGTH;
    private static final int COUNTER_LABEL_LENGTH = COUNTER_LABEL.length();
    private static final long CLIENT_LIVENESS_TIMEOUT_NS = clientLivenessTimeoutNs();
    private static final long PUBLICATION_LINGER_TIMEOUT_NS = publicationLingerTimeoutNs();
    private static final int MTU_LENGTH = Configuration.mtuLength();

    private final ByteBuffer conductorBuffer = ByteBuffer.allocate(CONDUCTOR_BUFFER_LENGTH_DEFAULT);
    private final UnsafeBuffer counterKeyAndLabel = new UnsafeBuffer(new byte[BUFFER_LENGTH]);

    private final RawLogFactory mockRawLogFactory = mock(RawLogFactory.class);

    private final RingBuffer toDriverCommands = new ManyToOneRingBuffer(new UnsafeBuffer(conductorBuffer));
    private final ClientProxy mockClientProxy = mock(ClientProxy.class);

    private final ErrorHandler mockErrorHandler = mock(ErrorHandler.class);
    private final AtomicCounter mockErrorCounter = mock(AtomicCounter.class);

    private final SenderProxy senderProxy = mock(SenderProxy.class);
    private final ReceiverProxy receiverProxy = mock(ReceiverProxy.class);
    private final DriverConductorProxy driverConductorProxy = mock(DriverConductorProxy.class);

    private long currentTimeMs;
    private final EpochClock epochClock = () -> currentTimeMs;
    private long currentTimeNs;
    private final NanoClock nanoClock = () -> currentTimeNs;

    private CountersManager spyCountersManager;

    private DriverProxy driverProxy;

    private DriverConductor driverConductor;

    private final Answer<Void> closeChannelEndpointAnswer = (invocation) ->
    {
        final Object[] args = invocation.getArguments();
        final ReceiveChannelEndpoint channelEndpoint = (ReceiveChannelEndpoint)args[0];
        channelEndpoint.close();

        return null;
    };

    @Before
    public void setUp()
    {
        // System GC required in order to ensure that the direct byte buffers get cleaned and avoid OOM.
        System.gc();

        when(mockRawLogFactory.newNetworkPublication(
            anyString(), anyInt(), anyInt(), anyLong(), anyInt(), anyBoolean()))
            .thenReturn(LogBufferHelper.newTestLogBuffers(TERM_BUFFER_LENGTH));

        when(mockRawLogFactory.newNetworkedImage(
            anyString(), anyInt(), anyInt(), anyLong(), eq(TERM_BUFFER_LENGTH), anyBoolean()))
            .thenReturn(LogBufferHelper.newTestLogBuffers(TERM_BUFFER_LENGTH));

        when(mockRawLogFactory.newIpcPublication(anyInt(), anyInt(), anyLong(), anyInt(), anyBoolean()))
            .thenReturn(LogBufferHelper.newTestLogBuffers(TERM_BUFFER_LENGTH));

        currentTimeNs = 0;

        counterKeyAndLabel.putInt(COUNTER_KEY_OFFSET, 42);
        counterKeyAndLabel.putStringAscii(COUNTER_LABEL_OFFSET, COUNTER_LABEL);

        final UnsafeBuffer counterBuffer = new UnsafeBuffer(ByteBuffer.allocate(BUFFER_LENGTH));
        spyCountersManager = spy(new CountersManager(
            new UnsafeBuffer(ByteBuffer.allocate(BUFFER_LENGTH * 2)), counterBuffer, StandardCharsets.US_ASCII));

        final MediaDriver.Context ctx = new MediaDriver.Context()
            .tempBuffer(new UnsafeBuffer(new byte[METADATA_LENGTH]))
            .publicationTermBufferLength(TERM_BUFFER_LENGTH)
            .ipcTermBufferLength(TERM_BUFFER_LENGTH)
            .applicationSpecificFeedback(Configuration.applicationSpecificFeedback())
            .unicastFlowControlSupplier(Configuration.unicastFlowControlSupplier())
            .multicastFlowControlSupplier(Configuration.multicastFlowControlSupplier())
            .driverCommandQueue(new ManyToOneConcurrentArrayQueue<>(Configuration.CMD_QUEUE_CAPACITY))
            .errorHandler(mockErrorHandler)
            .rawLogBuffersFactory(mockRawLogFactory)
            .countersManager(spyCountersManager)
            .epochClock(epochClock)
            .nanoClock(nanoClock)
            .cachedEpochClock(new CachedEpochClock())
            .cachedNanoClock(new CachedNanoClock())
            .sendChannelEndpointSupplier(Configuration.sendChannelEndpointSupplier())
            .receiveChannelEndpointSupplier(Configuration.receiveChannelEndpointSupplier())
            .congestControlSupplier(Configuration.congestionControlSupplier());

        ctx.toDriverCommands(toDriverCommands)
            .clientProxy(mockClientProxy)
            .countersValuesBuffer(counterBuffer);

        final SystemCounters mockSystemCounters = mock(SystemCounters.class);
        when(mockSystemCounters.get(any())).thenReturn(mockErrorCounter);

        ctx.systemCounters(mockSystemCounters)
            .receiverProxy(receiverProxy)
{'fraction_non_alphanumeric': 0.08118015727546858, 'fraction_numerical': 0.012061727665109679, 'mean_word_length': 4.5748187211601845, 'pattern_counts': {'":': 0, '<': 7, '<?xml version=': 0, '>': 13, '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': '4333259', 'n_tokens_mistral': 4878, 'n_tokens_neox': 4612, 'n_words': 1069}