modules/tomcat-lite/java/org/apache/tomcat/lite/http/SpdyConnection.java

/*
 */
package org.apache.tomcat.lite.http;

import java.io.IOException;
import java.util.Collection;
import java.util.HashMap;
import java.util.Map;
import java.util.concurrent.atomic.AtomicInteger;
import java.util.logging.Level;
import java.util.logging.Logger;

import org.apache.tomcat.lite.http.HttpConnectionPool.RemoteServer;
import org.apache.tomcat.lite.http.HttpMessage.HttpMessageBytes;
import org.apache.tomcat.lite.io.BBucket;
import org.apache.tomcat.lite.io.BBuffer;
import org.apache.tomcat.lite.io.CBuffer;
import org.apache.tomcat.lite.io.DumpChannel;
import org.apache.tomcat.lite.io.IOBuffer;
import org.apache.tomcat.lite.io.IOChannel;
import org.apache.tomcat.lite.io.IOConnector;

/*
 * TODO: expectations ?
 * Fix docs - order matters
 * Crashes in chrome
 *
 * Test with unit tests or:
 *  google-chrome --use-flip=no-ssl
 *    --user-data-dir=/home/$USER/.config/google-chrome/Test
 *    http://localhost:8802/hello
 */

public class SpdyConnection extends HttpConnector.HttpConnection
        implements IOConnector.ConnectedCallback {


    /** Use compression for headers. Will magically turn to false
     * if the first request doesn't have x8xx ( i.e. compress header )
     */
    boolean headerCompression = true;
    boolean firstFrame = true;

    public static long DICT_ID = 3751956914L;
    private static String SPDY_DICT_S =
        "optionsgetheadpostputdeletetraceacceptaccept-charsetaccept-encodingaccept-" +
        "languageauthorizationexpectfromhostif-modified-sinceif-matchif-none-matchi" +
        "f-rangeif-unmodifiedsincemax-forwardsproxy-authorizationrangerefererteuser" +
        "-agent10010120020120220320420520630030130230330430530630740040140240340440" +
        "5406407408409410411412413414415416417500501502503504505accept-rangesageeta" +
        "glocationproxy-authenticatepublicretry-afterservervarywarningwww-authentic" +
        "ateallowcontent-basecontent-encodingcache-controlconnectiondatetrailertran" +
        "sfer-encodingupgradeviawarningcontent-languagecontent-lengthcontent-locati" +
        "oncontent-md5content-rangecontent-typeetagexpireslast-modifiedset-cookieMo" +
        "ndayTuesdayWednesdayThursdayFridaySaturdaySundayJanFebMarAprMayJunJulAugSe" +
        "pOctNovDecchunkedtext/htmlimage/pngimage/jpgimage/gifapplication/xmlapplic" +
        "ation/xhtmltext/plainpublicmax-agecharset=iso-8859-1utf-8gzipdeflateHTTP/1" +
        ".1statusversionurl ";
    public static byte[] SPDY_DICT = SPDY_DICT_S.getBytes();
    // C code uses this - not in the spec
    static {
        SPDY_DICT[SPDY_DICT.length - 1] = (byte) 0;
    }


    protected static Logger log = Logger.getLogger("SpdyConnection");

    /**
     * @param spdyConnector
     * @param remoteServer
     */
    SpdyConnection(HttpConnector spdyConnector, RemoteServer remoteServer) {
        this.httpConnector = spdyConnector;
        this.remoteHost = remoteServer;
        this.target = remoteServer.target;
    }

    AtomicInteger streamErrors = new AtomicInteger();

    AtomicInteger lastInStream = new AtomicInteger();
    AtomicInteger lastOutStream = new AtomicInteger();

    // TODO: use int map
    Map<Integer, HttpChannel> channels = new HashMap();

    SpdyConnection.Frame currentInFrame = null;

    SpdyConnection.Frame lastFrame = null; // for debug

    BBuffer outFrameBuffer = BBuffer.allocate();
    BBuffer inFrameBuffer = BBuffer.allocate();

    BBuffer headW = BBuffer.wrapper();

    CompressFilter headCompressIn = new CompressFilter()
        .setDictionary(SPDY_DICT, DICT_ID);

    CompressFilter headCompressOut = new CompressFilter()
        .setDictionary(SPDY_DICT, DICT_ID);

    IOBuffer headerCompressBuffer = new IOBuffer();
    IOBuffer headerDeCompressBuffer = new IOBuffer();

    AtomicInteger inFrames = new AtomicInteger();
    AtomicInteger inDataFrames = new AtomicInteger();
    AtomicInteger inSyncStreamFrames = new AtomicInteger();
    AtomicInteger inBytes = new AtomicInteger();

    AtomicInteger outFrames = new AtomicInteger();
    AtomicInteger outDataFrames = new AtomicInteger();
    AtomicInteger outBytes = new AtomicInteger();


    volatile boolean connecting = false;
    volatile boolean connected = false;

    // TODO: detect if it's spdy or http based on bit 8

    @Override
    public void withExtraBuffer(BBuffer received) {
        inFrameBuffer = received;
    }

    @Override
    public void dataReceived(IOBuffer iob) throws IOException {
        // Only one thread doing receive at a time.
        synchronized (inFrameBuffer) {
            while (true) {
                int avail = iob.available();
                if (avail == 0) {
                    return;
                }
                if (currentInFrame == null) {
                    if (inFrameBuffer.remaining() + avail < 8) {
                        return;
                    }
                    if (inFrameBuffer.remaining() < 8) {
                        int headRest = 8 - inFrameBuffer.remaining();
                        int rd = iob.read(inFrameBuffer, headRest);
                    }
                    currentInFrame = new SpdyConnection.Frame(); // TODO: reuse
                    currentInFrame.parse(this, inFrameBuffer);
                }
                if (iob.available() < currentInFrame.length) {
                    return;
                }
                // We have a full frame. Process it.
                onFrame(iob);

                // TODO: extra checks, make sure the frame is correct and
                // it consumed all data.
                currentInFrame = null;
            }
        }
    }


    /**
     * Frame received. Must consume all data for the frame.
     *
     * @param iob
     * @throws IOException
     */
    protected void onFrame(IOBuffer iob) throws IOException {
        // TODO: make sure we have enough data.
        lastFrame = currentInFrame;
        inFrames.incrementAndGet();
        inBytes.addAndGet(currentInFrame.length + 8);

        if (currentInFrame.c) {
            if (currentInFrame.type == SpdyConnection.Frame.TYPE_HELO) {
                // receivedHello = currentInFrame;
            } else if (currentInFrame.type == SpdyConnection.Frame.TYPE_SYN_STREAM) {
                inSyncStreamFrames.incrementAndGet();
                HttpChannel ch = new HttpChannel(); // TODO: reuse
                ch.channelId = SpdyConnection.readInt(iob);
                ch.setConnection(this);
                ch.httpConnector = this.httpConnector;
                if (serverMode) {
                    ch.serverMode(true);
                }
                if (this.httpConnector.defaultService != null) {
                    ch.setHttpService(this.httpConnector.defaultService);
                }

                synchronized (channels) {
                        channels.put(ch.channelId, ch);
                }

                try {
                    // pri and unused
                    SpdyConnection.readShort(iob);

                    HttpMessageBytes reqBytes = ch.getRequest().getMsgBytes();

                    processHeaders(iob, ch, reqBytes);
                } catch (Throwable t) {
                    log.log(Level.SEVERE, "Error parsing head", t);
                    abort("Error reading headers " + t);
                    return;
                }
                ch.getRequest().processReceivedHeaders();

                ch.handleHeadersReceived(ch.getRequest());

                if ((currentInFrame.flags & SpdyConnection.Frame.FLAG_HALF_CLOSE) != 0) {
                    ch.getIn().close();
                    ch.handleEndReceive();
                }
            } else if (currentInFrame.type == SpdyConnection.Frame.TYPE_SYN_REPLY) {
                int chId = SpdyConnection.readInt(iob);
                HttpChannel ch;
                synchronized (channels) {
                    ch = channels.get(chId);
                    if (ch == null) {
                        abort("Channel not found");
                    }
                }
                try {
                    SpdyConnection.readShort(iob);

                    HttpMessageBytes resBytes = ch.getResponse().getMsgBytes();

                    BBuffer head = processHeaders(iob, ch, resBytes);
                } catch (Throwable t) {
                    log.log(Level.SEVERE, "Error parsing head", t);
                    abort("Error reading headers " + t);
                    return;
                }
                ch.getResponse().processReceivedHeaders();

                ch.handleHeadersReceived(ch.getResponse());

                if ((currentInFrame.flags & SpdyConnection.Frame.FLAG_HALF_CLOSE) != 0) {
                    ch.getIn().close();
                    ch.handleEndReceive();
                }
            } else {
                log.warning("Unknown frame type " + currentInFrame.type);
                iob.advance(currentInFrame.length);
            }
        } else {
            inDataFrames.incrementAndGet();
            // data frame - part of an existing stream
            HttpChannel ch;
            synchronized (channels) {
                ch = channels.get(currentInFrame.streamId);
            }
            if (ch == null) {
                log.warning("Unknown stream ");
                net.close();
                net.startSending();
                return;
            }
            int len = currentInFrame.length;
            while (len > 0) {
                BBucket bb = iob.peekFirst();
                if (bb == null) {
                    // we should have all data
                    abort("Unexpected short read");
                    return;
                }
                if (len > bb.remaining()) {
                    ch.getIn().append(bb);
                    len -= bb.remaining();
                    bb.position(bb.limit());
                } else {
                    ch.getIn().append(bb, len);
                    bb.position(bb.position() + len);
                    len = 0;
                }
            }
            ch.sendHandleReceivedCallback();

            if ((currentInFrame.flags & SpdyConnection.Frame.FLAG_HALF_CLOSE) != 0) {
                ch.handleEndReceive();
            }
        }
        firstFrame = false;
    }

    /**
     * On frame error.
     */
    private void abort(String msg) throws IOException {
        streamErrors.incrementAndGet();
        synchronized(channels) {
            for (HttpChannel ch : channels.values()) {
                ch.abort(msg);
            }
        }
        close();
    }

    private BBuffer processHeaders(IOBuffer iob, HttpChannel ch,
            HttpMessageBytes reqBytes) throws IOException {
        int nvCount = 0;
        if (firstFrame) {
            int res = iob.peek() & 0xFF;
            if ((res & 0x0F) !=  8) {
                headerCompression = false;
            }
        }
        headRecvBuf.recycle();
        if (headerCompression) {
            // 0x800 headers seems a bit too much - assume compressed.
            // I wish this was a flag...
            headerDeCompressBuffer.recycle();
            // stream id ( 4 ) + unused ( 2 )
            // nvCount is compressed in impl - spec is different
            headCompressIn.decompress(iob, headerDeCompressBuffer,
                    currentInFrame.length - 6);
            headerDeCompressBuffer.copyAll(headRecvBuf);
            headerDeCompressBuffer.recycle();
            nvCount = readShort(headRecvBuf);
        } else {
            nvCount = readShort(iob);
            // 8 = stream Id (4) + pri/unused (2) + nvCount (2)
            // we know we have enough data
            int rd = iob.read(headRecvBuf, currentInFrame.length - 8);
            if (rd != currentInFrame.length - 8) {
                abort("Unexpected incomplete read");
            }
        }
        // Wrapper - so we don't change position in head
        headRecvBuf.wrapTo(headW);

        BBuffer nameBuf = BBuffer.wrapper();
        BBuffer valBuf = BBuffer.wrapper();

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

            int nameLen = SpdyConnection.readShort(headW);
            if (nameLen > headW.remaining()) {
                abort("Name too long");
            }

            nameBuf.setBytes(headW.array(), headW.position(),
                            nameLen);
            headW.advance(nameLen);

            int valueLen = SpdyConnection.readShort(headW);
            valBuf.setBytes(headW.array(), headW.position(),
                            valueLen);
            headW.advance(valueLen);

            // TODO: no need to send version, method if default

            if (nameBuf.equals("method")) {
                valBuf.wrapTo(reqBytes.method());
            } else if (nameBuf.equals("version")) {
                valBuf.wrapTo(reqBytes.protocol());
            } else if (nameBuf.equals("url")) {
                valBuf.wrapTo(reqBytes.url());
                // TODO: spdy uses full URL, we may want to trim
                // also no host header
            } else {
                int idx = reqBytes.addHeader();
                nameBuf.wrapTo(reqBytes.getHeaderName(idx));
                valBuf.wrapTo(reqBytes.getHeaderValue(idx));
            }

            // TODO: repeated values are separated by a 0
        }
        return headW;
    }

    @Override
    protected synchronized void sendRequest(HttpChannel http) throws IOException {
        if (serverMode) {
            throw new IOException("Only in client mode");
        }
        if (!checkConnection(http)) {
            return;
        }
        MultiMap mimeHeaders = http.getRequest().getMimeHeaders();

        BBuffer headBuf = BBuffer.allocate();
        SpdyConnection.appendShort(headBuf, mimeHeaders.size() + 3);
        serializeMime(mimeHeaders, headBuf);

        // TODO: url - with host prefix , method
        // optimize...
        SpdyConnection.appendAsciiHead(headBuf, "version");
        SpdyConnection.appendAsciiHead(headBuf, "HTTP/1.1");

        SpdyConnection.appendAsciiHead(headBuf, "method");
        SpdyConnection.appendAsciiHead(headBuf, http.getRequest().getMethod());

        SpdyConnection.appendAsciiHead(headBuf, "url");
        // TODO: url
        SpdyConnection.appendAsciiHead(headBuf, http.getRequest().requestURL());

        if (headerCompression && httpConnector.compression) {
            headerCompressBuffer.recycle();
            headCompressOut.compress(headBuf, headerCompressBuffer, false);
            headBuf.recycle();
            headerCompressBuffer.copyAll(headBuf);
        }

        // Frame head - 8
        BBuffer out = BBuffer.allocate();
        // Syn-reply
        out.putByte(0x80);
        out.putByte(0x01);
        out.putByte(0x00);
        out.putByte(0x01);

        CBuffer method = http.getRequest().method();
        if (method.equals("GET") || method.equals("HEAD")) {
            http.getOut().close();
        }

        if (http.getOut().isAppendClosed()) {
            out.putByte(0x01); // closed
        } else {
            out.putByte(0x00);
        }

        // Length, channel id (4) + unused (2) - headBuf has header count
        // and headers
        SpdyConnection.append24(out, headBuf.remaining() + 6);

        if (serverMode) {
            http.channelId = 2 * lastOutStream.incrementAndGet();
        } else {
            http.channelId = 2 * lastOutStream.incrementAndGet() + 1;
        }
        SpdyConnection.appendInt(out, http.channelId);

        http.setConnection(this);

        synchronized (channels) {
            channels.put(http.channelId, http);
        }

        out.putByte(0x00); // no priority
        out.putByte(0x00);

        sendFrame(out, headBuf);

        if (http.outMessage.state == HttpMessage.State.HEAD) {
            http.outMessage.state = HttpMessage.State.BODY_DATA;
        }
        if (http.getOut().isAppendClosed()) {
            http.handleEndSent();
        }

        // Any existing data
        //sendData(http);
    }


    public synchronized Collection<HttpChannel> getActives() {
        synchronized(channels) {
            return channels.values();
        }
    }

    @Override
    protected synchronized void sendResponseHeaders(HttpChannel http) throws IOException {
        if (!serverMode) {
            throw new IOException("Only in server mode");
        }

        if (http.getResponse().isCommitted()) {
            return;
        }
        http.getResponse().setCommitted(true);

        MultiMap mimeHeaders = http.getResponse().getMimeHeaders();

        BBuffer headBuf = BBuffer.allocate();


        //mimeHeaders.remove("content-length");
        BBuffer headers = headBuf;
        if (headerCompression) {
            headers = BBuffer.allocate();
        }

        //SpdyConnection.appendInt(headers, http.channelId);
        //headers.putByte(0);
        //headers.putByte(0);
        SpdyConnection.appendShort(headers, mimeHeaders.size() + 2);

        // chrome will crash if we don't send the header
        serializeMime(mimeHeaders, headers);

        // Must be at the end
        SpdyConnection.appendAsciiHead(headers, "status");
        SpdyConnection.appendAsciiHead(headers,
                Integer.toString(http.getResponse().getStatus()));

        SpdyConnection.appendAsciiHead(headers, "version");
        SpdyConnection.appendAsciiHead(headers, "HTTP/1.1");

        if (headerCompression) {
            headerCompressBuffer.recycle();
            headCompressOut.compress(headers, headerCompressBuffer, false);
            headerCompressBuffer.copyAll(headBuf);
            headerCompressBuffer.recycle();
        }

        BBuffer frameHead = BBuffer.allocate();
        // Syn-reply
        frameHead.putByte(0x80); // Control
        frameHead.putByte(0x01); // version
        frameHead.putByte(0x00); // 00 02 - SYN_REPLY
        frameHead.putByte(0x02);

        // It seems piggibacking data is not allowed
        frameHead.putByte(0x00);

        int len = headBuf.remaining() + 6;
        SpdyConnection.append24(frameHead, len);

//        // Stream-Id, unused
        SpdyConnection.appendInt(frameHead, http.channelId);
        frameHead.putByte(0);
        frameHead.putByte(0);

        sendFrame(frameHead, headBuf);
    }


    public void startSending(HttpChannel http) throws IOException {
        http.send(); // if needed

        if (net != null) {
            sendData(http);
            net.startSending();
        }
    }

    private void sendData(HttpChannel http) throws IOException {
        int avail = http.getOut().available();
        boolean closed = http.getOut().isAppendClosed();
        if (avail > 0 || closed) {
            sendDataFrame(http.getOut(), avail,
                    http.channelId, closed);
            if (avail > 0) {
                getOut().advance(avail);
            }
        }
        if (closed) {
            http.handleEndSent();
        }
    }

    private BBuffer serializeMime(MultiMap mimeHeaders, BBuffer headBuf)
            throws IOException {

        // TODO: duplicated headers not allowed
        for (int i = 0; i < mimeHeaders.size(); i++) {
            CBuffer name = mimeHeaders.getName(i);
            CBuffer value = mimeHeaders.getValue(i);
            if (name.length() == 0 || value.length() == 0) {
                continue;
            }
            SpdyConnection.appendShort(headBuf, name.length());
            name.toAscii(headBuf);
            SpdyConnection.appendShort(headBuf, value.length());
            value.toAscii(headBuf);
        }
        return headBuf;
    }


    private synchronized void sendFrame(BBuffer out, BBuffer headBuf)
            throws IOException {
        if (net == null) {
            return; // unit test
        }
        outBytes.addAndGet(out.remaining());
        net.getOut().append(out);
        if (headBuf != null) {
            net.getOut().append(headBuf);
            outBytes.addAndGet(headBuf.remaining());
        }
        net.startSending();
        outFrames.incrementAndGet();
    }

    public synchronized void sendDataFrame(IOBuffer out2, int avail,
            int channelId, boolean last) throws IOException {
        if (net == null) {
            return; // unit test
        }
        outFrameBuffer.recycle();
        SpdyConnection.appendInt(outFrameBuffer, channelId); // first bit 0 ?
        if (last) {
            outFrameBuffer.putByte(0x01); // closed
        } else {
            outFrameBuffer.putByte(0x00);
        }

        // TODO: chunk if too much data ( at least at 24 bits)
        SpdyConnection.append24(outFrameBuffer, avail);

        outBytes.addAndGet(outFrameBuffer.remaining() + avail);
        net.getOut().append(outFrameBuffer);

        if (avail > 0) {
            net.getOut().append(out2, avail);
        }
        net.startSending();
        outDataFrames.incrementAndGet();
    }

    static void appendInt(BBuffer headBuf, int length) throws IOException {
        headBuf.putByte((length & 0xFF000000) >> 24);
        headBuf.putByte((length & 0xFF0000) >> 16);
        headBuf.putByte((length & 0xFF00) >> 8);
        headBuf.putByte((length & 0xFF));
    }

    static void append24(BBuffer headBuf, int length) throws IOException {
        headBuf.putByte((length & 0xFF0000) >> 16);
        headBuf.putByte((length & 0xFF00) >> 8);
        headBuf.putByte((length & 0xFF));
    }

    static void appendAsciiHead(BBuffer headBuf, CBuffer s) throws IOException {
        appendShort(headBuf, s.length());
        for (int i = 0; i < s.length(); i++) {
            headBuf.append(s.charAt(i));
        }
    }

    static void appendShort(BBuffer headBuf, int length) throws IOException {
        if (length > 0xFFFF) {
            throw new IOException("Too long");
        }
        headBuf.putByte((length & 0xFF00) >> 8);
        headBuf.putByte((length & 0xFF));
    }

    static void appendAsciiHead(BBuffer headBuf, String s) throws IOException {
        SpdyConnection.appendShort(headBuf, s.length());
        for (int i = 0; i < s.length(); i++) {
            headBuf.append(s.charAt(i));
        }
    }

    static int readShort(BBuffer iob) throws IOException {
        int res = iob.readByte();
        return res << 8 | iob.readByte();
    }

    static int readShort(IOBuffer iob) throws IOException {
        int res = iob.read();
        return res << 8 | iob.read();
    }

    static int readInt(IOBuffer iob) throws IOException {
        int res = 0;
        for (int i = 0; i < 4; i++) {
            int b0 = iob.read();
            res = res << 8 | b0;
        }
        return res;
    }

    public static class Frame {
        int flags;

        int length;

        boolean c; // for control

        int version;

        int type;

        int streamId; // for data

        static int TYPE_HELO = 4;

        static int TYPE_SYN_STREAM = 1;

        static int TYPE_SYN_REPLY = 2;

        static int FLAG_HALF_CLOSE = 1;

        public void parse(SpdyConnection spdyConnection,
                BBuffer iob) throws IOException {
            int b0 = iob.read();
            if (b0 < 128) {
                // data frame
                c = false;
                streamId = b0;
                for (int i = 0; i < 3; i++) {
                    b0 = iob.read();
                    streamId = streamId << 8 | b0;
                }
            } else {
                c = true;
                b0 -= 128;
                version = ((b0 << 8) | iob.read());
                if (version != 1) {
                    spdyConnection.abort("Wrong version");
                    return;
                }
                b0 = iob.read();
                type = ((b0 << 8) | iob.read());
            }

            flags = iob.read();
            for (int i = 0; i < 3; i++) {
                b0 = iob.read();
                length = length << 8 | b0;
            }

            iob.recycle();
        }

    }

    @Override
    protected void endSendReceive(HttpChannel http) throws IOException {
        synchronized (channels) {
            HttpChannel doneHttp = channels.remove(http.channelId);
            if (doneHttp != http) {
                log.severe("Error removing " + doneHttp + " " + http);
            }
        }
        httpConnector.cpool.afterRequest(http, this, true);
    }

    /**
     * Framing error, client interrupt, etc.
     */
    public void abort(HttpChannel http, String t) throws IOException {
        // TODO: send interrupt signal

    }


    private boolean checkConnection(HttpChannel http) throws IOException {
        synchronized(this) {
            if (net == null || !isOpen()) {
                connected = false;
            }

            if (!connected) {
                if (!connecting) {
                    // TODO: secure set at start ?
                    connecting = true;
                    httpConnector.cpool.httpConnect(http,
                            target.toString(),
                            http.getRequest().isSecure(), this);
                }

                synchronized (remoteHost) {
                    remoteHost.pending.add(http);
                    httpConnector.cpool.queued.incrementAndGet();
                }
                return false;
            }
        }

        return true;
    }

    @Override
    public void handleConnected(IOChannel net) throws IOException {
        HttpChannel httpCh = null;
        if (!net.isOpen()) {
            while (true) {
                synchronized (remoteHost) {
                    if (remoteHost.pending.size() == 0) {
                        return;
                    }
                    httpCh = remoteHost.pending.remove();
                }
                httpCh.abort("Can't connect");
            }
        }

        synchronized (remoteHost) {
            httpCh = remoteHost.pending.peek();
        }
        if (httpCh != null) {
            secure = httpCh.getRequest().isSecure();
            if (secure) {
                if (httpConnector.debugHttp) {
                    net = DumpChannel.wrap("SPDY-SSL", net);
                }
                String[] hostPort = httpCh.getTarget().split(":");

                IOChannel ch1 = httpConnector.sslProvider.channel(net,
                        hostPort[0], Integer.parseInt(hostPort[1]));
                //net.setHead(ch1);
                net = ch1;
            }
        }
        if (httpConnector.debugHttp) {
            net = DumpChannel.wrap("SPDY", net);
        }

        setSink(net);

        synchronized(this) {
            connecting = false;
            connected = true;
        }

        while (true) {
            synchronized (remoteHost) {
                if (remoteHost.pending.size() == 0) {
                    return;
                }
                httpCh = remoteHost.pending.remove();
            }
            sendRequest(httpCh);
        }

    }
}