<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
 *
 * https://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.*;
import io.aeron.status.ChannelEndpointStatus;
import org.agrona.DirectBuffer;
import org.agrona.LangUtil;
import org.agrona.ManagedResource;
import org.agrona.collections.ArrayListUtil;
import org.agrona.collections.Long2ObjectHashMap;
import org.agrona.concurrent.*;
import org.agrona.concurrent.status.CountersManager;
import org.agrona.concurrent.status.CountersReader;
import org.agrona.concurrent.status.UnsafeBufferPosition;

import java.util.ArrayList;
import java.util.concurrent.locks.Lock;

import static io.aeron.Aeron.Configuration.IDLE_SLEEP_MS;
import static io.aeron.Aeron.Configuration.IDLE_SLEEP_NS;
import static java.nio.charset.StandardCharsets.US_ASCII;
import static java.util.concurrent.TimeUnit.MILLISECONDS;
import static java.util.concurrent.TimeUnit.NANOSECONDS;

/**
 * Client conductor receives responses and notifications from Media Driver and acts on them in addition to forwarding
 * commands from the Client API to the Media Driver conductor.
 */
class ClientConductor implements Agent, DriverEventsListener
{
    private static final long NO_CORRELATION_ID = Aeron.NULL_VALUE;

    private final long closeLingerDurationMs;
    private final long keepAliveIntervalNs;
    private final long driverTimeoutMs;
    private final long driverTimeoutNs;
    private final long interServiceTimeoutNs;
    private long timeOfLastKeepAliveNs;
    private long timeOfLastServiceNs;
    private boolean isClosed;
    private boolean isInCallback;
    private boolean isTerminating;
    private String stashedChannel;
    private RegistrationException driverException;

    private final Aeron.Context ctx;
    private final Aeron aeron;
    private final Lock clientLock;
    private final EpochClock epochClock;
    private final NanoClock nanoClock;
    private final DriverEventsAdapter driverEventsAdapter;
    private final LogBuffersFactory logBuffersFactory;
    private final Long2ObjectHashMap<LogBuffers> logBuffersByIdMap = new Long2ObjectHashMap<>();
    private final Long2ObjectHashMap<Object> resourceByRegIdMap = new Long2ObjectHashMap<>();
    private final ArrayList<ManagedResource> lingeringResources = new ArrayList<>();
    private final AvailableImageHandler defaultAvailableImageHandler;
    private final UnavailableImageHandler defaultUnavailableImageHandler;
    private final ArrayList<AvailableCounterHandler> availableCounterHandlers = new ArrayList<>();
    private final ArrayList<UnavailableCounterHandler> unavailableCounterHandlers = new ArrayList<>();
    private final ArrayList<Runnable> closeHandlers = new ArrayList<>();
    private final DriverProxy driverProxy;
    private final AgentInvoker driverAgentInvoker;
    private final UnsafeBuffer counterValuesBuffer;
    private final CountersReader countersReader;

    ClientConductor(final Aeron.Context ctx, final Aeron aeron)
    {
        this.ctx = ctx;
        this.aeron = aeron;

        clientLock = ctx.clientLock();
        epochClock = ctx.epochClock();
        nanoClock = ctx.nanoClock();
        driverProxy = ctx.driverProxy();
        logBuffersFactory = ctx.logBuffersFactory();
        keepAliveIntervalNs = ctx.keepAliveIntervalNs();
        driverTimeoutMs = ctx.driverTimeoutMs();
        driverTimeoutNs = MILLISECONDS.toNanos(driverTimeoutMs);
        closeLingerDurationMs = NANOSECONDS.toMillis(ctx.closeLingerDurationNs());
        interServiceTimeoutNs = ctx.interServiceTimeoutNs();
        defaultAvailableImageHandler = ctx.availableImageHandler();
        defaultUnavailableImageHandler = ctx.unavailableImageHandler();
        driverEventsAdapter = new DriverEventsAdapter(ctx.toClientBuffer(), ctx.clientId(), this);
        driverAgentInvoker = ctx.driverAgentInvoker();
        counterValuesBuffer = ctx.countersValuesBuffer();
        countersReader = new CountersReader(ctx.countersMetaDataBuffer(), ctx.countersValuesBuffer(), US_ASCII);

        if (null != ctx.availableCounterHandler())
        {
            availableCounterHandlers.add(ctx.availableCounterHandler());
        }

        if (null != ctx.unavailableCounterHandler())
        {
            unavailableCounterHandlers.add(ctx.unavailableCounterHandler());
        }

        if (null != ctx.closeHandler())
        {
            closeHandlers.add(ctx.closeHandler());
        }

        final long nowNs = nanoClock.nanoTime();
        timeOfLastKeepAliveNs = nowNs;
        timeOfLastServiceNs = nowNs;
    }

    public void onClose()
    {
        clientLock.lock();
        try
        {
            if (!isClosed)
            {
                isClosed = true;
                forceCloseResources();

                for (int i = closeHandlers.size() - 1; i >= 0; i--)
                {
                    try
                    {
                        closeHandlers.get(i).run();
                    }
                    catch (final Exception ex)
                    {
                        handleError(ex);
                    }
                }

                try
                {
                    if (isTerminating)
                    {
                        aeron.internalClose();
                        Thread.sleep(IDLE_SLEEP_MS);
                    }

                    Thread.sleep(closeLingerDurationMs);
                }
                catch (final InterruptedException ignore)
                {
                    Thread.currentThread().interrupt();
                }

                for (int i = 0, size = lingeringResources.size(); i < size; i++)
                {
                    lingeringResources.get(i).delete();
                }

                driverProxy.clientClose();
                ctx.close();
            }
        }
        finally
        {
            clientLock.unlock();
        }
    }

    public int doWork()
    {
        int workCount = 0;<commit_msg>[Java] Close Aeron client before calling close handlers for consistent behaviour.<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
 *
 * https://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.*;
import io.aeron.status.ChannelEndpointStatus;
import org.agrona.DirectBuffer;
import org.agrona.LangUtil;
import org.agrona.ManagedResource;
import org.agrona.collections.ArrayListUtil;
import org.agrona.collections.Long2ObjectHashMap;
import org.agrona.concurrent.*;
import org.agrona.concurrent.status.CountersManager;
import org.agrona.concurrent.status.CountersReader;
import org.agrona.concurrent.status.UnsafeBufferPosition;

import java.util.ArrayList;
import java.util.concurrent.locks.Lock;

import static io.aeron.Aeron.Configuration.IDLE_SLEEP_MS;
import static io.aeron.Aeron.Configuration.IDLE_SLEEP_NS;
import static java.nio.charset.StandardCharsets.US_ASCII;
import static java.util.concurrent.TimeUnit.MILLISECONDS;
import static java.util.concurrent.TimeUnit.NANOSECONDS;

/**
 * Client conductor receives responses and notifications from Media Driver and acts on them in addition to forwarding
 * commands from the Client API to the Media Driver conductor.
 */
class ClientConductor implements Agent, DriverEventsListener
{
    private static final long NO_CORRELATION_ID = Aeron.NULL_VALUE;

    private final long closeLingerDurationMs;
    private final long keepAliveIntervalNs;
    private final long driverTimeoutMs;
    private final long driverTimeoutNs;
    private final long interServiceTimeoutNs;
    private long timeOfLastKeepAliveNs;
    private long timeOfLastServiceNs;
    private boolean isClosed;
    private boolean isInCallback;
    private boolean isTerminating;
    private String stashedChannel;
    private RegistrationException driverException;

    private final Aeron.Context ctx;
    private final Aeron aeron;
    private final Lock clientLock;
    private final EpochClock epochClock;
    private final NanoClock nanoClock;
    private final DriverEventsAdapter driverEventsAdapter;
    private final LogBuffersFactory logBuffersFactory;
    private final Long2ObjectHashMap<LogBuffers> logBuffersByIdMap = new Long2ObjectHashMap<>();
    private final Long2ObjectHashMap<Object> resourceByRegIdMap = new Long2ObjectHashMap<>();
    private final ArrayList<ManagedResource> lingeringResources = new ArrayList<>();
    private final AvailableImageHandler defaultAvailableImageHandler;
    private final UnavailableImageHandler defaultUnavailableImageHandler;
    private final ArrayList<AvailableCounterHandler> availableCounterHandlers = new ArrayList<>();
    private final ArrayList<UnavailableCounterHandler> unavailableCounterHandlers = new ArrayList<>();
    private final ArrayList<Runnable> closeHandlers = new ArrayList<>();
    private final DriverProxy driverProxy;
    private final AgentInvoker driverAgentInvoker;
    private final UnsafeBuffer counterValuesBuffer;
    private final CountersReader countersReader;

    ClientConductor(final Aeron.Context ctx, final Aeron aeron)
    {
        this.ctx = ctx;
        this.aeron = aeron;

        clientLock = ctx.clientLock();
        epochClock = ctx.epochClock();
        nanoClock = ctx.nanoClock();
        driverProxy = ctx.driverProxy();
        logBuffersFactory = ctx.logBuffersFactory();
        keepAliveIntervalNs = ctx.keepAliveIntervalNs();
        driverTimeoutMs = ctx.driverTimeoutMs();
        driverTimeoutNs = MILLISECONDS.toNanos(driverTimeoutMs);
        closeLingerDurationMs = NANOSECONDS.toMillis(ctx.closeLingerDurationNs());
        interServiceTimeoutNs = ctx.interServiceTimeoutNs();
        defaultAvailableImageHandler = ctx.availableImageHandler();
        defaultUnavailableImageHandler = ctx.unavailableImageHandler();
        driverEventsAdapter = new DriverEventsAdapter(ctx.toClientBuffer(), ctx.clientId(), this);
        driverAgentInvoker = ctx.driverAgentInvoker();
        counterValuesBuffer = ctx.countersValuesBuffer();
        countersReader = new CountersReader(ctx.countersMetaDataBuffer(), ctx.countersValuesBuffer(), US_ASCII);

        if (null != ctx.availableCounterHandler())
        {
            availableCounterHandlers.add(ctx.availableCounterHandler());
        }

        if (null != ctx.unavailableCounterHandler())
        {
            unavailableCounterHandlers.add(ctx.unavailableCounterHandler());
        }

        if (null != ctx.closeHandler())
        {
            closeHandlers.add(ctx.closeHandler());
        }

        final long nowNs = nanoClock.nanoTime();
        timeOfLastKeepAliveNs = nowNs;
        timeOfLastServiceNs = nowNs;
    }

    public void onClose()
    {
        clientLock.lock();
        try
        {
            if (!isClosed)
            {
                isClosed = true;
                if (isTerminating)
                {
                    aeron.internalClose();
                }

                forceCloseResources();

                for (int i = closeHandlers.size() - 1; i >= 0; i--)
                {
                    try
                    {
                        closeHandlers.get(i).run();
                    }
                    catch (final Exception ex)
                    {
                        handleError(ex);
                    }
                }

                try
                {
                    if (isTerminating)
                    {
                        Thread.sleep(IDLE_SLEEP_MS);
                    }

                    Thread.sleep(closeLingerDurationMs);
                }
                catch (final InterruptedException ignore)
                {
                    Thread.currentThread().interrupt();
                }

                for (int i = 0, size = lingeringResources.size(); i < size; i++)
                {
                    lingeringResources.get(i).delete();
                }

                driverProxy.clientClose();
                ctx.close();
            }
        }
        finally
        {
            clientLock.unlock();
        }
    }

    public int doWork()
    {
        int workCount = 0;
{'fraction_non_alphanumeric': 0.07667173252279635, 'fraction_numerical': 0.0031914893617021275, 'mean_word_length': 2.5695687550854354, 'pattern_counts': {'":': 0, '<': 29, '<?xml version=': 0, '>': 29, 'https://': 2, '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': '1885992', 'n_tokens_mistral': 3379, 'n_tokens_neox': 3227, 'n_words': 847}