/*
 * Licensed to the Apache Software Foundation (ASF) under one or more
 * contributor license agreements.  See the NOTICE file distributed with
 * this work for additional information regarding copyright ownership.
 * The ASF licenses this file to You 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 org.apache.qpid.protonj2.client.impl;

import java.io.IOException;
import java.io.UncheckedIOException;
import java.security.Principal;
import java.util.Map;
import java.util.Objects;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.ExecutionException;
import java.util.concurrent.Future;
import java.util.concurrent.LinkedBlockingQueue;
import java.util.concurrent.RejectedExecutionException;
import java.util.concurrent.ScheduledExecutorService;
import java.util.concurrent.ThreadPoolExecutor;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicIntegerFieldUpdater;
import java.util.concurrent.atomic.AtomicReferenceFieldUpdater;
import java.util.function.BiConsumer;

import org.apache.qpid.protonj2.buffer.ProtonBuffer;
import org.apache.qpid.protonj2.client.Client;
import org.apache.qpid.protonj2.client.Connection;
import org.apache.qpid.protonj2.client.ConnectionEvent;
import org.apache.qpid.protonj2.client.ConnectionOptions;
import org.apache.qpid.protonj2.client.DisconnectionEvent;
import org.apache.qpid.protonj2.client.ErrorCondition;
import org.apache.qpid.protonj2.client.Message;
import org.apache.qpid.protonj2.client.NextReceiverPolicy;
import org.apache.qpid.protonj2.client.Receiver;
import org.apache.qpid.protonj2.client.ReceiverOptions;
import org.apache.qpid.protonj2.client.ReconnectLocation;
import org.apache.qpid.protonj2.client.Sender;
import org.apache.qpid.protonj2.client.SenderOptions;
import org.apache.qpid.protonj2.client.Session;
import org.apache.qpid.protonj2.client.SessionOptions;
import org.apache.qpid.protonj2.client.StreamReceiver;
import org.apache.qpid.protonj2.client.StreamReceiverOptions;
import org.apache.qpid.protonj2.client.StreamSender;
import org.apache.qpid.protonj2.client.StreamSenderOptions;
import org.apache.qpid.protonj2.client.Tracker;
import org.apache.qpid.protonj2.client.exceptions.ClientConnectionRemotelyClosedException;
import org.apache.qpid.protonj2.client.exceptions.ClientConnectionSecurityException;
import org.apache.qpid.protonj2.client.exceptions.ClientConnectionSecuritySaslException;
import org.apache.qpid.protonj2.client.exceptions.ClientException;
import org.apache.qpid.protonj2.client.exceptions.ClientIOException;
import org.apache.qpid.protonj2.client.exceptions.ClientIllegalStateException;
import org.apache.qpid.protonj2.client.exceptions.ClientOperationTimedOutException;
import org.apache.qpid.protonj2.client.exceptions.ClientUnsupportedOperationException;
import org.apache.qpid.protonj2.client.futures.ClientFuture;
import org.apache.qpid.protonj2.client.futures.ClientFutureFactory;
import org.apache.qpid.protonj2.client.transport.NettyIOContext;
import org.apache.qpid.protonj2.client.transport.Transport;
import org.apache.qpid.protonj2.client.util.ReconnectLocationPool;
import org.apache.qpid.protonj2.client.util.TrackableThreadFactory;
import org.apache.qpid.protonj2.engine.Engine;
import org.apache.qpid.protonj2.engine.EngineFactory;
import org.apache.qpid.protonj2.engine.sasl.client.SaslAuthenticator;
import org.apache.qpid.protonj2.engine.sasl.client.SaslCredentialsProvider;
import org.apache.qpid.protonj2.engine.sasl.client.SaslMechanismSelector;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

/**
 * A {@link Connection} implementation that uses the Proton engine for AMQP protocol support.
 */
public final class ClientConnection implements Connection {

    private static final Logger LOG = LoggerFactory.getLogger(ClientConnection.class);

    private static final int UNLIMITED = -1;
    private static final int UNDEFINED = -1;

    // Future tracking of Closing. Closed. Failed state vs just simple boolean is intended here
    // later on we may decide this is overly optimized.
    private static final AtomicIntegerFieldUpdater<ClientConnection> CLOSED_UPDATER =
            AtomicIntegerFieldUpdater.newUpdater(ClientConnection.class, "closed");
    private static final AtomicReferenceFieldUpdater<ClientConnection, ClientIOException> FAILURE_CAUSE_UPDATER =
            AtomicReferenceFieldUpdater.newUpdater(ClientConnection.class, ClientIOException.class, "failureCause");

    private final ClientInstance client;
    private final ConnectionOptions options;
    private final ClientConnectionCapabilities capabilities = new ClientConnectionCapabilities();
    private final ClientFutureFactory futureFactory;
    private final ClientSessionBuilder sessionBuilder;
    private final ReconnectLocationPool reconnectPool = new ReconnectLocationPool();
    private final NettyIOContext ioContext;
    private final String connectionId;
    private final ScheduledExecutorService executor;
    private final Map<ClientFuture<?>, Object> requests = new ConcurrentHashMap<>();
    private final ThreadPoolExecutor notifications;

    private Engine engine;
    private org.apache.qpid.protonj2.engine.Connection protonConnection;
    private ClientSession connectionSession;
    private ClientSender connectionSender;
    private Transport transport;
    private boolean autoFlush = true;
    private ClientFuture<Connection> openFuture;
    private ClientFuture<Connection> closeFuture;
    private volatile int closed;
    private volatile ClientIOException failureCause;
    private long totalConnections;
    private long reconnectAttempts;
    private long nextReconnectDelay = -1;

    /**
     * Create a connection and define the initial configuration used to manage the
     * connection to the remote.
     *
     * @param host
     * 		the host that this connection is connecting to.
     * @param port
     * 		the port on the remote host where this connection attaches.
     * @param client
     *      the {@link Client} that this connection resides within.
     * @param options
     *      the connection options that configure this {@link Connection} instance.
     */
    ClientConnection(ClientInstance client, String host, int port, ConnectionOptions options) {
        this.client = client;
        this.options = options;
        this.connectionId = client.nextConnectionId();
        this.futureFactory = ClientFutureFactory.create(client.options().futureType());
        this.openFuture = futureFactory.createFuture();
        this.closeFuture = futureFactory.createFuture();
        this.sessionBuilder = new ClientSessionBuilder(this);
        this.ioContext = new NettyIOContext(options.transportOptions(),
                                            options.sslOptions(),
                                            "ClientConnection :(" + connectionId + "): I/O Thread");
        this.executor = ioContext.eventLoop();

        // This executor can be used for dispatching asynchronous tasks that might block or result
        // in reentrant calls to this Connection that could block.
        notifications = new ThreadPoolExecutor(1, 1, 1, TimeUnit.MINUTES, new LinkedBlockingQueue<Runnable>(),
            new TrackableThreadFactory("protonj2 Client Connection Executor: " + getId(), true));
        notifications.setRejectedExecutionHandler(new ThreadPoolExecutor.DiscardOldestPolicy());

        reconnectPool.add(new ReconnectLocation(host, port));
        reconnectPool.addAll(options.reconnectOptions().reconnectLocations());
    }

    @Override
    public ClientInstance client() {
        return client;
    }

    @Override
    public Future<Connection> openFuture() {
        return openFuture;
    }

    @Override
    public void close() {
        try {
            doClose(null).get();
        } catch (InterruptedException | ExecutionException e) {
            Thread.interrupted();
        }
    }

    @Override
    public void close(ErrorCondition error) {
        try {
            doClose(error).get();
        } catch (InterruptedException | ExecutionException e) {
            Thread.interrupted();
        }
    }

    @Override
    public Future<Connection> closeAsync() {
        return doClose(null);
    }

    @Override
    public Future<Connection> closeAsync(ErrorCondition error) {
        Objects.requireNonNull(error, "Error supplied cannot be null");

        return doClose(error);
    }

    private Future<Connection> doClose(ErrorCondition error) {
        if (CLOSED_UPDATER.compareAndSet(this, 0, 1)) {
            try {
                // Already closed by failure or shutdown so no need to queue task
                if (!closeFuture.isDone()) {
                    executor.execute(() -> {
                        LOG.trace("Close requested for connection: {}", this);

                        if (protonConnection.isLocallyOpen()) {
                            protonConnection.setCondition(ClientErrorCondition.asProtonErrorCondition(error));

                            try {
                                protonConnection.close();
                            } catch (Throwable ignored) {
                                // Engine error handler will kick in if the write of Close fails
                            }
                        } else {
                            engine.shutdown();
                        }
                    });
                }
            } catch (RejectedExecutionException rje) {
                // If the engine already shutdown due to the remote dropping then we
                // can just ignore that and continue as if everything already was closed.
                LOG.trace("Close task rejected from the event loop", rje);
            }
        }

        return closeFuture;
    }

    @Override
    public Session defaultSession() throws ClientException {
        checkClosedOrFailed();
        final ClientFuture<Session> defaultSession = getFutureFactory().createFuture();

        executor.execute(() -> {
            try {
                checkClosedOrFailed();
                defaultSession.complete(lazyCreateConnectionSession());
            } catch (Throwable error) {
                defaultSession.failed(ClientExceptionSupport.createNonFatalOrPassthrough(error));
            }
        });

        return request(this, defaultSession);
    }

    @Override
    public Session openSession() throws ClientException {
        return openSession(null);
    }

    @Override
    public Session openSession(SessionOptions sessionOptions) throws ClientException {
        checkClosedOrFailed();
        final ClientFuture<Session> createSession = getFutureFactory().createFuture();

        executor.execute(() -> {
            try {
                checkClosedOrFailed();
                createSession.complete(sessionBuilder.session(sessionOptions).open());
            } catch (Throwable error) {
                createSession.failed(ClientExceptionSupport.createNonFatalOrPassthrough(error));
            }
        });

        return request(this, createSession);
    }

    @Override
    public Receiver openReceiver(String address) throws ClientException {
        return openReceiver(address, null);
    }

    @Override
    public Receiver openReceiver(String address, ReceiverOptions receiverOptions) throws ClientException {
        checkClosedOrFailed();
        Objects.requireNonNull(address, "Cannot create a receiver with a null address");
        final ClientFuture<Receiver> createReceiver = getFutureFactory().createFuture();

        executor.execute(() -> {
            try {
                checkClosedOrFailed();
                createReceiver.complete(lazyCreateConnectionSession().internalOpenReceiver(address, receiverOptions));
            } catch (Throwable error) {
                createReceiver.failed(ClientExceptionSupport.createNonFatalOrPassthrough(error));
            }
        });

        return request(this, createReceiver);
    }

    @Override
    public Receiver openDurableReceiver(String address, String subscriptionName) throws ClientException {
        return openDurableReceiver(address, subscriptionName, null);
    }

    @Override
    public Receiver openDurableReceiver(String address, String subscriptionName, ReceiverOptions receiverOptions) throws ClientException {
        checkClosedOrFailed();
        Objects.requireNonNull(address, "Cannot create a receiver with a null address");
        final ClientFuture<Receiver> createReceiver = getFutureFactory().createFuture();

        executor.execute(() -> {
            try {
                checkClosedOrFailed();
                createReceiver.complete(lazyCreateConnectionSession().internalOpenDurableReceiver(address, subscriptionName, receiverOptions));
            } catch (Throwable error) {
                createReceiver.failed(ClientExceptionSupport.createNonFatalOrPassthrough(error));
            }
        });

        return request(this, createReceiver);
    }

    @Override
    public Receiver openDynamicReceiver() throws ClientException {
        return openDynamicReceiver(null, null);
    }

    @Override
    public Receiver openDynamicReceiver(Map<String, Object> dynamicNodeProperties) throws ClientException {
        return openDynamicReceiver(dynamicNodeProperties, null);
    }

    @Override
    public Receiver openDynamicReceiver(ReceiverOptions receiverOptions) throws ClientException {
        return openDynamicReceiver(null, receiverOptions);
    }

    @Override
    public Receiver openDynamicReceiver(Map<String, Object> dynamicNodeProperties, ReceiverOptions receiverOptions) throws ClientException {
        checkClosedOrFailed();
        final ClientFuture<Receiver> createReceiver = getFutureFactory().createFuture();

        executor.execute(() -> {
            try {
                checkClosedOrFailed();
                createReceiver.complete(lazyCreateConnectionSession().internalOpenDynamicReceiver(dynamicNodeProperties, receiverOptions));
            } catch (Throwable error) {
                createReceiver.failed(ClientExceptionSupport.createNonFatalOrPassthrough(error));
            }
        });

        return request(this, createReceiver);
    }

    @Override
    public StreamReceiver openStreamReceiver(String address) throws ClientException {
        return openStreamReceiver(address, null);
    }

    @Override
    public StreamReceiver openStreamReceiver(String address, StreamReceiverOptions receiverOptions) throws ClientException {
        checkClosedOrFailed();
        final ClientFuture<StreamReceiver> createRequest = getFutureFactory().createFuture();

        executor.execute(() -> {
            try {
                int sessionCapacity = StreamReceiverOptions.DEFAULT_READ_BUFFER_SIZE;
                if (receiverOptions != null) {
                    sessionCapacity = receiverOptions.readBufferSize() / 2;
                }

                // Session capacity cannot be smaller than one frame size so we adjust to the lower bound
                sessionCapacity = (int) Math.max(sessionCapacity, protonConnection.getMaxFrameSize());

                checkClosedOrFailed();
                SessionOptions sessionOptions = new SessionOptions(sessionBuilder.getDefaultSessionOptions());
                ClientStreamSession session = (ClientStreamSession) sessionBuilder.streamSession(sessionOptions.incomingCapacity(sessionCapacity)).open();
                createRequest.complete(session.internalOpenStreamReceiver(address, receiverOptions));
            } catch (Throwable error) {
                createRequest.failed(ClientExceptionSupport.createNonFatalOrPassthrough(error));
            }
        });

        return request(this, createRequest);
    }

    @Override
    public Sender defaultSender() throws ClientException {
        checkClosedOrFailed();
        final ClientFuture<Sender> defaultSender = getFutureFactory().createFuture();

        executor.execute(() -> {
            try {
                checkClosedOrFailed();
                defaultSender.complete(lazyCreateConnectionSender());
            } catch (Throwable error) {
                defaultSender.failed(ClientExceptionSupport.createNonFatalOrPassthrough(error));
            }
        });

        return request(this, defaultSender);
    }

    @Override
    public Sender openSender(String address) throws ClientException {
        return openSender(address, null);
    }

    @Override
    public Sender openSender(String address, SenderOptions senderOptions) throws ClientException {
        checkClosedOrFailed();
        Objects.requireNonNull(address, "Cannot create a sender with a null address");
        final ClientFuture<Sender> createSender = getFutureFactory().createFuture();

        executor.execute(() -> {
            try {
                checkClosedOrFailed();
                createSender.complete(lazyCreateConnectionSession().internalOpenSender(address, senderOptions));
            } catch (Throwable error) {
                createSender.failed(ClientExceptionSupport.createNonFatalOrPassthrough(error));
            }
        });

        return request(this, createSender);
    }

    @Override
    public Sender openAnonymousSender() throws ClientException {
        return openAnonymousSender(null);
    }

    @Override
    public Sender openAnonymousSender(SenderOptions senderOptions) throws ClientException {
        checkClosedOrFailed();
        final ClientFuture<Sender> createRequest = getFutureFactory().createFuture();

        executor.execute(() -> {
            try {
                checkClosedOrFailed();
                createRequest.complete(lazyCreateConnectionSession().internalOpenAnonymousSender(senderOptions));
            } catch (Throwable error) {
                createRequest.failed(ClientExceptionSupport.createNonFatalOrPassthrough(error));
            }
        });

        return request(this, createRequest);
    }

    @Override
    public StreamSender openStreamSender(String address) throws ClientException {
        return openStreamSender(address, null);
    }

    @Override
    public StreamSender openStreamSender(String address, StreamSenderOptions senderOptions) throws ClientException {
        checkClosedOrFailed();
        Objects.requireNonNull(address, "Cannot create a sender with a null address");
        final ClientFuture<StreamSender> createRequest = getFutureFactory().createFuture();

        executor.execute(() -> {
            try {
                int sessionCapacity = StreamSenderOptions.DEFAULT_PENDING_WRITES_BUFFER_SIZE;
                if (senderOptions != null) {
                    sessionCapacity = senderOptions.pendingWritesBufferSize();
                }

                // Session capacity cannot be smaller than one frame size so we adjust to the lower bound
                sessionCapacity = (int) Math.max(sessionCapacity, protonConnection.getMaxFrameSize());

                checkClosedOrFailed();
                SessionOptions sessionOptions = new SessionOptions(sessionBuilder.getDefaultSessionOptions());
                ClientStreamSession session = (ClientStreamSession) sessionBuilder.streamSession(sessionOptions.outgoingCapacity(sessionCapacity)).open();
                createRequest.complete(session.internalOpenStreamSender(address, senderOptions));
            } catch (Throwable error) {
                createRequest.failed(ClientExceptionSupport.createNonFatalOrPassthrough(error));
            }
        });

        return request(this, createRequest);
    }

    @Override
    public Tracker send(Message<?> message) throws ClientException {
        checkClosedOrFailed();
        Objects.requireNonNull(message, "Cannot send a null message");
        final ClientFuture<Sender> result = getFutureFactory().createFuture();

        executor.execute(() -> {
            try {
                checkClosedOrFailed();
                result.complete(lazyCreateConnectionSender());
            } catch (Throwable error) {
                result.failed(ClientExceptionSupport.createNonFatalOrPassthrough(error));
            }
        });

        return request(this, result).send(message);
    }

    @Override
    public Receiver nextReceiver() throws ClientException {
        checkClosedOrFailed();
        return defaultSession().nextReceiver();
    }

    @Override
    public Receiver nextReceiver(long timeout, TimeUnit unit) throws ClientException {
        checkClosedOrFailed();
        return defaultSession().nextReceiver(timeout, unit);
    }

    @Override
    public Receiver nextReceiver(NextReceiverPolicy policy) throws ClientException {
        checkClosedOrFailed();
        return defaultSession().nextReceiver(policy);
    }

    @Override
    public Receiver nextReceiver(NextReceiverPolicy policy, long timeout, TimeUnit unit) throws ClientException {
        checkClosedOrFailed();
        return defaultSession().nextReceiver(policy, timeout, unit);
    }

    @Override
    public Map<String, Object> properties() throws ClientException {
        waitForOpenToComplete();
        return ClientConversionSupport.toStringKeyedMap(protonConnection.getRemoteProperties());
    }

    @Override
    public String[] offeredCapabilities() throws ClientException {
        waitForOpenToComplete();
        return ClientConversionSupport.toStringArray(protonConnection.getRemoteOfferedCapabilities());
    }

    @Override
    public String[] desiredCapabilities() throws ClientException {
        waitForOpenToComplete();
        return ClientConversionSupport.toStringArray(protonConnection.getRemoteDesiredCapabilities());
    }

    @Override
    public String toString() {
        return "ClientConnection:[" + getId() + "]";
    }

    //----- Internal API

    String getId() {
        return connectionId;
    }

    Engine getEngine() {
        return engine;
    }

    ClientConnection connect() throws ClientException {
        try {
            final ReconnectLocation remoteLocation = reconnectPool.getNext();

            // Initial configuration validation happens here, if this step fails then the
            // user most likely configured something incorrect or that violates some constraint
            // like an invalid SASL mechanism etc.
            initializeProtonResources(remoteLocation);
            scheduleReconnect(remoteLocation);

            return this;
        } catch (Exception ex) {
            CLOSED_UPDATER.set(this, 1);
            FAILURE_CAUSE_UPDATER.compareAndSet(this, null, ClientExceptionSupport.createOrPassthroughFatal(ex));
            openFuture.failed(failureCause);
            closeFuture.complete(this);
            ioContext.shutdown();

            throw failureCause;
        }
    }

    boolean isClosed() {
        return closed > 0;
    }

    ScheduledExecutorService getScheduler() {
        return executor;
    }

    ClientFutureFactory getFutureFactory() {
        return futureFactory;
    }

    ConnectionOptions getOptions() {
        return options;
    }

    ClientConnectionCapabilities getCapabilities() {
        return capabilities;
    }

    org.apache.qpid.protonj2.engine.Connection getProtonConnection() {
        return protonConnection;
    }

    <T> T request(Object requestor, ClientFuture<T> request) throws ClientException {
        requests.put(request, requestor);

        try {
            return request.get();
        } catch (Throwable error) {
            request.cancel(false);
            throw ClientExceptionSupport.createNonFatalOrPassthrough(error);
        } finally {
            requests.remove(request);
        }
    }

    void failAllPendingRequests(Object requestor, ClientException cause) {
        requests.entrySet().removeIf(entry -> {
            if (entry.getValue() == requestor) {
                entry.getKey().failed(cause);
                return true;
            }

            return false;
        });
    }

    boolean autoFlushOff() {
        boolean oldState = autoFlush;
        autoFlush = false;
        return oldState;
    }

    void autoFlushOn() {
        autoFlush = true;
    }

    void flush() {
        try {
            transport.flush();
        } catch (IOException e) {
            LOG.debug("Error while flushing engine output to transport: ", e.getMessage());
            throw new UncheckedIOException(e);
        }
    }

    //----- Private implementation events handlers and utility methods

    private void handleLocalOpen(org.apache.qpid.protonj2.engine.Connection connection) {
        connection.tickAuto(getScheduler());

        if (options.openTimeout() > 0) {
            executor.schedule(() -> {
                if (!openFuture.isDone()) {
                    // Ensure a close write is attempted and then force failure regardless
                    // as we don't expect the remote to respond given it hasn't done so yet.
                    try {
                        connection.close();
                    } catch (Throwable ignore) {}

                    connection.getEngine().engineFailed(new ClientOperationTimedOutException(
                        "Connection Open timed out waiting for remote to open"));
                }
            }, options.openTimeout(), TimeUnit.MILLISECONDS);
        }
    }

    private void handleLocalClose(org.apache.qpid.protonj2.engine.Connection connection) {
        if (connection.isRemotelyClosed()) {
            final ClientException failureCause;

            if (engine.connection().getRemoteCondition() != null) {
                failureCause = ClientExceptionSupport.convertToConnectionClosedException(connection.getRemoteCondition());
            } else {
                failureCause = new ClientConnectionRemotelyClosedException("Unknown error led to connection disconnect");
            }

            try {
                connection.getEngine().engineFailed(failureCause);
            } catch (Throwable ignore) {
            }
        } else if (!engine.isShutdown() || !engine.isFailed()) {
            // Ensure engine gets shut down and future completed if remote doesn't respond.
            executor.schedule(() -> {
                try {
                    connection.getEngine().shutdown();
                } catch (Throwable ignore) {
                }
            }, options.closeTimeout(), TimeUnit.MILLISECONDS);
        }
    }

    private void handleRemoteOpen(org.apache.qpid.protonj2.engine.Connection connection) {
        connectionEstablished();
        capabilities.determineCapabilities(connection);

        if (totalConnections == 1) {
            LOG.info("Connection {} connected to server: {}:{}", getId(), transport.getHost(), transport.getPort());
            submitConnectionEvent(options.connectedHandler(), transport.getHost(), transport.getPort(), null);
        } else {
            LOG.info("Connection {} reconnected to server: {}:{}", getId(), transport.getHost(), transport.getPort());
            submitConnectionEvent(options.reconnectedHandler(), transport.getHost(), transport.getPort(), null);
        }

        openFuture.complete(this);
    }

    private void handleRemoteClose(org.apache.qpid.protonj2.engine.Connection connection) {
        // When the connection is already locally closed this implies the application requested
        // a close of this connection so this is normal, if not then the remote is closing for
        // some reason and we should react as if the connection has failed which we will determine
        // in the local close handler based on state.
        if (connection.isLocallyClosed()) {
            try {
                connection.getEngine().shutdown();
            } catch (Throwable ignore) {
                LOG.debug("Unexpected exception thrown from engine shutdown: ", ignore);
            }
        } else {
            try {
                connection.close();
            } catch (Throwable ignored) {
                // Engine handlers will ensure we close down if not already locally closed.
            }
        }
    }

    private void handleEngineOutput(ProtonBuffer output, Runnable ioComplete) {
        try {
            if (autoFlush) {
                transport.writeAndFlush(output, ioComplete);
            } else {
                transport.write(output, ioComplete);
            }
        } catch (IOException e) {
            LOG.debug("Error while writing engine output to transport: ", e.getMessage());
            throw new UncheckedIOException(e);
        }
    }

    /*
     * When an engine fails we check if we can reconnect or not and act accordingly.
     */
    private void handleEngineFailure(Engine engine) {
        final ClientIOException failureCause;

        if (engine.connection().getRemoteCondition() != null) {
            failureCause = ClientExceptionSupport.convertToConnectionClosedException(engine.connection().getRemoteCondition());
        } else if (engine.failureCause() != null) {
            failureCause = ClientExceptionSupport.convertToConnectionClosedException(engine.failureCause());
        } else {
            failureCause = new ClientConnectionRemotelyClosedException("Unknown error led to connection disconnect");
        }

        LOG.trace("Engine reports failure with error: {}", failureCause.getMessage());

        if (isReconnectAllowed(failureCause)) {
            // Disconnect the failed engine for this connection's event handling
            // to prevent cleanup processing of that engine instance from triggering
            // normal connection shutdown processing.
            engine.shutdownHandler(null);

            LOG.info("Connection {} interrupted to server: {}:{}", getId(), transport.getHost(), transport.getPort());
            submitDisconnectionEvent(options.interruptedHandler(), transport.getHost(), transport.getPort(), failureCause);

            // Initial configuration validation happens here, if this step fails then the
            // user most likely configured something incorrect or that violates some constraint
            // like an invalid SASL mechanism etc.
            try {
                final ReconnectLocation remoteLocation = reconnectPool.getNext();

                initializeProtonResources(remoteLocation);
                scheduleReconnect(remoteLocation);
            } catch (ClientException initError) {
                FAILURE_CAUSE_UPDATER.compareAndSet(this, null, ClientExceptionSupport.createOrPassthroughFatal(initError));
                this.engine.shutdown();  // Close down the engine created for reconnect
            } finally {
                engine.shutdown(); // Failed instance gets cleaned up.
            }
        } else {
            FAILURE_CAUSE_UPDATER.compareAndSet(this, null, failureCause);
            engine.shutdown();
        }
    }

    /*
     * Handle normal engine shutdown which should only happen when the connection is closed
     * by the user, all other cases should lead to engine failed event first which will deal
     * with reconnect cases and avoid this event unless reconnect cannot proceed.
     */
    private void handleEngineShutdown(Engine engine) {
        try {
            protonConnection.close();
        } catch (Exception ignore) {}

        try {
            transport.flush();
        } catch (Exception ignore) {}

        try {
            transport.close();
        } catch (Exception ignore) {}

        ioContext.shutdownAsync();

        if (failureCause != null)
        {
            openFuture.failed(failureCause);
            closeFuture.complete(this);

            LOG.warn("Connection {} has failed due to: {}", getId(), failureCause != null ?
                    failureCause.getClass().getSimpleName() + " -> " + failureCause.getMessage() : "No failure details provided.");

           submitDisconnectionEvent(options.disconnectedHandler(), transport.getHost(), transport.getPort(), failureCause);
        }
        else
        {
            openFuture.complete(this);
            closeFuture.complete(this);
        }

        client.unregisterConnection(this);
    }

    private void submitConnectionEvent(BiConsumer<Connection, ConnectionEvent> handler, String host, int port, ClientIOException cause) {
        if (handler != null) {
            try {
                notifications.submit(() -> {
                    try {
                        handler.accept(this, new ConnectionEvent(host, port));
                    } catch (Exception ex) {
                        LOG.trace("User supplied connection life-cycle event handler threw: ", ex);
                    }
                });
            } catch (Exception ex) {
                LOG.trace("Error thrown while attempting to submit event notification ", ex);
            }
        }
    }

    private void submitDisconnectionEvent(BiConsumer<Connection, DisconnectionEvent> handler, String host, int port, ClientIOException cause) {
        if (handler != null) {
            try {
                notifications.submit(() -> {
                    try {
                        handler.accept(this, new DisconnectionEvent(host, port, cause));
                    } catch (Exception ex) {
                        LOG.trace("User supplied disconnection life-cycle event handler threw: ", ex);
                    }
                });
            } catch (Exception ex) {
                LOG.trace("Error thrown while attempting to submit event notification ", ex);
            }
        }
    }

    private Engine configureEngineSaslSupport() {
        if (options.saslOptions().saslEnabled()) {
            SaslMechanismSelector mechSelector =
                new SaslMechanismSelector(ClientConversionSupport.toSymbolSet(options.saslOptions().allowedMechanisms()));

            engine.saslDriver().client().setListener(new SaslAuthenticator(mechSelector, new SaslCredentialsProvider() {

                @Override
                public String vhost() {
                    return options.virtualHost();
                }

                @Override
                public String username() {
                    return options.user();
                }

                @Override
                public String password() {
                    return options.password();
                }

                @Override
                public Principal localPrincipal() {
                    return transport.getLocalPrincipal();
                }
            }));
        }

        return engine;
    }

    private void initializeProtonResources(ReconnectLocation location) throws ClientException {
        if (options.saslOptions().saslEnabled()) {
            engine = EngineFactory.PROTON.createEngine();
        } else {
            engine = EngineFactory.PROTON.createNonSaslEngine();
        }

        if (options.traceFrames()) {
            engine.configuration().setTraceFrames(true);
            if (!engine.configuration().isTraceFrames()) {
                LOG.warn("Connection {} frame tracing was enabled but protocol engine does not support it", getId());
            }
        }

        engine.outputHandler(this::handleEngineOutput)
              .shutdownHandler(this::handleEngineShutdown)
              .errorHandler(this::handleEngineFailure);

        protonConnection = engine.connection();

        if (client.containerId() != null) {
            protonConnection.setContainerId(client.containerId());
        } else {
            protonConnection.setContainerId(connectionId);
        }

        protonConnection.setLinkedResource(this);
        protonConnection.setChannelMax(options.channelMax());
        protonConnection.setMaxFrameSize(options.maxFrameSize());
        protonConnection.setHostname(location.getHost());
        protonConnection.setIdleTimeout((int) options.idleTimeout());
        protonConnection.setOfferedCapabilities(ClientConversionSupport.toSymbolArray(options.offeredCapabilities()));
        protonConnection.setDesiredCapabilities(ClientConversionSupport.toSymbolArray(options.desiredCapabilities()));
        protonConnection.setProperties(ClientConversionSupport.toSymbolKeyedMap(options.properties()));
        protonConnection.localOpenHandler(this::handleLocalOpen)
                        .localCloseHandler(this::handleLocalClose)
                        .openHandler(this::handleRemoteOpen)
                        .closeHandler(this::handleRemoteClose);

        configureEngineSaslSupport();
    }

    private ClientSession lazyCreateConnectionSession() throws ClientException {
        if (connectionSession == null) {
            connectionSession = sessionBuilder.session(null).open();
        }

        return connectionSession;
    }

    private Sender lazyCreateConnectionSender() throws ClientException {
        if (connectionSender == null) {
            if (openFuture.isComplete()) {
                checkAnonymousRelaySupported();
            }

            connectionSender = lazyCreateConnectionSession().internalOpenAnonymousSender(null);
            connectionSender.remotelyClosedHandler((sender) -> {
                try {
                    sender.closeAsync();
                } catch (Throwable ignore) {}

                // Clear the old closed sender, a lazy create needs to construct a new sender.
                connectionSender = null;
            });
        }

        return connectionSender;
    }

    void checkAnonymousRelaySupported() throws ClientUnsupportedOperationException {
        if (!capabilities.anonymousRelaySupported()) {
            throw new ClientUnsupportedOperationException("Anonymous relay support not available from this connection");
        }
    }

    protected void checkClosedOrFailed() throws ClientException {
        if (closed > 0) {
            throw new ClientIllegalStateException("The Connection was explicitly closed", failureCause);
        } else if (failureCause != null) {
            throw failureCause;
        }
    }

    private void waitForOpenToComplete() throws ClientException {
        if (!openFuture.isComplete() || openFuture.isFailed()) {
            try {
                openFuture.get();
            } catch (ExecutionException | InterruptedException e) {
                Thread.interrupted();
                if (failureCause != null) {
                    throw failureCause;
                } else {
                    throw ClientExceptionSupport.createNonFatalOrPassthrough(e.getCause());
                }
            }
        }
    }

    //----- Reconnection related internal API

    private void attemptConnection(ReconnectLocation location) {
        try {
            reconnectAttempts++;
            transport = ioContext.newTransport();
            LOG.trace("Connection {} Attempting connection to remote {}:{}", getId(), location.getHost(), location.getPort());
            transport.connect(location.getHost(), location.getPort(), new ClientTransportListener(this, engine));
        } catch (Throwable error) {
            engine.engineFailed(ClientExceptionSupport.createOrPassthroughFatal(error));
        }
    }

    private void scheduleReconnect(ReconnectLocation location) {
        // Warn of ongoing connection attempts if configured.
        int warnInterval = options.reconnectOptions().warnAfterReconnectAttempts();
        if (reconnectAttempts > 0 && warnInterval > 0 && (reconnectAttempts % warnInterval) == 0) {
            LOG.warn("Connection {}: Failed to connect after: {} attempt(s) continuing to retry.", getId(), reconnectAttempts);
        }

        // If no connection recovery required then we have never fully connected to a remote
        // so we proceed down the connect with one immediate connection attempt and then follow
        // on delayed attempts based on configuration.
        if (totalConnections == 0) {
            if (reconnectAttempts == 0) {
                LOG.trace("Initial connect attempt will be performed immediately");
                executor.execute(() -> attemptConnection(location));
            } else {
                long delay = nextReconnectDelay();
                LOG.trace("Next connect attempt will be in {} milliseconds", delay);
                executor.schedule(() -> attemptConnection(location), delay, TimeUnit.MILLISECONDS);
            }
        } else if (reconnectAttempts == 0) {
            LOG.trace("Initial reconnect attempt will be performed immediately");
            executor.execute(() -> attemptConnection(location));
        } else {
            long delay = nextReconnectDelay();
            LOG.trace("Next reconnect attempt will be in {} milliseconds", delay);
            executor.schedule(() -> attemptConnection(location), delay, TimeUnit.MILLISECONDS);
        }
    }

    private void connectionEstablished() {
        totalConnections++;
        nextReconnectDelay = -1;
        reconnectAttempts = 0;
    }

    private boolean isLimitExceeded() {
        int reconnectLimit = reconnectAttemptLimit();
        if (reconnectLimit != UNLIMITED && reconnectAttempts >= reconnectLimit) {
            return true;
        }

        return false;
    }

    private boolean isReconnectAllowed(ClientException cause) {
        if (options.reconnectOptions().reconnectEnabled() && !isClosed()) {
            // If a connection attempts fail due to Security errors than we abort
            // reconnection as there is a configuration issue and we want to avoid
            // a spinning reconnect cycle that can never complete.
            if (isStoppageCause(cause)) {
                return false;
            }

            return !isLimitExceeded();
        } else {
            return false;
        }
    }

    private boolean isStoppageCause(ClientException cause) {
        if (cause instanceof ClientConnectionSecuritySaslException) {
            ClientConnectionSecuritySaslException saslFailure = (ClientConnectionSecuritySaslException) cause;
            return !saslFailure.isSysTempFailure();
        } else if (cause instanceof ClientConnectionSecurityException ) {
            return true;
        }

        return false;
    }

    private int reconnectAttemptLimit() {
        int maxReconnectValue = options.reconnectOptions().maxReconnectAttempts();
        if (totalConnections == 0 && options.reconnectOptions().maxInitialConnectionAttempts() != UNDEFINED) {
            // If this is the first connection attempt and a specific startup retry limit
            // is configured then use it, otherwise use the main reconnect limit
            maxReconnectValue = options.reconnectOptions().maxInitialConnectionAttempts();
        }

        return maxReconnectValue;
    }

    private long nextReconnectDelay() {
        if (nextReconnectDelay == UNDEFINED) {
            nextReconnectDelay = options.reconnectOptions().reconnectDelay();
        }

        if (options.reconnectOptions().useReconnectBackOff() && reconnectAttempts > 1) {
            // Exponential increment of reconnect delay.
            nextReconnectDelay *= options.reconnectOptions().reconnectBackOffMultiplier();
            if (nextReconnectDelay > options.reconnectOptions().maxReconnectDelay()) {
                nextReconnectDelay = options.reconnectOptions().maxReconnectDelay();
            }
        }

        return nextReconnectDelay;
    }
}