/*
 * Copyright (c) 2012, 2017 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.tyrus.servlet;

import java.io.IOException;
import java.nio.ByteBuffer;
import java.util.concurrent.CountDownLatch;
import java.util.logging.Level;
import java.util.logging.Logger;

import javax.websocket.CloseReason;

import javax.servlet.ReadListener;
import javax.servlet.ServletInputStream;
import javax.servlet.ServletOutputStream;
import javax.servlet.http.HttpUpgradeHandler;
import javax.servlet.http.WebConnection;

import org.glassfish.tyrus.core.CloseReasons;
import org.glassfish.tyrus.spi.Connection;
import org.glassfish.tyrus.spi.WebSocketEngine;
import org.glassfish.tyrus.spi.Writer;

/**
 * {@link HttpUpgradeHandler} and {@link ReadListener} implementation.
 * <p>
 * Reads data from {@link ServletInputStream} and passes it further to the Tyrus runtime.
 *
 * @author Jitendra Kotamraju
 * @author Pavel Bucek (pavel.bucek at oracle.com)
 */
public class TyrusHttpUpgradeHandler implements HttpUpgradeHandler, ReadListener {

    public static final String FRAME_BUFFER_SIZE = "org.glassfish.tyrus.servlet.incoming-buffer-size";

    private final CountDownLatch connectionLatch = new CountDownLatch(1);

    private ServletInputStream is;
    private ServletOutputStream os;
    private WebConnection wc;
    private ByteBuffer buf;

    private volatile boolean closed = false;
    private int incomingBufferSize = 4194315; // 4M (payload) + 11 (frame overhead)

    private static final Logger LOGGER = Logger.getLogger(TyrusHttpUpgradeHandler.class.getName());

    private Connection connection;
    private WebSocketEngine.UpgradeInfo upgradeInfo;
    private Writer writer;


    private boolean authenticated = false;

    @Override
    public void init(WebConnection wc) {
        LOGGER.config("Servlet 3.1 Upgrade");
        try {
            is = wc.getInputStream();
            os = wc.getOutputStream();
            this.wc = wc;
        } catch (IOException ioe) {
            throw new RuntimeException(ioe);
        }

        try {
            is.setReadListener(this);
        } catch (IllegalStateException e) {
            LOGGER.log(Level.WARNING, e.getMessage(), e);
        }

        connection = upgradeInfo.createConnection(writer, new Connection.CloseListener() {
            @Override
            public void close(CloseReason reason) {
                try {
                    TyrusHttpUpgradeHandler.this.getWebConnection().close();
                } catch (Exception e) {
                    LOGGER.log(Level.FINE, e.getMessage(), e);
                }
            }
        });

        connectionLatch.countDown();
    }

    public void preInit(WebSocketEngine.UpgradeInfo upgradeInfo, Writer writer, boolean authenticated) {
        this.upgradeInfo = upgradeInfo;
        this.writer = writer;
        this.authenticated = authenticated;
    }

    @Override
    public void onDataAvailable() {
        try {
            connectionLatch.await();
        } catch (InterruptedException e) {
            // do nothing.
        }

        do {
            try {
                int available = is.available();
                if (available == 0) {
                    // tomcat impl returns always 0
                    available = 16384;
                }

                int toRead;
                if (buf == null) {
                    if (available > incomingBufferSize) {
                        toRead = incomingBufferSize;
                    } else {
                        toRead = available;
                    }
                } else {
                    if (buf.remaining() + available > incomingBufferSize) {
                        toRead = incomingBufferSize - buf.remaining();
                    } else {
                        toRead = buf.remaining() + available;
                    }
                }

                if (toRead == 0) {
                    throw new IOException(String.format("Tyrus input buffer exceeded. Current buffer size is %s bytes.",
                                                        incomingBufferSize));
                }

                available -= fillBuf(toRead);

                if (buf != null) {

                    LOGGER.finest(String.format("Remaining Data = %d", buf.remaining()));

                    if (buf.hasRemaining()) {
                        connection.getReadHandler().handle(buf);
                    }
                }
            } catch (IOException e) {
                connection.close(CloseReasons.CANNOT_ACCEPT.getCloseReason());
            }
        } while (!closed && is.isReady());
    }

    /**
     * Fill the buf with some more websocket protocol data.
     *
     * @param length length of data available to read.
     * @return legth of actually read data.
     * @throws IOException if some other I/O error occurs.
     */
    private int fillBuf(int length) throws IOException {
        byte[] data = new byte[length];
        int len = is.read(data);
        if (len == 0) {
            return 0;
        }
        if (buf == null) {
            LOGGER.finest("No Buffer. Allocating new one");
            buf = ByteBuffer.wrap(data);
            buf.limit(len);
        } else {
            int limit = buf.limit();
            int capacity = buf.capacity();
            int remaining = buf.remaining();

            if (capacity - limit >= len) {
                // Remaining data need not be changed. New data is just appended
                LOGGER.finest("Remaining data need not be moved. New data is just appended");
                buf.mark();
                buf.position(limit);
                buf.limit(capacity);
                buf.put(data, 0, len);
                buf.limit(limit + len);
                buf.reset();
            } else if (remaining + len < capacity) {
                // Remaining data is moved to left. Then new data is appended
                LOGGER.finest("Remaining data is moved to left. Then new data is appended");
                buf.compact();
                buf.put(data, 0, len);
                buf.flip();
            } else {
                // Remaining data + new > capacity. So allocate new one
                LOGGER.finest("Remaining data + new > capacity. So allocate new one");
                byte[] array = new byte[remaining + len];
                buf.get(array, 0, remaining);
                System.arraycopy(data, 0, array, remaining, len);
                buf = ByteBuffer.wrap(array);
                buf.limit(remaining + len);
            }
        }

        return len;
    }

    @Override
    public void onAllDataRead() {
        close(CloseReason.CloseCodes.NORMAL_CLOSURE.getCode(), null);
    }

    @Override
    public void onError(Throwable t) {
        close(CloseReason.CloseCodes.CLOSED_ABNORMALLY.getCode(),
              t.getMessage() == null ? "No reason given." : t.getMessage());
    }

    @Override
    public void destroy() {
        close(CloseReason.CloseCodes.CLOSED_ABNORMALLY.getCode(), "No reason given.");
    }

    /**
     * Called when related {@link javax.servlet.http.HttpSession} is destroyed or invalidated.
     * <p>
     * Implementation is required to call onClose() on server-side with corresponding close code (1008, see
     * WebSocket spec 7.2) - only when there is an authorized user for this session.
     */
    public void sessionDestroyed() {
        if (authenticated) {
            // websocket spec 7.2 [WSC-7.2-3]
            httpSessionForcedClose(CloseReason.CloseCodes.VIOLATED_POLICY.getCode(), "No reason given.");
        }

        // else do nothing.
    }

    @Override
    public String toString() {
        final StringBuilder sb = new StringBuilder("TyrusHttpUpgradeHandler{");
        sb.append("is=").append(is);
        sb.append(", os=").append(os);
        sb.append(", wc=").append(wc);
        sb.append(", closed=").append(closed);
        sb.append('}');
        return sb.toString();
    }

    public void setIncomingBufferSize(int incomingBufferSize) {
        this.incomingBufferSize = incomingBufferSize;
    }

    private void httpSessionForcedClose(int closeCode, String closeReason) {
        if (!closed) {
            try {
                // TODO
                // initiates connection close without sending close frame to the client - session is already invalidated
                // so we should not send anything.
                // ((TyrusWebSocket) ((TyrusWebSocketEngine) engine).getWebSocketHolder(writer).webSocket).setClosed();
                connection.close(new CloseReason(CloseReason.CloseCodes.getCloseCode(closeCode), closeReason));
                closed = true;
                wc.close();
            } catch (Exception e) {
                LOGGER.log(Level.CONFIG, e.getMessage(), e);
            }
        }
    }

    private void close(int closeCode, String closeReason) {
        if (!closed) {
            try {
                connection.close(new CloseReason(CloseReason.CloseCodes.getCloseCode(closeCode), closeReason));
                closed = true;
                wc.close();
            } catch (Exception e) {
                LOGGER.log(Level.CONFIG, e.getMessage(), e);
            }
        }
    }

    WebConnection getWebConnection() {
        if (wc == null) {
            throw new IllegalStateException();
        }
        return wc;
    }
}
