<commit_before>/*
 * Copyright 2014 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 uk.co.real_logic.aeron.mediadriver;

import uk.co.real_logic.aeron.mediadriver.buffer.BufferManagement;
import uk.co.real_logic.aeron.mediadriver.buffer.BufferRotator;
import uk.co.real_logic.aeron.util.*;
import uk.co.real_logic.aeron.util.collections.Long2ObjectHashMap;
import uk.co.real_logic.aeron.util.command.*;
import uk.co.real_logic.aeron.util.concurrent.AtomicBuffer;
import uk.co.real_logic.aeron.util.concurrent.ringbuffer.ManyToOneRingBuffer;
import uk.co.real_logic.aeron.util.concurrent.ringbuffer.RingBuffer;
import uk.co.real_logic.aeron.util.protocol.DataHeaderFlyweight;
import uk.co.real_logic.aeron.util.protocol.ErrorHeaderFlyweight;

import java.nio.ByteBuffer;
import java.util.concurrent.ThreadLocalRandom;
import java.util.concurrent.TimeUnit;
import java.util.function.Supplier;

import static java.nio.ByteBuffer.allocateDirect;
import static uk.co.real_logic.aeron.mediadriver.MediaDriver.*;
import static uk.co.real_logic.aeron.util.command.ControlProtocolEvents.*;

/**
 * Media Conductor to take commands from Publishers and Subscribers as well as handle NAKs and retransmissions
 */
public class MediaConductor extends Agent
{
    public static final int WRITE_BUFFER_CAPACITY = 512;
    public static final int HEADER_LENGTH = DataHeaderFlyweight.HEADER_LENGTH;
    public static final int HEARTBEAT_TIMEOUT_MS = 100;

    private final RingBuffer mediaCommandBuffer;
    private final ReceiverProxy receiverProxy;
    private final NioSelector nioSelector;
    private final Receiver receiver;
    private final Sender sender;
    private final BufferManagement bufferManagement;
    private final RingBuffer toMediaDriverBuffer;
    private final RingBuffer toClientBuffer;
    private final Long2ObjectHashMap<ControlFrameHandler> srcDestinationMap = new Long2ObjectHashMap<>();
    private final AtomicBuffer writeBuffer = new AtomicBuffer(allocateDirect(WRITE_BUFFER_CAPACITY));
    private final TimerWheel timerWheel;

    private final Supplier<SenderControlStrategy> senderFlowControl;

    private final ThreadLocalRandom rng = ThreadLocalRandom.current();
    private final ChannelMessageFlyweight channelMessage = new ChannelMessageFlyweight();
    private final SubscriberMessageFlyweight receiverMessage = new SubscriberMessageFlyweight();
    private final ErrorHeaderFlyweight errorHeader = new ErrorHeaderFlyweight();
    private final QualifiedMessageFlyweight qualifiedMessage = new QualifiedMessageFlyweight();
    private final NewBufferMessageFlyweight newBufferMessage = new NewBufferMessageFlyweight();

    private final int mtuLength;
    private final ConductorByteBuffers conductorByteBuffers;
    private final TimerWheel.Timer heartbeatTimer;

    public MediaConductor(final Context ctx, final Receiver receiver, final Sender sender)
    {
        super(SELECT_TIMEOUT);

        this.mediaCommandBuffer = ctx.mediaCommandBuffer();
        this.receiverProxy = new ReceiverProxy(ctx.receiverCommandBuffer(), ctx.receiverNioSelector());
        this.bufferManagement = ctx.bufferManagement();
        this.nioSelector = ctx.conductorNioSelector();
        this.mtuLength = ctx.mtuLength();
        this.receiver = receiver;
        this.sender = sender;
        this.senderFlowControl = ctx.senderFlowControl();

        newBufferMessage.wrap(writeBuffer, 0);

        timerWheel = ctx.conductorTimerWheel() != null ?
            ctx.conductorTimerWheel() :
            new TimerWheel(MEDIA_CONDUCTOR_TICK_DURATION_US,
                           TimeUnit.MICROSECONDS,
                           MEDIA_CONDUCTOR_TICKS_PER_WHEEL);

        try
        {
            conductorByteBuffers = ctx.conductorByteBuffers();
            final ByteBuffer toMediaDriver = conductorByteBuffers.toMediaDriver();
            final ByteBuffer toClient = conductorByteBuffers.toClient();
            toMediaDriverBuffer = new ManyToOneRingBuffer(new AtomicBuffer(toMediaDriver));
            toClientBuffer = new ManyToOneRingBuffer(new AtomicBuffer(toClient));
        }
        catch (final Exception ex)
        {
            throw new IllegalStateException("Unable to create the conductor media buffers", ex);
        }

        heartbeatTimer = newTimeout(HEARTBEAT_TIMEOUT_MS, TimeUnit.MILLISECONDS, this::onHeartbeatCheck);
    }

    public ControlFrameHandler frameHandler(final UdpDestination destination)
    {
        return srcDestinationMap.get(destination.consistentHash());
    }

    public void process()
    {
        try
        {
            nioSelector.processKeys();
        }
        catch (final Exception ex)
        {
            // TODO: error
            ex.printStackTrace();
        }

        sender.processBufferRotation();
        receiver.processBufferRotation();
        processReceiveBuffer();
        processCommandBuffer();
        processTimers();
    }
<commit_msg>[Java]: Order cohesive statements.<commit_after>/*
 * Copyright 2014 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 uk.co.real_logic.aeron.mediadriver;

import uk.co.real_logic.aeron.mediadriver.buffer.BufferManagement;
import uk.co.real_logic.aeron.mediadriver.buffer.BufferRotator;
import uk.co.real_logic.aeron.util.*;
import uk.co.real_logic.aeron.util.collections.Long2ObjectHashMap;
import uk.co.real_logic.aeron.util.command.*;
import uk.co.real_logic.aeron.util.concurrent.AtomicBuffer;
import uk.co.real_logic.aeron.util.concurrent.ringbuffer.ManyToOneRingBuffer;
import uk.co.real_logic.aeron.util.concurrent.ringbuffer.RingBuffer;
import uk.co.real_logic.aeron.util.protocol.DataHeaderFlyweight;
import uk.co.real_logic.aeron.util.protocol.ErrorHeaderFlyweight;

import java.nio.ByteBuffer;
import java.util.concurrent.ThreadLocalRandom;
import java.util.concurrent.TimeUnit;
import java.util.function.Supplier;

import static java.nio.ByteBuffer.allocateDirect;
import static uk.co.real_logic.aeron.mediadriver.MediaDriver.*;
import static uk.co.real_logic.aeron.util.command.ControlProtocolEvents.*;

/**
 * Media Conductor to take commands from Publishers and Subscribers as well as handle NAKs and retransmissions
 */
public class MediaConductor extends Agent
{
    public static final int WRITE_BUFFER_CAPACITY = 512;
    public static final int HEADER_LENGTH = DataHeaderFlyweight.HEADER_LENGTH;
    public static final int HEARTBEAT_TIMEOUT_MS = 100;

    private final RingBuffer mediaCommandBuffer;
    private final ReceiverProxy receiverProxy;
    private final NioSelector nioSelector;
    private final Receiver receiver;
    private final Sender sender;
    private final BufferManagement bufferManagement;
    private final RingBuffer toMediaDriverBuffer;
    private final RingBuffer toClientBuffer;
    private final Long2ObjectHashMap<ControlFrameHandler> srcDestinationMap = new Long2ObjectHashMap<>();
    private final AtomicBuffer writeBuffer = new AtomicBuffer(allocateDirect(WRITE_BUFFER_CAPACITY));
    private final TimerWheel timerWheel;

    private final Supplier<SenderControlStrategy> senderFlowControl;

    private final ThreadLocalRandom rng = ThreadLocalRandom.current();
    private final ChannelMessageFlyweight channelMessage = new ChannelMessageFlyweight();
    private final SubscriberMessageFlyweight receiverMessage = new SubscriberMessageFlyweight();
    private final ErrorHeaderFlyweight errorHeader = new ErrorHeaderFlyweight();
    private final QualifiedMessageFlyweight qualifiedMessage = new QualifiedMessageFlyweight();
    private final NewBufferMessageFlyweight newBufferMessage = new NewBufferMessageFlyweight();

    private final int mtuLength;
    private final ConductorByteBuffers conductorByteBuffers;
    private final TimerWheel.Timer heartbeatTimer;

    public MediaConductor(final Context ctx, final Receiver receiver, final Sender sender)
    {
        super(SELECT_TIMEOUT);

        this.mediaCommandBuffer = ctx.mediaCommandBuffer();
        this.receiverProxy = new ReceiverProxy(ctx.receiverCommandBuffer(), ctx.receiverNioSelector());
        this.bufferManagement = ctx.bufferManagement();
        this.nioSelector = ctx.conductorNioSelector();
        this.mtuLength = ctx.mtuLength();
        this.receiver = receiver;
        this.sender = sender;
        this.senderFlowControl = ctx.senderFlowControl();

        newBufferMessage.wrap(writeBuffer, 0);

        timerWheel = ctx.conductorTimerWheel() != null ?
            ctx.conductorTimerWheel() :
            new TimerWheel(MEDIA_CONDUCTOR_TICK_DURATION_US,
                           TimeUnit.MICROSECONDS,
                           MEDIA_CONDUCTOR_TICKS_PER_WHEEL);

        heartbeatTimer = newTimeout(HEARTBEAT_TIMEOUT_MS, TimeUnit.MILLISECONDS, this::onHeartbeatCheck);

        try
        {
            conductorByteBuffers = ctx.conductorByteBuffers();
            final ByteBuffer toMediaDriver = conductorByteBuffers.toMediaDriver();
            final ByteBuffer toClient = conductorByteBuffers.toClient();
            toMediaDriverBuffer = new ManyToOneRingBuffer(new AtomicBuffer(toMediaDriver));
            toClientBuffer = new ManyToOneRingBuffer(new AtomicBuffer(toClient));
        }
        catch (final Exception ex)
        {
            throw new IllegalStateException("Unable to create the conductor media buffers", ex);
        }
    }

    public ControlFrameHandler frameHandler(final UdpDestination destination)
    {
        return srcDestinationMap.get(destination.consistentHash());
    }

    public void process()
    {
        try
        {
            nioSelector.processKeys();
        }
        catch (final Exception ex)
        {
            // TODO: error
            ex.printStackTrace();
        }

        sender.processBufferRotation();
        receiver.processBufferRotation();
        processReceiveBuffer();
        processCommandBuffer();
        processTimers();
    }

{'fraction_non_alphanumeric': 0.0740403027263609, 'fraction_numerical': 0.003282574997720434, 'mean_word_length': 3.7296248382923674, 'pattern_counts': {'":': 0, '<': 9, '<?xml version=': 0, '>': 9, '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': '7253930', 'n_tokens_mistral': 2994, 'n_tokens_neox': 2829, 'n_words': 725}