<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.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 MSG_BUFFER_CAPACITY = 4096;
    public static final int HEADER_LENGTH = DataHeaderFlyweight.HEADER_LENGTH;
    public static final int HEARTBEAT_TIMEOUT_MS = 100;

    private final RingBuffer commandBuffer;
    private final ReceiverProxy receiverProxy;
    private final NioSelector nioSelector;
    private final Receiver receiver;
    private final Sender sender;
    private final BufferManagement bufferManagement;
    private final RingBuffer toMediaBuffer;
    private final RingBuffer toClientBuffer;
    private final Long2ObjectHashMap<ControlFrameHandler> srcDestinationMap = new Long2ObjectHashMap<>();
    private final AtomicBuffer msgBuffer = new AtomicBuffer(allocateDirect(MSG_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 ConductorShmBuffers conductorShmBuffers;
    private final TimerWheel.Timer heartbeatTimer;

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

        this.commandBuffer = ctx.mediaCommandBuffer();
        this.receiverProxy = ctx.receiverProxy();
        this.bufferManagement = ctx.bufferManagement();
        this.nioSelector = ctx.conductorNioSelector();
        this.mtuLength = ctx.mtuLength();
        this.receiver = receiver;
        this.sender = sender;
        this.senderFlowControl = ctx.senderFlowControl();

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

        conductorShmBuffers = ctx.conductorShmBuffers();
        toMediaBuffer = new ManyToOneRingBuffer(new AtomicBuffer(conductorShmBuffers.toDriver()));
        toClientBuffer = new ManyToOneRingBuffer(new AtomicBuffer(conductorShmBuffers.toClient()));

        newBufferMessage.wrap(msgBuffer, 0);
    }

    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();
    }

    public void close()
    {
        stop();
        wakeup();

        srcDestinationMap.forEach((hash, frameHandler) -> frameHandler.close());

        conductorShmBuffers.close();
    }

    public void wakeup()
    {
        // TODO
    }

    private void processCommandBuffer()
    {
        commandBuffer.read(
            (msgTypeId, buffer, index, length) ->
            {
                switch (msgTypeId)
                {
                    case CREATE_TERM_BUFFER:
                        qualifiedMessage.wrap(buffer, index);
                        onCreateSubscriberTermBuffer(qualifiedMessage);
                        break;

                    case REMOVE_TERM_BUFFER:
                        qualifiedMessage.wrap(buffer, index);
                        onRemoveSubscriberTermBuffer(qualifiedMessage);
                        break;

                    case ERROR_RESPONSE:
                        errorHeader.wrap(buffer, index);
                        toClientBuffer.write(msgTypeId, buffer, index, length);
                        break;
                }
            });
    }

    private void processReceiveBuffer()
    {
        toMediaBuffer.read(
            (msgTypeId, buffer, index, length) ->
            {
                Flyweight flyweight = channelMessage;

                try
                {
                    switch (msgTypeId)
                    {
                        case ADD_CHANNEL:
                            channelMessage.wrap(buffer, index);
                            flyweight = channelMessage;
                            onAddChannel(channelMessage);
                            break;

                        case REMOVE_CHANNEL:
                            channelMessage.wrap(buffer, index);
                            flyweight = channelMessage;<commit_msg>[Java]: Rename command buffers in MediaConductor.<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.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 MSG_BUFFER_CAPACITY = 4096;
    public static final int HEADER_LENGTH = DataHeaderFlyweight.HEADER_LENGTH;
    public static final int HEARTBEAT_TIMEOUT_MS = 100;

    private final RingBuffer localCommandBuffer;
    private final ReceiverProxy receiverProxy;
    private final NioSelector nioSelector;
    private final Receiver receiver;
    private final Sender sender;
    private final BufferManagement bufferManagement;
    private final RingBuffer clientCommandBuffer;
    private final RingBuffer toClientBuffer;
    private final Long2ObjectHashMap<ControlFrameHandler> srcDestinationMap = new Long2ObjectHashMap<>();
    private final AtomicBuffer msgBuffer = new AtomicBuffer(allocateDirect(MSG_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 ConductorShmBuffers conductorShmBuffers;
    private final TimerWheel.Timer heartbeatTimer;

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

        this.localCommandBuffer = ctx.mediaCommandBuffer();
        this.receiverProxy = ctx.receiverProxy();
        this.bufferManagement = ctx.bufferManagement();
        this.nioSelector = ctx.conductorNioSelector();
        this.mtuLength = ctx.mtuLength();
        this.receiver = receiver;
        this.sender = sender;
        this.senderFlowControl = ctx.senderFlowControl();

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

        conductorShmBuffers = ctx.conductorShmBuffers();
        clientCommandBuffer = new ManyToOneRingBuffer(new AtomicBuffer(conductorShmBuffers.toDriver()));
        toClientBuffer = new ManyToOneRingBuffer(new AtomicBuffer(conductorShmBuffers.toClient()));

        newBufferMessage.wrap(msgBuffer, 0);
    }

    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();
        processClientCommandBuffer();
        processCommandBuffer();
        processTimers();
    }

    public void close()
    {
        stop();
        wakeup();

        srcDestinationMap.forEach((hash, frameHandler) -> frameHandler.close());

        conductorShmBuffers.close();
    }

    public void wakeup()
    {
        // TODO
    }

    private void processCommandBuffer()
    {
        localCommandBuffer.read(
            (msgTypeId, buffer, index, length) ->
            {
                switch (msgTypeId)
                {
                    case CREATE_TERM_BUFFER:
                        qualifiedMessage.wrap(buffer, index);
                        onCreateSubscriberTermBuffer(qualifiedMessage);
                        break;

                    case REMOVE_TERM_BUFFER:
                        qualifiedMessage.wrap(buffer, index);
                        onRemoveSubscriberTermBuffer(qualifiedMessage);
                        break;

                    case ERROR_RESPONSE:
                        errorHeader.wrap(buffer, index);
                        toClientBuffer.write(msgTypeId, buffer, index, length);
                        break;
                }
            });
    }

    private void processClientCommandBuffer()
    {
        clientCommandBuffer.read(
            (msgTypeId, buffer, index, length) ->
            {
                Flyweight flyweight = channelMessage;

                try
                {
                    switch (msgTypeId)
                    {
                        case ADD_CHANNEL:
                            channelMessage.wrap(buffer, index);
                            flyweight = channelMessage;
                            onAddChannel(channelMessage);
                            break;

                        case REMOVE_CHANNEL:
                            channelMessage.wrap(buffer, index);
                            flyweight = channelMessage;
{'fraction_non_alphanumeric': 0.07405191113770664, 'fraction_numerical': 0.002842396589124093, 'mean_word_length': 2.6242884250474385, 'pattern_counts': {'":': 0, '<': 9, '<?xml version=': 0, '>': 15, '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': '7154765', 'n_tokens_mistral': 3507, 'n_tokens_neox': 3283, 'n_words': 804}