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

import io.aeron.exceptions.DriverTimeoutException;
import io.aeron.logbuffer.FragmentHandler;
import org.agrona.DirectBuffer;
import org.agrona.ErrorHandler;
import org.agrona.IoUtil;
import org.agrona.concurrent.*;
import org.agrona.concurrent.broadcast.BroadcastReceiver;
import org.agrona.concurrent.broadcast.CopyBroadcastReceiver;
import org.agrona.concurrent.ringbuffer.ManyToOneRingBuffer;
import org.agrona.concurrent.ringbuffer.RingBuffer;
import org.agrona.concurrent.status.CountersReader;

import java.io.File;
import java.nio.MappedByteBuffer;
import java.util.concurrent.ThreadFactory;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicIntegerFieldUpdater;
import java.util.concurrent.locks.Lock;
import java.util.concurrent.locks.ReentrantLock;

import static java.util.concurrent.atomic.AtomicIntegerFieldUpdater.newUpdater;
import static org.agrona.IoUtil.mapExistingFile;

/**
 * Aeron entry point for communicating to the Media Driver for creating {@link Publication}s and {@link Subscription}s.
 * Use an {@link Aeron.Context} to configure the Aeron object.
 * <p>
 * A client application requires only one Aeron object per Media Driver.
 * <p>
 * <b>Note:</b> If {@link Aeron.Context#errorHandler(ErrorHandler)} is not set and a {@link DriverTimeoutException}
 * occurs then the process will face the wrath of {@link System#exit(int)}. See {@link #DEFAULT_ERROR_HANDLER}.
 */
public class Aeron implements AutoCloseable
{
    /**
     * Using an integer because there is no support for boolean. 1 is closed and 0 is not closed.
     */
    private static final AtomicIntegerFieldUpdater<Aeron> IS_CLOSED_UPDATER = newUpdater(Aeron.class, "isClosed");

    /**
     * The Default handler for Aeron runtime exceptions.
     * When a {@link io.aeron.exceptions.DriverTimeoutException} is encountered, this handler will
     * exit the program.
     * <p>
     * The error handler can be overridden by supplying an {@link Aeron.Context} with a custom handler.
     *
     * @see Aeron.Context#errorHandler(ErrorHandler)
     */
    public static final ErrorHandler DEFAULT_ERROR_HANDLER =
        (throwable) ->
        {
            throwable.printStackTrace();
            if (throwable instanceof DriverTimeoutException)
            {
                System.err.printf(
                    "%n***%n*** Timeout from the MediaDriver - is it currently running? Exiting.%n***%n");
                System.exit(-1);
            }
        };

    /**
     * Duration in milliseconds for which the client conductor will sleep between duty cycles.
     */
    public static final long IDLE_SLEEP_MS = 16;

    /**
     * Duration in nanoseconds for which the client conductor will sleep between duty cycles.
     */
    public static final long IDLE_SLEEP_NS = TimeUnit.MILLISECONDS.toNanos(IDLE_SLEEP_MS);

    /**
     * Default interval between sending keepalive control messages to the driver.
     */
    public static final long KEEPALIVE_INTERVAL_NS = TimeUnit.MILLISECONDS.toNanos(500);

    @SuppressWarnings("unused") private volatile int isClosed;
    private final long clientId;
    private final ClientConductor conductor;
    private final RingBuffer commandBuffer;
    private final AgentInvoker conductorInvoker;
    private final AgentRunner conductorRunner;
    private final Context ctx;

    Aeron(final Context ctx)
    {
        ctx.conclude();

        this.ctx = ctx;
        clientId = ctx.clientId();
        commandBuffer = ctx.toDriverBuffer();
        conductor = new ClientConductor(ctx);

        if (ctx.useConductorAgentInvoker())
        {
            conductorInvoker = new AgentInvoker(ctx.errorHandler(), null, conductor);
            conductorRunner = null;
        }
        else
        {
            conductorInvoker = null;
            conductorRunner = new AgentRunner(ctx.idleStrategy(), ctx.errorHandler(), null, conductor);
        }
    }

    /**
     * Create an Aeron instance and connect to the media driver with a default {@link Context}.
     * <p>
     * Threads required for interacting with the media driver are created and managed within the Aeron instance.
     *
     * @return the new {@link Aeron} instance connected to the Media Driver.
     */
    public static Aeron connect()
    {
        return connect(new Context());
    }

    /**
     * Create an Aeron instance and connect to the media driver.
     * <p>
     * Threads required for interacting with the media driver are created and managed within the Aeron instance.
     * <p>
     * If an exception occurs while trying to establish a connection then the {@link Context#close()} method
     * will be called on the passed context.
     *
     * @param ctx for configuration of the client.
     * @return the new {@link Aeron} instance connected to the Media Driver.
     */
    public static Aeron connect(final Context ctx)
    {
        try
        {
            final Aeron aeron = new Aeron(ctx);

            if (ctx.useConductorAgentInvoker())
            {
                aeron.conductorInvoker.start();
            }
            else
            {
                AgentRunner.startOnThread(aeron.conductorRunner, ctx.threadFactory);
            }

            return aeron;
        }
        catch (final Exception ex)
        {
            ctx.close();
            throw ex;
        }
    }

    /**
     * Print out the values from {@link #countersReader()} to {@link System#out} which can be useful for debugging.
     */
    public void printCounters()
    {
        final CountersReader counters = countersReader();
        counters.forEach((id, label) -> System.out.println(counters.getCounterValue(id) + "\t: " + label));
    }

    /**
     * Has the client been closed? If not then the CnC file may not be unmapped.
     *
     * @return true if the client has been explicitly closed otherwise false.
     */
    public boolean isClosed()
    {
        return 1 == isClosed;
    }

    /**
     * Get the {@link Aeron.Context} that is used by this client.
     *
     * @return the {@link Aeron.Context} that is use by this client.
     */
    public Context context()
    {
        return ctx;
    }

    /**
     * Get the client identity that has been allocated for communicating with the media driver.<commit_msg>[Java] Format debug output of counters to be like AeronStat.<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;

import io.aeron.exceptions.DriverTimeoutException;
import io.aeron.logbuffer.FragmentHandler;
import org.agrona.DirectBuffer;
import org.agrona.ErrorHandler;
import org.agrona.IoUtil;
import org.agrona.concurrent.*;
import org.agrona.concurrent.broadcast.BroadcastReceiver;
import org.agrona.concurrent.broadcast.CopyBroadcastReceiver;
import org.agrona.concurrent.ringbuffer.ManyToOneRingBuffer;
import org.agrona.concurrent.ringbuffer.RingBuffer;
import org.agrona.concurrent.status.CountersReader;

import java.io.File;
import java.nio.MappedByteBuffer;
import java.util.concurrent.ThreadFactory;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicIntegerFieldUpdater;
import java.util.concurrent.locks.Lock;
import java.util.concurrent.locks.ReentrantLock;

import static java.util.concurrent.atomic.AtomicIntegerFieldUpdater.newUpdater;
import static org.agrona.IoUtil.mapExistingFile;

/**
 * Aeron entry point for communicating to the Media Driver for creating {@link Publication}s and {@link Subscription}s.
 * Use an {@link Aeron.Context} to configure the Aeron object.
 * <p>
 * A client application requires only one Aeron object per Media Driver.
 * <p>
 * <b>Note:</b> If {@link Aeron.Context#errorHandler(ErrorHandler)} is not set and a {@link DriverTimeoutException}
 * occurs then the process will face the wrath of {@link System#exit(int)}. See {@link #DEFAULT_ERROR_HANDLER}.
 */
public class Aeron implements AutoCloseable
{
    /**
     * Using an integer because there is no support for boolean. 1 is closed and 0 is not closed.
     */
    private static final AtomicIntegerFieldUpdater<Aeron> IS_CLOSED_UPDATER = newUpdater(Aeron.class, "isClosed");

    /**
     * The Default handler for Aeron runtime exceptions.
     * When a {@link io.aeron.exceptions.DriverTimeoutException} is encountered, this handler will
     * exit the program.
     * <p>
     * The error handler can be overridden by supplying an {@link Aeron.Context} with a custom handler.
     *
     * @see Aeron.Context#errorHandler(ErrorHandler)
     */
    public static final ErrorHandler DEFAULT_ERROR_HANDLER =
        (throwable) ->
        {
            throwable.printStackTrace();
            if (throwable instanceof DriverTimeoutException)
            {
                System.err.printf(
                    "%n***%n*** Timeout from the MediaDriver - is it currently running? Exiting.%n***%n");
                System.exit(-1);
            }
        };

    /**
     * Duration in milliseconds for which the client conductor will sleep between duty cycles.
     */
    public static final long IDLE_SLEEP_MS = 16;

    /**
     * Duration in nanoseconds for which the client conductor will sleep between duty cycles.
     */
    public static final long IDLE_SLEEP_NS = TimeUnit.MILLISECONDS.toNanos(IDLE_SLEEP_MS);

    /**
     * Default interval between sending keepalive control messages to the driver.
     */
    public static final long KEEPALIVE_INTERVAL_NS = TimeUnit.MILLISECONDS.toNanos(500);

    @SuppressWarnings("unused") private volatile int isClosed;
    private final long clientId;
    private final ClientConductor conductor;
    private final RingBuffer commandBuffer;
    private final AgentInvoker conductorInvoker;
    private final AgentRunner conductorRunner;
    private final Context ctx;

    Aeron(final Context ctx)
    {
        ctx.conclude();

        this.ctx = ctx;
        clientId = ctx.clientId();
        commandBuffer = ctx.toDriverBuffer();
        conductor = new ClientConductor(ctx);

        if (ctx.useConductorAgentInvoker())
        {
            conductorInvoker = new AgentInvoker(ctx.errorHandler(), null, conductor);
            conductorRunner = null;
        }
        else
        {
            conductorInvoker = null;
            conductorRunner = new AgentRunner(ctx.idleStrategy(), ctx.errorHandler(), null, conductor);
        }
    }

    /**
     * Create an Aeron instance and connect to the media driver with a default {@link Context}.
     * <p>
     * Threads required for interacting with the media driver are created and managed within the Aeron instance.
     *
     * @return the new {@link Aeron} instance connected to the Media Driver.
     */
    public static Aeron connect()
    {
        return connect(new Context());
    }

    /**
     * Create an Aeron instance and connect to the media driver.
     * <p>
     * Threads required for interacting with the media driver are created and managed within the Aeron instance.
     * <p>
     * If an exception occurs while trying to establish a connection then the {@link Context#close()} method
     * will be called on the passed context.
     *
     * @param ctx for configuration of the client.
     * @return the new {@link Aeron} instance connected to the Media Driver.
     */
    public static Aeron connect(final Context ctx)
    {
        try
        {
            final Aeron aeron = new Aeron(ctx);

            if (ctx.useConductorAgentInvoker())
            {
                aeron.conductorInvoker.start();
            }
            else
            {
                AgentRunner.startOnThread(aeron.conductorRunner, ctx.threadFactory);
            }

            return aeron;
        }
        catch (final Exception ex)
        {
            ctx.close();
            throw ex;
        }
    }

    /**
     * Print out the values from {@link #countersReader()} to {@link System#out} which can be useful for debugging.
     */
    public void printCounters()
    {
        final CountersReader counters = countersReader();
        counters.forEach(
            (id, label) ->
            {
                final long value = counters.getCounterValue(id);
                System.out.format("%3d: %,20d - %s%n", id, value, label);
            });
    }

    /**
     * Has the client been closed? If not then the CnC file may not be unmapped.
     *
     * @return true if the client has been explicitly closed otherwise false.
     */
    public boolean isClosed()
    {
        return 1 == isClosed;
    }

    /**
     * Get the {@link Aeron.Context} that is used by this client.
     *
     * @return the {@link Aeron.Context} that is use by this client.
     */
    public Context context()
    {
        return ctx;
    }

    /**
     * Get the client identity that has been allocated for communicating with the media driver.
{'fraction_non_alphanumeric': 0.08582222539026303, 'fraction_numerical': 0.00320764131442013, 'mean_word_length': 2.678552700576822, 'pattern_counts': {'":': 0, '<': 21, '<?xml version=': 0, '>': 25, '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': '125458', 'n_tokens_mistral': 3866, 'n_tokens_neox': 3722, 'n_words': 1294}