Move parsing logic to onIngress from onIngressEOF and add chunking support on boundaries

Summary:
Context: https://docs.google.com/document/d/1rtw4JJJojO9NpUKlIUO0_oCRjWftusTXbajEJ0Zf8mo/edit

Before, our assumption was that the entire binary message would be given to our codec and then `onIngressEOF` would be called at the end when the result was needed. As a result, we did all of our parsing logic in `onIngressEOF`.

However, now requests will start streaming in on a chunkwise basis and `onIngress` will be called multiple times, each with a chunk of the message.

In this diff, I move the parsing logic from `onIngressEOF` into `onIngress` to support this new workflow. I additionally started adding support for HTTP binary chunks. This v1 implementation for binary chunks only supports message breaks on boundaries (ex. breaking after a section like the framing indicator, but not within a section).

Reviewed By: lnicco

Differential Revision: D64381365

fbshipit-source-id: 3ba303b89245a6c06e03a08d794bed3e755ab77d

Author: Shiv Kushwah

proxygen/lib/http/codec/HTTPBinaryCodec.cpp

@@ -253,8 +253,7 @@ ParseResult HTTPBinaryCodec::parseHeaders(folly::io::Cursor& cursor,
 }
 
 ParseResult HTTPBinaryCodec::parseContent(folly::io::Cursor& cursor,
-                                          size_t remaining,
-                                          HTTPMessage& msg) {
+                                          size_t remaining) {
   size_t parsed = 0;
 
   // Parse the contentLength and advance cursor
@@ -288,20 +287,14 @@ ParseResult HTTPBinaryCodec::parseTrailers(folly::io::Cursor& cursor,
 }
 
 size_t HTTPBinaryCodec::onIngress(const folly::IOBuf& buf) {
-  auto len = bufferedIngress_.chainLength();
   bufferedIngress_.append(buf.clone());
-  return bufferedIngress_.chainLength() - len;
-}
+  const auto len = bufferedIngress_.chainLength();
 
-void HTTPBinaryCodec::onIngressEOF() {
   size_t parsedTot = 0;
   folly::io::Cursor cursor(bufferedIngress_.front());
   auto bufLen = bufferedIngress_.chainLength();
-  if (!bufLen) {
-    parseError_ = "Empty buffer provided!";
-  }
 
-  while (!parseError_ && parsedTot < bufLen && !parserPaused_) {
+  while (!parseError_ && !parserPaused_ && parsedTot < bufLen) {
     size_t parsed = 0;
     ParseResult parseResult(ParseResultState::INITIALIZED);
     switch (state_) {
@@ -374,17 +367,20 @@ void HTTPBinaryCodec::onIngressEOF() {
         parsed += parseResult.bytesParsed_;
         state_ = ParseState::CONTENT;
         msg_ = std::move(decodeInfo_.msg);
+        callback_->onHeadersComplete(ingressTxnID_, std::move(msg_));
         break;
 
       case ParseState::CONTENT:
-        CHECK(msg_);
-        parseResult = parseContent(cursor, bufLen - parsedTot, *msg_);
+        parseResult = parseContent(cursor, bufLen - parsedTot);
         if (parseResult.parseResultState_ == ParseResultState::ERROR) {
           parseError_ = parseResult.error_;
           break;
         }
         parsed += parseResult.bytesParsed_;
         state_ = ParseState::TRAILERS_SECTION;
+        if (msgBody_) {
+          callback_->onBody(ingressTxnID_, std::move(msgBody_), 0);
+        }
         break;
 
       case ParseState::TRAILERS_SECTION:
@@ -402,6 +398,9 @@ void HTTPBinaryCodec::onIngressEOF() {
             std::make_unique<HTTPHeaders>(decodeInfo_.msg->getHeaders());
         parsed += parseResult.bytesParsed_;
         state_ = ParseState::PADDING;
+        if (trailers_) {
+          callback_->onTrailersComplete(ingressTxnID_, std::move(trailers_));
+        }
         break;
 
       case ParseState::PADDING:
@@ -422,30 +421,33 @@ void HTTPBinaryCodec::onIngressEOF() {
         ingressTxnID_,
         HTTPException(HTTPException::Direction::INGRESS,
                       fmt::format("Invalid Message: {}", *parseError_)));
-  } else {
+  }
 
-    if (!msg_) {
-      if (state_ == ParseState::HEADERS_SECTION) {
-        // Case where the sent message only contains control data
-        msg_ = std::move(decodeInfo_.msg);
-      } else {
-        callback_->onError(
-            ingressTxnID_,
-            HTTPException(
-                HTTPException::Direction::INGRESS,
-                fmt::format("Message not formed (incomplete binary data)")));
-        return;
-      }
-    }
-    callback_->onHeadersComplete(ingressTxnID_, std::move(msg_));
-    if (msgBody_) {
-      callback_->onBody(ingressTxnID_, std::move(msgBody_), 0);
-    }
-    if (trailers_) {
-      callback_->onTrailersComplete(ingressTxnID_, std::move(trailers_));
-    }
+  // We can trim the amount of bufferedIngress_ that we were successfully able
+  // to parse
+  bufferedIngress_.trimStartAtMost(parsedTot);
+
+  return len;
+}
+
+void HTTPBinaryCodec::onIngressEOF() {
+  if (!parseError_ && !bufferedIngress_.empty()) {
+    // Case where the ingress EOF is received before the entire message is
+    // parsed
+    callback_->onError(ingressTxnID_,
+                       HTTPException(HTTPException::Direction::INGRESS,
+                                     "Incomplete message received"));
+    return;
+  }
+  if (state_ == ParseState::HEADERS_SECTION) {
+    // Case where the sent message only contains control data and no headers
+    // nor body
+    callback_->onHeadersComplete(ingressTxnID_, std::move(decodeInfo_.msg));
+  }
+  if (!parseError_ && !parserPaused_) {
     callback_->onMessageComplete(ingressTxnID_, false);
   }
+  return;
 }
 
 size_t HTTPBinaryCodec::generateHeaderHelper(folly::io::QueueAppender& appender,

proxygen/lib/http/codec/HTTPBinaryCodec.h

@@ -163,9 +163,7 @@ class HTTPBinaryCodec : public HTTPCodec {
   ParseResult parseHeaders(folly::io::Cursor& cursor,
                            size_t remaining,
                            HeaderDecodeInfo& decodeInfo);
-  ParseResult parseContent(folly::io::Cursor& cursor,
-                           size_t remaining,
-                           HTTPMessage& msg);
+  ParseResult parseContent(folly::io::Cursor& cursor, size_t remaining);
   ParseResult parseTrailers(folly::io::Cursor& cursor,
                             size_t remaining,
                             HeaderDecodeInfo& decodeInfo);

proxygen/lib/http/codec/test/HTTPBinaryCodecTest.cpp

@@ -44,10 +44,8 @@ class HTTPBinaryCodecForTest : public HTTPBinaryCodec {
     return HTTPBinaryCodec::parseHeaders(cursor, remaining, decodeInfo);
   }
 
-  ParseResult parseContent(folly::io::Cursor& cursor,
-                           size_t remaining,
-                           HTTPMessage& msg) {
-    return HTTPBinaryCodec::parseContent(cursor, remaining, msg);
+  ParseResult parseContent(folly::io::Cursor& cursor, size_t remaining) {
+    return HTTPBinaryCodec::parseContent(cursor, remaining);
   }
 
   folly::IOBuf& getMsgBody() {
@@ -360,9 +358,9 @@ TEST_F(HTTPBinaryCodecTest, testParseContentSuccess) {
   folly::io::Cursor cursor(contentIOBuf.get());
 
   HTTPMessage msg;
-  EXPECT_EQ(upstreamBinaryCodec_->parseContent(cursor, content.size(), msg)
-                .bytesParsed_,
-            content.size());
+  EXPECT_EQ(
+      upstreamBinaryCodec_->parseContent(cursor, content.size()).bytesParsed_,
+      content.size());
   EXPECT_EQ(upstreamBinaryCodec_->getMsgBody().to<std::string>(), "hello\r\n");
 }
 
@@ -376,8 +374,7 @@ TEST_F(HTTPBinaryCodecTest, testParseContentFailure) {
 
   HTTPMessage msg;
   EXPECT_EQ(
-      upstreamBinaryCodec_->parseContent(cursor, contentInvalid.size(), msg)
-          .error_,
+      upstreamBinaryCodec_->parseContent(cursor, contentInvalid.size()).error_,
       "Failure to parse content");
 }
 
@@ -449,6 +446,57 @@ TEST_F(HTTPBinaryCodecTest, testOnIngressSuccessForControlData) {
   EXPECT_EQ(callback.msg->getURL(), "/");
 }
 
+TEST_F(HTTPBinaryCodecTest, testOnIngressSuccessChunkedOnBoundaryMessage) {
+  // Format is chunk1 = `..GET`, chunk2 =
+  // `.https.www.example.com./hello.txt..user-agent.curl/7.16.3 libcurl/7.16.3
+  // OpenSSL/0.9.7l zlib/1.2.3.host.www.example.com.accept-language.en, mi`
+  const std::vector<uint8_t> binaryHTTPMessageChunk1{
+      0x00, 0x03, 0x47, 0x45, 0x54, 0x05, 0x68, 0x74, 0x74, 0x70,
+      0x73, 0x0f, 0x77, 0x77, 0x77, 0x2e, 0x65, 0x78, 0x61, 0x6d,
+      0x70, 0x6c, 0x65, 0x2e, 0x63, 0x6f, 0x6d, 0x0a, 0x2f, 0x68,
+      0x65, 0x6c, 0x6c, 0x6f, 0x2e, 0x74, 0x78, 0x74};
+  const std::vector<uint8_t> binaryHTTPMessageChunk2{
+      0x40, 0x6c, 0x0a, 0x75, 0x73, 0x65, 0x72, 0x2d, 0x61, 0x67, 0x65, 0x6e,
+      0x74, 0x34, 0x63, 0x75, 0x72, 0x6c, 0x2f, 0x37, 0x2e, 0x31, 0x36, 0x2e,
+      0x33, 0x20, 0x6c, 0x69, 0x62, 0x63, 0x75, 0x72, 0x6c, 0x2f, 0x37, 0x2e,
+      0x31, 0x36, 0x2e, 0x33, 0x20, 0x4f, 0x70, 0x65, 0x6e, 0x53, 0x53, 0x4c,
+      0x2f, 0x30, 0x2e, 0x39, 0x2e, 0x37, 0x6c, 0x20, 0x7a, 0x6c, 0x69, 0x62,
+      0x2f, 0x31, 0x2e, 0x32, 0x2e, 0x33, 0x04, 0x68, 0x6f, 0x73, 0x74, 0x0f,
+      0x77, 0x77, 0x77, 0x2e, 0x65, 0x78, 0x61, 0x6d, 0x70, 0x6c, 0x65, 0x2e,
+      0x63, 0x6f, 0x6d, 0x0f, 0x61, 0x63, 0x63, 0x65, 0x70, 0x74, 0x2d, 0x6c,
+      0x61, 0x6e, 0x67, 0x75, 0x61, 0x67, 0x65, 0x06, 0x65, 0x6e, 0x2c, 0x20,
+      0x6d, 0x69, 0x00, 0x00};
+
+  auto binaryHTTPMessageIOBufChunk1 = folly::IOBuf::wrapBuffer(folly::ByteRange(
+      binaryHTTPMessageChunk1.data(), binaryHTTPMessageChunk1.size()));
+  folly::io::Cursor cursor1(binaryHTTPMessageIOBufChunk1.get());
+  auto binaryHTTPMessageIOBufChunk2 = folly::IOBuf::wrapBuffer(folly::ByteRange(
+      binaryHTTPMessageChunk2.data(), binaryHTTPMessageChunk2.size()));
+  folly::io::Cursor cursor2(binaryHTTPMessageIOBufChunk2.get());
+
+  FakeHTTPCodecCallback callback;
+  upstreamBinaryCodec_->setCallback(&callback);
+  upstreamBinaryCodec_->onIngress(*binaryHTTPMessageIOBufChunk1);
+  upstreamBinaryCodec_->onIngress(*binaryHTTPMessageIOBufChunk2);
+  upstreamBinaryCodec_->onIngressEOF();
+
+  // Check onError was not called for the callback
+  EXPECT_EQ(callback.lastParseError, nullptr);
+
+  // Check msg and header fields
+  EXPECT_EQ(callback.msg->isSecure(), true);
+  EXPECT_EQ(callback.msg->getMethod(), proxygen::HTTPMethod::GET);
+  EXPECT_EQ(callback.msg->getURL(), "/hello.txt");
+  HTTPHeaders httpHeaders = callback.msg->getHeaders();
+  EXPECT_EQ(httpHeaders.exists("user-agent"), true);
+  EXPECT_EQ(httpHeaders.exists("host"), true);
+  EXPECT_EQ(httpHeaders.exists("accept-language"), true);
+  EXPECT_EQ(httpHeaders.getSingleOrEmpty("user-agent"),
+            "curl/7.16.3 libcurl/7.16.3 OpenSSL/0.9.7l zlib/1.2.3");
+  EXPECT_EQ(httpHeaders.getSingleOrEmpty("host"), "www.example.com");
+  EXPECT_EQ(httpHeaders.getSingleOrEmpty("accept-language"), "en, mi");
+}
+
 TEST_F(HTTPBinaryCodecTest, testOnIngressFailureMalformedMessage) {
   // Format is `..GET.https.www.example.com./hello.txt..user-agent.curl/7.16.3
   // libcurl/7.16.3 OpenSSL/0.9.7l
@@ -482,24 +530,6 @@ TEST_F(HTTPBinaryCodecTest, testOnIngressFailureMalformedMessage) {
             "Invalid Message: Failure to parse: headerValue");
 }
 
-TEST_F(HTTPBinaryCodecTest, testOnIngressFailureIncompleteMessage) {
-  // Message is incomplete and has only 1 byte
-  const std::vector<uint8_t> binaryInvalidHTTPMessage{0x00};
-  auto binaryInvalidHTTPMessageIOBuf =
-      folly::IOBuf::wrapBuffer(folly::ByteRange(
-          binaryInvalidHTTPMessage.data(), binaryInvalidHTTPMessage.size()));
-  folly::io::Cursor cursor(binaryInvalidHTTPMessageIOBuf.get());
-
-  FakeHTTPCodecCallback callback;
-  upstreamBinaryCodec_->setCallback(&callback);
-  upstreamBinaryCodec_->onIngress(*binaryInvalidHTTPMessageIOBuf);
-  upstreamBinaryCodec_->onIngressEOF();
-
-  // Check onError was called with the correct error
-  EXPECT_EQ(std::string(callback.lastParseError.get()->what()),
-            "Message not formed (incomplete binary data)");
-}
-
 TEST_F(HTTPBinaryCodecTest, testGenerateHeaders) {
   // Create HTTPMessage and encode it to a buffer
   HTTPMessage msgEncoded;
@@ -545,8 +575,7 @@ TEST_F(HTTPBinaryCodecTest, testGenerateBody) {
   // Decode Test Body and check
   folly::io::Cursor cursor(writeBuffer.front());
   HTTPMessage msg;
-  EXPECT_EQ(upstreamBinaryCodec_->parseContent(cursor, 18, msg).bytesParsed_,
-            18);
+  EXPECT_EQ(upstreamBinaryCodec_->parseContent(cursor, 18).bytesParsed_, 18);
   EXPECT_EQ(upstreamBinaryCodec_->getMsgBody().to<std::string>(),
             "Sample Test Body!");
 }