//
//  ========================================================================
//  Copyright (c) 1995-2013 Mort Bay Consulting Pty. Ltd.
//  ------------------------------------------------------------------------
//  All rights reserved. This program and the accompanying materials
//  are made available under the terms of the Eclipse Public License v1.0
//  and Apache License v2.0 which accompanies this distribution.
//
//      The Eclipse Public License is available at
//      http://www.eclipse.org/legal/epl-v10.html
//
//      The Apache License v2.0 is available at
//      http://www.opensource.org/licenses/apache2.0.php
//
//  You may elect to redistribute this code under either of these licenses.
//  ========================================================================
//

package org.eclipse.jetty.websocket.common;

import java.nio.ByteBuffer;
import java.util.List;

import org.eclipse.jetty.io.ByteBufferPool;
import org.eclipse.jetty.util.BufferUtil;
import org.eclipse.jetty.util.log.Log;
import org.eclipse.jetty.util.log.Logger;
import org.eclipse.jetty.websocket.api.MessageTooLargeException;
import org.eclipse.jetty.websocket.api.ProtocolException;
import org.eclipse.jetty.websocket.api.WebSocketBehavior;
import org.eclipse.jetty.websocket.api.WebSocketException;
import org.eclipse.jetty.websocket.api.WebSocketPolicy;
import org.eclipse.jetty.websocket.api.extensions.Extension;
import org.eclipse.jetty.websocket.api.extensions.Frame;
import org.eclipse.jetty.websocket.api.extensions.IncomingFrames;
import org.eclipse.jetty.websocket.common.io.payload.CloseReasonValidator;
import org.eclipse.jetty.websocket.common.io.payload.DeMaskProcessor;
import org.eclipse.jetty.websocket.common.io.payload.NoOpValidator;
import org.eclipse.jetty.websocket.common.io.payload.PayloadProcessor;
import org.eclipse.jetty.websocket.common.io.payload.UTF8Validator;

/**
 * Parsing of a frames in WebSocket land.
 */
public class Parser
{
    private enum State
    {
        START,
        FINOP,
        PAYLOAD_LEN,
        PAYLOAD_LEN_BYTES,
        MASK,
        MASK_BYTES,
        PAYLOAD
    }

    private static final Logger LOG = Log.getLogger(Parser.class);
    private final WebSocketPolicy policy;
    private final ByteBufferPool bufferPool;

    // State specific
    private State state = State.START;
    private int cursor = 0;
    // Frame
    private WebSocketFrame frame;
    private Frame priorDataFrame;
    private byte lastDataOpcode;
    // payload specific
    private ByteBuffer payload;
    private int payloadLength;
    private PayloadProcessor maskProcessor = new DeMaskProcessor();
    private PayloadProcessor strictnessProcessor;

    /** Is there an extension using RSV1 */
    private boolean rsv1InUse = false;
    /** Is there an extension using RSV2 */
    private boolean rsv2InUse = false;
    /** Is there an extension using RSV3 */
    private boolean rsv3InUse = false;
    /** Is there an extension that processes invalid UTF8 text messages (such as compressed content) */
    private boolean isTextFrameValidated = true;

    private IncomingFrames incomingFramesHandler;

    public Parser(WebSocketPolicy wspolicy, ByteBufferPool bufferPool)
    {
        this.bufferPool = bufferPool;
        this.policy = wspolicy;
    }

    private void assertSanePayloadLength(long len)
    {
        LOG.debug("Payload Length: " + len);
        // Since we use ByteBuffer so often, having lengths over Integer.MAX_VALUE is really impossible.
        if (len > Integer.MAX_VALUE)
        {
            // OMG! Sanity Check! DO NOT WANT! Won't anyone think of the memory!
            throw new MessageTooLargeException("[int-sane!] cannot handle payload lengths larger than " + Integer.MAX_VALUE);
        }
        policy.assertValidMessageSize((int)len);

        switch (frame.getOpCode())
        {
            case OpCode.CLOSE:
                if (len == 1)
                {
                    throw new ProtocolException("Invalid close frame payload length, [" + payloadLength + "]");
                }
                // fall thru
            case OpCode.PING:
            case OpCode.PONG:
                if (len > WebSocketFrame.MAX_CONTROL_PAYLOAD)
                {
                    throw new ProtocolException("Invalid control frame payload length, [" + payloadLength + "] cannot exceed ["
                            + WebSocketFrame.MAX_CONTROL_PAYLOAD + "]");
                }
                break;
        }
    }

    public void configureFromExtensions(List<? extends Extension> exts)
    {
        // default
        this.rsv1InUse = false;
        this.rsv2InUse = false;
        this.rsv3InUse = false;
        this.isTextFrameValidated = true;

        // configure from list of extensions in use
        for (Extension ext : exts)
        {
            if (ext.isRsv1User())
            {
                this.rsv1InUse = true;
            }
            if (ext.isRsv2User())
            {
                this.rsv2InUse = true;
            }
            if (ext.isRsv3User())
            {
                this.rsv3InUse = true;
            }
            if (ext.isTextDataDecoder())
            {
                this.isTextFrameValidated = false;
            }
        }
    }

    public IncomingFrames getIncomingFramesHandler()
    {
        return incomingFramesHandler;
    }

    public WebSocketPolicy getPolicy()
    {
        return policy;
    }

    public boolean isRsv1InUse()
    {
        return rsv1InUse;
    }

    public boolean isRsv2InUse()
    {
        return rsv2InUse;
    }

    public boolean isRsv3InUse()
    {
        return rsv3InUse;
    }

    protected void notifyFrame(final Frame f)
    {
        if (LOG.isDebugEnabled())
        {
            LOG.debug("{} Notify {}",policy.getBehavior(),incomingFramesHandler);
        }

        if (policy.getBehavior() == WebSocketBehavior.SERVER)
        {
            // Parsing on server?
            // Then you MUST make sure all incoming frames are masked!
            if (f.isMasked() == false)
            {
                throw new ProtocolException("Client frames MUST be masked (RFC-6455)");
            }
        }

        if (incomingFramesHandler == null)
        {
            return;
        }
        try
        {
            incomingFramesHandler.incomingFrame(f);
        }
        catch (WebSocketException e)
        {
            notifyWebSocketException(e);
        }
        catch (Throwable t)
        {
            LOG.warn(t);
            notifyWebSocketException(new WebSocketException(t));
        }
    }

    protected void notifyWebSocketException(WebSocketException e)
    {
        LOG.warn(e);
        if (incomingFramesHandler == null)
        {
            return;
        }
        incomingFramesHandler.incomingError(e);
    }

    public synchronized void parse(ByteBuffer buffer)
    {
        if (buffer.remaining() <= 0)
        {
            return;
        }
        try
        {
            // TODO: create DebugBuffer

            // parse through all the frames in the buffer
            while (parseFrame(buffer))
            {
                LOG.debug("{} Parsed Frame: {}",policy.getBehavior(),frame);
                notifyFrame(frame);
                if (frame.isDataFrame() && frame.isFin())
                {
                    priorDataFrame = null;
                }
                else
                {
                    priorDataFrame = frame;
                }
            }
        }
        catch (WebSocketException e)
        {
            buffer.position(buffer.limit()); // consume remaining
            this.payload = null; // reset
            notifyWebSocketException(e);
        }
        catch (Throwable t)
        {
            buffer.position(buffer.limit()); // consume remaining
            this.payload = null; // reset
            notifyWebSocketException(new WebSocketException(t));
        }
    }

    /**
     * Parse the base framing protocol buffer.
     * <p>
     * Note the first byte (fin,rsv1,rsv2,rsv3,opcode) are parsed by the {@link Parser#parse(ByteBuffer)} method
     * <p>
     * Not overridable
     * 
     * @param buffer
     *            the buffer to parse from.
     * @return true if done parsing base framing protocol and ready for parsing of the payload. false if incomplete parsing of base framing protocol.
     */
    private boolean parseFrame(ByteBuffer buffer)
    {
        if (buffer.remaining() <= 0)
        {
            return false;
        }

        LOG.debug("{} Parsing {} bytes",policy.getBehavior(),buffer.remaining());
        while (buffer.hasRemaining())
        {
            switch (state)
            {
                case START:
                {
                    if ((frame != null) && (frame.isFin()))
                    {
                        frame.reset();
                    }

                    state = State.FINOP;
                    break;
                }
                case FINOP:
                {
                    // peek at byte
                    byte b = buffer.get();
                    boolean fin = ((b & 0x80) != 0);
                    boolean rsv1 = ((b & 0x40) != 0);
                    boolean rsv2 = ((b & 0x20) != 0);
                    boolean rsv3 = ((b & 0x10) != 0);
                    byte opc = (byte)(b & 0x0F);
                    byte opcode = opc;

                    if (!OpCode.isKnown(opcode))
                    {
                        throw new ProtocolException("Unknown opcode: " + opc);
                    }

                    if (LOG.isDebugEnabled())
                    {
                        LOG.debug("OpCode {}, fin={} rsv={}{}{}",OpCode.name(opcode),fin,(rsv1?'1':'.'),(rsv2?'1':'.'),(rsv3?'1':'.'));
                    }

                    /*
                     * RFC 6455 Section 5.2
                     * 
                     * MUST be 0 unless an extension is negotiated that defines meanings for non-zero values. If a nonzero value is received and none of the
                     * negotiated extensions defines the meaning of such a nonzero value, the receiving endpoint MUST _Fail the WebSocket Connection_.
                     */
                    if (!rsv1InUse && rsv1)
                    {
                        throw new ProtocolException("RSV1 not allowed to be set");
                    }

                    if (!rsv2InUse && rsv2)
                    {
                        throw new ProtocolException("RSV2 not allowed to be set");
                    }

                    if (!rsv3InUse && rsv3)
                    {
                        throw new ProtocolException("RSV3 not allowed to be set");
                    }

                    boolean isContinuation = false;

                    switch (opcode)
                    {
                        case OpCode.TEXT:
                            if (isTextFrameValidated)
                            {
                                strictnessProcessor = new UTF8Validator();
                            }
                            else
                            {
                                strictnessProcessor = NoOpValidator.INSTANCE;
                            }
                            break;
                        case OpCode.CLOSE:
                            strictnessProcessor = new CloseReasonValidator();
                            break;
                        default:
                            strictnessProcessor = NoOpValidator.INSTANCE;
                            break;
                    }

                    if (OpCode.isControlFrame(opcode))
                    {
                        // control frame validation
                        if (!fin)
                        {
                            throw new ProtocolException("Fragmented Control Frame [" + OpCode.name(opcode) + "]");
                        }
                    }
                    else if (opcode == OpCode.CONTINUATION)
                    {
                        isContinuation = true;
                        // continuation validation
                        if (priorDataFrame == null)
                        {
                            throw new ProtocolException("CONTINUATION frame without prior !FIN");
                        }
                        // Be careful to use the original opcode
                        opcode = lastDataOpcode;
                    }
                    else if (OpCode.isDataFrame(opcode))
                    {
                        // data validation
                        if ((priorDataFrame != null) && (!priorDataFrame.isFin()))
                        {
                            throw new ProtocolException("Unexpected " + OpCode.name(opcode) + " frame, was expecting CONTINUATION");
                        }
                    }

                    // base framing flags
                    frame = new WebSocketFrame(opcode);
                    frame.setFin(fin);
                    frame.setRsv1(rsv1);
                    frame.setRsv2(rsv2);
                    frame.setRsv3(rsv3);
                    frame.setContinuation(isContinuation);

                    if (frame.isDataFrame())
                    {
                        lastDataOpcode = opcode;
                    }

                    state = State.PAYLOAD_LEN;
                    break;
                }
                case PAYLOAD_LEN:
                {
                    byte b = buffer.get();
                    frame.setMasked((b & 0x80) != 0);
                    payloadLength = (byte)(0x7F & b);

                    if (payloadLength == 127) // 0x7F
                    {
                        // length 8 bytes (extended payload length)
                        payloadLength = 0;
                        state = State.PAYLOAD_LEN_BYTES;
                        cursor = 8;
                        break; // continue onto next state
                    }
                    else if (payloadLength == 126) // 0x7E
                    {
                        // length 2 bytes (extended payload length)
                        payloadLength = 0;
                        state = State.PAYLOAD_LEN_BYTES;
                        cursor = 2;
                        break; // continue onto next state
                    }

                    assertSanePayloadLength(payloadLength);
                    if (frame.isMasked())
                    {
                        state = State.MASK;
                    }
                    else
                    {
                        // special case for empty payloads (no more bytes left in buffer)
                        if (payloadLength == 0)
                        {
                            state = State.START;
                            return true;
                        }

                        maskProcessor.reset(frame);
                        state = State.PAYLOAD;
                    }

                    break;
                }
                case PAYLOAD_LEN_BYTES:
                {
                    byte b = buffer.get();
                    --cursor;
                    payloadLength |= (b & 0xFF) << (8 * cursor);
                    if (cursor == 0)
                    {
                        assertSanePayloadLength(payloadLength);
                        if (frame.isMasked())
                        {
                            state = State.MASK;
                        }
                        else
                        {
                            // special case for empty payloads (no more bytes left in buffer)
                            if (payloadLength == 0)
                            {
                                state = State.START;
                                return true;
                            }

                            maskProcessor.reset(frame);
                            state = State.PAYLOAD;
                        }
                    }
                    break;
                }
                case MASK:
                {
                    byte m[] = new byte[4];
                    frame.setMask(m);
                    if (buffer.remaining() >= 4)
                    {
                        buffer.get(m,0,4);
                        // special case for empty payloads (no more bytes left in buffer)
                        if (payloadLength == 0)
                        {
                            state = State.START;
                            return true;
                        }

                        maskProcessor.reset(frame);
                        state = State.PAYLOAD;
                    }
                    else
                    {
                        state = State.MASK_BYTES;
                        cursor = 4;
                    }
                    break;
                }
                case MASK_BYTES:
                {
                    byte b = buffer.get();
                    frame.getMask()[4 - cursor] = b;
                    --cursor;
                    if (cursor == 0)
                    {
                        // special case for empty payloads (no more bytes left in buffer)
                        if (payloadLength == 0)
                        {
                            state = State.START;
                            return true;
                        }

                        maskProcessor.reset(frame);
                        state = State.PAYLOAD;
                    }
                    break;
                }
                case PAYLOAD:
                {
                    if (parsePayload(buffer))
                    {
                        // special check for close
                        if (frame.getOpCode() == OpCode.CLOSE)
                        {
                            new CloseInfo(frame);
                        }
                        state = State.START;
                        // we have a frame!
                        return true;
                    }
                    break;
                }
            }
        }

        return false;
    }

    /**
     * Implementation specific parsing of a payload
     * 
     * @param buffer
     *            the payload buffer
     * @return true if payload is done reading, false if incomplete
     */
    private boolean parsePayload(ByteBuffer buffer)
    {
        if (payloadLength == 0)
        {
            return true;
        }

        while (buffer.hasRemaining())
        {
            if (payload == null)
            {
                frame.assertValid();
                payload = bufferPool.acquire(payloadLength,false);
                BufferUtil.clearToFill(payload);
            }

            // Create a small window of the incoming buffer to work with.
            // this should only show the payload itself, and not any more
            // bytes that could belong to the start of the next frame.
            ByteBuffer window = buffer.slice();
            int bytesExpected = payloadLength - payload.position();
            int bytesAvailable = buffer.remaining();
            int windowBytes = Math.min(bytesAvailable,bytesExpected);
            window.limit(window.position() + windowBytes);

            if (LOG.isDebugEnabled())
            {
                LOG.debug("Window: {}",BufferUtil.toDetailString(window));
            }

            maskProcessor.process(window);
            strictnessProcessor.process(window);
            int len = BufferUtil.put(window,payload);

            buffer.position(buffer.position() + len); // update incoming buffer position

            if (payload.position() >= payloadLength)
            {
                BufferUtil.flipToFlush(payload,0);
                frame.setPayload(payload);
                this.payload = null;
                return true;
            }
        }
        return false;
    }

    public void setIncomingFramesHandler(IncomingFrames incoming)
    {
        this.incomingFramesHandler = incoming;
    }

    @Override
    public String toString()
    {
        StringBuilder builder = new StringBuilder();
        builder.append("Parser[");
        if (incomingFramesHandler == null)
        {
            builder.append("NO_HANDLER");
        }
        else
        {
            builder.append(incomingFramesHandler.getClass().getSimpleName());
        }
        builder.append(",s=");
        builder.append(state);
        builder.append(",c=");
        builder.append(cursor);
        builder.append(",len=");
        builder.append(payloadLength);
        builder.append(",f=");
        builder.append(frame);
        builder.append("]");
        return builder.toString();
    }
}
