/*
 * Copyright (c) 2008, 2020 Oracle and/or its affiliates. All rights reserved.
 *
 * This program and the accompanying materials are made available under the
 * terms of the Eclipse Public License v. 2.0, which is available at
 * http://www.eclipse.org/legal/epl-2.0.
 *
 * This Source Code may also be made available under the following Secondary
 * Licenses when the conditions for such availability set forth in the
 * Eclipse Public License v. 2.0 are satisfied: GNU General Public License,
 * version 2 with the GNU Classpath Exception, which is available at
 * https://www.gnu.org/software/classpath/license.html.
 *
 * SPDX-License-Identifier: EPL-2.0 OR GPL-2.0 WITH Classpath-exception-2.0
 */

package org.glassfish.grizzly.nio;

import java.io.IOException;
import java.net.SocketAddress;
import java.util.logging.Level;
import java.util.logging.Logger;

import org.glassfish.grizzly.AbstractWriter;
import org.glassfish.grizzly.CompletionHandler;
import org.glassfish.grizzly.Connection;
import org.glassfish.grizzly.Context;
import org.glassfish.grizzly.Grizzly;
import org.glassfish.grizzly.IOEvent;
import org.glassfish.grizzly.WriteHandler;
import org.glassfish.grizzly.WriteResult;
import org.glassfish.grizzly.asyncqueue.AsyncQueueWriter;
import org.glassfish.grizzly.asyncqueue.AsyncWriteQueueRecord;
import org.glassfish.grizzly.asyncqueue.MessageCloner;
import org.glassfish.grizzly.asyncqueue.PushBackHandler;
import org.glassfish.grizzly.asyncqueue.RecordWriteResult;
import org.glassfish.grizzly.asyncqueue.TaskQueue;
import org.glassfish.grizzly.asyncqueue.WritableMessage;

The AsyncQueueWriter implementation, based on the Java NIO 
Author:
Alexey Stashok, Ryan Lubke, Gustav Trede/**
 * The {@link AsyncQueueWriter} implementation, based on the Java NIO
 *
 * @author Alexey Stashok
 * @author Ryan Lubke
 * @author Gustav Trede
 */
@SuppressWarnings({ "unchecked", "deprecation" })
public abstract class AbstractNIOAsyncQueueWriter extends AbstractWriter<SocketAddress> implements AsyncQueueWriter<SocketAddress> {

    private final static Logger LOGGER = Grizzly.logger(AbstractNIOAsyncQueueWriter.class);

    protected final NIOTransport transport;

    protected volatile int maxPendingBytes = AUTO_SIZE;

    protected volatile int maxWriteReentrants = 10;

    private volatile boolean isAllowDirectWrite = true;

    public AbstractNIOAsyncQueueWriter(NIOTransport transport) {
        this.transport = transport;
    }

    
{@inheritDoc}
/**
     * {@inheritDoc}
     */
    @Deprecated
    @Override
    public boolean canWrite(final Connection<SocketAddress> connection, final int size) {
        return canWrite(connection);
    }

    @Override
    public boolean canWrite(final Connection<SocketAddress> connection) {
        final NIOConnection nioConnection = (NIOConnection) connection;
        final int connectionMaxPendingBytes = nioConnection.getMaxAsyncWriteQueueSize();

        if (connectionMaxPendingBytes < 0) {
            return true;
        }

        final TaskQueue<AsyncWriteQueueRecord> connectionQueue = nioConnection.getAsyncWriteQueue();
        final int size = connectionQueue.spaceInBytes();

        return size == 0 || size < connectionMaxPendingBytes;
    }

    
{@inheritDoc}
/**
     * {@inheritDoc}
     */
    @Deprecated
    @Override
    public void notifyWritePossible(final Connection<SocketAddress> connection, final WriteHandler writeHandler, final int size) {
        notifyWritePossible(connection, writeHandler);
    }

    @Override
    public void notifyWritePossible(final Connection<SocketAddress> connection, final WriteHandler writeHandler) {
        ((NIOConnection) connection).getAsyncWriteQueue().notifyWritePossible(writeHandler);
    }

    
{@inheritDoc}
/**
     * {@inheritDoc}
     */
    @Override
    public void setMaxPendingBytesPerConnection(final int maxPendingBytes) {
        this.maxPendingBytes = maxPendingBytes < AUTO_SIZE ? AUTO_SIZE : maxPendingBytes;
    }

    
{@inheritDoc}
/**
     * {@inheritDoc}
     */
    @Override
    public int getMaxPendingBytesPerConnection() {
        return maxPendingBytes;
    }

    
{@inheritDoc}
/**
     * {@inheritDoc}
     */
    @Override
    public boolean isAllowDirectWrite() {
        return isAllowDirectWrite;
    }

    
{@inheritDoc}
/**
     * {@inheritDoc}
     */
    @Override
    public void setAllowDirectWrite(final boolean isAllowDirectWrite) {
        this.isAllowDirectWrite = isAllowDirectWrite;
    }

    @Override
    public void write(final Connection<SocketAddress> connection, final SocketAddress dstAddress, final WritableMessage message,
            final CompletionHandler<WriteResult<WritableMessage, SocketAddress>> completionHandler, final MessageCloner<WritableMessage> cloner) {
        write(connection, dstAddress, message, completionHandler, null, cloner);
    }

    @Override
    @Deprecated
    public void write(final Connection<SocketAddress> connection, SocketAddress dstAddress, final WritableMessage message,
            final CompletionHandler<WriteResult<WritableMessage, SocketAddress>> completionHandler, final PushBackHandler pushBackHandler) {
        write(connection, dstAddress, message, completionHandler, pushBackHandler, null);
    }

    
{@inheritDoc}
/**
     * {@inheritDoc}
     */
    @Override
    @Deprecated
    public void write(final Connection<SocketAddress> connection, final SocketAddress dstAddress, final WritableMessage message,
            final CompletionHandler<WriteResult<WritableMessage, SocketAddress>> completionHandler, final PushBackHandler pushBackHandler,
            final MessageCloner<WritableMessage> cloner) {

        final NIOConnection nioConnection = (NIOConnection) connection;

        // create and initialize the write queue record
        final AsyncWriteQueueRecord queueRecord = createRecord(nioConnection, message, completionHandler, dstAddress, pushBackHandler,
                !message.hasRemaining() || message.isExternal());

        if (nioConnection == null) {
            queueRecord.notifyFailure(new IOException("Connection is null"));
            return;
        }

        if (!nioConnection.isOpen()) {
            onWriteFailure(nioConnection, queueRecord, nioConnection.getCloseReason().getCause());
            return;
        }

        // Get connection async write queue
        final TaskQueue<AsyncWriteQueueRecord> writeTaskQueue = nioConnection.getAsyncWriteQueue();

        // For empty buffer reserve 1 byte space
        final int bytesToReserve = (int) queueRecord.getBytesToReserve();

        final int pendingBytes = writeTaskQueue.reserveSpace(bytesToReserve);
        final boolean isCurrent = pendingBytes == bytesToReserve;

        final boolean isLogFine = LOGGER.isLoggable(Level.FINEST);

        if (isLogFine) {
            doFineLog(
                    "AsyncQueueWriter.write connection={0}, record={1}, " + "directWrite={2}, size={3}, isUncountable={4}, "
                            + "bytesToReserve={5}, pendingBytes={6}",
                    nioConnection, queueRecord, isCurrent, queueRecord.remaining(), queueRecord.isUncountable(), bytesToReserve, pendingBytes);
        }

        final Reentrant reentrants = Reentrant.getWriteReentrant();

        try {
            if (!reentrants.inc()) {
                // Max number of reentrants is reached

                queueRecord.setMessage(cloneRecordIfNeeded(nioConnection, cloner, message));

                if (isCurrent) { // current but can't write because of maxReentrants limit
                    writeTaskQueue.setCurrentElement(queueRecord);
                    nioConnection.simulateIOEvent(IOEvent.WRITE);
                } else {
                    writeTaskQueue.offer(queueRecord);
                }

                return;
            }

            if (isCurrent && isAllowDirectWrite) {

                // If we can write directly - do it w/o creating queue record (simple)
                final RecordWriteResult writeResult = write0(nioConnection, queueRecord);
                final int bytesToRelease = (int) writeResult.bytesToReleaseAfterLastWrite();

                final boolean isFinished = queueRecord.isFinished();

                final int pendingBytesAfterRelease = writeTaskQueue.releaseSpaceAndNotify(bytesToRelease);

                final boolean isQueueEmpty = pendingBytesAfterRelease == 0;

                if (isLogFine) {
                    doFineLog(
                            "AsyncQueueWriter.write directWrite connection={0}, record={1}, " + "isFinished={2}, remaining={3}, isUncountable={4}, "
                                    + "bytesToRelease={5}, pendingBytesAfterRelease={6}",
                            nioConnection, queueRecord, isFinished, queueRecord.remaining(), queueRecord.isUncountable(), bytesToRelease,
                            pendingBytesAfterRelease);
                }

                if (isFinished) {
                    queueRecord.notifyCompleteAndRecycle();
                    if (!isQueueEmpty) {
                        nioConnection.simulateIOEvent(IOEvent.WRITE);
                    }
                    return;
                }
            }

            queueRecord.setMessage(cloneRecordIfNeeded(nioConnection, cloner, message));

            if (isLogFine) {
                doFineLog("AsyncQueueWriter.write queuing connection={0}, record={1}, " + "size={2}, isUncountable={3}", nioConnection, queueRecord,
                        queueRecord.remaining(), queueRecord.isUncountable());
            }

            if (isCurrent) { // current but not finished.
                writeTaskQueue.setCurrentElement(queueRecord);
                onReadyToWrite(nioConnection);
            } else {
                writeTaskQueue.offer(queueRecord);
            }
        } catch (IOException e) {
            if (isLogFine) {
                LOGGER.log(Level.FINEST, "AsyncQueueWriter.write exception. connection=" + nioConnection + " record=" + queueRecord, e);
            }

            onWriteFailure(nioConnection, queueRecord, e);
        } finally {
            reentrants.dec();
        }
    }

    
{@inheritDoc}
/**
     * {@inheritDoc}
     */
    @Override
    public AsyncResult processAsync(final Context context) {
        final boolean isLogFine = LOGGER.isLoggable(Level.FINEST);
        final NIOConnection nioConnection = (NIOConnection) context.getConnection();
        if (!nioConnection.isOpen()) {
            return AsyncResult.COMPLETE;
        }

        final TaskQueue<AsyncWriteQueueRecord> writeTaskQueue = nioConnection.getAsyncWriteQueue();

        int bytesReleased = 0;

        boolean done = true;

        AsyncWriteQueueRecord queueRecord = null;
        try {
            while ((queueRecord = aggregate(writeTaskQueue)) != null) {
                if (isLogFine) {
                    doFineLog("AsyncQueueWriter.processAsync beforeWrite " + "connection={0} record={1}", nioConnection, queueRecord);
                }

                final RecordWriteResult writeResult = write0(nioConnection, queueRecord);
                final int bytesToRelease = (int) writeResult.bytesToReleaseAfterLastWrite();

                done = queueRecord.isFinished();

                bytesReleased += bytesToRelease;

                if (isLogFine) {
                    doFineLog("AsyncQueueWriter.processAsync written " + "connection={0}, written={1}, done={2}, " + "bytesToRelease={3}, bytesReleased={4}",
                            nioConnection, writeResult.lastWrittenBytes(), done, bytesToRelease, bytesReleased);
                }

                if (done) {
                    finishQueueRecord(nioConnection, queueRecord);
                } else { // if there is still some data in current message
                    queueRecord.notifyIncomplete();
                    writeTaskQueue.setCurrentElement(queueRecord);
                    if (isLogFine) {
                        doFineLog("AsyncQueueWriter.processAsync onReadyToWrite " + "connection={0} peekRecord={1}", nioConnection, queueRecord);
                    }

                    // If connection is closed - this will fail,
                    // and onWriteFailure called properly
                    break;
                }
            }

            boolean isComplete = false;

            // Notify completed records' handlers (if any)
            if (bytesReleased > 0) {
                // Is here a chance that queue becomes empty?
                // If yes - we need to switch to manual io event processing
                // mode to *disable WRITE interest for SameThreadStrategy*,
                // so we don't have either neverending WRITE events processing
                // or stuck, when other thread tried to add data to the queue.
                if (done && !context.isManualIOEventControl() && writeTaskQueue.spaceInBytes() - bytesReleased <= 0) {
                    if (isLogFine) {
                        doFineLog("AsyncQueueWriter.processAsync setManualIOEventControl " + "connection={0}", nioConnection);
                    }

                    context.setManualIOEventControl();
                }

                isComplete = writeTaskQueue.releaseSpace(bytesReleased) == 0;
            }

            if (isLogFine) {
                doFineLog("AsyncQueueWriter.processAsync exit " + "connection={0}, done={1}, isComplete={2}, " + "bytesReleased={3}, queueSize={4}",
                        nioConnection, done, isComplete, bytesReleased, writeTaskQueue.size());
            }

            final AsyncResult result = !done ? AsyncResult.INCOMPLETE : !isComplete ? AsyncResult.EXPECTING_MORE : AsyncResult.COMPLETE;

            if (bytesReleased > 0) {
                // Finish the context processing (enable OP_WRITE if needed),
                // so following notification calls will not block the async write
                // queue write process
                context.complete(result.toProcessorResult());

                writeTaskQueue.doNotify();

                return AsyncResult.TERMINATE;
            }

            return result;
        } catch (IOException e) {
            if (isLogFine) {
                LOGGER.log(Level.FINEST, "AsyncQueueWriter.processAsync " + "exception connection=" + nioConnection + " peekRecord=" + queueRecord, e);
            }
            onWriteFailure(nioConnection, queueRecord, e);
        }

        return AsyncResult.COMPLETE;
    }

    private static void finishQueueRecord(final NIOConnection nioConnection, final AsyncWriteQueueRecord queueRecord) {
        final boolean isLogFine = LOGGER.isLoggable(Level.FINEST);

        if (isLogFine) {
            doFineLog("AsyncQueueWriter.processAsync finished " + "connection={0} record={1}", nioConnection, queueRecord);
        }

        if (queueRecord != null) {
            queueRecord.notifyCompleteAndRecycle();
        }

        if (isLogFine) {
            doFineLog("AsyncQueueWriter.processAsync finishQueueRecord " + "connection={0} queueRecord={1}", nioConnection, queueRecord);
        }
    }

    private static WritableMessage cloneRecordIfNeeded(final Connection connection, final MessageCloner<WritableMessage> cloner,
            final WritableMessage message) {

        if (LOGGER.isLoggable(Level.FINEST)) {
            LOGGER.log(Level.FINEST, "AsyncQueueWriter.write clone. connection={0} cloner={1} size={2}",
                    new Object[] { connection, cloner, message.remaining() });
        }

        return cloner == null ? message : cloner.clone(connection, message);
    }

    protected AsyncWriteQueueRecord createRecord(final Connection connection, final WritableMessage message,
            final CompletionHandler<WriteResult<WritableMessage, SocketAddress>> completionHandler, final SocketAddress dstAddress,
            final PushBackHandler pushBackHandler, final boolean isUncountable) {
        return AsyncWriteQueueRecord.create(connection, message, completionHandler, dstAddress, pushBackHandler, isUncountable);
    }

    
{@inheritDoc}
/**
     * {@inheritDoc}
     */
    @Override
    public final boolean isReady(final Connection connection) {
        final TaskQueue connectionQueue = ((NIOConnection) connection).getAsyncWriteQueue();

        return connectionQueue != null && !connectionQueue.isEmpty();
    }

    private static void doFineLog(final String msg, final Object... params) {
        LOGGER.log(Level.FINEST, msg, params);
    }

    
{@inheritDoc}
/**
     * {@inheritDoc}
     */
    @Override
    public void onClose(final Connection connection) {
        final NIOConnection nioConnection = (NIOConnection) connection;
        final TaskQueue<AsyncWriteQueueRecord> writeQueue = nioConnection.getAsyncWriteQueue();
        writeQueue.onClose(nioConnection.getCloseReason().getCause());
    }

    
{@inheritDoc}
/**
     * {@inheritDoc}
     */
    @Override
    public final void close() {
    }

    protected static void onWriteFailure(final Connection connection, final AsyncWriteQueueRecord failedRecord, final Throwable e) {

        failedRecord.notifyFailure(e);
        connection.closeSilently();
    }

    protected abstract RecordWriteResult write0(NIOConnection connection, AsyncWriteQueueRecord queueRecord) throws IOException;

    protected abstract void onReadyToWrite(NIOConnection connection) throws IOException;

    
Aggregates records in a queue to be written as one chunk.
/**
     * Aggregates records in a queue to be written as one chunk.
     */
    protected AsyncWriteQueueRecord aggregate(final TaskQueue<AsyncWriteQueueRecord> connectionQueue) {
        return connectionQueue.poll();
    }
}