/*

 * DO NOT ALTER OR REMOVE COPYRIGHT NOTICES OR THIS FILE HEADER.

 *

 * This code is free software; you can redistribute it and/or modify it

 * under the terms of the GNU General Public License version 2 only, as

 * published by the Free Software Foundation.  Oracle designates this

 * particular file as subject to the "Classpath" exception as provided

 * by Oracle in the LICENSE file that accompanied this code.

 *

 * This code 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 for more details (a copy is included in the LICENSE file that

 * accompanied this code).

 *

 * You should have received a copy of the GNU General Public License version

 * 2 along with this work; if not, write to the Free Software Foundation,

 * Inc., 51 Franklin St, Fifth Floor, Boston, MA 02110-1301 USA.

 *

 * Please contact Oracle, 500 Oracle Parkway, Redwood Shores, CA 94065 USA

 * or visit www.oracle.com if you need additional information or have any

 * questions.

 */

package jdk.incubator.http.internal.frame;

import jdk.incubator.http.internal.common.Log;

import jdk.incubator.http.internal.common.Utils;

import java.io.IOException;

import java.lang.System.Logger.Level;

import java.nio.ByteBuffer;

import java.util.ArrayDeque;

import java.util.ArrayList;

import java.util.List;

import java.util.Queue;

/**

 * Frames Decoder

 * <p>

 * collect buffers until frame decoding is possible,

 * all decoded frames are passed to the FrameProcessor callback in order of decoding.

 *

 * It's a stateful class due to the fact that FramesDecoder stores buffers inside.

 * Should be allocated only the single instance per connection.

 */

public class FramesDecoder {

    static final boolean DEBUG = Utils.DEBUG; // Revisit: temporary dev flag.

    static final System.Logger DEBUG_LOGGER =

            Utils.getDebugLogger("FramesDecoder"::toString, DEBUG);

    @FunctionalInterface

    public interface FrameProcessor {

        void processFrame(Http2Frame frame) throws IOException;

    }

    private final FrameProcessor frameProcessor;

    private final int maxFrameSize;

    private ByteBuffer currentBuffer; // current buffer either null or hasRemaining

    private final ArrayDeque<ByteBuffer> tailBuffers = new ArrayDeque<>();

    private int tailSize = 0;

    private boolean slicedToDataFrame = false;

    private final List<ByteBuffer> prepareToRelease = new ArrayList<>();

    // if true  - Frame Header was parsed (9 bytes consumed) and subsequent fields have meaning

    // otherwise - stopped at frames boundary

    private boolean frameHeaderParsed = false;

    private int frameLength;

    private int frameType;

    private int frameFlags;

    private int frameStreamid;

    private boolean closed;

    /**

     * Creates Frame Decoder

     *

     * @param frameProcessor - callback for decoded frames

     */

    public FramesDecoder(FrameProcessor frameProcessor) {

        this(frameProcessor, 16 * 1024);

    }

    /**

     * Creates Frame Decoder

     * @param frameProcessor - callback for decoded frames

     * @param maxFrameSize - maxFrameSize accepted by this decoder

     */

    public FramesDecoder(FrameProcessor frameProcessor, int maxFrameSize) {

        this.frameProcessor = frameProcessor;

        this.maxFrameSize = Math.min(Math.max(16 * 1024, maxFrameSize), 16 * 1024 * 1024 - 1);

    }

    /** Threshold beyond which data is no longer copied into the current buffer,

     * if that buffer has enough unused space. */

    private static final int COPY_THRESHOLD = 8192;

    /**

     * Adds the data from the given buffer, and performs frame decoding if

     * possible.   Either 1) appends the data from the given buffer to the

     * current buffer ( if there is enough unused space ), or 2) adds it to the

     * next buffer in the queue.

     *

     * If there is enough data to perform frame decoding then, all buffers are

     * decoded and the FrameProcessor is invoked.

     */

    public void decode(ByteBuffer inBoundBuffer) throws IOException {

        if (closed) {

            DEBUG_LOGGER.log(Level.DEBUG, "closed: ignoring buffer (%s bytes)",

                    inBoundBuffer.remaining());

            inBoundBuffer.position(inBoundBuffer.limit());

            return;

        }

        int remaining = inBoundBuffer.remaining();

        DEBUG_LOGGER.log(Level.DEBUG, "decodes: %d", remaining);

        if (remaining > 0) {

            if (currentBuffer == null) {

                currentBuffer = inBoundBuffer;

            } else {

                ByteBuffer b = currentBuffer;

                if (!tailBuffers.isEmpty()) {

                    b = tailBuffers.getLast();

                }

                int limit = b.limit();

                int freeSpace = b.capacity() - limit;

                if (remaining <= COPY_THRESHOLD && freeSpace >= remaining) {

                    // append the new data to the unused space in the current buffer

                    int position = b.position();

                    b.position(limit);

                    b.limit(limit + inBoundBuffer.remaining());

                    b.put(inBoundBuffer);

                    b.position(position);

                    if (b != currentBuffer)

                        tailSize += remaining;

                    DEBUG_LOGGER.log(Level.DEBUG, "copied: %d", remaining);

                } else {

                    DEBUG_LOGGER.log(Level.DEBUG, "added: %d", remaining);

                    tailBuffers.add(inBoundBuffer);

                    tailSize += remaining;

                }

            }

        }

        DEBUG_LOGGER.log(Level.DEBUG, "Tail size is now: %d, current=",

                tailSize,

                (currentBuffer == null ? 0 :

                   currentBuffer.remaining()));

        Http2Frame frame;

        while ((frame = nextFrame()) != null) {

            DEBUG_LOGGER.log(Level.DEBUG, "Got frame: %s", frame);

            frameProcessor.processFrame(frame);

            frameProcessed();

        }

    }

    private Http2Frame nextFrame() throws IOException {

        while (true) {

            if (currentBuffer == null) {

                return null; // no data at all

            }

            long available = currentBuffer.remaining() + tailSize;

            if (!frameHeaderParsed) {

                if (available >= Http2Frame.FRAME_HEADER_SIZE) {

                    parseFrameHeader();

                    if (frameLength > maxFrameSize) {

                        // connection error

                        return new MalformedFrame(ErrorFrame.FRAME_SIZE_ERROR,

                                "Frame type("+frameType+") "

                                +"length("+frameLength

                                +") exceeds MAX_FRAME_SIZE("

                                + maxFrameSize+")");

                    }

                    frameHeaderParsed = true;

                } else {

                    DEBUG_LOGGER.log(Level.DEBUG,

                            "Not enough data to parse header, needs: %d, has: %d",

                            Http2Frame.FRAME_HEADER_SIZE, available);

                    return null;

                }

            }

            available = currentBuffer == null ? 0 : currentBuffer.remaining() + tailSize;

            if ((frameLength == 0) ||

                    (currentBuffer != null && available >= frameLength)) {

                Http2Frame frame = parseFrameBody();

                frameHeaderParsed = false;

                // frame == null means we have to skip this frame and try parse next

                if (frame != null) {

                    return frame;

                }

            } else {

                DEBUG_LOGGER.log(Level.DEBUG,

                        "Not enough data to parse frame body, needs: %d,  has: %d",

                        frameLength, available);

                return null;  // no data for the whole frame header

            }

        }

    }

    private void frameProcessed() {

        prepareToRelease.clear();

    }

    private void parseFrameHeader() throws IOException {

        int x = getInt();

        this.frameLength = (x >>> 8) & 0x00ffffff;

        this.frameType = x & 0xff;

        this.frameFlags = getByte();

        this.frameStreamid = getInt() & 0x7fffffff;

        // R: A reserved 1-bit field.  The semantics of this bit are undefined,

        // MUST be ignored when receiving.

    }

    // move next buffer from tailBuffers to currentBuffer if required

    private void nextBuffer() {

        if (!currentBuffer.hasRemaining()) {

            if (!slicedToDataFrame) {

                prepareToRelease.add(currentBuffer);

            }

            slicedToDataFrame = false;

            currentBuffer = tailBuffers.poll();

            if (currentBuffer != null) {

                tailSize -= currentBuffer.remaining();

            }

        }

    }

    public int getByte() {

        int res = currentBuffer.get() & 0xff;

        nextBuffer();

        return res;

    }

    public int getShort() {

        if (currentBuffer.remaining() >= 2) {

            int res = currentBuffer.getShort() & 0xffff;

            nextBuffer();

            return res;

        }

        int val = getByte();

        val = (val << 8) + getByte();

        return val;

    }

    public int getInt() {

        if (currentBuffer.remaining() >= 4) {

            int res = currentBuffer.getInt();

            nextBuffer();

            return res;

        }

        int val = getByte();

        val = (val << 8) + getByte();

        val = (val << 8) + getByte();

        val = (val << 8) + getByte();

        return val;

    }

    public byte[] getBytes(int n) {

        byte[] bytes = new byte[n];

        int offset = 0;

        while (n > 0) {

            int length = Math.min(n, currentBuffer.remaining());

            currentBuffer.get(bytes, offset, length);

            offset += length;

            n -= length;

            nextBuffer();

        }

        return bytes;

    }

    private List<ByteBuffer> getBuffers(boolean isDataFrame, int bytecount) {

        List<ByteBuffer> res = new ArrayList<>();

        while (bytecount > 0) {

            int remaining = currentBuffer.remaining();

            int extract = Math.min(remaining, bytecount);

            ByteBuffer extractedBuf;

            if (isDataFrame) {

                slicedToDataFrame = true;

            } else {

                // Header frames here

                // HPACK decoding should performed under lock and immediately after frame decoding.

                // in that case it is safe to release original buffer,

                // because of sliced buffer has a very short life

            }

            res.add(extractedBuf);

            bytecount -= extract;

            nextBuffer();

        }

        return res;

    }

    public void close(String msg) {

        closed = true;

        tailBuffers.clear();

        int bytes = tailSize;

        ByteBuffer b = currentBuffer;

        if (b != null) {

            bytes += b.remaining();

            b.position(b.limit());

        }

        tailSize = 0;

        currentBuffer = null;

        DEBUG_LOGGER.log(Level.DEBUG, "closed %s, ignoring %d bytes", msg, bytes);

    }

    public void skipBytes(int bytecount) {

        while (bytecount > 0) {

            int remaining = currentBuffer.remaining();

            int extract = Math.min(remaining, bytecount);

            currentBuffer.position(currentBuffer.position() + extract);

            bytecount -= remaining;

            nextBuffer();

        }

    }

    private Http2Frame parseFrameBody() throws IOException {

        assert frameHeaderParsed;

        switch (frameType) {

            case DataFrame.TYPE:

                return parseDataFrame(frameLength, frameStreamid, frameFlags);

            case HeadersFrame.TYPE:

                return parseHeadersFrame(frameLength, frameStreamid, frameFlags);

            case PriorityFrame.TYPE:

                return parsePriorityFrame(frameLength, frameStreamid, frameFlags);

            case ResetFrame.TYPE:

                return parseResetFrame(frameLength, frameStreamid, frameFlags);

            case SettingsFrame.TYPE:

                return parseSettingsFrame(frameLength, frameStreamid, frameFlags);

            case PushPromiseFrame.TYPE:

                return parsePushPromiseFrame(frameLength, frameStreamid, frameFlags);

            case PingFrame.TYPE:

                return parsePingFrame(frameLength, frameStreamid, frameFlags);

            case GoAwayFrame.TYPE:

                return parseGoAwayFrame(frameLength, frameStreamid, frameFlags);

            case WindowUpdateFrame.TYPE:

                return parseWindowUpdateFrame(frameLength, frameStreamid, frameFlags);

            case ContinuationFrame.TYPE:

                return parseContinuationFrame(frameLength, frameStreamid, frameFlags);

            default:

                // RFC 7540 4.1

                // Implementations MUST ignore and discard any frame that has a type that is unknown.

                Log.logTrace("Unknown incoming frame type: {0}", frameType);

                skipBytes(frameLength);

                return null;

        }

    }

    private Http2Frame parseDataFrame(int frameLength, int streamid, int flags) {

        // non-zero stream

        if (streamid == 0) {

            return new MalformedFrame(ErrorFrame.PROTOCOL_ERROR,

                                      "zero streamId for DataFrame");

        }

        int padLength = 0;

        if ((flags & DataFrame.PADDED) != 0) {

            padLength = getByte();

            if (padLength >= frameLength) {

                return new MalformedFrame(ErrorFrame.PROTOCOL_ERROR,

                        "the length of the padding is the length of the frame payload or greater");

            }

            frameLength--;

        }

        DataFrame df = new DataFrame(streamid, flags,

                getBuffers(true, frameLength - padLength), padLength);

        skipBytes(padLength);

        return df;

    }

    private Http2Frame parseHeadersFrame(int frameLength, int streamid, int flags) {

        // non-zero stream

        if (streamid == 0) {

            return new MalformedFrame(ErrorFrame.PROTOCOL_ERROR,

                                      "zero streamId for HeadersFrame");

        }

        int padLength = 0;

        if ((flags & HeadersFrame.PADDED) != 0) {

            padLength = getByte();

            frameLength--;

        }

        boolean hasPriority = (flags & HeadersFrame.PRIORITY) != 0;

        boolean exclusive = false;

        int streamDependency = 0;

        int weight = 0;

        if (hasPriority) {

            int x = getInt();

            exclusive = (x & 0x80000000) != 0;

            streamDependency = x & 0x7fffffff;

            weight = getByte();

            frameLength -= 5;

        }

        if(frameLength < padLength) {

            return new MalformedFrame(ErrorFrame.PROTOCOL_ERROR,

                    "Padding exceeds the size remaining for the header block");

        }

        HeadersFrame hf = new HeadersFrame(streamid, flags,

                getBuffers(false, frameLength - padLength), padLength);

        skipBytes(padLength);

        if (hasPriority) {

            hf.setPriority(streamDependency, exclusive, weight);

        }

        return hf;

    }

    private Http2Frame parsePriorityFrame(int frameLength, int streamid, int flags) {

        // non-zero stream; no flags

        if (streamid == 0) {

            return new MalformedFrame(ErrorFrame.PROTOCOL_ERROR,

                    "zero streamId for PriorityFrame");

        }

        if(frameLength != 5) {

            skipBytes(frameLength);

            return new MalformedFrame(ErrorFrame.FRAME_SIZE_ERROR, streamid,

                    "PriorityFrame length is "+ frameLength+", expected 5");

        }

        int x = getInt();

        int weight = getByte();

        return new PriorityFrame(streamid, x & 0x7fffffff, (x & 0x80000000) != 0, weight);

    }

    private Http2Frame parseResetFrame(int frameLength, int streamid, int flags) {

        // non-zero stream; no flags

        if (streamid == 0) {

            return new MalformedFrame(ErrorFrame.PROTOCOL_ERROR,

                    "zero streamId for ResetFrame");

        }

        if(frameLength != 4) {

            return new MalformedFrame(ErrorFrame.FRAME_SIZE_ERROR,

                    "ResetFrame length is "+ frameLength+", expected 4");

        }

        return new ResetFrame(streamid, getInt());

    }

    private Http2Frame parseSettingsFrame(int frameLength, int streamid, int flags) {

        // only zero stream

        if (streamid != 0) {

            return new MalformedFrame(ErrorFrame.PROTOCOL_ERROR,

                    "non-zero streamId for SettingsFrame");

        }

        if ((SettingsFrame.ACK & flags) != 0 && frameLength > 0) {

            // RFC 7540 6.5

            // Receipt of a SETTINGS frame with the ACK flag set and a length

            // field value other than 0 MUST be treated as a connection error

            return new MalformedFrame(ErrorFrame.FRAME_SIZE_ERROR,

                    "ACK SettingsFrame is not empty");

        }

        if (frameLength % 6 != 0) {

            return new MalformedFrame(ErrorFrame.FRAME_SIZE_ERROR,

                    "invalid SettingsFrame size: "+frameLength);

        }

        SettingsFrame sf = new SettingsFrame(flags);

        int n = frameLength / 6;

        for (int i=0; i<n; i++) {

            int id = getShort();

            int val = getInt();

            if (id > 0 && id <= SettingsFrame.MAX_PARAM) {

                // a known parameter. Ignore otherwise

                sf.setParameter(id, val); // TODO parameters validation

            }

        }

        return sf;

    }

    private Http2Frame parsePushPromiseFrame(int frameLength, int streamid, int flags) {

        // non-zero stream

        if (streamid == 0) {

            return new MalformedFrame(ErrorFrame.PROTOCOL_ERROR,

                    "zero streamId for PushPromiseFrame");

        }

        int padLength = 0;

        if ((flags & PushPromiseFrame.PADDED) != 0) {

            padLength = getByte();

            frameLength--;

        }

        int promisedStream = getInt() & 0x7fffffff;

        frameLength -= 4;

        if(frameLength < padLength) {

            return new MalformedFrame(ErrorFrame.PROTOCOL_ERROR,

                    "Padding exceeds the size remaining for the PushPromiseFrame");

        }

        PushPromiseFrame ppf = new PushPromiseFrame(streamid, flags, promisedStream,

                getBuffers(false, frameLength - padLength), padLength);

        skipBytes(padLength);

        return ppf;

    }

    private Http2Frame parsePingFrame(int frameLength, int streamid, int flags) {

        // only zero stream

        if (streamid != 0) {

            return new MalformedFrame(ErrorFrame.PROTOCOL_ERROR,

                    "non-zero streamId for PingFrame");

        }

        if(frameLength != 8) {

            return new MalformedFrame(ErrorFrame.FRAME_SIZE_ERROR,

                    "PingFrame length is "+ frameLength+", expected 8");

        }

        return new PingFrame(flags, getBytes(8));

    }

    private Http2Frame parseGoAwayFrame(int frameLength, int streamid, int flags) {

        // only zero stream; no flags

        if (streamid != 0) {

            return new MalformedFrame(ErrorFrame.PROTOCOL_ERROR,

                    "non-zero streamId for GoAwayFrame");

        }

        if (frameLength < 8) {

            return new MalformedFrame(ErrorFrame.FRAME_SIZE_ERROR,

                    "Invalid GoAway frame size");