/*
 * Copyright (c) 2015, 2020, Oracle and/or its affiliates.
 *
 * This program is free software; you can redistribute it and/or modify it under
 * the terms of the GNU General Public License, version 2.0, as published by the
 * Free Software Foundation.
 *
 * This program is also distributed with certain software (including but not
 * limited to OpenSSL) that is licensed under separate terms, as designated in a
 * particular file or component or in included license documentation. The
 * authors of MySQL hereby grant you an additional permission to link the
 * program and your derivative works with the separately licensed software that
 * they have included with MySQL.
 *
 * Without limiting anything contained in the foregoing, this file, which is
 * part of MySQL Connector/J, is also subject to the Universal FOSS Exception,
 * version 1.0, a copy of which can be found at
 * http://oss.oracle.com/licenses/universal-foss-exception.
 *
 * This program is distributed in the hope that it will be useful, but WITHOUT
 * ANY WARRANTY; without even the implied warranty of MERCHANTABILITY or FITNESS
 * FOR A PARTICULAR PURPOSE. See the GNU General Public License, version 2.0,
 * for more details.
 *
 * You should have received a copy of the GNU General Public License along with
 * this program; if not, write to the Free Software Foundation, Inc.,
 * 51 Franklin St, Fifth Floor, Boston, MA 02110-1301  USA
 */

package com.mysql.cj.protocol.x;

import java.io.IOException;
import java.util.ArrayList;
import java.util.LinkedList;
import java.util.List;
import java.util.Optional;
import java.util.concurrent.BlockingQueue;
import java.util.concurrent.LinkedBlockingQueue;
import java.util.concurrent.TimeUnit;

import com.google.protobuf.GeneratedMessageV3;
import com.google.protobuf.InvalidProtocolBufferException;
import com.google.protobuf.Parser;
import com.mysql.cj.exceptions.CJCommunicationsException;
import com.mysql.cj.exceptions.MysqlErrorNumbers;
import com.mysql.cj.exceptions.WrongArgumentException;
import com.mysql.cj.protocol.FullReadInputStream;
import com.mysql.cj.protocol.MessageListener;
import com.mysql.cj.protocol.MessageReader;
import com.mysql.cj.protocol.Protocol.ProtocolEventHandler;
import com.mysql.cj.protocol.Protocol.ProtocolEventListener.EventType;
import com.mysql.cj.protocol.x.Notice.XWarning;
import com.mysql.cj.x.protobuf.Mysqlx.Error;
import com.mysql.cj.x.protobuf.Mysqlx.ServerMessages;
import com.mysql.cj.x.protobuf.MysqlxNotice.Frame;

/**
 * Synchronous-only implementation of {@link MessageReader}. This implementation wraps a {@link java.io.InputStream}.
 */
public class SyncMessageReader implements MessageReader<XMessageHeader, XMessage> {
    /** Stream as a source of messages. */
    private FullReadInputStream inputStream;

    LinkedList<XMessageHeader> headersQueue = new LinkedList<>();
    LinkedList<GeneratedMessageV3> messagesQueue = new LinkedList<>();

    /** Queue of <code>MessageListener</code>s waiting to process messages. */
    BlockingQueue<MessageListener<XMessage>> messageListenerQueue = new LinkedBlockingQueue<>();

    /** Lock to protect the pending message. */
    Object dispatchingThreadMonitor = new Object();
    /** Lock to protect async reads from sync ones. */
    Object waitingSyncOperationMonitor = new Object();

    Thread dispatchingThread = null;

    private ProtocolEventHandler protocolEventHandler = null;

    public SyncMessageReader(FullReadInputStream inputStream, ProtocolEventHandler protocolEventHandler) {
        this.inputStream = inputStream;
        this.protocolEventHandler = protocolEventHandler;
    }

    @Override
    public XMessageHeader readHeader() throws IOException {
        // waiting for ListenersDispatcher completion to perform sync call
        synchronized (this.waitingSyncOperationMonitor) {
            XMessageHeader header;
            if ((header = this.headersQueue.peek()) == null) {
                header = readHeaderLocal();
            }
            if (header.getMessageType() == ServerMessages.Type.ERROR_VALUE) {
                throw new XProtocolError(readMessageLocal(Error.class, true));
            }
            return header;
        }
    }

    public int getNextNonNoticeMessageType() throws IOException {
        synchronized (this.waitingSyncOperationMonitor) {
            if (!this.headersQueue.isEmpty()) {
                for (XMessageHeader hdr : this.headersQueue) {
                    if (hdr.getMessageType() != ServerMessages.Type.NOTICE_VALUE) {
                        return hdr.getMessageType();
                    }
                }
            }

            XMessageHeader header;
            do {
                header = readHeaderLocal();
                if (header.getMessageType() == ServerMessages.Type.ERROR_VALUE) {
                    Error msg;
                    this.messagesQueue.addLast(msg = readMessageLocal(Error.class, false));
                    throw new XProtocolError(msg);

                } else if (header.getMessageType() == ServerMessages.Type.NOTICE_VALUE) {
                    this.messagesQueue.addLast(readMessageLocal(Frame.class, false));
                }
            } while (header.getMessageType() == ServerMessages.Type.NOTICE_VALUE);

            return header.getMessageType();
        }
    }

    private XMessageHeader readHeaderLocal() throws IOException {

        XMessageHeader header;
        try {
            /*
             * Note that the "header" per-se is the size of all data following the header. This currently includes the message type tag (1 byte) and the
             * message bytes. However since we know the type tag is present we also read it as part of the header. This may change in the future if session
             * multiplexing is supported by the protocol. The protocol will be able to accommodate it but we will have to separate reading data after the
             * header (size).
             */
            byte[] buf = new byte[5];
            this.inputStream.readFully(buf);
            header = new XMessageHeader(buf);
            this.headersQueue.add(header);
        } catch (IOException ex) {
            // TODO close socket?
            throw new CJCommunicationsException("Cannot read packet header", ex);
        }
        return header;
    }

    @SuppressWarnings("unchecked")
    private <T extends GeneratedMessageV3> T readMessageLocal(Class<T> messageClass, boolean fromQueue) {

        XMessageHeader header;
        if (fromQueue) {
            header = this.headersQueue.poll();
            T msg = (T) this.messagesQueue.poll();
            if (msg != null) {
                return msg;
            }
        } else {
            header = this.headersQueue.getLast();
        }

        Parser<T> parser = (Parser<T>) MessageConstants.MESSAGE_CLASS_TO_PARSER.get(messageClass);
        byte[] packet = new byte[header.getMessageSize()];

        try {
            this.inputStream.readFully(packet);
        } catch (IOException ex) {
            // TODO close socket?
            throw new CJCommunicationsException("Cannot read packet payload", ex);
        }

        try {
            T msg = parser.parseFrom(packet);

            if (msg instanceof Frame && ((Frame) msg).getType() == Frame.Type.WARNING_VALUE && ((Frame) msg).getScope() == Frame.Scope.GLOBAL) {
                XWarning w = new XWarning((Frame) msg);
                int code = (int) w.getCode();
                if (code == MysqlErrorNumbers.ER_SERVER_SHUTDOWN || code == MysqlErrorNumbers.ER_IO_READ_ERROR
                        || code == MysqlErrorNumbers.ER_SESSION_WAS_KILLED) {

                    CJCommunicationsException ex = new CJCommunicationsException(w.getMessage());
                    ex.setVendorCode(code);
                    if (this.protocolEventHandler != null) {
                        this.protocolEventHandler.invokeListeners(
                                code == MysqlErrorNumbers.ER_SERVER_SHUTDOWN ? EventType.SERVER_SHUTDOWN : EventType.SERVER_CLOSED_SESSION, ex);
                    }

                    throw ex;
                }
            }

            return msg;

        } catch (InvalidProtocolBufferException ex) {
            throw new WrongArgumentException(ex);
        }
    }

    @Override
    public XMessage readMessage(Optional<XMessage> reuse, XMessageHeader hdr) throws IOException {
        return readMessage(reuse, hdr.getMessageType());
    }

    @Override
    public XMessage readMessage(Optional<XMessage> reuse, int expectedType) throws IOException {
        // waiting for ListenersDispatcher completion to perform sync call
        synchronized (this.waitingSyncOperationMonitor) {
            try {
                Class<? extends GeneratedMessageV3> expectedClass = MessageConstants.getMessageClassForType(expectedType);

                List<Notice> notices = null;
                XMessageHeader hdr;
                while ((hdr = readHeader()).getMessageType() == ServerMessages.Type.NOTICE_VALUE && expectedType != ServerMessages.Type.NOTICE_VALUE) {
                    if (notices == null) {
                        notices = new ArrayList<>();
                    }
                    notices.add(Notice
                            .getInstance(new XMessage(readMessageLocal(MessageConstants.getMessageClassForType(ServerMessages.Type.NOTICE_VALUE), true))));
                }

                Class<? extends GeneratedMessageV3> messageClass = MessageConstants.getMessageClassForType(hdr.getMessageType());
                // ensure that parsed message class matches incoming tag
                if (expectedClass != messageClass) {
                    throw new WrongArgumentException("Unexpected message class. Expected '" + expectedClass.getSimpleName() + "' but actually received '"
                            + messageClass.getSimpleName() + "'");
                }

                return new XMessage(readMessageLocal(messageClass, true)).addNotices(notices);
            } catch (IOException e) {
                throw new XProtocolError(e.getMessage(), e);
            }
        }
    }

    public void pushMessageListener(final MessageListener<XMessage> listener) {
        try {
            this.messageListenerQueue.put(listener);
        } catch (InterruptedException e) {
            throw new CJCommunicationsException("Cannot queue message listener.", e);
        }

        synchronized (this.dispatchingThreadMonitor) {
            if (this.dispatchingThread == null) {
                ListenersDispatcher ld = new ListenersDispatcher();
                this.dispatchingThread = new Thread(ld, "Message listeners dispatching thread");
                this.dispatchingThread.start();

                // We must ensure that ListenersDispatcher is really started before leaving
                // the synchronized block. Otherwise the race condition is possible: if next
                // operation is executed synchronously it could consume results of the previous
                // asynchronous operation.
                int millis = 5000; // TODO expose via properties ?
                while (!ld.started) {
                    try {
                        Thread.sleep(10);
                        millis = millis - 10;
                    } catch (InterruptedException e) {
                        throw new XProtocolError(e.getMessage(), e);
                    }
                    if (millis <= 0) {
                        throw new XProtocolError("Timeout for starting ListenersDispatcher exceeded.");
                    }
                }
            }
        }
    }

    private class ListenersDispatcher implements Runnable {
        /**
         * The timeout value for queue.poll(timeout, unit) defining the time after which we close and unregister the dispatching thread.
         * On the other hand, a bigger timeout value allows us to keep dispatcher thread running while multiple concurrent asynchronous
         * read operations are pending, thus avoiding the delays for new dispatching threads creation.
         */
        private static final long POLL_TIMEOUT = 100; // TODO expose via connection property
        boolean started = false;

        public ListenersDispatcher() {
        }

        @Override
        public void run() {
            synchronized (SyncMessageReader.this.waitingSyncOperationMonitor) {
                this.started = true;
                try {
                    while (true) {
                        MessageListener<XMessage> l;
                        if ((l = SyncMessageReader.this.messageListenerQueue.poll(POLL_TIMEOUT, TimeUnit.MILLISECONDS)) == null) {
                            synchronized (SyncMessageReader.this.dispatchingThreadMonitor) {
                                if (SyncMessageReader.this.messageListenerQueue.peek() == null) {
                                    SyncMessageReader.this.dispatchingThread = null;
                                    break;
                                }
                            }
                        } else {
                            try {
                                XMessage msg = null;
                                do {
                                    XMessageHeader hdr = readHeader();
                                    msg = readMessage(null, hdr);
                                } while (!l.processMessage(msg));
                            } catch (Throwable t) {
                                l.error(t);
                            }
                        }
                    }
                } catch (InterruptedException e) {
                    throw new CJCommunicationsException("Read operation interrupted.", e);
                }
            }
        }
    }
}
