Start to process Headers frame

git-svn-id: https://svn.apache.org/repos/asf/tomcat/trunk@1679270 13f79535-47bb-0310-9956-ffa450edef68

java/org/apache/coyote/http2/AbstractStream.java

@@ -21,15 +21,13 @@
 import java.util.Set;
 
 import org.apache.juli.logging.Log;
-import org.apache.juli.logging.LogFactory;
 import org.apache.tomcat.util.res.StringManager;
 
 /**
  * Used to managed prioritisation.
  */
 abstract class AbstractStream {
 
-    private static final Log log = LogFactory.getLog(AbstractStream.class);
     private static final StringManager sm = StringManager.getManager(AbstractStream.class);
 
     private final Integer identifier;
@@ -49,8 +47,8 @@ public AbstractStream(Integer identifier) {
 
 
     public void rePrioritise(AbstractStream parent, boolean exclusive, int weight) {
-        if (log.isDebugEnabled()) {
-            log.debug(sm.getString("abstractStream.reprioritisation.debug", identifier,
+        if (getLog().isDebugEnabled()) {
+            getLog().debug(sm.getString("abstractStream.reprioritisation.debug", identifier,
                     Boolean.toString(exclusive), parent.getIdentifier(), Integer.toString(weight)));
         }
 
@@ -120,4 +118,6 @@ void setParentStream(AbstractStream parentStream) {
     Set<AbstractStream> getChildStreams() {
         return childStreams;
     }
+
+    protected abstract Log getLog();
 }

java/org/apache/coyote/http2/ConnectionSettings.java

@@ -27,20 +27,20 @@ public class ConnectionSettings {
     private final Log log = LogFactory.getLog(ConnectionSettings.class);
     private final StringManager sm = StringManager.getManager(ConnectionSettings.class);
 
-    public static final long DEFAULT_WINDOW_SIZE = (1 << 16) - 1;
-    private static final long UNLIMITED = 1 << 32; // Use the maximum possible
-    private static final long MAX_WINDOW_SIZE = (1 << 31) - 1;
-    private static final long MIN_MAX_FRAME_SIZE = 1 << 14;
-    private static final long MAX_MAX_FRAME_SIZE = (1 << 24) - 1;
-
-    private volatile long headerTableSize = 4096;
-    private volatile long enablePush = 1;
-    private volatile long maxConcurrentStreams = UNLIMITED;
-    private volatile long initialWindowSize = DEFAULT_WINDOW_SIZE;
-    private volatile long maxFrameSize = MIN_MAX_FRAME_SIZE;
-    private volatile long maxHeaderListSize = UNLIMITED;
-
-    public void set(int parameterId, long value) throws IOException {
+    public static final int DEFAULT_WINDOW_SIZE = (1 << 16) - 1;
+    private static final int UNLIMITED = 1 << 32; // Use the maximum possible
+    private static final int MAX_WINDOW_SIZE = (1 << 31) - 1;
+    private static final int MIN_MAX_FRAME_SIZE = 1 << 14;
+    private static final int MAX_MAX_FRAME_SIZE = (1 << 24) - 1;
+
+    private volatile int headerTableSize = 4096;
+    private volatile int enablePush = 1;
+    private volatile int maxConcurrentStreams = UNLIMITED;
+    private volatile int initialWindowSize = DEFAULT_WINDOW_SIZE;
+    private volatile int maxFrameSize = MIN_MAX_FRAME_SIZE;
+    private volatile int maxHeaderListSize = UNLIMITED;
+
+    public void set(int parameterId, int value) throws IOException {
         if (log.isDebugEnabled()) {
             log.debug(sm.getString("connectionSettings.debug",
                     Integer.toString(parameterId), Long.toString(value)));
@@ -73,18 +73,18 @@ public void set(int parameterId, long value) throws IOException {
     }
 
 
-    public long getHeaderTableSize() {
+    public int getHeaderTableSize() {
         return headerTableSize;
     }
-    public void setHeaderTableSize(long headerTableSize) {
+    public void setHeaderTableSize(int headerTableSize) {
         this.headerTableSize = headerTableSize;
     }
 
 
-    public long getEnablePush() {
+    public int getEnablePush() {
         return enablePush;
     }
-    public void setEnablePush(long enablePush) throws IOException {
+    public void setEnablePush(int enablePush) throws IOException {
         // Can't be less than zero since the result of the byte->long conversion
         // will never be negative
         if (enablePush > 1) {
@@ -95,18 +95,18 @@ public void setEnablePush(long enablePush) throws IOException {
     }
 
 
-    public long getMaxConcurrentStreams() {
+    public int getMaxConcurrentStreams() {
         return maxConcurrentStreams;
     }
-    public void setMaxConcurrentStreams(long maxConcurrentStreams) {
+    public void setMaxConcurrentStreams(int maxConcurrentStreams) {
         this.maxConcurrentStreams = maxConcurrentStreams;
     }
 
 
-    public long getInitialWindowSize() {
+    public int getInitialWindowSize() {
         return initialWindowSize;
     }
-    public void setInitialWindowSize(long initialWindowSize) throws IOException {
+    public void setInitialWindowSize(int initialWindowSize) throws IOException {
         if (initialWindowSize > MAX_WINDOW_SIZE) {
             throw new Http2Exception(sm.getString("connectionSettings.windowSizeTooBig",
                     Long.toString(initialWindowSize), Long.toString(MAX_WINDOW_SIZE)),
@@ -116,10 +116,10 @@ public void setInitialWindowSize(long initialWindowSize) throws IOException {
     }
 
 
-    public long getMaxFrameSize() {
+    public int getMaxFrameSize() {
         return maxFrameSize;
     }
-    public void setMaxFrameSize(long maxFrameSize) throws IOException {
+    public void setMaxFrameSize(int maxFrameSize) throws IOException {
         if (maxFrameSize < MIN_MAX_FRAME_SIZE || maxFrameSize > MAX_MAX_FRAME_SIZE) {
             throw new Http2Exception(sm.getString("connectionSettings.maxFrameSizeInvalid",
                     Long.toString(maxFrameSize), Long.toString(MIN_MAX_FRAME_SIZE),
@@ -129,10 +129,10 @@ public void setMaxFrameSize(long maxFrameSize) throws IOException {
     }
 
 
-    public long getMaxHeaderListSize() {
+    public int getMaxHeaderListSize() {
         return maxHeaderListSize;
     }
-    public void setMaxHeaderListSize(long maxHeaderListSize) {
+    public void setMaxHeaderListSize(int maxHeaderListSize) {
         this.maxHeaderListSize = maxHeaderListSize;
     }
 }

java/org/apache/coyote/http2/Http2UpgradeHandler.java

@@ -18,6 +18,7 @@
 
 import java.io.EOFException;
 import java.io.IOException;
+import java.nio.ByteBuffer;
 import java.nio.charset.StandardCharsets;
 import java.util.HashMap;
 import java.util.Map;
@@ -52,6 +53,7 @@ public class Http2UpgradeHandler extends AbstractStream implements InternalHttpU
 
     private static final Integer STREAM_ID_ZERO = Integer.valueOf(0);
 
+    private static final int FRAME_TYPE_HEADERS = 1;
     private static final int FRAME_TYPE_PRIORITY = 2;
     private static final int FRAME_TYPE_SETTINGS = 4;
     private static final int FRAME_TYPE_WINDOW_UPDATE = 8;
@@ -67,11 +69,15 @@ public class Http2UpgradeHandler extends AbstractStream implements InternalHttpU
     private volatile boolean firstFrame = true;
 
     private final ConnectionSettings remoteSettings = new ConnectionSettings();
+    private final ConnectionSettings localSettings = new ConnectionSettings();
     private volatile long flowControlWindowSize = ConnectionSettings.DEFAULT_WINDOW_SIZE;
     private volatile int maxRemoteStreamId = 0;
 
-    private final Map<Integer,Stream> streams = new HashMap<>();
+    private HpackDecoder hpackDecoder;
+    private ByteBuffer headerReadBuffer = ByteBuffer.allocate(1024);
+    private HpackEncoder hpackEncoder;
 
+    private final Map<Integer,Stream> streams = new HashMap<>();
 
     public Http2UpgradeHandler() {
         super (STREAM_ID_ZERO);
@@ -188,6 +194,9 @@ private boolean processFrame() throws IOException {
         int payloadSize = getPayloadSize(streamId, frameHeader);
 
         switch (frameType) {
+        case FRAME_TYPE_HEADERS:
+            processFrameHeaders(flags, streamId, payloadSize);
+            break;
         case FRAME_TYPE_PRIORITY:
             processFramePriority(flags, streamId, payloadSize);
             break;
@@ -205,6 +214,85 @@ private boolean processFrame() throws IOException {
     }
 
 
+    private void processFrameHeaders(int flags, int streamId, int payloadSize) throws IOException {
+        if (log.isDebugEnabled()) {
+            log.debug(sm.getString("upgradeHandler.processFrame",
+                    Integer.toString(FRAME_TYPE_HEADERS), Integer.toString(flags),
+                    Integer.toString(streamId), Integer.toString(payloadSize)));
+        }
+
+        // Validate the stream
+        if (streamId == 0) {
+            throw new Http2Exception(sm.getString("upgradeHandler.processFrameHeaders.invalidStream"),
+                    0, Http2Exception.PROTOCOL_ERROR);
+        }
+
+        // Process the stream
+        // TODO Handle end of headers flag
+        // TODO Handle end of stream flag
+        // TODO Handle continutation frames
+        Stream stream = getStream(streamId);
+        int padLength = 0;
+
+        boolean padding = (flags & 0x08) > 0;
+        boolean priority = (flags & 0x20) > 0;
+        int optionalLen = 0;
+        if (padding) {
+            optionalLen = 1;
+        }
+        if (priority) {
+            optionalLen += 5;
+        }
+        if (optionalLen > 0) {
+            byte[] optional = new byte[optionalLen];
+            readFully(optional);
+            int optionalPos = 0;
+            if (padding) {
+                padLength = ByteUtil.getOneByte(optional, optionalPos++);
+            }
+            if (priority) {
+                boolean exclusive = ByteUtil.isBit7Set(optional[optionalPos]);
+                int parentStreamId = ByteUtil.get31Bits(optional, optionalPos);
+                int weight = ByteUtil.getOneByte(optional, optionalPos + 4) + 1;
+                AbstractStream parentStream = getStream(parentStreamId);
+                if (parentStream == null) {
+                    parentStream = this;
+                }
+                stream.rePrioritise(parentStream, exclusive, weight);
+            }
+
+            payloadSize -= optionalLen;
+        }
+
+        hpackDecoder.setHeaderEmitter(stream);
+        while (payloadSize > 0) {
+            int toRead = Math.min(headerReadBuffer.remaining(), payloadSize);
+            // headerReadBuffer in write mode
+            readFully(headerReadBuffer, toRead);
+            // switch to read mode
+            headerReadBuffer.flip();
+            try {
+                hpackDecoder.decode(headerReadBuffer);
+            } catch (HpackException hpe) {
+                // TODO i18n
+                throw new Http2Exception("", 0, Http2Exception.PROTOCOL_ERROR);
+            }
+            // switches to write mode
+            headerReadBuffer.compact();
+            payloadSize -= toRead;
+        }
+        // Should be empty at this point
+        if (headerReadBuffer.position() > 0) {
+            // TODO i18n
+            throw new Http2Exception("", 0, Http2Exception.PROTOCOL_ERROR);
+        }
+
+        if (padLength > 0) {
+            swallowPayload(padLength);
+        }
+    }
+
+
     private void processFramePriority(int flags, int streamId, int payloadSize) throws IOException {
         if (log.isDebugEnabled()) {
             log.debug(sm.getString("upgradeHandler.processFrame",
@@ -280,10 +368,15 @@ private void processFrameSettings(int flags, int streamId, int payloadSize) thro
             for (int i = 0; i < payloadSize / 6; i++) {
                 readFully(setting);
                 int id = ByteUtil.getTwoBytes(setting, 0);
-                long value = ByteUtil.getFourBytes(setting, 2);
+                int value = ByteUtil.getFourBytes(setting, 2);
                 remoteSettings.set(id, value);
             }
         }
+        if (firstFrame) {
+            firstFrame = false;
+            hpackDecoder = new HpackDecoder(remoteSettings.getHeaderTableSize());
+            hpackEncoder = new HpackEncoder(localSettings.getHeaderTableSize());
+        }
 
         // Acknowledge the settings
         // TODO Need to coordinate writes with other threads
@@ -344,7 +437,7 @@ private void processFrameUnknown(int streamId, int type, int payloadSize) throws
 
     private void swallowPayload(int payloadSize) throws IOException {
         int read = 0;
-        byte[] buffer = new byte[8 * 1024];
+        byte[] buffer = new byte[1024];
         while (read < payloadSize) {
             int toRead = Math.min(buffer.length, payloadSize - read);
             int thisTime = socketWrapper.read(true, buffer, 0, toRead);
@@ -386,8 +479,6 @@ private int getFrameType(byte[] frameHeader) throws IOException {
             if (frameType != FRAME_TYPE_SETTINGS) {
                 throw new Http2Exception(sm.getString("upgradeHandler.receivePrefaceNotSettings"),
                         0, Http2Exception.PROTOCOL_ERROR);
-            } else {
-                firstFrame = false;
             }
         }
         return frameType;
@@ -463,6 +554,19 @@ private void readFully(byte[] dest) throws IOException {
     }
 
 
+    private void readFully(ByteBuffer dest, int len) throws IOException {
+        int read = 0;
+        while (read < len) {
+            int thisTime = socketWrapper.read(true, dest.array(), dest.arrayOffset(), len -read);
+            if (thisTime == -1) {
+                throw new EOFException(sm.getString("upgradeHandler.unexpectedEos"));
+            }
+            read += thisTime;
+        }
+        dest.position(dest.position() + read);
+    }
+
+
     private Stream getStream(int streamId) {
         Integer key = Integer.valueOf(streamId);
 
@@ -485,4 +589,10 @@ private void close() {
             log.debug(sm.getString("upgradeHandler.socketCloseFailed"), ioe);
         }
     }
+
+
+    @Override
+    protected final Log getLog() {
+        return log;
+    }
 }

java/org/apache/coyote/http2/LocalStrings.properties

@@ -30,12 +30,14 @@ hpackdecoder.zeroNotValidHeaderTableIndex=Zero is not a valid header table index
 
 hpackhuffman.huffmanEncodedHpackValueDidNotEndWithEOS=Huffman encoded value in HPACK headers did not end with EOS padding
 
+stream.header.debug=Stream [{0}] recieved HTTP header [{1}] with value [{2}]
 upgradeHandler.connectionError=An error occurred that requires the HTTP/2 connection to be closed.
 upgradeHandler.payloadTooBig=The payload is [{0}] bytes long but the maximum frame size is [{1}]
 upgradeHandler.processFrame=Processing frame of type [{0}] for stream [{2}] with flags [{1}] and payload size [{3}]
 upgradeHandler.processFrame.ioerror=An I/O error occurred while reading an incoming HTTP/2 frame
-upgradeHandler.processFramePriority.invalidPayloadSize=Settings frame received with an invalid payload size of [{0}] (should be 5)
-upgradeHandler.processFramePriority.invalidStream=Settings frame received for stream [0]
+upgradeHandler.processFrameHeaders.invalidStream=Headers frame received for stream [0]
+upgradeHandler.processFramePriority.invalidPayloadSize=Priority frame received with an invalid payload size of [{0}] (should be 5)
+upgradeHandler.processFramePriority.invalidStream=Priority frame received for stream [0]
 upgradeHandler.processFrameSettings.ackWithNonZeroPayload=Settings frame received with the ACK flag set and payload present
 upgradeHandler.processFrameSettings.invalidPayloadSize=Settings frame received with a payload size of [{0}] which is not a multiple of 6
 upgradeHandler.processFrameSettings.invalidStream=Settings frame received for stream [{0}]

java/org/apache/coyote/http2/Stream.java

@@ -16,7 +16,15 @@
  */
 package org.apache.coyote.http2;
 
-public class Stream extends AbstractStream {
+import org.apache.coyote.http2.HpackDecoder.HeaderEmitter;
+import org.apache.juli.logging.Log;
+import org.apache.juli.logging.LogFactory;
+import org.apache.tomcat.util.res.StringManager;
+
+public class Stream extends AbstractStream implements HeaderEmitter {
+
+    private static final Log log = LogFactory.getLog(Stream.class);
+    private static final StringManager sm = StringManager.getManager(Stream.class);
 
     private volatile long flowControlWindowSize;
 
@@ -31,4 +39,20 @@ public Stream(Integer identifier, Http2UpgradeHandler handler) {
     public void incrementWindowSize(int windowSizeIncrement) {
         flowControlWindowSize += windowSizeIncrement;
     }
+
+
+    @Override
+    public void emitHeader(String name, String value, boolean neverIndex) {
+        if (log.isDebugEnabled()) {
+            log.debug(sm.getString("stream.header.debug", getIdentifier(), name, value));
+        }
+
+        // TODO: Do something with these headers
+    }
+
+
+    @Override
+    protected final Log getLog() {
+        return log;
+    }
 }