Merge pull request #184 from jpinner/spdy_http_layer_3_2

SPDY Protocol HTTP Layer

src/main/java/org/jboss/netty/handler/codec/spdy/SpdyCodecUtil.java

@@ -139,6 +139,14 @@ final class SpdyCodecUtil {
                       (buf.getByte(offset + 3) & 0xFF));
     }
 
+    /**
+     * Returns {@code true} if ID is for a server initiated stream or ping.
+     */
+    static boolean isServerID(int ID) {
+        // Server initiated streams and pings have even IDs
+        return ID % 2 == 0;
+    }
+
     /**
      * Validate a SPDY header name.
      */

src/main/java/org/jboss/netty/handler/codec/spdy/SpdyFrameCodec.java

@@ -43,10 +43,27 @@ import org.jboss.netty.channel.ChannelUpstreamHandler;
 public class SpdyFrameCodec implements ChannelUpstreamHandler,
        ChannelDownstreamHandler {
 
-    private final SpdyFrameDecoder decoder = new SpdyFrameDecoder();
+    private final SpdyFrameDecoder decoder;
-    private final SpdyFrameEncoder encoder = new SpdyFrameEncoder();
+    private final SpdyFrameEncoder encoder;
 
+    /**
+     * Creates a new instance with the default decoder and encoder options
+     * ({@code maxChunkSize (8192)}, {@code maxFrameSize (65536)},
+     * {@code maxHeaderSize (16384)}, {@code compressionLevel (6)},
+     * {@code windowBits (15)}, and {@code memLevel (8)}).
+     */
     public SpdyFrameCodec() {
+        this(8192, 65536, 16384, 6, 15, 8);
+    }
+
+    /**
+     * Creates a new instance with the specified decoder and encoder options.
+     */
+    public SpdyFrameCodec(
+            int maxChunkSize, int maxFrameSize, int maxHeaderSize,
+            int compressionLevel, int windowBits, int memLevel) {
+        decoder = new SpdyFrameDecoder(maxChunkSize, maxFrameSize, maxHeaderSize);
+        encoder = new SpdyFrameEncoder(compressionLevel, windowBits, memLevel);
     }
 
     public void handleUpstream(ChannelHandlerContext ctx, ChannelEvent e)

src/main/java/org/jboss/netty/handler/codec/spdy/SpdyFrameDecoder.java

@@ -44,11 +44,42 @@ import static org.jboss.netty.handler.codec.spdy.SpdyCodecUtil.*;
  */
 public class SpdyFrameDecoder extends FrameDecoder {
 
+    private final int maxChunkSize;
+    private final int maxFrameSize;
+    private final int maxHeaderSize;
+
     private final DecoderEmbedder<ChannelBuffer> headerBlockDecompressor =
         new DecoderEmbedder<ChannelBuffer>(new ZlibDecoder(SPDY_DICT));
 
+    /**
+     * Creates a new instance with the default {@code maxChunkSize (8192)},
+     * {@code maxFrameSize (65536)}, and {@code maxHeaderSize (16384)}.
+     */
     public SpdyFrameDecoder() {
-        super();
+        this(8192, 65536, 16384);
+    }
+
+    /**
+     * Creates a new instance with the specified parameters.
+     */
+    public SpdyFrameDecoder(
+            int maxChunkSize, int maxFrameSize, int maxHeaderSize) {
+        super(true); // Enable unfold for data frames
+        if (maxChunkSize <= 0) {
+            throw new IllegalArgumentException(
+                    "maxChunkSize must be a positive integer: " + maxChunkSize);
+        }
+        if (maxFrameSize <= 0) {
+            throw new IllegalArgumentException(
+                    "maxFrameSize must be a positive integer: " + maxFrameSize);
+        }
+        if (maxHeaderSize <= 0) {
+            throw new IllegalArgumentException(
+                    "maxHeaderSize must be a positive integer: " + maxHeaderSize);
+        }
+        this.maxChunkSize = maxChunkSize;
+        this.maxFrameSize = maxFrameSize;
+        this.maxHeaderSize = maxHeaderSize;
     }
 
     @Override
@@ -67,6 +98,12 @@ public class SpdyFrameDecoder extends FrameDecoder {
         int dataLength   = getUnsignedMedium(buffer, lengthOffset);
         int frameLength  = SPDY_HEADER_SIZE + dataLength;
 
+        // Throw exception if frameLength exceeds maxFrameSize
+        if (frameLength > maxFrameSize) {
+            throw new SpdyProtocolException(
+                    "Frame length exceeds " + maxFrameSize + ": " + frameLength);
+        }
+
         // Wait until entire frame is readable
         if (buffer.readableBytes() < frameLength) {
             return null;
@@ -98,12 +135,25 @@ public class SpdyFrameDecoder extends FrameDecoder {
             int streamID = getUnsignedInt(buffer, frameOffset);
             buffer.skipBytes(SPDY_HEADER_SIZE);
 
+            // Generate data frames that do not exceed maxChunkSize
+            int numFrames = dataLength / maxChunkSize;
+            if (dataLength % maxChunkSize != 0) {
+                numFrames ++;
+            }
+            SpdyDataFrame[] frames = new SpdyDataFrame[numFrames];
+            for (int i = 0; i < numFrames; i++) {
+                int chunkSize = Math.min(maxChunkSize, dataLength);
                 SpdyDataFrame spdyDataFrame = new DefaultSpdyDataFrame(streamID);
-            spdyDataFrame.setLast((flags & SPDY_DATA_FLAG_FIN) != 0);
                 spdyDataFrame.setCompressed((flags & SPDY_DATA_FLAG_COMPRESS) != 0);
-            spdyDataFrame.setData(buffer.readBytes(dataLength));
+                spdyDataFrame.setData(buffer.readBytes(chunkSize));
+                dataLength -= chunkSize;
+                if (dataLength == 0) {
+                    spdyDataFrame.setLast((flags & SPDY_DATA_FLAG_FIN) != 0);
+                }
+                frames[i] = spdyDataFrame;
+            }
 
-            return spdyDataFrame;
+            return frames;
         }
     }
 
@@ -276,6 +326,7 @@ public class SpdyFrameDecoder extends FrameDecoder {
             throw new SpdyProtocolException(
                     "Received invalid header block");
         }
+        int headerSize = 0;
         int numEntries = getUnsignedShort(headerBlock, headerBlock.readerIndex());
         headerBlock.skipBytes(2);
         for (int i = 0; i < numEntries; i ++) {
@@ -289,6 +340,11 @@ public class SpdyFrameDecoder extends FrameDecoder {
                 headerFrame.setInvalid();
                 return;
             }
+            headerSize += nameLength;
+            if (headerSize > maxHeaderSize) {
+                throw new SpdyProtocolException(
+                        "Header block exceeds " + maxHeaderSize);
+            }
             if (headerBlock.readableBytes() < nameLength) {
                 throw new SpdyProtocolException(
                         "Received invalid header block");
@@ -310,6 +366,11 @@ public class SpdyFrameDecoder extends FrameDecoder {
                 headerFrame.setInvalid();
                 return;
             }
+            headerSize += valueLength;
+            if (headerSize > maxHeaderSize) {
+                throw new SpdyProtocolException(
+                        "Header block exceeds " + maxHeaderSize);
+            }
             if (headerBlock.readableBytes() < valueLength) {
                 throw new SpdyProtocolException(
                         "Received invalid header block");

src/main/java/org/jboss/netty/handler/codec/spdy/SpdyFrameEncoder.java

@@ -48,11 +48,23 @@ import static org.jboss.netty.handler.codec.spdy.SpdyCodecUtil.*;
  */
 public class SpdyFrameEncoder extends OneToOneEncoder {
 
-    private final EncoderEmbedder<ChannelBuffer> headerBlockCompressor =
+    private final EncoderEmbedder<ChannelBuffer> headerBlockCompressor;
-        new EncoderEmbedder<ChannelBuffer>(new ZlibEncoder(9, SPDY_DICT));
 
+    /**
+     * Creates a new instance with the default {@code compressionLevel (6)},
+     * {@code windowBits (15)}, and {@code memLevel (8)}.
+     */
     public SpdyFrameEncoder() {
+        this(6, 15, 8);
+    }
+
+    /**
+     * Creates a new instance with the specified parameters.
+     */
+    public SpdyFrameEncoder(int compressionLevel, int windowBits, int memLevel) {
         super();
+        headerBlockCompressor = new EncoderEmbedder<ChannelBuffer>(
+                new ZlibEncoder(compressionLevel, windowBits, memLevel, SPDY_DICT));
     }
 
     @Override

src/main/java/org/jboss/netty/handler/codec/spdy/SpdyHttpCodec.java

@@ -0,0 +1,64 @@
+/*
+ * Copyright 2012 The Netty Project
+ *
+ * The Netty Project licenses this file to you under the Apache License,
+ * version 2.0 (the "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at:
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS, WITHOUT
+ * WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the
+ * License for the specific language governing permissions and limitations
+ * under the License.
+ */
+/*
+ * Copyright 2012 Twitter, Inc.
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License"); you may
+ * not use this file except in compliance with the License. You may obtain
+ * a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.jboss.netty.handler.codec.spdy;
+
+import org.jboss.netty.channel.ChannelDownstreamHandler;
+import org.jboss.netty.channel.ChannelEvent;
+import org.jboss.netty.channel.ChannelHandlerContext;
+import org.jboss.netty.channel.ChannelUpstreamHandler;
+
+/**
+ * A combination of {@link SpdyHttpDecoder} and {@link SpdyHttpEncoder}
+ * @apiviz.has org.jboss.netty.handler.codec.sdpy.SpdyHttpDecoder
+ * @apiviz.has org.jboss.netty.handler.codec.spdy.SpdyHttpEncoder
+ */
+public class SpdyHttpCodec implements ChannelUpstreamHandler, ChannelDownstreamHandler {
+
+    private final SpdyHttpDecoder decoder;
+    private final SpdyHttpEncoder encoder = new SpdyHttpEncoder();
+
+    /**
+     * Creates a new instance with the specified decoder options.
+     */
+    public SpdyHttpCodec(int maxContentLength) {
+        decoder = new SpdyHttpDecoder(maxContentLength);
+    }
+
+    public void handleUpstream(ChannelHandlerContext ctx, ChannelEvent e)
+            throws Exception {
+        decoder.handleUpstream(ctx, e);
+    }
+
+    public void handleDownstream(ChannelHandlerContext ctx, ChannelEvent e)
+            throws Exception {
+        encoder.handleDownstream(ctx, e);
+    }
+}

src/main/java/org/jboss/netty/handler/codec/spdy/SpdyHttpDecoder.java

@@ -0,0 +1,296 @@
+/*
+ * Copyright 2012 The Netty Project
+ *
+ * The Netty Project licenses this file to you under the Apache License,
+ * version 2.0 (the "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at:
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS, WITHOUT
+ * WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the
+ * License for the specific language governing permissions and limitations
+ * under the License.
+ */
+/*
+ * Copyright 2012 Twitter, Inc.
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License"); you may
+ * not use this file except in compliance with the License. You may obtain
+ * a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.jboss.netty.handler.codec.spdy;
+
+import java.util.HashMap;
+import java.util.List;
+import java.util.Map;
+
+import org.jboss.netty.buffer.ChannelBuffer;
+import org.jboss.netty.buffer.ChannelBuffers;
+import org.jboss.netty.channel.Channel;
+import org.jboss.netty.channel.ChannelHandlerContext;
+import org.jboss.netty.channel.Channels;
+import org.jboss.netty.handler.codec.frame.TooLongFrameException;
+import org.jboss.netty.handler.codec.http.DefaultHttpRequest;
+import org.jboss.netty.handler.codec.http.DefaultHttpResponse;
+import org.jboss.netty.handler.codec.http.HttpHeaders;
+import org.jboss.netty.handler.codec.http.HttpMessage;
+import org.jboss.netty.handler.codec.http.HttpMethod;
+import org.jboss.netty.handler.codec.http.HttpRequest;
+import org.jboss.netty.handler.codec.http.HttpResponse;
+import org.jboss.netty.handler.codec.http.HttpResponseStatus;
+import org.jboss.netty.handler.codec.http.HttpVersion;
+import org.jboss.netty.handler.codec.oneone.OneToOneDecoder;
+
+/**
+ * Decodes {@link SpdySynStreamFrame}s, {@link SpdySynReplyFrame}s,
+ * and {@link SpdyDataFrame}s into {@link HttpRequest}s and {@link HttpResponse}s.
+ */
+public class SpdyHttpDecoder extends OneToOneDecoder {
+
+    private final int maxContentLength;
+    private final Map<Integer, HttpMessage> messageMap = new HashMap<Integer, HttpMessage>();
+
+    /**
+     * Creates a new instance.
+     *
+     * @param maxContentLength the maximum length of the message content.
+     *        If the length of the message content exceeds this value,
+     *        a {@link TooLongFrameException} will be raised.
+     */
+    public SpdyHttpDecoder(int maxContentLength) {
+        super();
+        if (maxContentLength <= 0) {
+            throw new IllegalArgumentException(
+                    "maxContentLength must be a positive integer: " + maxContentLength);
+        }
+        this.maxContentLength = maxContentLength;
+    }
+
+    @Override
+    protected Object decode(ChannelHandlerContext ctx, Channel channel, Object msg)
+            throws Exception {
+
+        if (msg instanceof SpdySynStreamFrame) {
+
+            // HTTP requests/responses are mapped one-to-one to SPDY streams.
+            SpdySynStreamFrame spdySynStreamFrame = (SpdySynStreamFrame) msg;
+            int streamID = spdySynStreamFrame.getStreamID();
+
+            if (SpdyCodecUtil.isServerID(streamID)) {
+                // SYN_STREAM frames inititated by the server are pushed resources
+                int associatedToStreamID = spdySynStreamFrame.getAssociatedToStreamID();
+
+                // If a client receives a SYN_STREAM with an Associated-To-Stream-ID of 0
+                // it must reply with a RST_STREAM with error code INVALID_STREAM
+                if (associatedToStreamID == 0) {
+                    SpdyRstStreamFrame spdyRstStreamFrame =
+                        new DefaultSpdyRstStreamFrame(streamID, SpdyStreamStatus.INVALID_STREAM);
+                    Channels.write(ctx, Channels.future(channel), spdyRstStreamFrame);
+                }
+
+                String URL = SpdyHeaders.getUrl(spdySynStreamFrame);
+
+                // If a client receives a SYN_STREAM without a 'url' header
+                // it must reply with a RST_STREAM with error code PROTOCOL_ERROR
+                if (URL == null) {
+                    SpdyRstStreamFrame spdyRstStreamFrame =
+                        new DefaultSpdyRstStreamFrame(streamID, SpdyStreamStatus.PROTOCOL_ERROR);
+                    Channels.write(ctx, Channels.future(channel), spdyRstStreamFrame);
+                }
+
+                try {
+                    HttpResponse httpResponse = createHttpResponse(spdySynStreamFrame);
+
+                    // Set the Stream-ID, Associated-To-Stream-ID, Priority, and URL as headers
+                    SpdyHttpHeaders.setStreamID(httpResponse, streamID);
+                    SpdyHttpHeaders.setAssociatedToStreamID(httpResponse, associatedToStreamID);
+                    SpdyHttpHeaders.setPriority(httpResponse, spdySynStreamFrame.getPriority());
+                    SpdyHttpHeaders.setUrl(httpResponse, URL);
+
+                    if (spdySynStreamFrame.isLast()) {
+                        HttpHeaders.setContentLength(httpResponse, 0);
+                        return httpResponse;
+                    } else {
+                        // Response body will follow in a series of Data Frames
+                        messageMap.put(new Integer(streamID), httpResponse);
+                    }
+                } catch (Exception e) {
+                    SpdyRstStreamFrame spdyRstStreamFrame =
+                        new DefaultSpdyRstStreamFrame(streamID, SpdyStreamStatus.PROTOCOL_ERROR);
+                    Channels.write(ctx, Channels.future(channel), spdyRstStreamFrame);
+                }
+
+            } else {
+                // SYN_STREAM frames initiated by the client are HTTP requests
+                try {
+                    HttpRequest httpRequest = createHttpRequest(spdySynStreamFrame);
+
+                    // Set the Stream-ID as a header
+                    SpdyHttpHeaders.setStreamID(httpRequest, streamID);
+
+                    if (spdySynStreamFrame.isLast()) {
+                        return httpRequest;
+                    } else {
+                        // Request body will follow in a series of Data Frames
+                        messageMap.put(new Integer(streamID), httpRequest);
+                    }
+                } catch (Exception e) {
+                    // If a client sends a SYN_STREAM without method, url, and version headers
+                    // the server must reply with a HTTP 400 BAD REQUEST reply
+                    // Also sends HTTP 400 BAD REQUEST reply if header name/value pairs are invalid
+                    SpdySynReplyFrame spdySynReplyFrame = new DefaultSpdySynReplyFrame(streamID);
+                    spdySynReplyFrame.setLast(true);
+                    SpdyHeaders.setStatus(spdySynReplyFrame, HttpResponseStatus.BAD_REQUEST);
+                    SpdyHeaders.setVersion(spdySynReplyFrame, HttpVersion.HTTP_1_0);
+                    Channels.write(ctx, Channels.future(channel), spdySynReplyFrame);
+                }
+            }
+
+        } else if (msg instanceof SpdySynReplyFrame) {
+
+            SpdySynReplyFrame spdySynReplyFrame = (SpdySynReplyFrame) msg;
+            int streamID = spdySynReplyFrame.getStreamID();
+
+            try {
+                HttpResponse httpResponse = createHttpResponse(spdySynReplyFrame);
+
+                // Set the Stream-ID as a header
+                SpdyHttpHeaders.setStreamID(httpResponse, streamID);
+
+                if (spdySynReplyFrame.isLast()) {
+                    HttpHeaders.setContentLength(httpResponse, 0);
+                    return httpResponse;
+                } else {
+                    // Response body will follow in a series of Data Frames
+                    messageMap.put(new Integer(streamID), httpResponse);
+                }
+            } catch (Exception e) {
+                // If a client receives a SYN_REPLY without valid status and version headers
+                // the client must reply with a RST_STREAM frame indicating a PROTOCOL_ERROR
+                SpdyRstStreamFrame spdyRstStreamFrame =
+                    new DefaultSpdyRstStreamFrame(streamID, SpdyStreamStatus.PROTOCOL_ERROR);
+                Channels.write(ctx, Channels.future(channel), spdyRstStreamFrame);
+            }
+
+        } else if (msg instanceof SpdyHeadersFrame) {
+
+            SpdyHeadersFrame spdyHeadersFrame = (SpdyHeadersFrame) msg;
+            Integer streamID = new Integer(spdyHeadersFrame.getStreamID());
+            HttpMessage httpMessage = messageMap.get(streamID);
+
+            // If message is not in map discard HEADERS frame.
+            // SpdySessionHandler should prevent this from happening.
+            if (httpMessage == null) {
+                return null;
+            }
+
+            for (Map.Entry<String, String> e: spdyHeadersFrame.getHeaders()) {
+                httpMessage.addHeader(e.getKey(), e.getValue());
+            }
+
+        } else if (msg instanceof SpdyDataFrame) {
+
+            SpdyDataFrame spdyDataFrame = (SpdyDataFrame) msg;
+            Integer streamID = new Integer(spdyDataFrame.getStreamID());
+            HttpMessage httpMessage = messageMap.get(streamID);
+
+            // If message is not in map discard Data Frame.
+            // SpdySessionHandler should prevent this from happening.
+            if (httpMessage == null) {
+                return null;
+            }
+
+            ChannelBuffer content = httpMessage.getContent();
+            if (content.readableBytes() > maxContentLength - spdyDataFrame.getData().readableBytes()) {
+                messageMap.remove(streamID);
+                throw new TooLongFrameException(
+                        "HTTP content length exceeded " + maxContentLength + " bytes.");
+            }
+
+            if (content == ChannelBuffers.EMPTY_BUFFER) {
+                content = ChannelBuffers.dynamicBuffer(channel.getConfig().getBufferFactory());
+                content.writeBytes(spdyDataFrame.getData());
+                httpMessage.setContent(content);
+            } else {
+                content.writeBytes(spdyDataFrame.getData());
+            }
+
+            if (spdyDataFrame.isLast()) {
+                HttpHeaders.setContentLength(httpMessage, content.readableBytes());
+                messageMap.remove(streamID);
+                return httpMessage;
+            }
+        }
+
+        return null;
+    }
+
+    private HttpRequest createHttpRequest(SpdyHeaderBlock requestFrame)
+            throws Exception {
+        // Create the first line of the request from the name/value pairs
+        HttpMethod  method  = SpdyHeaders.getMethod(requestFrame);
+        String      url     = SpdyHeaders.getUrl(requestFrame);
+        HttpVersion version = SpdyHeaders.getVersion(requestFrame);
+        SpdyHeaders.removeMethod(requestFrame);
+        SpdyHeaders.removeUrl(requestFrame);
+        SpdyHeaders.removeVersion(requestFrame);
+
+        HttpRequest httpRequest = new DefaultHttpRequest(version, method, url);
+        for (Map.Entry<String, String> e: requestFrame.getHeaders()) {
+            httpRequest.addHeader(e.getKey(), e.getValue());
+        }
+
+        // Chunked encoding is no longer valid
+        List<String> encodings = httpRequest.getHeaders(HttpHeaders.Names.TRANSFER_ENCODING);
+        encodings.remove(HttpHeaders.Values.CHUNKED);
+        if (encodings.isEmpty()) {
+            httpRequest.removeHeader(HttpHeaders.Names.TRANSFER_ENCODING);
+        } else {
+            httpRequest.setHeader(HttpHeaders.Names.TRANSFER_ENCODING, encodings);
+        }
+
+        // The Connection and Keep-Alive headers are no longer valid
+        HttpHeaders.setKeepAlive(httpRequest, true);
+
+        return httpRequest;
+    }
+
+    private HttpResponse createHttpResponse(SpdyHeaderBlock responseFrame)
+            throws Exception {
+        // Create the first line of the response from the name/value pairs
+        HttpResponseStatus status = SpdyHeaders.getStatus(responseFrame);
+        HttpVersion version = SpdyHeaders.getVersion(responseFrame);
+        SpdyHeaders.removeStatus(responseFrame);
+        SpdyHeaders.removeVersion(responseFrame);
+
+        HttpResponse httpResponse = new DefaultHttpResponse(version, status);
+        for (Map.Entry<String, String> e: responseFrame.getHeaders()) {
+            httpResponse.addHeader(e.getKey(), e.getValue());
+        }
+
+        // Chunked encoding is no longer valid
+        List<String> encodings = httpResponse.getHeaders(HttpHeaders.Names.TRANSFER_ENCODING);
+        encodings.remove(HttpHeaders.Values.CHUNKED);
+        if (encodings.isEmpty()) {
+            httpResponse.removeHeader(HttpHeaders.Names.TRANSFER_ENCODING);
+        } else {
+            httpResponse.setHeader(HttpHeaders.Names.TRANSFER_ENCODING, encodings);
+        }
+        httpResponse.removeHeader(HttpHeaders.Names.TRAILER);
+
+        // The Connection and Keep-Alive headers are no longer valid
+        HttpHeaders.setKeepAlive(httpResponse, true);
+
+        return httpResponse;
+    }
+}

src/main/java/org/jboss/netty/handler/codec/spdy/SpdyHttpEncoder.java

@@ -0,0 +1,327 @@
+/*
+ * Copyright 2012 The Netty Project
+ *
+ * The Netty Project licenses this file to you under the Apache License,
+ * version 2.0 (the "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at:
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS, WITHOUT
+ * WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the
+ * License for the specific language governing permissions and limitations
+ * under the License.
+ */
+/*
+ * Copyright 2012 Twitter, Inc.
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License"); you may
+ * not use this file except in compliance with the License. You may obtain
+ * a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.jboss.netty.handler.codec.spdy;
+
+import java.util.List;
+import java.util.Map;
+
+import org.jboss.netty.buffer.ChannelBuffer;
+import org.jboss.netty.buffer.ChannelBuffers;
+import org.jboss.netty.channel.ChannelDownstreamHandler;
+import org.jboss.netty.channel.ChannelEvent;
+import org.jboss.netty.channel.ChannelFuture;
+import org.jboss.netty.channel.ChannelFutureListener;
+import org.jboss.netty.channel.ChannelHandlerContext;
+import org.jboss.netty.channel.Channels;
+import org.jboss.netty.channel.MessageEvent;
+import org.jboss.netty.handler.codec.http.HttpChunk;
+import org.jboss.netty.handler.codec.http.HttpChunkTrailer;
+import org.jboss.netty.handler.codec.http.HttpHeaders;
+import org.jboss.netty.handler.codec.http.HttpMessage;
+import org.jboss.netty.handler.codec.http.HttpRequest;
+import org.jboss.netty.handler.codec.http.HttpResponse;
+
+/**
+ * Encodes {@link HttpRequest}s, {@link HttpResponse}s, and {@link HttpChunk}s
+ * into {@link SpdySynStreamFrame}s and {@link SpdySynReplyFrame}s.
+ *
+ * <h3>Request Annotations</h3>
+ *
+ * SPDY specific headers must be added to {@link HttpRequest}s:
+ * <table border=1>
+ * <tr>
+ * <th>Header Name</th><th>Header Value</th>
+ * </tr>
+ * <tr>
+ * <td>{@code "X-SPDY-Stream-ID"}</td>
+ * <td>The Stream-ID for this request.
+ * Stream-IDs must be odd, positive integers, and must increase monotonically.</td>
+ * </tr>
+ * <tr>
+ * <td>{@code "X-SPDY-Priority"}</td>
+ * <td>The priority value for this request.
+ * The priority should be between 0 and 3 inclusive.
+ * 0 represents the highest priority and 3 represents the lowest.
+ * This header is optional and defaults to 0.</td>
+ * </tr>
+ * </table>
+ *
+ * <h3>Response Annotations</h3>
+ *
+ * SPDY specific headers must be added to {@link HttpResponse}s:
+ * <table border=1>
+ * <tr>
+ * <th>Header Name</th><th>Header Value</th>
+ * </tr>
+ * <tr>
+ * <td>{@code "X-SPDY-Stream-ID"}</td>
+ * <td>The Stream-ID of the request corresponding to this response.</td>
+ * </tr>
+ * </table>
+ *
+ * <h3>Pushed Resource Annotations</h3>
+ *
+ * SPDY specific headers must be added to pushed {@link HttpResponse}s:
+ * <table border=1>
+ * <tr>
+ * <th>Header Name</th><th>Header Value</th>
+ * </tr>
+ * <tr>
+ * <td>{@code "X-SPDY-Stream-ID"}</td>
+ * <td>The Stream-ID for this resource.
+ * Stream-IDs must be even, positive integers, and must increase monotonically.</td>
+ * </tr>
+ * <tr>
+ * <td>{@code "X-SPDY-Associated-To-Stream-ID"}</td>
+ * <td>The Stream-ID of the request that inititated this pushed resource.</td>
+ * </tr>
+ * <tr>
+ * <td>{@code "X-SPDY-Priority"}</td>
+ * <td>The priority value for this resource.
+ * The priority should be between 0 and 3 inclusive.
+ * 0 represents the highest priority and 3 represents the lowest.
+ * This header is optional and defaults to 0.</td>
+ * </tr>
+ * <tr>
+ * <td>{@code "X-SPDY-URL"}</td>
+ * <td>The full URL for the resource being pushed.</td>
+ * </tr>
+ * </table>
+ *
+ * <h3>Chunked Content</h3>
+ *
+ * This encoder associates all {@link HttpChunk}s that it receives
+ * with the most recently received 'chunked' {@link HttpRequest}
+ * or {@link HttpResponse}.
+ *
+ * <h3>Pushed Resources</h3>
+ *
+ * All pushed resources should be sent before sending the response
+ * that corresponds to the initial request.
+ */
+public class SpdyHttpEncoder implements ChannelDownstreamHandler {
+
+    private volatile int currentStreamID;
+
+    public SpdyHttpEncoder() {
+    }
+
+    public void handleDownstream(ChannelHandlerContext ctx, ChannelEvent evt)
+            throws Exception {
+        if (!(evt instanceof MessageEvent)) {
+            ctx.sendDownstream(evt);
+            return;
+        }
+
+        MessageEvent e = (MessageEvent) evt;
+        Object msg = e.getMessage();
+
+        if (msg instanceof HttpRequest) {
+
+            HttpRequest httpRequest = (HttpRequest) msg;
+            SpdySynStreamFrame spdySynStreamFrame = createSynStreamFrame(httpRequest);
+            int streamID = spdySynStreamFrame.getStreamID();
+            ChannelFuture future = getContentFuture(ctx, e, streamID, httpRequest);
+            Channels.write(ctx, future, spdySynStreamFrame, e.getRemoteAddress());
+
+        } else if (msg instanceof HttpResponse) {
+
+            HttpResponse httpResponse = (HttpResponse) msg;
+            if (httpResponse.containsHeader(SpdyHttpHeaders.Names.ASSOCIATED_TO_STREAM_ID)) {
+                SpdySynStreamFrame spdySynStreamFrame = createSynStreamFrame(httpResponse);
+                int streamID = spdySynStreamFrame.getStreamID();
+                ChannelFuture future = getContentFuture(ctx, e, streamID, httpResponse);
+                Channels.write(ctx, future, spdySynStreamFrame, e.getRemoteAddress());
+            } else {
+                SpdySynReplyFrame spdySynReplyFrame = createSynReplyFrame(httpResponse);
+                int streamID = spdySynReplyFrame.getStreamID();
+                ChannelFuture future = getContentFuture(ctx, e, streamID, httpResponse);
+                Channels.write(ctx, future, spdySynReplyFrame, e.getRemoteAddress());
+            }
+
+        } else if (msg instanceof HttpChunk) {
+
+            HttpChunk chunk = (HttpChunk) msg;
+            SpdyDataFrame spdyDataFrame = new DefaultSpdyDataFrame(currentStreamID);
+            spdyDataFrame.setData(chunk.getContent());
+            spdyDataFrame.setLast(chunk.isLast());
+
+            if (chunk instanceof HttpChunkTrailer) {
+                HttpChunkTrailer trailer = (HttpChunkTrailer) chunk;
+                List<Map.Entry<String, String>> trailers = trailer.getHeaders();
+                if (trailers.isEmpty()) {
+                    Channels.write(ctx, e.getFuture(), spdyDataFrame, e.getRemoteAddress());
+                } else {
+                    // Create SPDY HEADERS frame out of trailers
+                    SpdyHeadersFrame spdyHeadersFrame = new DefaultSpdyHeadersFrame(currentStreamID);
+                    for (Map.Entry<String, String> entry: trailers) {
+                        spdyHeadersFrame.addHeader(entry.getKey(), entry.getValue());
+                    }
+
+                    // Write HEADERS frame and append Data Frame
+                    ChannelFuture future = Channels.future(e.getChannel());
+                    future.addListener(new SpdyFrameWriter(ctx, e, spdyDataFrame));
+                    Channels.write(ctx, future, spdyHeadersFrame, e.getRemoteAddress());
+                }
+            } else {
+                Channels.write(ctx, e.getFuture(), spdyDataFrame, e.getRemoteAddress());
+            }
+        } else {
+            // Unknown message type
+            ctx.sendDownstream(evt);
+        }
+    }
+
+    private ChannelFuture getContentFuture(
+            ChannelHandlerContext ctx, MessageEvent e, int streamID, HttpMessage httpMessage) {
+        if (httpMessage.getContent().readableBytes() == 0) {
+            return e.getFuture();
+        }
+
+        // Create SPDY Data Frame out of message content
+        SpdyDataFrame spdyDataFrame = new DefaultSpdyDataFrame(streamID);
+        spdyDataFrame.setData(httpMessage.getContent());
+        spdyDataFrame.setLast(true);
+
+        // Create new future and add listener
+        ChannelFuture future = Channels.future(e.getChannel());
+        future.addListener(new SpdyFrameWriter(ctx, e, spdyDataFrame));
+
+        return future;
+    }
+
+    private class SpdyFrameWriter implements ChannelFutureListener {
+
+        private final ChannelHandlerContext ctx;
+        private final MessageEvent e;
+        private final Object spdyFrame;
+
+        SpdyFrameWriter(ChannelHandlerContext ctx, MessageEvent e, Object spdyFrame) {
+            this.ctx = ctx;
+            this.e = e;
+            this.spdyFrame = spdyFrame;
+        }
+
+        public void operationComplete(ChannelFuture future) throws Exception {
+            if (future.isSuccess()) {
+                Channels.write(ctx, e.getFuture(), spdyFrame, e.getRemoteAddress());
+            } else if (future.isCancelled()) {
+                e.getFuture().cancel();
+            } else {
+                e.getFuture().setFailure(future.getCause());
+            }
+        }
+    }
+
+    private SpdySynStreamFrame createSynStreamFrame(HttpMessage httpMessage)
+            throws Exception {
+        boolean chunked = httpMessage.isChunked();
+
+        // Get the Stream-ID, Associated-To-Stream-ID, Priority, and URL from the headers
+        int streamID = SpdyHttpHeaders.getStreamID(httpMessage);
+        int associatedToStreamID = SpdyHttpHeaders.getAssociatedToStreamID(httpMessage);
+        byte priority = SpdyHttpHeaders.getPriority(httpMessage);
+        String URL = SpdyHttpHeaders.getUrl(httpMessage);
+        SpdyHttpHeaders.removeStreamID(httpMessage);
+        SpdyHttpHeaders.removeAssociatedToStreamID(httpMessage);
+        SpdyHttpHeaders.removePriority(httpMessage);
+        SpdyHttpHeaders.removeUrl(httpMessage);
+
+        // The Connection, Keep-Alive, Proxy-Connection, and Transfer-Encoding
+        // headers are not valid and MUST not be sent.
+        httpMessage.removeHeader(HttpHeaders.Names.CONNECTION);
+        httpMessage.removeHeader("Keep-Alive");
+        httpMessage.removeHeader("Proxy-Connection");
+        httpMessage.removeHeader(HttpHeaders.Names.TRANSFER_ENCODING);
+
+        SpdySynStreamFrame spdySynStreamFrame = new DefaultSpdySynStreamFrame(streamID, associatedToStreamID, priority);
+        for (Map.Entry<String, String> entry: httpMessage.getHeaders()) {
+            spdySynStreamFrame.addHeader(entry.getKey(), entry.getValue());
+        }
+
+        // Unfold the first line of the message into name/value pairs
+        SpdyHeaders.setVersion(spdySynStreamFrame, httpMessage.getProtocolVersion());
+        if (httpMessage instanceof HttpRequest) {
+            HttpRequest httpRequest = (HttpRequest) httpMessage;
+            SpdyHeaders.setMethod(spdySynStreamFrame, httpRequest.getMethod());
+            SpdyHeaders.setUrl(spdySynStreamFrame, httpRequest.getUri());
+        }
+        if (httpMessage instanceof HttpResponse) {
+            HttpResponse httpResponse = (HttpResponse) httpMessage;
+            SpdyHeaders.setStatus(spdySynStreamFrame, httpResponse.getStatus());
+            SpdyHeaders.setUrl(spdySynStreamFrame, URL);
+            spdySynStreamFrame.setUnidirectional(true);
+        }
+
+        if (chunked) {
+            currentStreamID = streamID;
+            spdySynStreamFrame.setLast(false);
+        } else {
+            spdySynStreamFrame.setLast(httpMessage.getContent().readableBytes() == 0);
+        }
+
+        return spdySynStreamFrame;
+    }
+
+    private SpdySynReplyFrame createSynReplyFrame(HttpResponse httpResponse)
+            throws Exception {
+        boolean chunked = httpResponse.isChunked();
+
+        // Get the Stream-ID from the headers
+        int streamID = SpdyHttpHeaders.getStreamID(httpResponse);
+        SpdyHttpHeaders.removeStreamID(httpResponse);
+
+        // The Connection, Keep-Alive, Proxy-Connection, and Transfer-ENcoding
+        // headers are not valid and MUST not be sent.
+        httpResponse.removeHeader(HttpHeaders.Names.CONNECTION);
+        httpResponse.removeHeader("Keep-Alive");
+        httpResponse.removeHeader("Proxy-Connection");
+        httpResponse.removeHeader(HttpHeaders.Names.TRANSFER_ENCODING);
+
+        SpdySynReplyFrame spdySynReplyFrame = new DefaultSpdySynReplyFrame(streamID);
+        for (Map.Entry<String, String> entry: httpResponse.getHeaders()) {
+            spdySynReplyFrame.addHeader(entry.getKey(), entry.getValue());
+        }
+
+        // Unfold the first line of the repsonse into name/value pairs
+        SpdyHeaders.setStatus(spdySynReplyFrame, httpResponse.getStatus());
+        SpdyHeaders.setVersion(spdySynReplyFrame, httpResponse.getProtocolVersion());
+
+        if (chunked) {
+            currentStreamID = streamID;
+            spdySynReplyFrame.setLast(false);
+        } else {
+            spdySynReplyFrame.setLast(httpResponse.getContent().readableBytes() == 0);
+        }
+
+        return spdySynReplyFrame;
+    }
+}

src/main/java/org/jboss/netty/handler/codec/spdy/SpdyHttpHeaders.java

@@ -0,0 +1,162 @@
+/*
+ * Copyright 2012 The Netty Project
+ *
+ * The Netty Project licenses this file to you under the Apache License,
+ * version 2.0 (the "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at:
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS, WITHOUT
+ * WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the
+ * License for the specific language governing permissions and limitations
+ * under the License.
+ */
+/*
+ * Copyright 2012 Twitter, Inc.
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License"); you may
+ * not use this file except in compliance with the License. You may obtain
+ * a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.jboss.netty.handler.codec.spdy;
+
+import org.jboss.netty.handler.codec.http.HttpHeaders;
+import org.jboss.netty.handler.codec.http.HttpMessage;
+
+/**
+ * Provides the constants for the header names and the utility methods
+ * used by the {@link SpdyHttpDecoder} and {@link SpdyHttpEncoder}.
+ * @apiviz.sterotype static
+ */
+public final class SpdyHttpHeaders {
+
+    /**
+     * SPDY HTTP header names
+     * @apiviz.sterotype static
+     */
+    public static final class Names {
+        /**
+         * {@code "X-SPDY-Stream-ID"}
+         */
+        public static final String STREAM_ID = "X-SPDY-Stream-ID";
+        /**
+         * {@code "X-SPDY-Associated-To-Stream-ID"}
+         */
+        public static final String ASSOCIATED_TO_STREAM_ID = "X-SPDY-Associated-To-Stream-ID";
+        /**
+         * {@code "X-SPDY-Priority"}
+         */
+        public static final String PRIORITY = "X-SPDY-Priority";
+        /**
+         * {@code "X-SPDY-URL"}
+         */
+        public static final String URL = "X-SPDY-URL";
+
+        private Names() {
+            super();
+        }
+    }
+
+    private SpdyHttpHeaders() {
+    }
+
+    /**
+     * Removes the {@code "X-SPDY-Stream-ID"} header.
+     */
+    public static void removeStreamID(HttpMessage message) {
+        message.removeHeader(Names.STREAM_ID);
+    }
+
+    /**
+     * Returns the value of the {@code "X-SPDY-Stream-ID"} header.
+     */
+    public static int getStreamID(HttpMessage message) {
+        return HttpHeaders.getIntHeader(message, Names.STREAM_ID);
+    }
+
+    /**
+     * Sets the {@code "X-SPDY-Stream-ID"} header.
+     */
+    public static void setStreamID(HttpMessage message, int streamID) {
+        HttpHeaders.setIntHeader(message, Names.STREAM_ID, streamID);
+    }
+
+    /**
+     * Removes the {@code "X-SPDY-Associated-To-Stream-ID"} header.
+     */
+    public static void removeAssociatedToStreamID(HttpMessage message) {
+        message.removeHeader(Names.ASSOCIATED_TO_STREAM_ID);
+    }
+
+    /**
+     * Returns the value of the {@code "X-SPDY-Associated-To-Stream-ID"} header.
+     *
+     * @return the header value or {@code 0} if there is no such header or
+     *         if the header value is not a number
+     */
+    public static int getAssociatedToStreamID(HttpMessage message) {
+        return HttpHeaders.getIntHeader(message, Names.ASSOCIATED_TO_STREAM_ID, 0);
+    }
+
+    /**
+     * Sets the {@code "X-SPDY-Associated-To-Stream-ID"} header.
+     */
+    public static void setAssociatedToStreamID(HttpMessage message, int associatedToStreamID) {
+        HttpHeaders.setIntHeader(message, Names.ASSOCIATED_TO_STREAM_ID, associatedToStreamID);
+    }
+
+    /**
+     * Removes the {@code "X-SPDY-Priority"} header.
+     */
+    public static void removePriority(HttpMessage message) {
+        message.removeHeader(Names.PRIORITY);
+    }
+
+    /**
+     * Returns the value of the {@code "X-SPDY-Priority"} header.
+     *
+     * @return the header value or {@code 0} if there is no such header or
+     *         if the header value is not a number
+     */
+    public static byte getPriority(HttpMessage message) {
+        return (byte) HttpHeaders.getIntHeader(message, Names.PRIORITY, 0);
+    }
+
+    /**
+     * Sets the {@code "X-SPDY-Priority"} header.
+     */
+    public static void setPriority(HttpMessage message, byte priority) {
+        HttpHeaders.setIntHeader(message, Names.PRIORITY, priority);
+    }
+
+    /**
+     * Removes the {@code "X-SPDY-URL"} header.
+     */
+    public static void removeUrl(HttpMessage message) {
+        message.removeHeader(Names.URL);
+    }
+
+    /**
+     * Returns the value of the {@code "X-SPDY-URL"} header.
+     */
+    public static String getUrl(HttpMessage message) {
+        return message.getHeader(Names.URL);
+    }
+
+    /**
+     * Sets the {@code "X-SPDY-URL"} header.
+     */
+    public static void setUrl(HttpMessage message, String url) {
+        message.setHeader(Names.URL, url);
+    }
+}

src/main/java/org/jboss/netty/handler/codec/spdy/SpdySessionHandler.java

@@ -393,12 +393,8 @@ public class SpdySessionHandler extends SimpleChannelUpstreamHandler
      * Helper functions
      */
 
-    private boolean isServerID(int ID) {
-        return ID % 2 == 0;
-    }
-
     private boolean isRemoteInitiatedID(int ID) {
-        boolean serverID = isServerID(ID);
+        boolean serverID = SpdyCodecUtil.isServerID(ID);
         return (server && !serverID) || (!server && serverID);
     }