HTTPDownstreamSessionTest.cpp

Go to the documentation of this file.

/*
 *  Copyright (c) 2015-present, Facebook, Inc.
 *  All rights reserved.
 *
 *  This source code is licensed under the BSD-style license found in the
 *  LICENSE file in the root directory of this source tree. An additional grant
 *  of patent rights can be found in the PATENTS file in the same directory.
 *
 */
#include <string>
#include <vector>

#include <folly/Conv.h>
#include <folly/Range.h>
#include <folly/futures/Promise.h>
#include <folly/io/Cursor.h>
#include <folly/io/async/EventBase.h>
#include <folly/io/async/EventBaseManager.h>
#include <folly/io/async/TimeoutManager.h>
#include <folly/io/async/test/MockAsyncTransport.h>
#include <folly/portability/GTest.h>
#include <proxygen/lib/http/codec/HTTPCodecFactory.h>
#include <proxygen/lib/http/codec/test/TestUtils.h>
#include <proxygen/lib/http/session/HTTPDirectResponseHandler.h>
#include <proxygen/lib/http/session/HTTPDownstreamSession.h>
#include <proxygen/lib/http/session/HTTPSession.h>
#include <proxygen/lib/http/session/test/HTTPSessionMocks.h>
#include <proxygen/lib/http/session/test/HTTPSessionTest.h>
#include <proxygen/lib/http/session/test/MockByteEventTracker.h>
#include <proxygen/lib/http/session/test/TestUtils.h>
#include <proxygen/lib/test/TestAsyncTransport.h>
#include <wangle/acceptor/ConnectionManager.h>

using namespace folly::io;
using namespace wangle;
using namespace folly;
using namespace proxygen;
using namespace std;
using namespace testing;
using namespace std::chrono;
using folly::Promise;

template <typename C>
class HTTPDownstreamTest : public testing::Test {
 public:
  explicit HTTPDownstreamTest(
    std::vector<int64_t> flowControl = { -1, -1, -1 },
    bool startImmediately = true)
    : eventBase_(),
      transport_(new TestAsyncTransport(&eventBase_)),
      transactionTimeouts_(makeTimeoutSet(&eventBase_)),
      flowControl_(flowControl) {
    EXPECT_CALL(mockController_, getGracefulShutdownTimeout())
      .WillRepeatedly(Return(std::chrono::milliseconds(0)));
    EXPECT_CALL(mockController_, attachSession(_))
      .WillRepeatedly(Invoke([&] (HTTPSessionBase* session) {
        session->setPrioritySampled(true);
      }));
    HTTPSession::setDefaultReadBufferLimit(65536);
    auto codec = makeServerCodec<typename C::Codec>(C::version);
    rawCodec_ = codec.get();

    // If the codec is H2, getHeaderIndexingStrategy will be called when setting
    // up the codec
    if (rawCodec_->getProtocol() == CodecProtocol::HTTP_2) {
      EXPECT_CALL(mockController_, getHeaderIndexingStrategy())
        .WillOnce(
          Return(&testH2IndexingStrat_)
      );
    }

    httpSession_ = new HTTPDownstreamSession(
      transactionTimeouts_.get(),
      std::move(AsyncTransportWrapper::UniquePtr(transport_)),
      localAddr, peerAddr,
      &mockController_,
      std::move(codec),
      mockTransportInfo /* no stats for now */,
      nullptr);
    for (auto& param: flowControl) {
      if (param < 0) {
        param = rawCodec_->getDefaultWindowSize();
      }
    }

    // Ensure the H2 header indexing strategy was setup correctly if applicable
    if (rawCodec_->getProtocol() == CodecProtocol::HTTP_2) {
      HTTP2Codec* recastedCodec = dynamic_cast<HTTP2Codec*>(rawCodec_);
      EXPECT_EQ(
        recastedCodec->getHeaderIndexingStrategy(), &testH2IndexingStrat_);
    }

    httpSession_->setFlowControl(flowControl[0], flowControl[1],
                                 flowControl[2]);
    httpSession_->setEgressSettings({{ SettingsId::MAX_CONCURRENT_STREAMS, 80 },
                                     { SettingsId::HEADER_TABLE_SIZE, 5555 },
                                     { SettingsId::ENABLE_PUSH, 1 },
                                     { SettingsId::ENABLE_EX_HEADERS, 1 }});
    if (startImmediately) {
      httpSession_->startNow();
    }
    clientCodec_ = makeClientCodec<typename C::Codec>(C::version);
    if (clientCodec_->getProtocol() == CodecProtocol::HTTP_2) {
      clientCodec_->getEgressSettings()->setSetting(
        SettingsId::ENABLE_EX_HEADERS, 1);
    }
    clientCodec_->generateConnectionPreface(requests_);
    clientCodec_->setCallback(&callbacks_);
  }

  HTTPCodec::StreamID sendRequest(const std::string& url = "/",
                                  int8_t priority = 0,
                                  bool eom = true) {
    auto req = getGetRequest();
    req.setURL(url);
    req.setPriority(priority);
    return sendRequest(req, eom);
  }

  HTTPCodec::StreamID sendRequest(const HTTPMessage& req, bool eom = true) {
    auto streamID = clientCodec_->createStream();
    clientCodec_->generateHeader(requests_, streamID, req, eom);
    return streamID;
  }

  HTTPCodec::StreamID sendHeader() {
    return sendRequest("/", 0, false);
  }

  Promise<Unit> sendRequestLater(HTTPMessage req, bool eof=false) {
    Promise<Unit> reqp;
    reqp.getFuture().then(&eventBase_, [=] {
        sendRequest(req);
        transport_->addReadEvent(requests_, milliseconds(0));
        if (eof) {
          transport_->addReadEOF(milliseconds(0));
        }
      });
    return reqp;
  }

  void SetUp() override {
    folly::EventBaseManager::get()->clearEventBase();
    HTTPSession::setDefaultWriteBufferLimit(65536);
    HTTP2PriorityQueue::setNodeLifetime(std::chrono::milliseconds(2));
  }

  void cleanup() {
    EXPECT_CALL(mockController_, detachSession(_));
    httpSession_->dropConnection();
  }


  std::unique_ptr<testing::StrictMock<MockHTTPHandler>>
  addSimpleStrictHandler() {
    std::unique_ptr<testing::StrictMock<MockHTTPHandler>> handler =
      std::make_unique<testing::StrictMock<MockHTTPHandler>>();

    // The ownership model here is suspect, but assume the callers won't destroy
    // handler before it's requested
    auto rawHandler = handler.get();
    EXPECT_CALL(mockController_, getRequestHandler(testing::_, testing::_))
      .WillOnce(testing::Return(rawHandler))
      .RetiresOnSaturation();

    EXPECT_CALL(*handler, setTransaction(testing::_))
      .WillOnce(testing::SaveArg<0>(&handler->txn_));

    return handler;
  }

  std::unique_ptr<testing::NiceMock<MockHTTPHandler>>
  addSimpleNiceHandler() {
    std::unique_ptr<testing::NiceMock<MockHTTPHandler>> handler =
      std::make_unique<testing::NiceMock<MockHTTPHandler>>();

    // See comment above
    auto rawHandler = handler.get();
    EXPECT_CALL(mockController_, getRequestHandler(testing::_, testing::_))
      .WillOnce(testing::Return(rawHandler))
      .RetiresOnSaturation();

    EXPECT_CALL(*handler, setTransaction(testing::_))
      .WillOnce(testing::SaveArg<0>(&handler->txn_));

    return handler;
  }

  void onEOMTerminateHandlerExpectShutdown(MockHTTPHandler& handler) {
    handler.expectEOM([&] { handler.terminate(); });
    handler.expectDetachTransaction();
    expectDetachSession();
  }

  void expectDetachSession() {
    EXPECT_CALL(mockController_, detachSession(testing::_));
  }

  void addSingleByteReads(const char* data, milliseconds delay={}) {
    for (const char* p = data; *p != '\0'; ++p) {
      transport_->addReadEvent(p, 1, delay);
    }
  }

  void flushRequestsAndLoop(
    bool eof=false, milliseconds eofDelay=milliseconds(0),
    milliseconds initialDelay=milliseconds(0),
    std::function<void()> extraEventsFn = std::function<void()>()) {
    flushRequests(eof, eofDelay, initialDelay, extraEventsFn);
    eventBase_.loop();
  }

  void flushRequestsAndLoopN(uint64_t n,
    bool eof=false, milliseconds eofDelay=milliseconds(0),
    milliseconds initialDelay=milliseconds(0),
    std::function<void()> extraEventsFn = std::function<void()>()) {
    flushRequests(eof, eofDelay, initialDelay, extraEventsFn);
    for (uint64_t i = 0; i < n; i++) {
      eventBase_.loopOnce();
    }
  }

  void flushRequests(
    bool eof=false, milliseconds eofDelay=milliseconds(0),
    milliseconds initialDelay=milliseconds(0),
    std::function<void()> extraEventsFn = std::function<void()>()) {
    transport_->addReadEvent(requests_, initialDelay);
    if (extraEventsFn) {
      extraEventsFn();
    }
    if (eof) {
      transport_->addReadEOF(eofDelay);
    }
    transport_->startReadEvents();
  }

  void testSimpleUpgrade(
    const std::string& upgradeHeader,
    CodecProtocol expectedProtocol,
    const std::string& expectedUpgradeHeader);

  void gracefulShutdown() {
    folly::DelayedDestruction::DestructorGuard g(httpSession_);
    clientCodec_->generateGoaway(this->requests_, 0, ErrorCode::NO_ERROR);
    expectDetachSession();
    flushRequestsAndLoop(true);
  }

  void testPriorities(uint32_t numPriorities);

  void testChunks(bool trailers);

  void expect101(CodecProtocol expectedProtocol,
                 const std::string& expectedUpgrade,
                 bool expect100 = false) {
    NiceMock<MockHTTPCodecCallback> callbacks;

    EXPECT_CALL(callbacks, onMessageBegin(_, _));
    EXPECT_CALL(callbacks, onNativeProtocolUpgrade(_, _, _, _))
      .WillOnce(
        Invoke([this, expectedUpgrade] (HTTPCodec::StreamID,
                                        CodecProtocol,
                                        const std::string&,
                                        HTTPMessage& msg) {
             EXPECT_EQ(msg.getStatusCode(), 101);
             EXPECT_EQ(msg.getStatusMessage(), "Switching Protocols");
             EXPECT_EQ(msg.getHeaders().getSingleOrEmpty(HTTP_HEADER_UPGRADE),
                       expectedUpgrade);
             // also connection and date
             EXPECT_EQ(msg.getHeaders().size(), 3);
             breakParseOutput_ = true;
             return true;
               }));
    // this comes before 101, but due to gmock this is backwards
    if (expect100) {
      EXPECT_CALL(callbacks, onMessageBegin(_, _))
        .RetiresOnSaturation();
      EXPECT_CALL(callbacks, onHeadersComplete(_, _))
        .WillOnce(Invoke([] (HTTPCodec::StreamID,
                             std::shared_ptr<HTTPMessage> msg) {
                 LOG(INFO) << "100 headers";
                 EXPECT_EQ(msg->getStatusCode(), 100);
                         }))
        .RetiresOnSaturation();
      EXPECT_CALL(callbacks, onMessageComplete(_, _))
        .RetiresOnSaturation();
    }
    clientCodec_->setCallback(&callbacks);
    parseOutput(*clientCodec_);
    clientCodec_ = HTTPCodecFactory::getCodec(expectedProtocol,
                                              TransportDirection::UPSTREAM);
  }
  void expectResponse(uint32_t code = 200,
                      ErrorCode errorCode = ErrorCode::NO_ERROR,
                      bool expect100 = false, bool expectGoaway = false) {
    expectResponses(1, code, errorCode, expect100, expectGoaway);
  }
  void expectResponses(uint32_t n, uint32_t code = 200,
                       ErrorCode errorCode = ErrorCode::NO_ERROR,
                       bool expect100 = false, bool expectGoaway = false) {
    clientCodec_->setCallback(&callbacks_);
    if (isParallelCodecProtocol(clientCodec_->getProtocol())) {
      EXPECT_CALL(callbacks_, onSettings(_))
        .WillOnce(Invoke([this] (const SettingsList& settings) {
              if (flowControl_[0] > 0) {
                bool foundInitialWindow = false;
                for (const auto& setting: settings) {
                  if (setting.id == SettingsId::INITIAL_WINDOW_SIZE) {
                    EXPECT_EQ(flowControl_[0], setting.value);
                    foundInitialWindow = true;
                  }
                }
                EXPECT_TRUE(foundInitialWindow);
              }
            }));
    }
    if (flowControl_[2] > 0) {
      int64_t sessionDelta =
        flowControl_[2] - clientCodec_->getDefaultWindowSize();
      if (clientCodec_->supportsSessionFlowControl() && sessionDelta) {
        EXPECT_CALL(callbacks_, onWindowUpdate(0, sessionDelta));
      }
    }
    if (flowControl_[1] > 0) {
      size_t initWindow = flowControl_[0] > 0 ?
        flowControl_[0] : clientCodec_->getDefaultWindowSize();
      int64_t streamDelta = flowControl_[1] - initWindow;
      if (clientCodec_->supportsStreamFlowControl() && streamDelta) {
        EXPECT_CALL(callbacks_, onWindowUpdate(1, streamDelta));
      }
    }

    if (expectGoaway) {
      EXPECT_CALL(callbacks_, onGoaway(HTTPCodec::StreamID(1),
                                      ErrorCode::NO_ERROR, _));
    }

    for (uint32_t i = 0; i < n; i++) {
      uint8_t times = (expect100) ? 2 : 1;
      EXPECT_CALL(callbacks_, onMessageBegin(_, _))
        .Times(times).RetiresOnSaturation();
      EXPECT_CALL(callbacks_, onHeadersComplete(_, _))
        .WillOnce(Invoke([code] (HTTPCodec::StreamID,
                                 std::shared_ptr<HTTPMessage> msg) {
                           EXPECT_EQ(msg->getStatusCode(), code);
                         }));
      if (expect100) {
        EXPECT_CALL(callbacks_, onHeadersComplete(_, _))
          .WillOnce(Invoke([] (HTTPCodec::StreamID,
                               std::shared_ptr<HTTPMessage> msg) {
                             EXPECT_EQ(msg->getStatusCode(), 100);
                           }))
          .RetiresOnSaturation();
      }
      if (errorCode != ErrorCode::NO_ERROR) {
        EXPECT_CALL(callbacks_, onAbort(_, _))
          .WillOnce(Invoke([errorCode] (HTTPCodec::StreamID,
                                        ErrorCode error) {
                             EXPECT_EQ(error, errorCode);
                         }));
      }
      EXPECT_CALL(callbacks_, onBody(_, _, _)).RetiresOnSaturation();
      EXPECT_CALL(callbacks_, onMessageComplete(_, _)).RetiresOnSaturation();
    }
    parseOutput(*clientCodec_);
  }

  void parseOutput(HTTPCodec& clientCodec) {
    auto writeEvents = transport_->getWriteEvents();
    while (!breakParseOutput_ &&
           (!writeEvents->empty() || !parseOutputStream_.empty())) {
      if (!writeEvents->empty()) {
        auto event = writeEvents->front();
        auto vec = event->getIoVec();
        for (size_t i = 0; i < event->getCount(); i++) {
          parseOutputStream_.append(
            IOBuf::copyBuffer(vec[i].iov_base, vec[i].iov_len));
        }
        writeEvents->pop_front();
      }
      uint32_t consumed = clientCodec.onIngress(*parseOutputStream_.front());
      parseOutputStream_.split(consumed);
    }
    if (!breakParseOutput_) {
      EXPECT_EQ(parseOutputStream_.chainLength(), 0);
    }
    breakParseOutput_ = false;
  }

  void resumeWritesInLoop() {
    eventBase_.runInLoop([this] { transport_->resumeWrites(); });
  }

  void resumeWritesAfterDelay(milliseconds delay) {
    eventBase_.runAfterDelay([this] { transport_->resumeWrites(); },
                             delay.count());
  }

  MockByteEventTracker* setMockByteEventTracker() {
    auto byteEventTracker = new MockByteEventTracker(nullptr);
    httpSession_->setByteEventTracker(
      std::unique_ptr<ByteEventTracker>(byteEventTracker));
    EXPECT_CALL(*byteEventTracker, preSend(_, _, _))
      .WillRepeatedly(Return(0));
    EXPECT_CALL(*byteEventTracker, drainByteEvents())
      .WillRepeatedly(Return(0));
    EXPECT_CALL(*byteEventTracker, processByteEvents(_, _))
      .WillRepeatedly(Invoke([]
                             (std::shared_ptr<ByteEventTracker> self,
                              uint64_t bytesWritten) {
                               return self->ByteEventTracker::processByteEvents(
                                 self,
                                 bytesWritten);
                             }));

    return byteEventTracker;
  }

 protected:
  EventBase eventBase_;
  TestAsyncTransport* transport_;  // invalid once httpSession_ is destroyed
  folly::HHWheelTimer::UniquePtr transactionTimeouts_;
  std::vector<int64_t> flowControl_;
  StrictMock<MockController> mockController_;
  HTTPDownstreamSession* httpSession_;
  IOBufQueue requests_{IOBufQueue::cacheChainLength()};
  unique_ptr<HTTPCodec> clientCodec_;
  NiceMock<MockHTTPCodecCallback> callbacks_;
  IOBufQueue parseOutputStream_{IOBufQueue::cacheChainLength()};
  bool breakParseOutput_{false};
  typename C::Codec* rawCodec_{nullptr};
  HeaderIndexingStrategy testH2IndexingStrat_;
};

// Uses TestAsyncTransport
using HTTPDownstreamSessionTest = HTTPDownstreamTest<HTTP1xCodecPair>;
using SPDY3DownstreamSessionTest = HTTPDownstreamTest<SPDY3CodecPair>;
namespace {
class HTTP2DownstreamSessionTest : public HTTPDownstreamTest<HTTP2CodecPair> {
 public:
  HTTP2DownstreamSessionTest()
      : HTTPDownstreamTest<HTTP2CodecPair>() {}

  void SetUp() override {
    HTTPDownstreamTest<HTTP2CodecPair>::SetUp();
  }

  void SetupControlStream(HTTPCodec::StreamID cStreamId) {
    // enable EX_HEADERS
    clientCodec_->getEgressSettings()->setSetting(
      SettingsId::ENABLE_EX_HEADERS, 1);
    clientCodec_->generateSettings(requests_);
    // create a control stream
    clientCodec_->generateHeader(requests_, cStreamId, getGetRequest("/cc"),
                                 true, nullptr);
  }

  void TearDown() override {
  }
};
}

namespace {
class HTTP2DownstreamSessionEarlyShutdownTest :
public HTTPDownstreamTest<HTTP2CodecPair> {
 public:
  HTTP2DownstreamSessionEarlyShutdownTest()
      : HTTPDownstreamTest<HTTP2CodecPair>({-1, -1, -1}, false) {}

  void SetUp() override {
    HTTPDownstreamTest<HTTP2CodecPair>::SetUp();
  }

  void TearDown() override {
  }
};
}

TEST_F(HTTP2DownstreamSessionEarlyShutdownTest, EarlyShutdown) {
  folly::DelayedDestruction::DestructorGuard g(httpSession_);

  // Try shutting down the session and then starting it. This should be properly
  // handled by the HTTPSession such that no HTTP/2 frames are sent in the
  // wrong order.
  StrictMock<MockHTTPCodecCallback> callbacks;
  clientCodec_->setCallback(&callbacks);
  EXPECT_CALL(callbacks, onFrameHeader(_, _, _, _, _)).Times(2);
  EXPECT_CALL(callbacks, onSettings(_)).Times(1);
  EXPECT_CALL(callbacks, onGoaway(_, _, _)).Times(1);
  expectDetachSession();
  httpSession_->notifyPendingShutdown();
  httpSession_->startNow();
  eventBase_.loop();
  parseOutput(*clientCodec_);
}

TEST_F(HTTPDownstreamSessionTest, ImmediateEof) {
  // Send EOF without any request data
  EXPECT_CALL(mockController_, getRequestHandler(_, _)).Times(0);
  expectDetachSession();

  flushRequestsAndLoop(true, milliseconds(0));
}

TEST_F(HTTPDownstreamSessionTest, Http10NoHeaders) {
  InSequence enforceOrder;

  auto handler = addSimpleNiceHandler();
  handler->expectHeaders([&] (std::shared_ptr<HTTPMessage> msg) {
      EXPECT_FALSE(msg->getIsChunked());
      EXPECT_FALSE(msg->getIsUpgraded());
      EXPECT_EQ("/", msg->getURL());
      EXPECT_EQ("/", msg->getPath());
      EXPECT_EQ("", msg->getQueryString());
      EXPECT_EQ(1, msg->getHTTPVersion().first);
      EXPECT_EQ(0, msg->getHTTPVersion().second);
    });
  onEOMTerminateHandlerExpectShutdown(*handler);

  auto req = getGetRequest();
  req.setHTTPVersion(1, 0);
  sendRequest(req);
  flushRequestsAndLoop();
}

TEST_F(HTTPDownstreamSessionTest, Http10NoHeadersEof) {
  InSequence enforceOrder;

  auto handler = addSimpleNiceHandler();
  handler->expectHeaders([&] (std::shared_ptr<HTTPMessage> msg) {
      EXPECT_FALSE(msg->getIsChunked());
      EXPECT_FALSE(msg->getIsUpgraded());
      EXPECT_EQ("http://example.com/foo?bar", msg->getURL());
      EXPECT_EQ("/foo", msg->getPath());
      EXPECT_EQ("bar", msg->getQueryString());
      EXPECT_EQ(1, msg->getHTTPVersion().first);
      EXPECT_EQ(0, msg->getHTTPVersion().second);
    });
  onEOMTerminateHandlerExpectShutdown(*handler);

  const char *req = "GET http://example.com/foo?bar HTTP/1.0\r\n\r\n";
  requests_.append(req, strlen(req));
  flushRequestsAndLoop(true, milliseconds(0));
}

TEST_F(HTTPDownstreamSessionTest, SingleBytes) {
  InSequence enforceOrder;

  auto handler = addSimpleNiceHandler();
  handler->expectHeaders([&] (std::shared_ptr<HTTPMessage> msg) {
      const HTTPHeaders& hdrs = msg->getHeaders();
      EXPECT_EQ(2, hdrs.size());
      EXPECT_TRUE(hdrs.exists("host"));
      EXPECT_TRUE(hdrs.exists("connection"));

      EXPECT_FALSE(msg->getIsChunked());
      EXPECT_FALSE(msg->getIsUpgraded());
      EXPECT_EQ("/somepath.php?param=foo", msg->getURL());
      EXPECT_EQ("/somepath.php", msg->getPath());
      EXPECT_EQ("param=foo", msg->getQueryString());
      EXPECT_EQ(1, msg->getHTTPVersion().first);
      EXPECT_EQ(1, msg->getHTTPVersion().second);
    });
  onEOMTerminateHandlerExpectShutdown(*handler);

  addSingleByteReads("GET /somepath.php?param=foo HTTP/1.1\r\n"
                     "Host: example.com\r\n"
                     "Connection: close\r\n"
                     "\r\n");
  transport_->addReadEOF(milliseconds(0));
  transport_->startReadEvents();
  eventBase_.loop();
}

TEST_F(HTTPDownstreamSessionTest, SingleBytesWithBody) {
  InSequence enforceOrder;

  auto handler = addSimpleNiceHandler();
  handler->expectHeaders([&] (std::shared_ptr<HTTPMessage> msg) {
      const HTTPHeaders& hdrs = msg->getHeaders();
      EXPECT_EQ(3, hdrs.size());
      EXPECT_TRUE(hdrs.exists("host"));
      EXPECT_TRUE(hdrs.exists("content-length"));
      EXPECT_TRUE(hdrs.exists("myheader"));

      EXPECT_FALSE(msg->getIsChunked());
      EXPECT_FALSE(msg->getIsUpgraded());
      EXPECT_EQ("/somepath.php?param=foo", msg->getURL());
      EXPECT_EQ("/somepath.php", msg->getPath());
      EXPECT_EQ("param=foo", msg->getQueryString());
      EXPECT_EQ(1, msg->getHTTPVersion().first);
      EXPECT_EQ(1, msg->getHTTPVersion().second);
    });
  EXPECT_CALL(*handler, onBody(_))
    .WillOnce(ExpectString("1"))
    .WillOnce(ExpectString("2"))
    .WillOnce(ExpectString("3"))
    .WillOnce(ExpectString("4"))
    .WillOnce(ExpectString("5"));
  onEOMTerminateHandlerExpectShutdown(*handler);

  addSingleByteReads("POST /somepath.php?param=foo HTTP/1.1\r\n"
                     "Host: example.com\r\n"
                     "MyHeader: FooBar\r\n"
                     "Content-Length: 5\r\n"
                     "\r\n"
                     "12345");
  transport_->addReadEOF(milliseconds(0));
  transport_->startReadEvents();
  eventBase_.loop();
}

TEST_F(HTTPDownstreamSessionTest, SplitBody) {
  InSequence enforceOrder;

  auto handler = addSimpleNiceHandler();
  handler->expectHeaders([&] (std::shared_ptr<HTTPMessage> msg) {
      const HTTPHeaders& hdrs = msg->getHeaders();
      EXPECT_EQ(2, hdrs.size());
    });
  EXPECT_CALL(*handler, onBody(_))
    .WillOnce(ExpectString("12345"))
    .WillOnce(ExpectString("abcde"));
  onEOMTerminateHandlerExpectShutdown(*handler);

  transport_->addReadEvent("POST / HTTP/1.1\r\n"
                           "Host: example.com\r\n"
                           "Content-Length: 10\r\n"
                           "\r\n"
                           "12345", milliseconds(0));
  transport_->addReadEvent("abcde", milliseconds(5));
  transport_->addReadEOF(milliseconds(0));
  transport_->startReadEvents();
  eventBase_.loop();
}

TEST_F(HTTPDownstreamSessionTest, PostChunked) {
  InSequence enforceOrder;

  auto handler = addSimpleNiceHandler();
  handler->expectHeaders([&] (std::shared_ptr<HTTPMessage> msg) {
      const HTTPHeaders& hdrs = msg->getHeaders();
      EXPECT_EQ(3, hdrs.size());
      EXPECT_TRUE(hdrs.exists("host"));
      EXPECT_TRUE(hdrs.exists("content-type"));
      EXPECT_TRUE(hdrs.exists("transfer-encoding"));
      EXPECT_TRUE(msg->getIsChunked());
      EXPECT_FALSE(msg->getIsUpgraded());
      EXPECT_EQ("http://example.com/cgi-bin/foo.aspx?abc&def",
                msg->getURL());
      EXPECT_EQ("/cgi-bin/foo.aspx", msg->getPath());
      EXPECT_EQ("abc&def", msg->getQueryString());
      EXPECT_EQ(1, msg->getHTTPVersion().first);
      EXPECT_EQ(1, msg->getHTTPVersion().second);
    });
  EXPECT_CALL(*handler, onChunkHeader(3));
  EXPECT_CALL(*handler, onBody(_))
    .WillOnce(ExpectString("bar"));
  EXPECT_CALL(*handler, onChunkComplete());
  EXPECT_CALL(*handler, onChunkHeader(0x22));
  EXPECT_CALL(*handler, onBody(_))
    .WillOnce(ExpectString("0123456789abcdef\nfedcba9876543210\n"));
  EXPECT_CALL(*handler, onChunkComplete());
  EXPECT_CALL(*handler, onChunkHeader(3));
  EXPECT_CALL(*handler, onBody(_))
    .WillOnce(ExpectString("foo"));
  EXPECT_CALL(*handler, onChunkComplete());
  onEOMTerminateHandlerExpectShutdown(*handler);

  transport_->addReadEvent("POST http://example.com/cgi-bin/foo.aspx?abc&def "
                           "HTTP/1.1\r\n"
                           "Host: example.com\r\n"
                           "Content-Type: text/pla", milliseconds(0));
  transport_->addReadEvent("in; charset=utf-8\r\n"
                           "Transfer-encoding: chunked\r\n"
                           "\r", milliseconds(2));
  transport_->addReadEvent("\n"
                           "3\r\n"
                           "bar\r\n"
                           "22\r\n"
                           "0123456789abcdef\n"
                           "fedcba9876543210\n"
                           "\r\n"
                           "3\r", milliseconds(3));
  transport_->addReadEvent("\n"
                           "foo\r\n"
                           "0\r\n\r\n", milliseconds(1));
  transport_->startReadEvents();
  eventBase_.loop();
}

TEST_F(HTTPDownstreamSessionTest, MultiMessage) {
  InSequence enforceOrder;

  auto handler1 = addSimpleNiceHandler();
  handler1->expectHeaders();
  EXPECT_CALL(*handler1, onBody(_))
    .WillOnce(ExpectString("foo"))
    .WillOnce(ExpectString("bar9876"));
  handler1->expectEOM([&] { handler1->sendReply(); });
  handler1->expectDetachTransaction();

  auto handler2 = addSimpleNiceHandler();
  handler2->expectHeaders();
  EXPECT_CALL(*handler2, onChunkHeader(0xa));
  EXPECT_CALL(*handler2, onBody(_))
    .WillOnce(ExpectString("some "))
    .WillOnce(ExpectString("data\n"));
  EXPECT_CALL(*handler2, onChunkComplete());
  onEOMTerminateHandlerExpectShutdown(*handler2);

  transport_->addReadEvent("POST / HTTP/1.1\r\n"
                           "Host: example.com\r\n"
                           "Content-Length: 10\r\n"
                           "\r\n"
                           "foo", milliseconds(0));
  transport_->addReadEvent("bar9876"
                           "POST /foo HTTP/1.1\r\n"
                           "Host: exa", milliseconds(2));
  transport_->addReadEvent("mple.com\r\n"
                           "Connection: close\r\n"
                           "Trans", milliseconds(0));
  transport_->addReadEvent("fer-encoding: chunked\r\n"
                           "\r\n", milliseconds(2));
  transport_->addReadEvent("a\r\nsome ", milliseconds(0));
  transport_->addReadEvent("data\n\r\n0\r\n\r\n", milliseconds(2));
  transport_->addReadEOF(milliseconds(0));
  transport_->startReadEvents();
  eventBase_.loop();
}

TEST_F(HTTPDownstreamSessionTest, Connect) {
  InSequence enforceOrder;

  auto handler = addSimpleStrictHandler();
  // Send HTTP 200 OK to accept the CONNECT request
  handler->expectHeaders([&handler] {
      handler->sendHeaders(200, 100);
    });

  EXPECT_CALL(*handler, onUpgrade(_));

  // Data should be received using onBody
  EXPECT_CALL(*handler, onBody(_))
    .WillOnce(ExpectString("12345"))
    .WillOnce(ExpectString("abcde"));
  onEOMTerminateHandlerExpectShutdown(*handler);

  transport_->addReadEvent("CONNECT test HTTP/1.1\r\n"
                           "\r\n"
                           "12345", milliseconds(0));
  transport_->addReadEvent("abcde", milliseconds(5));
  transport_->addReadEOF(milliseconds(0));
  transport_->startReadEvents();
  eventBase_.loop();
}

TEST_F(HTTPDownstreamSessionTest, ConnectRejected) {
  InSequence enforceOrder;

  auto handler = addSimpleStrictHandler();
  // Send HTTP 400 to reject the CONNECT request
  handler->expectHeaders([&handler] {
      handler->sendReplyCode(400);
    });

  onEOMTerminateHandlerExpectShutdown(*handler);

  transport_->addReadEvent("CONNECT test HTTP/1.1\r\n"
                           "\r\n"
                           "12345", milliseconds(0));
  transport_->addReadEvent("abcde", milliseconds(5));
  transport_->addReadEOF(milliseconds(0));
  transport_->startReadEvents();
  eventBase_.loop();
}

TEST_F(HTTPDownstreamSessionTest, HttpUpgrade) {
  InSequence enforceOrder;

  auto handler = addSimpleStrictHandler();
  // Send HTTP 101 Switching Protocls to accept the upgrade request
  handler->expectHeaders([&handler] {
      handler->sendHeaders(101, 100);
    });

  // Send the response in the new protocol after upgrade
  EXPECT_CALL(*handler, onUpgrade(_))
      .WillOnce(Invoke([&handler](UpgradeProtocol /*protocol*/) {
        handler->sendReplyCode(100);
      }));

  onEOMTerminateHandlerExpectShutdown(*handler);

  HTTPMessage req = getGetRequest();
  req.getHeaders().add(HTTP_HEADER_UPGRADE, "TEST/1.0");
  req.getHeaders().add(HTTP_HEADER_CONNECTION, "upgrade");
  sendRequest(req);
  flushRequestsAndLoop(true, milliseconds(0));
}

TEST(HTTPDownstreamTest, ParseErrorNoTxn) {
  // 1) Get a parse error on SYN_STREAM for streamID == 1
  // 2) Expect that the codec should be asked to generate an abort on
  //    streamID==1
  EventBase evb;

  // Setup the controller and its expecations.
  NiceMock<MockController> mockController;

  // Setup the codec, its callbacks, and its expectations.
  auto codec = makeDownstreamParallelCodec();
  HTTPCodec::Callback* codecCallback = nullptr;
  EXPECT_CALL(*codec, setCallback(_))
    .WillRepeatedly(SaveArg<0>(&codecCallback));
  // Expect egress abort for streamID == 1
  EXPECT_CALL(*codec, generateRstStream(_, 1, _));

  // Setup transport
  bool transportGood = true;
  auto transport = newMockTransport(&evb);
  EXPECT_CALL(*transport, good())
    .WillRepeatedly(ReturnPointee(&transportGood));
  EXPECT_CALL(*transport, closeNow())
    .WillRepeatedly(Assign(&transportGood, false));
  EXPECT_CALL(*transport, writeChain(_, _, _))
    .WillRepeatedly(
      Invoke([&] (folly::AsyncTransportWrapper::WriteCallback* callback,
                  const shared_ptr<IOBuf>&, WriteFlags) {
               callback->writeSuccess();
             }));

  // Create the downstream session, thus initializing codecCallback
  auto transactionTimeouts = makeInternalTimeoutSet(&evb);
  auto session = new HTTPDownstreamSession(
    transactionTimeouts.get(),
    AsyncTransportWrapper::UniquePtr(transport),
    localAddr, peerAddr,
    &mockController, std::move(codec),
    mockTransportInfo,
    nullptr);
  session->startNow();
  HTTPException ex(HTTPException::Direction::INGRESS_AND_EGRESS, "foo");
  ex.setProxygenError(kErrorParseHeader);
  ex.setCodecStatusCode(ErrorCode::REFUSED_STREAM);
  codecCallback->onError(HTTPCodec::StreamID(1), ex, true);

  // cleanup
  session->dropConnection();
  evb.loop();
}

TEST(HTTPDownstreamTest, ByteEventsDrained) {
  // Test that byte events are drained before socket is closed
  EventBase evb;

  NiceMock<MockController> mockController;
  auto codec = makeDownstreamParallelCodec();
  auto byteEventTracker = new MockByteEventTracker(nullptr);
  auto transport = newMockTransport(&evb);
  auto transactionTimeouts = makeInternalTimeoutSet(&evb);

  // Create the downstream session
  auto session = new HTTPDownstreamSession(
    transactionTimeouts.get(),
    AsyncTransportWrapper::UniquePtr(transport),
    localAddr, peerAddr,
    &mockController, std::move(codec),
    mockTransportInfo,
    nullptr);
  session->setByteEventTracker(
      std::unique_ptr<ByteEventTracker>(byteEventTracker));

  InSequence enforceOrder;

  session->startNow();

  // Byte events should be drained first
  EXPECT_CALL(*byteEventTracker, drainByteEvents())
    .Times(1);
  EXPECT_CALL(*transport, closeNow())
    .Times(AtLeast(1));

  // Close the socket
  session->dropConnection();
  evb.loop();
}

TEST_F(HTTPDownstreamSessionTest, HttpWithAckTiming) {
  // This is to test cases where holding a byte event to a finished HTTP/1.1
  // transaction does not masquerade as HTTP pipelining.
  auto byteEventTracker = setMockByteEventTracker();
  InSequence enforceOrder;

  auto handler1 = addSimpleStrictHandler();
  handler1->expectHeaders();
  handler1->expectEOM([&handler1] () {
      handler1->sendChunkedReplyWithBody(200, 100, 100, false);
    });
  // Hold a pending byte event
  EXPECT_CALL(*byteEventTracker, addLastByteEvent(_, _))
      .WillOnce(Invoke([] (HTTPTransaction* txn,
                           uint64_t /*byteNo*/) {
                         txn->incrementPendingByteEvents();
                       }));
  sendRequest();
  flushRequestsAndLoop();
  expectResponse();

  // Send the secode request after receiving the first response (eg: clearly
  // not pipelined)
  auto handler2 = addSimpleStrictHandler();
  handler2->expectHeaders();
  handler2->expectEOM([&handler2] () {
      handler2->sendChunkedReplyWithBody(200, 100, 100, false);
    });
  // This txn processed and destroyed before txn1
  EXPECT_CALL(*byteEventTracker, addLastByteEvent(_, _));
  handler2->expectDetachTransaction();

  sendRequest();
  flushRequestsAndLoop();
  expectResponse();

  // Now clear the pending byte event (simulate ack) and the first txn
  // goes away too
  handler1->expectDetachTransaction();
  handler1->txn_->decrementPendingByteEvents();
  gracefulShutdown();
}

TEST_F(HTTPDownstreamSessionTest, TestOnContentMismatch) {
  // Test the behavior when the reported content-length on the header
  // is different from the actual length of the body.
  // The expectation is simply to log the behavior, such as:
  // ".. HTTPTransaction.cpp ] Content-Length/body mismatch: expected: .. "
  folly::EventBase base;
  InSequence enforceOrder;
  auto handler1 = addSimpleNiceHandler();
  handler1->expectHeaders();
  handler1->expectEOM([&handler1] () {
        // over-estimate the content-length on the header
        handler1->sendHeaders(200, 105);
        handler1->sendBody(100);
        handler1->txn_->sendEOM();
      });
  sendRequest();
  flushRequestsAndLoop();

  auto handler2 = addSimpleNiceHandler();
  handler2->expectHeaders();
  handler2->expectEOM([&handler2] () {
        // under-estimate the content-length on the header
        handler2->sendHeaders(200, 95);
        handler2->sendBody(100);
        handler2->txn_->sendEOM();
      });
  sendRequest();
  flushRequestsAndLoop();
  gracefulShutdown();
}

TEST_F(HTTPDownstreamSessionTest, HttpWithAckTimingPipeline) {
  // Test a real pipelining case as well.  First request is done waiting for
  // ack, then receive two pipelined requests.
  auto byteEventTracker = setMockByteEventTracker();
  InSequence enforceOrder;

  auto handler1 = addSimpleStrictHandler();
  handler1->expectHeaders();
  handler1->expectEOM([&handler1] () {
      handler1->sendChunkedReplyWithBody(200, 100, 100, false);
    });
  EXPECT_CALL(*byteEventTracker, addLastByteEvent(_, _))
      .WillOnce(Invoke([] (HTTPTransaction* txn,
                           uint64_t /*byteNo*/) {
                         txn->incrementPendingByteEvents();
                       }));
  sendRequest();
  auto handler2 = addSimpleStrictHandler();
  handler2->expectHeaders();
  handler2->expectEOM([&handler2] () {
      handler2->sendChunkedReplyWithBody(200, 100, 100, false);
    });
  EXPECT_CALL(*byteEventTracker, addLastByteEvent(_, _));
  handler2->expectDetachTransaction();

  sendRequest();
  sendRequest();
  auto handler3 = addSimpleStrictHandler();
  handler3->expectHeaders();
  handler3->expectEOM([&handler3] () {
      handler3->sendChunkedReplyWithBody(200, 100, 100, false);
    });
  EXPECT_CALL(*byteEventTracker, addLastByteEvent(_, _));
  handler3->expectDetachTransaction();
  flushRequestsAndLoop();
  expectResponses(3);
  handler1->expectDetachTransaction();
  handler1->txn_->decrementPendingByteEvents();
  gracefulShutdown();
}

/*
 * The sequence of streams are generated in the following order:
 * - [client --> server] regular request 1st stream (getGetRequest())
 * - [server --> client] respond 1st stream (res, 100 bytes, without EOM)
 * - [server --> client] request 2nd stream (pub, 200 bytes, EOM)
 * - [client --> server] respond 2nd stream (OK, EOM)
 * - [client --> server] EOM on the 1st stream
 */
TEST_F(HTTP2DownstreamSessionTest, ExheaderFromServer) {
  auto cStreamId = HTTPCodec::StreamID(1);
  SetupControlStream(cStreamId);

  // Create a dummy request and a dummy response messages
  auto pub = getGetRequest("/sub/fyi");
  // set up the priority for fun
  pub.setHTTP2Priority(std::make_tuple(0, false, 7));

  InSequence handlerSequence;
  auto cHandler = addSimpleStrictHandler();
  StrictMock<MockHTTPHandler> pubHandler;

  cHandler->expectHeaders([&] {
      cHandler->txn_->pauseIngress();
      // Generate response for the control stream
      cHandler->txn_->sendHeaders(getResponse(200, 0));
      cHandler->txn_->sendBody(makeBuf(100));

      auto* pubTxn = cHandler->txn_->newExTransaction(&pubHandler);
      // Generate a pub request (encapsulated in EX_HEADERS frame)
      pubTxn->sendHeaders(pub);
      pubTxn->sendBody(makeBuf(200));
      pubTxn->sendEOM();
    });

  EXPECT_CALL(pubHandler, setTransaction(_));
  EXPECT_CALL(callbacks_, onMessageBegin(cStreamId, _));
  EXPECT_CALL(callbacks_, onHeadersComplete(cStreamId, _));
  EXPECT_CALL(callbacks_, onExMessageBegin(2, _, _, _));
  EXPECT_CALL(callbacks_, onHeadersComplete(2, _));
  EXPECT_CALL(callbacks_, onMessageComplete(2, _));

  EXPECT_CALL(pubHandler, onHeadersComplete(_));
  EXPECT_CALL(pubHandler, onEOM());
  EXPECT_CALL(pubHandler, detachTransaction());

  EXPECT_CALL(*cHandler, onEOM());
  EXPECT_CALL(*cHandler, detachTransaction());

  transport_->addReadEvent(requests_, milliseconds(0));
  transport_->startReadEvents();

  eventBase_.runAfterDelay([&] {
      parseOutput(*clientCodec_);
      // send a response from client to server
      clientCodec_->generateExHeader(requests_, 2, getResponse(200, 0),
                                     HTTPCodec::ExAttributes(cStreamId, false),
                                     true, nullptr);
      transport_->addReadEvent(requests_, milliseconds(0));
      transport_->startReadEvents();
      parseOutput(*clientCodec_);
      cHandler->txn_->resumeIngress();
      cHandler->txn_->sendEOM();
      transport_->addReadEOF(milliseconds(0));
    }, 100);

  HTTPSession::DestructorGuard g(httpSession_);
  expectDetachSession();
  eventBase_.loop();
}

/*
 * The sequence of streams are generated in the following order:
 * - [client --> server] regular request on control stream 1
 * - [client --> server] Pub request on stream 3
 * - [server --> client] response on stream 1 (OK, )
 * - [server --> client] response on stream 3 (OK, EOM)
 * - [server --> client] response on stream 1 (EOM)
 */
TEST_F(HTTP2DownstreamSessionTest, ExheaderFromClient) {
  auto cStreamId = HTTPCodec::StreamID(1);
  SetupControlStream(cStreamId);

  // generate an EX_HEADERS
  auto exStreamId = cStreamId + 2;
  clientCodec_->generateExHeader(requests_, exStreamId, getGetRequest("/pub"),
                                 HTTPCodec::ExAttributes(cStreamId, false),
                                 true, nullptr);

  auto cHandler = addSimpleStrictHandler();
  cHandler->expectHeaders([&] {
      // send back the response for control stream, but EOM
      cHandler->txn_->sendHeaders(getResponse(200, 0));
    });
  EXPECT_CALL(*cHandler, onEOM());

  StrictMock<MockHTTPHandler> pubHandler;
  EXPECT_CALL(*cHandler, onExTransaction(_))
    .WillOnce(Invoke([&pubHandler] (HTTPTransaction* exTxn) {
          exTxn->setHandler(&pubHandler);
          pubHandler.txn_ = exTxn;
        }));

  InSequence handlerSequence;
  EXPECT_CALL(pubHandler, setTransaction(_));
  pubHandler.expectHeaders([&] {
      // send back the response for the pub request
      pubHandler.txn_->sendHeadersWithEOM(getResponse(200, 0));
    });
  EXPECT_CALL(pubHandler, onEOM());
  EXPECT_CALL(pubHandler, detachTransaction());
  cHandler->expectDetachTransaction();

  EXPECT_CALL(callbacks_, onMessageBegin(cStreamId, _));
  EXPECT_CALL(callbacks_, onHeadersComplete(cStreamId, _));
  EXPECT_CALL(callbacks_, onExMessageBegin(exStreamId, _, _, _));
  EXPECT_CALL(callbacks_, onHeadersComplete(exStreamId, _));
  EXPECT_CALL(callbacks_, onMessageComplete(exStreamId, _));
  EXPECT_CALL(callbacks_, onMessageComplete(cStreamId, _));

  transport_->addReadEvent(requests_, milliseconds(0));
  transport_->startReadEvents();
  transport_->addReadEOF(milliseconds(0));
  eventBase_.loop();

  HTTPSession::DestructorGuard g(httpSession_);
  expectDetachSession();
  cHandler->txn_->sendEOM();
  eventBase_.loop();
  parseOutput(*clientCodec_);
}

/*
 * The sequence of streams are generated in the following order:
 * - [client --> server] regular request 1st stream (getGetRequest())
 * - [server --> client] request 2nd stream (unidirectional)
 * - [server --> client] response + EOM on the 1st stream
 */
TEST_F(HTTP2DownstreamSessionTest, UnidirectionalExTransaction) {
  auto cStreamId = HTTPCodec::StreamID(1);
  SetupControlStream(cStreamId);
  InSequence handlerSequence;
  auto cHandler = addSimpleStrictHandler();
  StrictMock<MockHTTPHandler> uniHandler;

  cHandler->expectHeaders([&] {
      auto* uniTxn = cHandler->txn_->newExTransaction(&uniHandler, true);
      EXPECT_TRUE(uniTxn->isIngressComplete());
      uniTxn->sendHeaders(getGetRequest("/uni"));
      uniTxn->sendEOM();

      // close control stream
      cHandler->txn_->sendHeadersWithEOM(getResponse(200, 0));
    });

  EXPECT_CALL(uniHandler, setTransaction(_));
  EXPECT_CALL(*cHandler, onEOM());
  EXPECT_CALL(uniHandler, detachTransaction());
  EXPECT_CALL(*cHandler, detachTransaction());

  transport_->addReadEvent(requests_, milliseconds(0));
  transport_->startReadEvents();
  eventBase_.runAfterDelay([&] {
      transport_->addReadEOF(milliseconds(0));
    }, 100);

  HTTPSession::DestructorGuard g(httpSession_);
  expectDetachSession();
  eventBase_.loop();
}

TEST_F(HTTP2DownstreamSessionTest, PauseResumeControlStream) {
  auto cStreamId = HTTPCodec::StreamID(1);
  SetupControlStream(cStreamId);

  // generate an EX_HEADERS
  clientCodec_->generateExHeader(requests_, cStreamId + 2, getGetRequest(),
                                 HTTPCodec::ExAttributes(cStreamId, false),
                                 true, nullptr);

  auto cHandler = addSimpleStrictHandler();
  cHandler->expectHeaders([&] {
      cHandler->txn_->pauseIngress();
      // send back the response for control stream, but EOM
      cHandler->txn_->sendHeaders(getResponse(200, 0));
    });
  EXPECT_CALL(*cHandler, onEOM());

  StrictMock<MockHTTPHandler> pubHandler;
  EXPECT_CALL(*cHandler, onExTransaction(_))
    .WillOnce(Invoke([&pubHandler] (HTTPTransaction* exTxn) {
          exTxn->setHandler(&pubHandler);
          pubHandler.txn_ = exTxn;
        }));

  InSequence handlerSequence;
  EXPECT_CALL(pubHandler, setTransaction(_));
  pubHandler.expectHeaders([&] {
      // send back the response for the pub request
      pubHandler.txn_->sendHeadersWithEOM(getResponse(200, 0));
    });
  EXPECT_CALL(pubHandler, onEOM());
  EXPECT_CALL(pubHandler, detachTransaction());
  cHandler->expectDetachTransaction();

  EXPECT_CALL(callbacks_, onMessageBegin(cStreamId, _));
  EXPECT_CALL(callbacks_, onHeadersComplete(cStreamId, _));
  EXPECT_CALL(callbacks_, onHeadersComplete(cStreamId + 2, _));
  EXPECT_CALL(callbacks_, onMessageComplete(cStreamId + 2, _));
  EXPECT_CALL(callbacks_, onMessageComplete(cStreamId, _));

  HTTPSession::DestructorGuard g(httpSession_);
  transport_->addReadEvent(requests_, milliseconds(0));
  transport_->addReadEOF(milliseconds(0));
  transport_->startReadEvents();
  eventBase_.loop();

  cHandler->txn_->resumeIngress();
  cHandler->txn_->sendEOM();
  eventBase_.loop();

  expectDetachSession();
  parseOutput(*clientCodec_);
}

TEST_F(HTTP2DownstreamSessionTest, InvalidControlStream) {
  auto cStreamId = HTTPCodec::StreamID(1);
  SetupControlStream(cStreamId);

  // generate an EX_HEADERS, but with a non-existing control stream
  clientCodec_->generateExHeader(requests_, cStreamId + 2, getGetRequest(),
                                 HTTPCodec::ExAttributes(cStreamId + 4, false),
                                 true, nullptr);

  auto cHandler = addSimpleStrictHandler();
  InSequence handlerSequence;
  cHandler->expectHeaders([&] {
      // send back the response for control stream, but EOM
      cHandler->txn_->sendHeaders(getResponse(200, 0));
    });
  EXPECT_CALL(*cHandler, onExTransaction(_)).Times(0);
  EXPECT_CALL(*cHandler, onEOM());
  cHandler->expectDetachTransaction();

  EXPECT_CALL(callbacks_, onMessageBegin(cStreamId, _));
  EXPECT_CALL(callbacks_, onHeadersComplete(cStreamId, _));
  EXPECT_CALL(callbacks_, onAbort(cStreamId + 2, _));

  HTTPSession::DestructorGuard g(httpSession_);
  transport_->addReadEvent(requests_, milliseconds(0));
  transport_->addReadEOF(milliseconds(0));
  transport_->startReadEvents();
  eventBase_.loop();

  cHandler->txn_->sendEOM();
  eventBase_.loop();

  expectDetachSession();
  parseOutput(*clientCodec_);
}

TEST_F(HTTP2DownstreamSessionTest, SetByteEventTracker) {
  InSequence enforceOrder;

  // Send two requests with writes paused, which will queue several byte events,
  // including last byte events which are holding a reference to the
  // transaction.
  transport_->pauseWrites();
  auto handler1 = addSimpleStrictHandler();
  handler1->expectHeaders();
  handler1->expectEOM([&handler1] () {
      handler1->sendReplyWithBody(200, 100);
    });
  auto handler2 = addSimpleStrictHandler();
  handler2->expectHeaders();
  handler2->expectEOM([&handler2] () {
      handler2->sendReplyWithBody(200, 100);
    });

  sendRequest();
  sendRequest();
  // Resume writes from the loop callback
  eventBase_.runInLoop([this] {
      transport_->resumeWrites();
    });

  // Graceful shutdown will notify of GOAWAY
  EXPECT_CALL(*handler1, onGoaway(ErrorCode::NO_ERROR));
  EXPECT_CALL(*handler2, onGoaway(ErrorCode::NO_ERROR));
  // The original byteEventTracker will process the last byte event of the
  // first transaction, and detach by deleting the event.  Swap out the tracker.
  handler1->expectDetachTransaction([this] {
      auto tracker = std::make_unique<ByteEventTracker>(httpSession_);
      httpSession_->setByteEventTracker(std::move(tracker));
    });
  // handler2 should also be detached immediately because the new
  // ByteEventTracker continues procesing where the old one left off.
  handler2->expectDetachTransaction();
  gracefulShutdown();
}

TEST_F(HTTPDownstreamSessionTest, TestTrackedByteEventTracker) {
  auto byteEventTracker = setMockByteEventTracker();
  InSequence enforceOrder;

  auto handler1 = addSimpleStrictHandler();
  size_t bytesToSend = 200;
  size_t expectedTrackedByteOffset = bytesToSend + 99;
  handler1->expectHeaders();
  handler1->expectEOM([&handler1, &bytesToSend] () {
    handler1->sendHeaders(200, 200);
    handler1->sendBodyWithLastByteTracking(bytesToSend);
    handler1->txn_->sendEOM();
    });

  EXPECT_CALL(*byteEventTracker,
    addTrackedByteEvent(_, expectedTrackedByteOffset))
      .WillOnce(Invoke([] (HTTPTransaction* txn,
                           uint64_t /*byteNo*/) {
                         txn->incrementPendingByteEvents();
                       }));
  sendRequest();
  flushRequestsAndLoop();
  handler1->expectDetachTransaction();
  handler1->txn_->decrementPendingByteEvents();
  gracefulShutdown();
}

TEST_F(HTTP2DownstreamSessionTest, Trailers) {
  InSequence enforceOrder;

  auto handler = addSimpleStrictHandler();
  handler->expectHeaders();
  handler->expectEOM([&handler]() {
    handler->sendReplyWithBody(
        200, 100, true /* keepalive */, true /* sendEOM */, true /*trailers*/);
  });
  handler->expectDetachTransaction();

  HTTPSession::DestructorGuard g(httpSession_);
  sendRequest();
  flushRequestsAndLoop(true, milliseconds(0));

  EXPECT_CALL(callbacks_, onMessageBegin(1, _)).Times(1);
  EXPECT_CALL(callbacks_, onHeadersComplete(1, _)).Times(1);
  EXPECT_CALL(callbacks_, onBody(1, _, _));
  EXPECT_CALL(callbacks_, onTrailersComplete(1, _));
  EXPECT_CALL(callbacks_, onMessageComplete(1, _));

  parseOutput(*clientCodec_);
  expectDetachSession();
}

TEST_F(HTTPDownstreamSessionTest, Trailers) {
  testChunks(true);
}

TEST_F(HTTPDownstreamSessionTest, ExplicitChunks) {
  testChunks(false);
}

template <class C>
void HTTPDownstreamTest<C>::testChunks(bool trailers) {
  InSequence enforceOrder;

  auto handler = addSimpleStrictHandler();
  handler->expectHeaders();
  handler->expectEOM([&handler, trailers] () {
      handler->sendChunkedReplyWithBody(200, 100, 17, trailers);
    });
  handler->expectDetachTransaction();

  HTTPSession::DestructorGuard g(httpSession_);
  sendRequest();
  flushRequestsAndLoop(true, milliseconds(0));

  EXPECT_CALL(callbacks_, onMessageBegin(1, _))
    .Times(1);
  EXPECT_CALL(callbacks_, onHeadersComplete(1, _))
    .Times(1);
  for (int i = 0; i < 6; i++) {
    EXPECT_CALL(callbacks_, onChunkHeader(1, _));
    EXPECT_CALL(callbacks_, onBody(1, _, _));
    EXPECT_CALL(callbacks_, onChunkComplete(1));
  }
  if (trailers) {
    EXPECT_CALL(callbacks_, onTrailersComplete(1, _));
  }
  EXPECT_CALL(callbacks_, onMessageComplete(1, _));

  parseOutput(*clientCodec_);
  expectDetachSession();
}

TEST_F(HTTPDownstreamSessionTest, HttpDrain) {
  InSequence enforceOrder;

  auto handler1 = addSimpleStrictHandler();
  handler1->expectHeaders([this, &handler1] {
      handler1->sendHeaders(200, 100);
      httpSession_->notifyPendingShutdown();
    });
  handler1->expectEOM([&handler1] {
      handler1->sendBody(100);
      handler1->txn_->sendEOM();
    });
  handler1->expectDetachTransaction();

  auto handler2 = addSimpleStrictHandler();
  handler2->expectHeaders([&handler2] {
      handler2->sendHeaders(200, 100);
    });
  handler2->expectEOM([&handler2] {
          handler2->sendBody(100);
          handler2->txn_->sendEOM();
    });
  handler2->expectDetachTransaction();

  expectDetachSession();

  sendRequest();
  sendRequest();
  flushRequestsAndLoop();
}

// 1) receive full request
// 2) notify pending shutdown
// 3) wait for session read timeout -> should be ignored
// 4) response completed
TEST_F(HTTPDownstreamSessionTest, HttpDrainLongRunning) {
  InSequence enforceSequence;

  auto handler = addSimpleStrictHandler();
  handler->expectHeaders([this, &handler] {
      httpSession_->notifyPendingShutdown();
      eventBase_.tryRunAfterDelay([this] {
          // simulate read timeout
          httpSession_->timeoutExpired();
        }, 100);
      eventBase_.tryRunAfterDelay([&handler] {
          handler->sendReplyWithBody(200, 100);
        }, 200);
    });
  handler->expectEOM();
  handler->expectDetachTransaction();

  expectDetachSession();

  sendRequest();
  flushRequestsAndLoop();
}

TEST_F(HTTPDownstreamSessionTest, EarlyAbort) {
  StrictMock<MockHTTPHandler> handler;

  InSequence enforceOrder;
  EXPECT_CALL(mockController_, getRequestHandler(_, _))
    .WillOnce(Return(&handler));

  EXPECT_CALL(handler, setTransaction(_))
    .WillOnce(Invoke([&] (HTTPTransaction* txn) {
          handler.txn_ = txn;
          handler.txn_->sendAbort();
        }));
  handler.expectDetachTransaction();
  expectDetachSession();

  addSingleByteReads("GET /somepath.php?param=foo HTTP/1.1\r\n"
                     "Host: example.com\r\n"
                     "Connection: close\r\n"
                     "\r\n");
  transport_->addReadEOF(milliseconds(0));
  transport_->startReadEvents();
  eventBase_.loop();
}

TEST_F(SPDY3DownstreamSessionTest, HttpPausedBuffered) {
  IOBufQueue rst{IOBufQueue::cacheChainLength()};
  auto s = sendRequest();
  clientCodec_->generateRstStream(rst, s, ErrorCode::CANCEL);
  sendRequest();

  InSequence handlerSequence;
  auto handler1 = addSimpleNiceHandler();
  handler1->expectHeaders();
  handler1->expectEOM([&handler1, this] {
      transport_->pauseWrites();
      handler1->sendHeaders(200, 65536 * 2);
      handler1->sendBody(65536 * 2);
    });
  handler1->expectEgressPaused();
  auto handler2 = addSimpleNiceHandler();
  handler2->expectEgressPaused();
  handler2->expectHeaders();
  handler2->expectEOM([&] {
      eventBase_.runInLoop([&] {
          transport_->addReadEvent(rst, milliseconds(0)); });
    });
  handler1->expectError([&] (const HTTPException& ex) {
      ASSERT_EQ(ex.getProxygenError(), kErrorStreamAbort);
      resumeWritesInLoop();
    });
  handler1->expectDetachTransaction();
  handler2->expectEgressResumed([&] {
      handler2->sendReplyWithBody(200, 32768);
    });
  handler2->expectDetachTransaction([this] {
      eventBase_.runInLoop([&] { transport_->addReadEOF(milliseconds(0)); });
    });
  expectDetachSession();

  flushRequestsAndLoop();
}

TEST_F(HTTPDownstreamSessionTest, HttpWritesDrainingTimeout) {
  sendRequest();
  sendHeader();

  InSequence handlerSequence;
  auto handler1 = addSimpleNiceHandler();
  handler1->expectHeaders();
  handler1->expectEOM([&handler1, this] {
      transport_->pauseWrites();
      handler1->sendHeaders(200, 1000);
    });
  handler1->expectError([&] (const HTTPException& ex) {
      ASSERT_EQ(ex.getProxygenError(), kErrorWriteTimeout);
      ASSERT_EQ(
        folly::to<std::string>("WriteTimeout on transaction id: ",
                               handler1->txn_->getID()),
        std::string(ex.what()));
      handler1->txn_->sendAbort();
    });
  handler1->expectDetachTransaction();
  expectDetachSession();

  flushRequestsAndLoop();
}

TEST_F(HTTPDownstreamSessionTest, HttpRateLimitNormal) {
  // The rate-limiting code grabs the event base from the EventBaseManager,
  // so we need to set it.
  folly::EventBaseManager::get()->setEventBase(&eventBase_, false);

  // Create a request
  sendRequest();

  InSequence handlerSequence;

  // Set a low rate-limit on the transaction
  auto handler1 = addSimpleNiceHandler();
  handler1->expectHeaders([&] {
      uint32_t rateLimit_kbps = 640;
      handler1->txn_->setEgressRateLimit(rateLimit_kbps * 1024);
    });
  // Send a somewhat big response that we know will get rate-limited
  handler1->expectEOM([&handler1] {
      // At 640kbps, this should take slightly over 800ms
      uint32_t rspLengthBytes = 100000;
      handler1->sendHeaders(200, rspLengthBytes);
      handler1->sendBody(rspLengthBytes);
      handler1->txn_->sendEOM();
    });
  handler1->expectDetachTransaction();

  // Keep the session around even after the event base loop completes so we can
  // read the counters on a valid object.
  HTTPSession::DestructorGuard g(httpSession_);
  flushRequestsAndLoop();

  proxygen::TimePoint timeFirstWrite =
    transport_->getWriteEvents()->front()->getTime();
  proxygen::TimePoint timeLastWrite =
    transport_->getWriteEvents()->back()->getTime();
  int64_t writeDuration =
    (int64_t)millisecondsBetween(timeLastWrite, timeFirstWrite).count();
  EXPECT_GE(writeDuration, 800);

  cleanup();
}

TEST_F(SPDY3DownstreamSessionTest, SpdyRateLimitNormal) {
  // The rate-limiting code grabs the event base from the EventBaseManager,
  // so we need to set it.
  folly::EventBaseManager::get()->setEventBase(&eventBase_, false);

  clientCodec_->getEgressSettings()->setSetting(SettingsId::INITIAL_WINDOW_SIZE,
                                                100000);
  clientCodec_->generateSettings(requests_);
  sendRequest();

  InSequence handlerSequence;
  auto handler1 = addSimpleNiceHandler();
  handler1->expectHeaders([&] {
      uint32_t rateLimit_kbps = 640;
      handler1->txn_->setEgressRateLimit(rateLimit_kbps * 1024);
    });

  handler1->expectEOM([&handler1] {
      // At 640kbps, this should take slightly over 800ms
      uint32_t rspLengthBytes = 100000;
      handler1->sendHeaders(200, rspLengthBytes);
      handler1->sendBody(rspLengthBytes);
      handler1->txn_->sendEOM();
    });
  handler1->expectDetachTransaction();

  // Keep the session around even after the event base loop completes so we can
  // read the counters on a valid object.
  HTTPSession::DestructorGuard g(httpSession_);
  flushRequestsAndLoop(true, milliseconds(50));

  proxygen::TimePoint timeFirstWrite =
    transport_->getWriteEvents()->front()->getTime();
  proxygen::TimePoint timeLastWrite =
    transport_->getWriteEvents()->back()->getTime();
  int64_t writeDuration =
    (int64_t)millisecondsBetween(timeLastWrite, timeFirstWrite).count();
  EXPECT_GE(writeDuration, 800);
  expectDetachSession();
}

TEST_F(SPDY3DownstreamSessionTest, SpdyRateLimitRst) {
  // The rate-limiting code grabs the event base from the EventBaseManager,
  // so we need to set it.
  folly::EventBaseManager::get()->setEventBase(&eventBase_, false);

  IOBufQueue rst{IOBufQueue::cacheChainLength()};
  clientCodec_->getEgressSettings()->setSetting(SettingsId::INITIAL_WINDOW_SIZE,
                                                100000);
  clientCodec_->generateSettings(requests_);
  auto streamID = sendRequest();
  clientCodec_->generateRstStream(rst, streamID, ErrorCode::CANCEL);

  InSequence handlerSequence;
  auto handler1 = addSimpleNiceHandler();
  handler1->expectHeaders([&] {
      uint32_t rateLimit_kbps = 640;
      handler1->txn_->setEgressRateLimit(rateLimit_kbps * 1024);
    });
  handler1->expectEOM([&handler1] {
      uint32_t rspLengthBytes = 100000;
      handler1->sendHeaders(200, rspLengthBytes);
      handler1->sendBody(rspLengthBytes);
      handler1->txn_->sendEOM();
    });
  handler1->expectError();
  handler1->expectDetachTransaction();
  expectDetachSession();

  flushRequestsAndLoop(true, milliseconds(50), milliseconds(0), [&] {
      transport_->addReadEvent(rst, milliseconds(10));
    });
}

// Send a 1.0 request, egress the EOM with the last body chunk on a paused
// socket, and let it timeout.  dropConnection()
// to removeTransaction with writesDraining_=true
TEST_F(HTTPDownstreamSessionTest, WriteTimeout) {
  HTTPMessage req = getGetRequest();
  req.setHTTPVersion(1, 0);
  sendRequest(req);

  InSequence handlerSequence;
  auto handler1 = addSimpleNiceHandler();
  handler1->expectHeaders();
  handler1->expectEOM([&handler1, this] {
      handler1->sendHeaders(200, 100);
      eventBase_.tryRunAfterDelay([&handler1, this] {
          transport_->pauseWrites();
          handler1->sendBody(100);
          handler1->txn_->sendEOM();
        }, 50);
    });
  handler1->expectError([&] (const HTTPException& ex) {
      ASSERT_EQ(ex.getProxygenError(), kErrorWriteTimeout);
      ASSERT_EQ(folly::to<std::string>("WriteTimeout on transaction id: ",
                                       handler1->txn_->getID()),
                std::string(ex.what()));
    });
  handler1->expectDetachTransaction();

  expectDetachSession();

  flushRequestsAndLoop();
}

// Send an abort from the write timeout path while pipelining
TEST_F(HTTPDownstreamSessionTest, WriteTimeoutPipeline) {
  const char* buf = "GET / HTTP/1.1\r\nHost: localhost\r\n\r\n"
    "GET / HTTP/1.1\r\nHost: localhost\r\n\r\n";
  requests_.append(buf, strlen(buf));

  InSequence handlerSequence;
  auto handler1 = addSimpleNiceHandler();
  handler1->expectHeaders();
  handler1->expectEOM([&handler1, this] {
      handler1->sendHeaders(200, 100);
      eventBase_.tryRunAfterDelay([&handler1, this] {
          transport_->pauseWrites();
          handler1->sendBody(100);
          handler1->txn_->sendEOM();
        }, 50);
    });
  auto handler2 = addSimpleNiceHandler();
  handler2->expectHeaders();
  handler2->expectEOM();
  handler1->expectError([&] (const HTTPException& ex) {
      ASSERT_EQ(ex.getProxygenError(), kErrorWriteTimeout);
      ASSERT_EQ(folly::to<std::string>("WriteTimeout on transaction id: ",
                                       handler1->txn_->getID()),
                std::string(ex.what()));
      handler1->txn_->sendAbort();
    });
  handler2->expectError([&] (const HTTPException& ex) {
      ASSERT_EQ(ex.getProxygenError(), kErrorWriteTimeout);
      ASSERT_EQ(folly::to<std::string>("WriteTimeout on transaction id: ",
                                       handler2->txn_->getID()),
                std::string(ex.what()));
      handler2->txn_->sendAbort();
    });
  handler2->expectDetachTransaction();
  handler1->expectDetachTransaction();
  expectDetachSession();

  flushRequestsAndLoop();
}

TEST_F(HTTPDownstreamSessionTest, BodyPacketization) {
  HTTPMessage req = getGetRequest();
  req.setHTTPVersion(1, 0);
  req.setWantsKeepalive(false);
  sendRequest(req);

  InSequence handlerSequence;
  auto handler1 = addSimpleNiceHandler();
  handler1->expectHeaders();
  handler1->expectEOM([&handler1] {
      handler1->sendReplyWithBody(200, 32768);
    });
  handler1->expectDetachTransaction();

  expectDetachSession();

  // Keep the session around even after the event base loop completes so we can
  // read the counters on a valid object.
  HTTPSession::DestructorGuard g(httpSession_);
  flushRequestsAndLoop();

  EXPECT_EQ(transport_->getWriteEvents()->size(), 1);
}

TEST_F(HTTPDownstreamSessionTest, HttpMalformedPkt1) {
  // Create a HTTP connection and keep sending just '\n' to the HTTP1xCodec.
  std::string data(90000, '\n');
  requests_.append(data.data(), data.length());

  expectDetachSession();

  flushRequestsAndLoop(true, milliseconds(0));
}

TEST_F(HTTPDownstreamSessionTest, BigExplcitChunkWrite) {
  // even when the handler does a massive write, the transport only gets small
  // writes
  sendRequest();

  auto handler = addSimpleNiceHandler();
  handler->expectHeaders([&handler] {
      handler->sendHeaders(200, 100, false);
      size_t len = 16 * 1024 * 1024;
      handler->txn_->sendChunkHeader(len);
      auto chunk = makeBuf(len);
      handler->txn_->sendBody(std::move(chunk));
      handler->txn_->sendChunkTerminator();
      handler->txn_->sendEOM();
    });
  handler->expectDetachTransaction();

  expectDetachSession();

  // Keep the session around even after the event base loop completes so we can
  // read the counters on a valid object.
  HTTPSession::DestructorGuard g(httpSession_);
  flushRequestsAndLoop();

  EXPECT_GT(transport_->getWriteEvents()->size(), 250);
}


// ==== upgrade tests ====

// Test upgrade to a protocol unknown to HTTPSession
TEST_F(HTTPDownstreamSessionTest, HttpUpgradeNonNative) {
  auto handler = addSimpleStrictHandler();

  handler->expectHeaders([&handler] {
      handler->sendHeaders(101, 0, true, {{"Upgrade", "blarf"}});
    });
  EXPECT_CALL(*handler, onUpgrade(UpgradeProtocol::TCP));
  handler->expectEOM([&handler] {
      handler->txn_->sendEOM();
    });
  handler->expectDetachTransaction();

  sendRequest(getUpgradeRequest("blarf"));
  expectDetachSession();
  flushRequestsAndLoop(true);
}

// Test upgrade to a protocol unknown to HTTPSession, but don't switch
// protocols
TEST_F(HTTPDownstreamSessionTest, HttpUpgradeNonNativeIgnore) {
  auto handler = addSimpleStrictHandler();

  handler->expectHeaders([&handler] {
      handler->sendReplyWithBody(200, 100);
    });
  handler->expectEOM();
  handler->expectDetachTransaction();

  sendRequest(getUpgradeRequest("blarf"));

  expectDetachSession();
  flushRequestsAndLoop(true);
}


// Test upgrade to a protocol unknown to HTTPSession
TEST_F(HTTPDownstreamSessionTest, HttpUpgradeNonNativePipeline) {
  auto handler1 = addSimpleStrictHandler();

  handler1->expectHeaders([&handler1] (std::shared_ptr<HTTPMessage> msg) {
      EXPECT_EQ(msg->getHeaders().getSingleOrEmpty(HTTP_HEADER_UPGRADE),
                "blarf");
      handler1->sendReplyWithBody(200, 100);
    });
  handler1->expectEOM();
  handler1->expectDetachTransaction();

  auto handler2 = addSimpleStrictHandler();
  handler2->expectHeaders([&handler2] {
      handler2->sendReplyWithBody(200, 100);
    });
  handler2->expectEOM();
  handler2->expectDetachTransaction();

  sendRequest(getUpgradeRequest("blarf"));
  transport_->addReadEvent("GET / HTTP/1.1\r\n"
                           "\r\n");
  expectDetachSession();
  flushRequestsAndLoop(true);
}

// Helper that does a simple upgrade test - request an upgrade, receive a 101
// and an upgraded response
template <class C>
void HTTPDownstreamTest<C>::testSimpleUpgrade(
  const std::string& upgradeHeader,
  CodecProtocol expectedProtocol,
  const std::string& expectedUpgradeHeader) {
  this->rawCodec_->setAllowedUpgradeProtocols({expectedUpgradeHeader});

  auto handler = addSimpleStrictHandler();

  HeaderIndexingStrategy testH2IndexingStrat;
  handler->expectHeaders();
  EXPECT_CALL(mockController_, onSessionCodecChange(httpSession_));
  handler->expectEOM(
    [&handler, expectedProtocol, expectedUpgradeHeader, &testH2IndexingStrat] {
      EXPECT_FALSE(handler->txn_->getSetupTransportInfo().secure);
      EXPECT_EQ(*handler->txn_->getSetupTransportInfo().appProtocol,
                expectedUpgradeHeader);
      if (expectedProtocol == CodecProtocol::HTTP_2) {
        const HTTP2Codec* codec = dynamic_cast<const HTTP2Codec*>(
          &handler->txn_->getTransport().getCodec());
        ASSERT_NE(codec, nullptr);
        EXPECT_EQ(codec->getHeaderIndexingStrategy(), &testH2IndexingStrat);
      }
      handler->sendReplyWithBody(200, 100);
    });
  handler->expectDetachTransaction();

  if (expectedProtocol == CodecProtocol::HTTP_2) {
    EXPECT_CALL(mockController_, getHeaderIndexingStrategy())
      .WillOnce(
        Return(&testH2IndexingStrat)
    );
  }

  HTTPMessage req = getUpgradeRequest(upgradeHeader);
  if (upgradeHeader == http2::kProtocolCleartextString) {
    HTTP2Codec::requestUpgrade(req);
  }
  sendRequest(req);
  flushRequestsAndLoop();

  expect101(expectedProtocol, expectedUpgradeHeader);
  expectResponse();

  gracefulShutdown();
}

// Upgrade to SPDY/3
TEST_F(HTTPDownstreamSessionTest, HttpUpgradeNative3) {
  testSimpleUpgrade("spdy/3", CodecProtocol::SPDY_3, "spdy/3");
}

// Upgrade to SPDY/3.1
TEST_F(HTTPDownstreamSessionTest, HttpUpgradeNative31) {
  testSimpleUpgrade("spdy/3.1", CodecProtocol::SPDY_3_1, "spdy/3.1");
}

// Upgrade to HTTP/2
TEST_F(HTTPDownstreamSessionTest, HttpUpgradeNativeH2) {
  testSimpleUpgrade("h2c", CodecProtocol::HTTP_2, "h2c");
}

class HTTPDownstreamSessionUpgradeFlowControlTest :
      public HTTPDownstreamSessionTest {
 public:
  HTTPDownstreamSessionUpgradeFlowControlTest()
      : HTTPDownstreamSessionTest({100000, 105000, 110000}) {}
};

// Upgrade to HTTP/2, with non-default flow control settings
TEST_F(HTTPDownstreamSessionUpgradeFlowControlTest, UpgradeH2Flowcontrol) {
  testSimpleUpgrade("h2c", CodecProtocol::HTTP_2, "h2c");
}

// Upgrade to SPDY/3.1 with a non-native proto in the list
TEST_F(HTTPDownstreamSessionTest, HttpUpgradeNativeUnknown) {
  // This is maybe weird, the client asked for non-native as first choice,
  // but we go native
  testSimpleUpgrade("blarf, spdy/3.1, spdy/3",
                    CodecProtocol::SPDY_3_1, "spdy/3.1");
}

// Upgrade header with extra whitespace
TEST_F(HTTPDownstreamSessionTest, HttpUpgradeNativeWhitespace) {
  testSimpleUpgrade(" \tspdy/3.1\t , spdy/3",
                    CodecProtocol::SPDY_3_1, "spdy/3.1");
}

// Upgrade header with random junk
TEST_F(HTTPDownstreamSessionTest, HttpUpgradeNativeJunk) {
  testSimpleUpgrade(",,,,   ,,\t~^%$(*&@(@$^^*(,spdy/3",
                    CodecProtocol::SPDY_3, "spdy/3");
}

// Attempt to upgrade on second txn
TEST_F(HTTPDownstreamSessionTest, HttpUpgradeNativeTxn2) {
  this->rawCodec_->setAllowedUpgradeProtocols({"spdy/3"});
  auto handler1 = addSimpleStrictHandler();
  handler1->expectHeaders();
  handler1->expectEOM([&handler1] {
      handler1->sendReplyWithBody(200, 100);
    });
  handler1->expectDetachTransaction();
  sendRequest(getGetRequest());
  flushRequestsAndLoop();
  expectResponse();

  auto handler2 = addSimpleStrictHandler();
  handler2->expectHeaders();
  handler2->expectEOM([&handler2] {
      handler2->sendReplyWithBody(200, 100);
    });
  handler2->expectDetachTransaction();

  sendRequest(getUpgradeRequest("spdy/3"));
  flushRequestsAndLoop();
  expectResponse();
  gracefulShutdown();
}

// Upgrade on POST
TEST_F(HTTPDownstreamSessionTest, HttpUpgradeNativePost) {
  this->rawCodec_->setAllowedUpgradeProtocols({"spdy/3"});
  auto handler = addSimpleStrictHandler();
  handler->expectHeaders();
  handler->expectBody();
  EXPECT_CALL(mockController_, onSessionCodecChange(httpSession_));
  handler->expectEOM([&handler] {
      handler->sendReplyWithBody(200, 100);
    });
  handler->expectDetachTransaction();

  HTTPMessage req = getUpgradeRequest("spdy/3", HTTPMethod::POST, 10);
  auto streamID = sendRequest(req, false);
  clientCodec_->generateBody(requests_, streamID, makeBuf(10),
                             HTTPCodec::NoPadding, true);
  // cheat and not sending EOM, it's a no-op
  flushRequestsAndLoop();
  expect101(CodecProtocol::SPDY_3, "spdy/3");
  expectResponse();
  gracefulShutdown();
}

// Upgrade on POST with a reply that comes before EOM, don't switch protocols
TEST_F(HTTPDownstreamSessionTest, HttpUpgradeNativePostEarlyResp) {
  this->rawCodec_->setAllowedUpgradeProtocols({"spdy/3"});
  auto handler = addSimpleStrictHandler();
  handler->expectHeaders([&handler] {
      handler->sendReplyWithBody(200, 100);
    });
  handler->expectBody();
  handler->expectEOM();
  handler->expectDetachTransaction();

  HTTPMessage req = getUpgradeRequest("spdy/3", HTTPMethod::POST, 10);
  auto streamID = sendRequest(req, false);
  clientCodec_->generateBody(requests_, streamID, makeBuf(10),
                             HTTPCodec::NoPadding, true);
  flushRequestsAndLoop();
  expectResponse();
  gracefulShutdown();
}

TEST_F(HTTPDownstreamSessionTest, HttpUpgradeNativePostEarlyPartialResp) {
  this->rawCodec_->setAllowedUpgradeProtocols({"spdy/3"});
  auto handler = addSimpleStrictHandler();
  handler->expectHeaders([&handler] {
      handler->sendHeaders(200, 100);
    });
  handler->expectBody();
  handler->expectEOM([&handler] {
      handler->sendBody(100);
      handler->txn_->sendEOM();
    });
  handler->expectDetachTransaction();

  HTTPMessage req = getUpgradeRequest("spdy/3", HTTPMethod::POST, 10);
  auto streamID = sendRequest(req, false);
  clientCodec_->generateBody(requests_, streamID, makeBuf(10),
                             HTTPCodec::NoPadding, true);
  flushRequestsAndLoop();
  expectResponse();
  gracefulShutdown();
}

// Upgrade but with a pipelined HTTP request.  It is parsed as SPDY and
// rejected
TEST_F(HTTPDownstreamSessionTest, HttpUpgradeNativeExtra) {
  this->rawCodec_->setAllowedUpgradeProtocols({"spdy/3"});
  auto handler = addSimpleStrictHandler();
  handler->expectHeaders();
  EXPECT_CALL(mockController_, onSessionCodecChange(httpSession_));
  handler->expectEOM([&handler] {
      handler->sendReplyWithBody(200, 100);
    });
  handler->expectDetachTransaction();

  sendRequest(getUpgradeRequest("spdy/3"));
  // It's a fatal to send this out on the HTTP1xCodec, so hack it manually
  transport_->addReadEvent("GET / HTTP/1.1\r\n"
                           "Upgrade: spdy/3\r\n"
                           "\r\n");
  flushRequestsAndLoop();
  expect101(CodecProtocol::SPDY_3, "spdy/3");
  expectResponse(200, ErrorCode::_SPDY_INVALID_STREAM);
  gracefulShutdown();
}

// Upgrade on POST with Expect: 100-Continue.  If the 100 goes out
// before the EOM is parsed, the 100 will be in HTTP.  This should be the normal
// case since the client *should* wait a bit for the 100 continue to come back
// before sending the POST.  But if the 101 is delayed beyond EOM, the 101
// will come via SPDY.
TEST_F(HTTPDownstreamSessionTest, HttpUpgradeNativePost100) {
  this->rawCodec_->setAllowedUpgradeProtocols({"spdy/3"});
  auto handler = addSimpleStrictHandler();
  handler->expectHeaders([&handler] {
      handler->sendHeaders(100, 0);
    });
  handler->expectBody();
  EXPECT_CALL(mockController_, onSessionCodecChange(httpSession_));
  handler->expectEOM([&handler] {
      handler->sendReplyWithBody(200, 100);
    });
  handler->expectDetachTransaction();

  HTTPMessage req = getUpgradeRequest("spdy/3", HTTPMethod::POST, 10);
  req.getHeaders().add(HTTP_HEADER_EXPECT, "100-continue");
  auto streamID = sendRequest(req, false);
  clientCodec_->generateBody(requests_, streamID, makeBuf(10),
                             HTTPCodec::NoPadding, true);
  flushRequestsAndLoop();
  expect101(CodecProtocol::SPDY_3, "spdy/3", true /* expect 100 continue */);
  expectResponse();
  gracefulShutdown();
}

TEST_F(HTTPDownstreamSessionTest, HttpUpgradeNativePost100Late) {
  this->rawCodec_->setAllowedUpgradeProtocols({"spdy/3"});
  auto handler = addSimpleStrictHandler();
  handler->expectHeaders();
  handler->expectBody();
  EXPECT_CALL(mockController_, onSessionCodecChange(httpSession_));
  handler->expectEOM([&handler] {
      handler->sendHeaders(100, 0);
      handler->sendReplyWithBody(200, 100);
    });
  handler->expectDetachTransaction();

  HTTPMessage req = getUpgradeRequest("spdy/3", HTTPMethod::POST, 10);
  req.getHeaders().add(HTTP_HEADER_EXPECT, "100-continue");
  auto streamID = sendRequest(req, false);
  clientCodec_->generateBody(requests_, streamID, makeBuf(10),
                             HTTPCodec::NoPadding, true);
  flushRequestsAndLoop();
  expect101(CodecProtocol::SPDY_3, "spdy/3");
  expectResponse(200, ErrorCode::NO_ERROR, true /* expect 100 via SPDY */);
  gracefulShutdown();
}


TEST_F(SPDY3DownstreamSessionTest, SpdyPrio) {
  testPriorities(8);

  cleanup();
}

// Test sending a GOAWAY while the downstream session is still processing
// the request that was an upgrade.  The reply GOAWAY should have last good
// stream = 1, not 0.
TEST_F(HTTPDownstreamSessionTest, HttpUpgradeGoawayDrain) {
  this->rawCodec_->setAllowedUpgradeProtocols({"h2c"});
  auto handler = addSimpleStrictHandler();
  handler->expectHeaders();
  handler->expectBody();
  EXPECT_CALL(mockController_, onSessionCodecChange(httpSession_));
  handler->expectEOM();
  handler->expectGoaway();
  handler->expectDetachTransaction();

  EXPECT_CALL(mockController_, getHeaderIndexingStrategy())
    .WillOnce(
      Return(&testH2IndexingStrat_)
  );

  HTTPMessage req = getUpgradeRequest("h2c", HTTPMethod::POST, 10);
  HTTP2Codec::requestUpgrade(req);
  auto streamID = sendRequest(req, false);
  clientCodec_->generateBody(requests_, streamID, makeBuf(10),
                             HTTPCodec::NoPadding, true);
  // cheat and not sending EOM, it's a no-op

  flushRequestsAndLoop();
  expect101(CodecProtocol::HTTP_2, "h2c");
  clientCodec_->generateConnectionPreface(requests_);
  clientCodec_->generateGoaway(requests_, 0, ErrorCode::NO_ERROR);
  flushRequestsAndLoop();
  eventBase_.runInLoop([&handler] {
      handler->sendReplyWithBody(200, 100);
    });
  HTTPSession::DestructorGuard g(httpSession_);
  eventBase_.loop();
  expectResponse(200, ErrorCode::NO_ERROR, false, true);
  expectDetachSession();
}

template <class C>
void HTTPDownstreamTest<C>::testPriorities(uint32_t numPriorities) {
  uint32_t iterations = 10;
  uint32_t maxPriority = numPriorities - 1;
  std::vector<std::unique_ptr<testing::NiceMock<MockHTTPHandler>>> handlers;
  for (int pri = numPriorities - 1; pri >= 0; pri--) {
    for (uint32_t i = 0; i < iterations; i++) {
      sendRequest("/", pri * (8 / numPriorities));
      InSequence handlerSequence;
      auto handler = addSimpleNiceHandler();
      auto rawHandler = handler.get();
      handlers.push_back(std::move(handler));
      rawHandler->expectHeaders();
      rawHandler->expectEOM([rawHandler] {
          rawHandler->sendReplyWithBody(200, 1000);
        });
      rawHandler->expectDetachTransaction([] {  });
    }
  }

  auto buf = requests_.move();
  buf->coalesce();
  requests_.append(std::move(buf));

  flushRequestsAndLoop();

  std::list<HTTPCodec::StreamID> streams;
  EXPECT_CALL(callbacks_, onMessageBegin(_, _))
    .Times(iterations * numPriorities);
  EXPECT_CALL(callbacks_, onHeadersComplete(_, _))
    .Times(iterations * numPriorities);
  // body is variable and hence ignored
  EXPECT_CALL(callbacks_, onMessageComplete(_, _))
      .Times(iterations * numPriorities)
      .WillRepeatedly(Invoke([&](HTTPCodec::StreamID stream, bool /*upgrade*/) {
        streams.push_back(stream);
      }));

  parseOutput(*clientCodec_);

  // transactions finish in priority order (higher streamIDs first)
  EXPECT_EQ(streams.size(), iterations * numPriorities);
  auto txn = streams.begin();
  for (int band = maxPriority; band >= 0; band--) {
    auto upperID = iterations * 2 * (band + 1);
    auto lowerID = iterations * 2 * band;
    for (uint32_t i = 0; i < iterations; i++) {
      EXPECT_LE(lowerID, (uint32_t)*txn);
      EXPECT_GE(upperID, (uint32_t)*txn);
      ++txn;
    }
  }
}

// Verifies that the read timeout is not running when no ingress is expected/
// required to proceed
TEST_F(SPDY3DownstreamSessionTest, SpdyTimeout) {
  sendRequest();
  sendRequest();

  httpSession_->setWriteBufferLimit(512);

  InSequence handlerSequence;
  auto handler1 = addSimpleStrictHandler();
  handler1->expectHeaders([this] { transport_->pauseWrites(); });
  handler1->expectEOM([&] {
      handler1->sendHeaders(200, 1000);
      handler1->sendBody(1000);
    });
  handler1->expectEgressPaused();
  auto handler2 = addSimpleStrictHandler();
  // handler2 is paused before it gets headers
  handler2->expectEgressPaused();
  handler2->expectHeaders();
  handler2->expectEOM([this] {
      // This transaction should start egress paused.  We've received the
      // EOM, so the timeout shouldn't be running delay 400ms and resume
      // writes, this keeps txn1 from getting a write timeout
      resumeWritesAfterDelay(milliseconds(400));
    });
  handler1->expectEgressResumed([&handler1] { handler1->txn_->sendEOM(); });
  handler2->expectEgressResumed([&handler2, this] {
      // delay an additional 200ms.  The total 600ms delay shouldn't fire
      // onTimeout
      eventBase_.tryRunAfterDelay([&handler2] {
          handler2->sendReplyWithBody(200, 400); }, 200
        );
    });
  handler1->expectDetachTransaction();
  handler2->expectDetachTransaction();

  flushRequestsAndLoop(false, milliseconds(0), milliseconds(10));

  cleanup();
}

// Verifies that the read timer is running while a transaction is blocked
// on a window update
TEST_F(SPDY3DownstreamSessionTest, SpdyTimeoutWin) {
  clientCodec_->getEgressSettings()->setSetting(SettingsId::INITIAL_WINDOW_SIZE,
                                                500);
  clientCodec_->generateSettings(requests_);
  auto streamID = sendRequest();

  InSequence handlerSequence;
  auto handler = addSimpleStrictHandler();
  handler->expectHeaders();
  handler->expectEOM([&] {
      handler->sendReplyWithBody(200, 1000);
    });
  handler->expectEgressPaused();
  handler->expectError([&] (const HTTPException& ex) {
      ASSERT_EQ(ex.getProxygenError(), kErrorWriteTimeout);
      ASSERT_EQ(
        folly::to<std::string>("ingress timeout, streamID=", streamID),
        std::string(ex.what()));
      handler->terminate();
    });
  handler->expectDetachTransaction();

  flushRequestsAndLoop();

  cleanup();
}

TYPED_TEST_CASE_P(HTTPDownstreamTest);

TYPED_TEST_P(HTTPDownstreamTest, TestWritesDraining) {
  auto badCodec =
    makeServerCodec<typename TypeParam::Codec>(TypeParam::version);
  this->sendRequest();
  badCodec->generatePushPromise(this->requests_, 2 /* bad */, getGetRequest(),
                                1);

  this->expectDetachSession();

  InSequence handlerSequence;
  auto handler1 = this->addSimpleNiceHandler();
  handler1->expectHeaders();
  handler1->expectEOM();
  handler1->expectError([&](const HTTPException& ex) {
    ASSERT_EQ(ex.getProxygenError(), kErrorEOF);
    ASSERT_TRUE(
        folly::StringPiece(ex.what()).startsWith("Shutdown transport: EOF"))
        << ex.what();
  });
  handler1->expectDetachTransaction();

  this->flushRequestsAndLoop();
}

TYPED_TEST_P(HTTPDownstreamTest, TestBodySizeLimit) {
  this->clientCodec_->generateWindowUpdate(this->requests_, 0, 65536);
  this->sendRequest();
  this->sendRequest();

  InSequence handlerSequence;
  auto handler1 = this->addSimpleNiceHandler();
  handler1->expectHeaders();
  handler1->expectEOM();
  auto handler2 = this->addSimpleNiceHandler();
  handler2->expectHeaders();
  handler2->expectEOM([&] {
      handler1->sendReplyWithBody(200, 33000);
      handler2->sendReplyWithBody(200, 33000);
    });
  handler1->expectDetachTransaction();
  handler2->expectDetachTransaction();

  this->flushRequestsAndLoop();

  std::list<HTTPCodec::StreamID> streams;
  EXPECT_CALL(this->callbacks_, onMessageBegin(1, _));
  EXPECT_CALL(this->callbacks_, onHeadersComplete(1, _));
  EXPECT_CALL(this->callbacks_, onMessageBegin(3, _));
  EXPECT_CALL(this->callbacks_, onHeadersComplete(3, _));
  for (uint32_t i = 0; i < 8; i++) {
    EXPECT_CALL(this->callbacks_, onBody(1, _, _));
    EXPECT_CALL(this->callbacks_, onBody(3, _, _));
  }
  EXPECT_CALL(this->callbacks_, onBody(1, _, _));
  EXPECT_CALL(this->callbacks_, onMessageComplete(1, _));
  EXPECT_CALL(this->callbacks_, onBody(3, _, _));
  EXPECT_CALL(this->callbacks_, onMessageComplete(3, _));

  this->parseOutput(*this->clientCodec_);

  this->cleanup();
}

#define IF_HTTP2(X) \
  if (this->clientCodec_->getProtocol() == CodecProtocol::HTTP_2) { X; }

TYPED_TEST_P(HTTPDownstreamTest, TestUniformPauseState) {
  this->httpSession_->setWriteBufferLimit(12000);
  this->clientCodec_->getEgressSettings()->setSetting(
    SettingsId::INITIAL_WINDOW_SIZE, 1000000);
  this->clientCodec_->generateSettings(this->requests_);
  this->clientCodec_->generateWindowUpdate(this->requests_, 0, 1000000);
  this->sendRequest("/", 1);
  this->sendRequest("/", 1);
  this->sendRequest("/", 2);

  InSequence handlerSequence;
  auto handler1 = this->addSimpleStrictHandler();
  handler1->expectHeaders();
  handler1->expectEOM();
  auto handler2 = this->addSimpleStrictHandler();
  handler2->expectHeaders();
  handler2->expectEOM([&] {
      handler1->sendHeaders(200, 24002);
      // triggers pause of all txns
      this->transport_->pauseWrites();
      handler1->txn_->sendBody(std::move(makeBuf(12001)));
      this->resumeWritesAfterDelay(milliseconds(50));
    });
  handler1->expectEgressPaused();
  handler2->expectEgressPaused();
  auto handler3 = this->addSimpleStrictHandler();
  handler3->expectEgressPaused();
  handler3->expectHeaders();
  handler3->expectEOM();

  handler1->expectEgressResumed([&] {
      // resume does not trigger another pause,
      handler1->txn_->sendBody(std::move(makeBuf(12001)));
    });
  // handler2 gets a fair shot, handler3 is not resumed
  // HTTP/2 priority is not implemented, so handler3 is like another 0 pri txn
  handler2->expectEgressResumed();
  IF_HTTP2(handler3->expectEgressResumed());
  handler1->expectEgressPaused();
  handler2->expectEgressPaused();
  IF_HTTP2(handler3->expectEgressPaused());

  handler1->expectEgressResumed();
  handler2->expectEgressResumed([&] {
      handler2->sendHeaders(200, 12001);
      handler2->txn_->sendBody(std::move(makeBuf(12001)));
      this->transport_->pauseWrites();
      this->resumeWritesAfterDelay(milliseconds(50));
    });
  // handler3 not resumed
  IF_HTTP2(handler3->expectEgressResumed());

  handler1->expectEgressPaused();
  handler2->expectEgressPaused();
  IF_HTTP2(handler3->expectEgressPaused());

  handler1->expectEgressResumed();
  handler2->expectEgressResumed([&] {
      handler1->txn_->sendEOM();
      handler2->txn_->sendEOM();
    });
  handler3->expectEgressResumed([&] {
      handler3->txn_->sendAbort();
    });

  handler3->expectDetachTransaction();
  handler1->expectDetachTransaction();
  handler2->expectDetachTransaction();

  this->flushRequestsAndLoop();

  this->cleanup();
}

// Test exceeding the MAX_CONCURRENT_STREAMS setting.  The txn should get
// REFUSED_STREAM, and other streams can complete normally
TYPED_TEST_P(HTTPDownstreamTest, TestMaxTxns) {
  auto settings = this->rawCodec_->getEgressSettings();
  auto maxTxns = settings->getSetting(SettingsId::MAX_CONCURRENT_STREAMS,
                                      100);
  std::list<unique_ptr<StrictMock<MockHTTPHandler>>> handlers;
  {
    InSequence enforceOrder;
    for (auto i = 0U; i < maxTxns; i++) {
      this->sendRequest();
      auto handler = this->addSimpleStrictHandler();
      handler->expectHeaders();
      handler->expectEOM();
      handlers.push_back(std::move(handler));
    }
    auto streamID = this->sendRequest();
    this->clientCodec_->generateGoaway(this->requests_, 0, ErrorCode::NO_ERROR);

    for (auto& handler: handlers) {
      EXPECT_CALL(*handler, onGoaway(ErrorCode::NO_ERROR));
    }

    this->flushRequestsAndLoop();

    EXPECT_CALL(this->callbacks_, onSettings(_));
    EXPECT_CALL(this->callbacks_, onAbort(streamID, ErrorCode::REFUSED_STREAM));

    this->parseOutput(*this->clientCodec_);
  }
  // handlers can finish out of order?
  for (auto& handler: handlers) {
    handler->sendReplyWithBody(200, 100);
    handler->expectDetachTransaction();
  }
  this->expectDetachSession();
  this->eventBase_.loop();
}

// Set max streams=1
// send two spdy requests a few ms apart.
// Block writes
// generate a complete response for txn=1 before parsing txn=3
// HTTPSession should allow the txn=3 to be served rather than refusing it
TEST_F(SPDY3DownstreamSessionTest, SpdyMaxConcurrentStreams) {
  HTTPMessage req = getGetRequest();
  req.setHTTPVersion(1, 0);
  req.setWantsKeepalive(false);
  sendRequest(req);
  auto req2p = sendRequestLater(req, true);

  httpSession_->setEgressSettings({{
        SettingsId::MAX_CONCURRENT_STREAMS, 1}});

  InSequence handlerSequence;
  auto handler1 = addSimpleStrictHandler();
  handler1->expectHeaders();
  handler1->expectEOM([&handler1, req, this, &req2p] {
      transport_->pauseWrites();
      handler1->sendReplyWithBody(200, 100);
      req2p.setValue();
    });
  auto handler2 = addSimpleStrictHandler();
  handler2->expectHeaders();
  handler2->expectEOM([&handler2, this] {
      handler2->sendReplyWithBody(200, 100);
      resumeWritesInLoop();
    });
  handler1->expectDetachTransaction();
  handler2->expectDetachTransaction();

  expectDetachSession();

  flushRequestsAndLoop();
}

REGISTER_TYPED_TEST_CASE_P(HTTPDownstreamTest,
                           TestWritesDraining, TestBodySizeLimit,
                           TestUniformPauseState, TestMaxTxns);

typedef ::testing::Types<SPDY3CodecPair, SPDY3_1CodecPair,
                         HTTP2CodecPair> ParallelCodecs;
INSTANTIATE_TYPED_TEST_CASE_P(ParallelCodecs,
                              HTTPDownstreamTest,
                              ParallelCodecs);

class SPDY31DownstreamTest : public HTTPDownstreamTest<SPDY3_1CodecPair> {
 public:
  SPDY31DownstreamTest()
      : HTTPDownstreamTest<SPDY3_1CodecPair>({-1, -1,
            2 * spdy::kInitialWindow}) {}
};

TEST_F(SPDY31DownstreamTest, TestSessionFlowControl) {
  eventBase_.loopOnce();

  InSequence sequence;
  EXPECT_CALL(callbacks_, onSettings(_));
  EXPECT_CALL(callbacks_, onWindowUpdate(0, spdy::kInitialWindow));
  parseOutput(*clientCodec_);

  cleanup();
}

TEST_F(SPDY3DownstreamSessionTest, TestEOFOnBlockedStream) {
  sendRequest();

  auto handler1 = addSimpleStrictHandler();

  InSequence handlerSequence;
  handler1->expectHeaders();
  handler1->expectEOM([&handler1] {
      handler1->sendReplyWithBody(200, 80000);
    });
  handler1->expectEgressPaused();

  handler1->expectError([&] (const HTTPException& ex) {
      // Not optimal to have a different error code here than the session
      // flow control case, but HTTPException direction is immutable and
      // building another one seems not future proof.
      EXPECT_EQ(ex.getDirection(), HTTPException::Direction::INGRESS);
    });
  handler1->expectDetachTransaction();

  expectDetachSession();

  flushRequestsAndLoop(true, milliseconds(10));
}

TEST_F(SPDY31DownstreamTest, TestEOFOnBlockedSession) {
  sendRequest();
  sendRequest();

  InSequence handlerSequence;
  auto handler1 = addSimpleStrictHandler();
  handler1->expectHeaders();
  handler1->expectEOM([&handler1] {
      handler1->sendHeaders(200, 40000);
      handler1->sendBody(32769);
    });
  auto handler2 = addSimpleStrictHandler();
  handler2->expectHeaders();
  handler2->expectEOM([&handler2, this] {
      handler2->sendHeaders(200, 40000);
      handler2->sendBody(32768);
      eventBase_.runInLoop([this] { transport_->addReadEOF(milliseconds(0)); });
    });

  handler1->expectEgressPaused();
  handler2->expectEgressPaused();
  handler1->expectEgressResumed();
  handler2->expectEgressResumed();
  handler1->expectError([&] (const HTTPException& ex) {
      EXPECT_EQ(ex.getDirection(),
                HTTPException::Direction::INGRESS_AND_EGRESS);
    });
  handler1->expectDetachTransaction();
  handler2->expectError([&] (const HTTPException& ex) {
      EXPECT_EQ(ex.getDirection(),
                HTTPException::Direction::INGRESS_AND_EGRESS);
    });
  handler2->expectDetachTransaction();

  expectDetachSession();

  flushRequestsAndLoop();
}


TEST_F(SPDY3DownstreamSessionTest, NewTxnEgressPaused) {
  // Send 1 request with prio=0
  // Have egress pause while sending the first response
  // Send a second request with prio=1
  //   -- the new txn should start egress paused
  // Finish the body and eom both responses
  // Unpause egress
  // The first txn should complete first

  sendRequest("/", 0);
  auto req2 = getGetRequest();
  req2.setPriority(1);
  auto req2p = sendRequestLater(req2, true);

  unique_ptr<StrictMock<MockHTTPHandler>> handler1;
  unique_ptr<StrictMock<MockHTTPHandler>> handler2;

  httpSession_->setWriteBufferLimit(200); // lower the per session buffer limit
  {
    InSequence handlerSequence;
    handler1 = addSimpleStrictHandler();
    handler1->expectHeaders();
    handler1->expectEOM([&handler1, this, &req2p] {
        this->transport_->pauseWrites();
        handler1->sendHeaders(200, 1000);
        handler1->sendBody(100); // headers + 100 bytes - over the limit
        req2p.setValue();
      });
    handler1->expectEgressPaused([] { LOG(INFO) << "paused 1"; });

    handler2 = addSimpleStrictHandler();
    handler2->expectEgressPaused(); // starts paused
    handler2->expectHeaders();
    handler2->expectEOM([&] {
        // Technically shouldn't send while handler is egress paused, but meh.
        handler1->sendBody(900);
        handler1->txn_->sendEOM();
        handler2->sendReplyWithBody(200, 1000);
        resumeWritesInLoop();
      });
    handler1->expectDetachTransaction();
    handler2->expectDetachTransaction();
  }
  HTTPSession::DestructorGuard g(httpSession_);
  flushRequestsAndLoop();

  std::list<HTTPCodec::StreamID> streams;
  EXPECT_CALL(callbacks_, onMessageBegin(_, _))
    .Times(2);
  EXPECT_CALL(callbacks_, onHeadersComplete(_, _))
    .Times(2);
  // body is variable and hence ignored;
  EXPECT_CALL(callbacks_, onMessageComplete(_, _))
      .WillRepeatedly(Invoke([&](HTTPCodec::StreamID stream, bool /*upgrade*/) {
        streams.push_back(stream);
      }));
  parseOutput(*clientCodec_);

  cleanup();
}

TEST_F(HTTP2DownstreamSessionTest, ZeroDeltaWindowUpdate) {
  // generateHeader() will create a session and a transaction
  auto streamID = sendHeader();
  // First generate a frame with delta=1 so as to pass the checks, and then
  // hack the frame so that delta=0 without modifying other checks
  clientCodec_->generateWindowUpdate(requests_, streamID, 1);
  requests_.trimEnd(http2::kFrameWindowUpdateSize);
  QueueAppender appender(&requests_, http2::kFrameWindowUpdateSize);
  appender.writeBE<uint32_t>(0);

  auto handler = addSimpleStrictHandler();

  InSequence handlerSequence;
  handler->expectHeaders();
  handler->expectError([&] (const HTTPException& ex) {
      ASSERT_EQ(ex.getCodecStatusCode(), ErrorCode::PROTOCOL_ERROR);
      ASSERT_EQ(
        "streamID=1 with HTTP2Codec stream error: window update delta=0",
        std::string(ex.what()));
    });
  handler->expectDetachTransaction();
  expectDetachSession();

  flushRequestsAndLoop();
}

TEST_F(HTTP2DownstreamSessionTest, PaddingFlowControl) {
  // generateHeader() will create a session and a transaction
  auto streamID = sendHeader();
  // This sends a total of 33kb including padding, so we should get a session
  // and stream window update
  for (auto i = 0; i < 129; i++) {
    clientCodec_->generateBody(requests_, streamID, makeBuf(1), 255, false);
  }

  auto handler = addSimpleStrictHandler();

  InSequence handlerSequence;
  handler->expectHeaders([&] {
      handler->txn_->pauseIngress();
      eventBase_.runAfterDelay([&] { handler->txn_->resumeIngress(); },
                               100);
    });
  EXPECT_CALL(*handler, onBody(_))
    .Times(129);
  handler->expectError();
  handler->expectDetachTransaction();

  HTTPSession::DestructorGuard g(httpSession_);
  flushRequestsAndLoop(false, milliseconds(0), milliseconds(0), [&] {
      clientCodec_->generateRstStream(requests_, streamID, ErrorCode::CANCEL);
      clientCodec_->generateGoaway(requests_, 0, ErrorCode::NO_ERROR);
      transport_->addReadEvent(requests_, milliseconds(110));
    });

  std::list<HTTPCodec::StreamID> streams;
  EXPECT_CALL(callbacks_, onWindowUpdate(0, _));
  EXPECT_CALL(callbacks_, onWindowUpdate(1, _));
  parseOutput(*clientCodec_);
  expectDetachSession();
}

TEST_F(HTTP2DownstreamSessionTest, GracefulDrainOnTimeout) {
  InSequence handlerSequence;
  std::chrono::milliseconds gracefulTimeout(200);
  httpSession_->enableDoubleGoawayDrain();
  EXPECT_CALL(mockController_, getGracefulShutdownTimeout())
    .WillOnce(InvokeWithoutArgs([&] {
          // Once session asks for graceful shutdown timeout, expect the client
          // to receive the first GOAWAY
          eventBase_.runInLoop([&] {
              EXPECT_CALL(callbacks_,
                          onGoaway(std::numeric_limits<int32_t>::max(),
                                   ErrorCode::NO_ERROR, _));
              parseOutput(*clientCodec_);
            });
          return gracefulTimeout;
        }));


  // Simulate ConnectionManager idle timeout
  eventBase_.runAfterDelay([&] { httpSession_->timeoutExpired(); },
                           transactionTimeouts_->getDefaultTimeout().count());
  HTTPSession::DestructorGuard g(httpSession_);
  auto start = getCurrentTime();
  eventBase_.loop();
  auto finish = getCurrentTime();
  auto minDuration =
    gracefulTimeout + transactionTimeouts_->getDefaultTimeout();
  EXPECT_GE((finish - start).count(), minDuration.count());
  EXPECT_CALL(callbacks_, onGoaway(0, ErrorCode::NO_ERROR, _));
  parseOutput(*clientCodec_);
  expectDetachSession();
}

/*
 * The sequence of streams are generated in the following order:
 * - [client --> server] request 1st stream (getGetRequest())
 * - [server --> client] respond 1st stream (res with length 100)
 * - [server --> client] request 2nd stream (req)
 * - [server --> client] respond 2nd stream (res with length 200 + EOM)
 * - [client --> server] RST_STREAM on the 1st stream
 */
TEST_F(HTTP2DownstreamSessionTest, ServerPush) {
  // Create a dummy request and a dummy response messages
  HTTPMessage req, res;
  req.getHeaders().set("HOST", "www.foo.com");
  req.setURL("https://www.foo.com/");
  res.setStatusCode(200);
  res.setStatusMessage("Ohai");

  // enable server push
  clientCodec_->getEgressSettings()->setSetting(SettingsId::ENABLE_PUSH, 1);
  clientCodec_->generateSettings(requests_);
  // generateHeader() will create a session and a transaction
  auto assocStreamId = HTTPCodec::StreamID(1);
  clientCodec_->generateHeader(requests_, assocStreamId, getGetRequest(),
                               false, nullptr);

  auto handler = addSimpleStrictHandler();
  StrictMock<MockHTTPPushHandler> pushHandler;

  InSequence handlerSequence;
  handler->expectHeaders([&] {
      // Generate response for the associated stream
      handler->txn_->sendHeaders(res);
      handler->txn_->sendBody(makeBuf(100));
      handler->txn_->pauseIngress();

      auto* pushTxn = handler->txn_->newPushedTransaction(&pushHandler);
      ASSERT_NE(pushTxn, nullptr);
      // Generate a push request (PUSH_PROMISE)
      auto outgoingStreams = httpSession_->getNumOutgoingStreams();
      pushTxn->sendHeaders(req);
      EXPECT_EQ(httpSession_->getNumOutgoingStreams(), outgoingStreams);
      // Generate a push response
      auto pri = handler->txn_->getPriority();
      res.setHTTP2Priority(std::make_tuple(pri.streamDependency,
                                           pri.exclusive, pri.weight));
      pushTxn->sendHeaders(res);
      EXPECT_EQ(httpSession_->getNumOutgoingStreams(), outgoingStreams + 1);
      pushTxn->sendBody(makeBuf(200));
      pushTxn->sendEOM();

      eventBase_.runAfterDelay([&] { handler->txn_->resumeIngress(); },
                               100);
    });
  EXPECT_CALL(pushHandler, setTransaction(_))
    .WillOnce(Invoke([&] (HTTPTransaction* txn) {
          pushHandler.txn_ = txn; }));
  EXPECT_CALL(pushHandler, detachTransaction());
  handler->expectError();
  handler->expectDetachTransaction();

  transport_->addReadEvent(requests_, milliseconds(0));
  clientCodec_->generateRstStream(requests_, assocStreamId, ErrorCode::CANCEL);
  clientCodec_->generateGoaway(requests_, 2, ErrorCode::NO_ERROR);
  transport_->addReadEvent(requests_, milliseconds(200));
  transport_->startReadEvents();
  HTTPSession::DestructorGuard g(httpSession_);
  eventBase_.loop();

  EXPECT_CALL(callbacks_, onMessageBegin(1, _));
  EXPECT_CALL(callbacks_, onHeadersComplete(1, _));
  EXPECT_CALL(callbacks_, onPushMessageBegin(2, 1, _));
  EXPECT_CALL(callbacks_, onHeadersComplete(2, _));
  EXPECT_CALL(callbacks_, onMessageBegin(2, _));
  EXPECT_CALL(callbacks_, onHeadersComplete(2, _));
  EXPECT_CALL(callbacks_, onMessageComplete(2, _));

  parseOutput(*clientCodec_);
  expectDetachSession();
}

TEST_F(HTTP2DownstreamSessionTest, ServerPushAbortPaused) {
  // Create a dummy request and a dummy response messages
  HTTPMessage req, res;
  req.getHeaders().set("HOST", "www.foo.com");
  req.setURL("https://www.foo.com/");
  res.setStatusCode(200);
  res.setStatusMessage("Ohai");

  // enable server push
  clientCodec_->getEgressSettings()->setSetting(SettingsId::ENABLE_PUSH, 1);
  clientCodec_->generateSettings(requests_);
  // generateHeader() will create a session and a transaction
  auto assocStreamId = HTTPCodec::StreamID(1);
  clientCodec_->generateHeader(requests_, assocStreamId, getGetRequest(),
                               false, nullptr);

  auto handler = addSimpleStrictHandler();
  StrictMock<MockHTTPPushHandler> pushHandler;

  InSequence handlerSequence;
  handler->expectHeaders([&] {
      // Generate response for the associated stream
      this->transport_->pauseWrites();
      handler->txn_->sendHeaders(res);
      handler->txn_->sendBody(makeBuf(100));
      handler->txn_->pauseIngress();

      auto* pushTxn = handler->txn_->newPushedTransaction(&pushHandler);
      ASSERT_NE(pushTxn, nullptr);
      // Generate a push request (PUSH_PROMISE)
      pushTxn->sendHeaders(req);
    });
  EXPECT_CALL(pushHandler, setTransaction(_))
    .WillOnce(Invoke([&] (HTTPTransaction* txn) {
          pushHandler.txn_ = txn; }));
  EXPECT_CALL(pushHandler, onError(_));
  EXPECT_CALL(pushHandler, detachTransaction());
  handler->expectError();
  handler->expectDetachTransaction();

  transport_->addReadEvent(requests_, milliseconds(0));
  // Cancels everything
  clientCodec_->generateRstStream(requests_, assocStreamId, ErrorCode::CANCEL);
  transport_->addReadEvent(requests_, milliseconds(10));
  transport_->startReadEvents();
  HTTPSession::DestructorGuard g(httpSession_);
  eventBase_.loop();

  parseOutput(*clientCodec_);
  expectDetachSession();
}

TEST_F(HTTP2DownstreamSessionTest, TestPriorityWeightsTinyRatio) {
  // Create a transaction with egress and a ratio small enough that
  // ratio*4096 < 1.
  //
  //     root
  //     /  \                                                 level 1
  //   256   1 (no egress)
  //        / \                                               level 2
  //      256  1  <-- has ratio (1/257)^2
  InSequence enforceOrder;
  auto req1 = getGetRequest();
  auto req2 = getGetRequest();
  req1.setHTTP2Priority(HTTPMessage::HTTPPriority{0, false, 255});
  req2.setHTTP2Priority(HTTPMessage::HTTPPriority{0, false, 0});

  sendRequest(req1);
  auto id2 = sendRequest(req2);
  req1.setHTTP2Priority(HTTPMessage::HTTPPriority{id2, false, 255});
  req2.setHTTP2Priority(HTTPMessage::HTTPPriority{id2, false, 0});
  sendRequest(req1);
  sendRequest(req2);

  auto handler1 = addSimpleStrictHandler();
  handler1->expectHeaders();
  handler1->expectEOM([&] {
      handler1->sendReplyWithBody(200, 4 * 1024);
    });
  auto handler2 = addSimpleStrictHandler();
  handler2->expectHeaders();
  handler2->expectEOM();
  auto handler3 = addSimpleStrictHandler();
  handler3->expectHeaders();
  handler3->expectEOM([&] {
      handler3->sendReplyWithBody(200, 15);
    });
  auto handler4 = addSimpleStrictHandler();
  handler4->expectHeaders();
  handler4->expectEOM([&] {
      handler4->sendReplyWithBody(200, 1);
    });

  handler1->expectDetachTransaction([&] {
      HTTPTransaction::PrioritySampleSummary summary;
      EXPECT_EQ(handler1->txn_->getPrioritySampleSummary(summary), true);
      EXPECT_EQ(handler1->txn_->getTransport().getHTTP2PrioritiesEnabled(),
                true);
      // id1 had no egress when id2 was running, so id1 was contending only with
      // id3 and id4. Average number of contentions for id1 is 3
      EXPECT_EQ(summary.contentions_.byTransactionBytes_, 3);
      EXPECT_EQ(summary.contentions_.bySessionBytes_, 3);
      // this is a first level transaction, depth == 1
      EXPECT_EQ(summary.depth_.byTransactionBytes_, 1);
      EXPECT_EQ(summary.depth_.bySessionBytes_, 1);
      // the expected relative weight is 256/257 ~= 0.9961
      EXPECT_GT(summary.expected_weight_, 0.996);
      EXPECT_LT(summary.expected_weight_, 0.9962);
      // the measured relative weight is 4096/(4096+15) ~= 0.99635
      // This value is higher than the expected relative weight of 0.9961.
      // Due to the arithmetical rounding to the lowest integer, the measured
      // relative weight tends to be higher for transactions with high relative
      // weights and lower for transactions with the low relative weights.
      EXPECT_GT(summary.measured_weight_, 0.9963);
      EXPECT_LT(summary.measured_weight_, 0.9964);
    });
  handler3->expectDetachTransaction([&] {
      HTTPTransaction::PrioritySampleSummary summary;
      EXPECT_EQ(handler3->txn_->getPrioritySampleSummary(summary), true);
      EXPECT_EQ(handler3->txn_->getTransport().getHTTP2PrioritiesEnabled(),
                true);
      // Similarly, id3 was contenting with id1 and id4
      // Average number of contentions for id3 is 3
      EXPECT_EQ(summary.contentions_.byTransactionBytes_, 3);
      EXPECT_EQ(summary.contentions_.bySessionBytes_, 3);
      // this is a second level transaction where parent has
      // no egress, depth == 2
      EXPECT_EQ(summary.depth_.byTransactionBytes_, 2);
      EXPECT_EQ(summary.depth_.bySessionBytes_, 2);
      // the expected relative weight should be
      // 1/257 * 256/257 ~= 0.00388. However, in the calculation of the
      // allowed bytes to send we rounding to the lowest positive integer.
      // Therefore, the measured relative weight tends to be less than
      // it should be.  In this example, the allowed bytes sent is
      // 4096 * 0.00388 ~= 15.89, which is rounded to 15. Hence the measured
      // relative weight is 15/(4096+15) ~= 0.00365
      EXPECT_GT(summary.expected_weight_, 0.00388);
      EXPECT_LT(summary.expected_weight_, 0.0039);
      EXPECT_GT(summary.measured_weight_, 0.00364);
      EXPECT_LT(summary.measured_weight_, 0.00366);
    });
  handler4->expectDetachTransaction([&] {
      HTTPTransaction::PrioritySampleSummary summary;
      EXPECT_EQ(handler4->txn_->getPrioritySampleSummary(summary), true);
      EXPECT_EQ(handler4->txn_->getTransport().getHTTP2PrioritiesEnabled(),
                true);
      // This is the priority-based blocking situation. id4 was blocked by
      // higher priority transactions id1 and id3. Only when id1 and id3
      // finished, id4 had a chance to transfer its data.
      // Average contention number weighted by transaction bytes is 1, which
      // means that when id4 had a chance to transfer bytes it did not contend
      // with any other transactions.
      // id4 was waiting for id1 and id3 during transfer of 4256 bytes (encoded)
      // after which it tranferred 10 bytes (encoded) without contention.
      // Therefore, the average number contentions weighted by session bytes is
      // (4111*3 + 1*1)/(4111 + 1) = 12334/4112 ~= 2.999
      // The difference in average contentions weighted by transaction and
      // session bytes tells that id4 was mostly blocked by rather than blocking
      // other transactions.
      EXPECT_EQ(summary.contentions_.byTransactionBytes_, 1);
      EXPECT_GT(summary.contentions_.bySessionBytes_, 2.99);
      EXPECT_LT(summary.contentions_.bySessionBytes_, 3.00);
      // this is a second level transaction where parent has
      // no egress, depth == 2
      EXPECT_EQ(summary.depth_.byTransactionBytes_, 2);
      EXPECT_EQ(summary.depth_.bySessionBytes_, 2);
      // the expected relative weight should be
      // 1/257 * 1/257 ~= 0.000015.
      // Because no bytes of this transaction were sent during the previous
      // egress, the expected relative weight was calculated as:
      // (0*4111 + 1*1)/(4111 + 1) ~= 0.000243
      EXPECT_GT(summary.expected_weight_, 0.000243);
      EXPECT_LT(summary.expected_weight_, 0.000244);
      // The measured weight is (0+1)/(4111+1)  ~= 0.000243
      // The difference between the theoretical value of 0.000015 and the
      // measured one is not because of the arithmetical rounding, but because
      // all other transactions are completed and the relative waight for the
      // only survived transaction was elevated to 1.0
      EXPECT_GT(summary.measured_weight_, 0.000243);
      EXPECT_LT(summary.measured_weight_, 0.000244);
      handler2->txn_->sendAbort();
    });
  handler2->expectDetachTransaction();
  flushRequestsAndLoop();
  httpSession_->closeWhenIdle();
  expectDetachSession();
  eventBase_.loop();
}

TEST_F(HTTP2DownstreamSessionTest, TestPriorityDependentTransactions) {
  // Create a dependent transaction to test the priority blocked by dependency.
  // ratio*4096 < 1.
  //
  //     root
  //      \                                                 level 1
  //      16
  //       \                                                level 2
  //       16
  InSequence enforceOrder;
  auto req1 = getGetRequest();
  req1.setHTTP2Priority(HTTPMessage::HTTPPriority{0, false, 15});
  auto id1 = sendRequest(req1);

  auto req2 = getGetRequest();
  req2.setHTTP2Priority(HTTPMessage::HTTPPriority{id1, false, 15});
  sendRequest(req2);

  auto handler1 = addSimpleStrictHandler();
  handler1->expectHeaders();
  handler1->expectEOM([&] {
      handler1->sendReplyWithBody(200, 1024);
    });
  auto handler2 = addSimpleStrictHandler();
  handler2->expectHeaders();
  handler2->expectEOM([&] {
      handler2->sendReplyWithBody(200, 1024);
    });

  handler1->expectDetachTransaction([&] {
      HTTPTransaction::PrioritySampleSummary summary;
      EXPECT_EQ(handler1->txn_->getPrioritySampleSummary(summary), true);
      EXPECT_EQ(handler1->txn_->getTransport().getHTTP2PrioritiesEnabled(),
                true);
      // id1 is contending with id2 during the entire transfer.
      // Average number of contentions for id1 is 2 in both cases.
      // The same number of average contentions weighted by both transaction
      // and session bytes tells that id1 was not blocked by any other
      // transaction during the entire transfer.
      EXPECT_EQ(summary.contentions_.byTransactionBytes_, 2);
      EXPECT_EQ(summary.contentions_.bySessionBytes_, 2);
      // this is a first level transaction, depth == 1
      EXPECT_EQ(summary.depth_.byTransactionBytes_, 1);
      EXPECT_EQ(summary.depth_.bySessionBytes_, 1);
      // dependent transaction is blocked, the parent is egressing on 100%
      EXPECT_EQ(summary.expected_weight_, 1);
      EXPECT_EQ(summary.measured_weight_, 1);
    });
  handler2->expectDetachTransaction([&] {
      HTTPTransaction::PrioritySampleSummary summary;
      EXPECT_EQ(handler2->txn_->getPrioritySampleSummary(summary), true);
      EXPECT_EQ(handler2->txn_->getTransport().getHTTP2PrioritiesEnabled(),
                true);
      // This is the dependency-based blocking. id2 is blocked by id1.
      // When id2 had a chance to transfer bytes, it was no longer contended
      // with any other transaction. Hence the average contention weighted by
      // transaction bytes is 1.
      // The average number of contentions weighted by the session bytes is
      // computed as (1024*2 + 1024*1)/(1024 + 1024) = 3072/2048 = 1.5
      EXPECT_EQ(summary.contentions_.byTransactionBytes_, 1);
      EXPECT_EQ(summary.contentions_.bySessionBytes_, 1.5);
      // The transaction transferred bytes only when its parent transaction
      // completed. At that time its level decreased to 1. The average depth
      // weighted by session bytes is (2*1024 + 1*1024)/(2048) = 1.5.
      EXPECT_EQ(summary.depth_.byTransactionBytes_, 1);
      EXPECT_EQ(summary.depth_.bySessionBytes_, 1.5);
      // this dependent transaction was bloted, so it was egressiong only 1/2
      // of the session bytes.
      EXPECT_EQ(summary.expected_weight_, 0.5);
      EXPECT_EQ(summary.measured_weight_, 0.5);
      handler2->txn_->sendAbort();
    });
  flushRequestsAndLoop();
  httpSession_->closeWhenIdle();
  expectDetachSession();
  eventBase_.loop();
}

TEST_F(HTTP2DownstreamSessionTest, TestDisablePriorities) {
  // turn off HTTP2 priorities
  httpSession_->setHTTP2PrioritiesEnabled(false);

  InSequence enforceOrder;
  HTTPMessage req1 = getGetRequest();
  req1.setHTTP2Priority(HTTPMessage::HTTPPriority{0, false, 0});
  sendRequest(req1);

  HTTPMessage req2 = getGetRequest();
  req2.setHTTP2Priority(HTTPMessage::HTTPPriority{0, false, 255});
  sendRequest(req2);

  auto handler1 = addSimpleStrictHandler();
  handler1->expectHeaders();
  handler1->expectEOM([&] {
      handler1->sendReplyWithBody(200, 4 * 1024);
    });

  auto handler2 = addSimpleStrictHandler();
  handler2->expectHeaders();
  handler2->expectEOM([&] {
      handler2->sendReplyWithBody(200, 4 * 1024);
    });

  // expecting handler 1 to finish first irrespective of
  // request 2 having higher weight
  handler1->expectDetachTransaction();
  handler2->expectDetachTransaction();

  flushRequestsAndLoop();
  httpSession_->closeWhenIdle();
  expectDetachSession();
  eventBase_.loop();
}

TEST_F(HTTP2DownstreamSessionTest, TestPriorityWeights) {
  // virtual priority node with pri=4
  auto priGroupID = clientCodec_->createStream();
  clientCodec_->generatePriority(
    requests_, priGroupID, HTTPMessage::HTTPPriority(0, false, 3));
  // Both txn's are at equal pri=16
  auto id1 = sendRequest();
  auto id2 = sendRequest();

  auto handler1 = addSimpleStrictHandler();

  handler1->expectHeaders();
  handler1->expectEOM([&] {
      handler1->sendHeaders(200, 12 * 1024);
      handler1->txn_->sendBody(makeBuf(4 * 1024));
    });
  auto handler2 = addSimpleStrictHandler();
  handler2->expectHeaders();
  handler2->expectEOM([&] {
      handler2->sendHeaders(200, 12 * 1024);
      handler2->txn_->sendBody(makeBuf(4 * 1024));
    });

  // twice- once to send and once to receive
  flushRequestsAndLoopN(2);
  EXPECT_CALL(callbacks_, onSettings(_));
  EXPECT_CALL(callbacks_, onMessageBegin(id1, _));
  EXPECT_CALL(callbacks_, onHeadersComplete(id1, _));
  EXPECT_CALL(callbacks_, onMessageBegin(id2, _));
  EXPECT_CALL(callbacks_, onHeadersComplete(id2, _));
  EXPECT_CALL(callbacks_, onBody(id1, _, _))
    .WillOnce(ExpectBodyLen(4 * 1024));
  EXPECT_CALL(callbacks_, onBody(id2, _, _))
    .WillOnce(ExpectBodyLen(4 * 1024));
  parseOutput(*clientCodec_);

  // update handler2 to be in the pri-group (which has lower weight)
  clientCodec_->generatePriority(
    requests_, id2, HTTPMessage::HTTPPriority(priGroupID, false, 15));

  eventBase_.runInLoop([&] {
      handler1->txn_->sendBody(makeBuf(4 * 1024));
      handler2->txn_->sendBody(makeBuf(4 * 1024));
    });
  flushRequestsAndLoopN(2);

  EXPECT_CALL(callbacks_, onBody(id1, _, _))
    .WillOnce(ExpectBodyLen(4 * 1024));
  EXPECT_CALL(callbacks_, onBody(id2, _, _))
    .WillOnce(ExpectBodyLen(1 * 1024))
    .WillOnce(ExpectBodyLen(3 * 1024));
  parseOutput(*clientCodec_);

  // update vnode weight to match txn1 weight
  clientCodec_->generatePriority(requests_, priGroupID,
                                 HTTPMessage::HTTPPriority(0, false, 15));
  eventBase_.runInLoop([&] {
      handler1->txn_->sendBody(makeBuf(4 * 1024));
      handler1->txn_->sendEOM();
      handler2->txn_->sendBody(makeBuf(4 * 1024));
      handler2->txn_->sendEOM();
    });
  handler1->expectDetachTransaction();
  handler2->expectDetachTransaction();
  flushRequestsAndLoopN(2);

  // expect 32/32
  EXPECT_CALL(callbacks_, onBody(id1, _, _))
    .WillOnce(ExpectBodyLen(4 * 1024));
  EXPECT_CALL(callbacks_, onMessageComplete(id1, _));
  EXPECT_CALL(callbacks_, onBody(id2, _, _))
    .WillOnce(ExpectBodyLen(4 * 1024));
  EXPECT_CALL(callbacks_, onMessageComplete(id2, _));
  parseOutput(*clientCodec_);

  httpSession_->closeWhenIdle();
  expectDetachSession();
  this->eventBase_.loop();
}

TEST_F(HTTP2DownstreamSessionTest, TestPriorityWeightsTinyWindow) {
  httpSession_->setWriteBufferLimit(2 * 65536);
  InSequence enforceOrder;
  auto id1 = sendRequest();
  auto id2 = sendRequest();

  auto handler1 = addSimpleStrictHandler();

  handler1->expectHeaders();
  handler1->expectEOM([&] {
      handler1->sendReplyWithBody(200, 32 * 1024);
    });
  auto handler2 = addSimpleStrictHandler();
  handler2->expectHeaders();
  handler2->expectEOM([&] {
      handler2->sendReplyWithBody(200, 32 * 1024);
    });

  handler1->expectDetachTransaction();

  // twice- once to send and once to receive
  flushRequestsAndLoopN(2);
  EXPECT_CALL(callbacks_, onSettings(_));
  EXPECT_CALL(callbacks_, onMessageBegin(id1, _));
  EXPECT_CALL(callbacks_, onHeadersComplete(id1, _));
  EXPECT_CALL(callbacks_, onMessageBegin(id2, _));
  EXPECT_CALL(callbacks_, onHeadersComplete(id2, _));
  for (auto i = 0; i < 7; i++) {
    EXPECT_CALL(callbacks_, onBody(id1, _, _))
      .WillOnce(ExpectBodyLen(4 * 1024));
    EXPECT_CALL(callbacks_, onBody(id2, _, _))
      .WillOnce(ExpectBodyLen(4 * 1024));
  }
  EXPECT_CALL(callbacks_, onBody(id1, _, _))
    .WillOnce(ExpectBodyLen(4 * 1024 - 1));
  EXPECT_CALL(callbacks_, onBody(id2, _, _))
    .WillOnce(ExpectBodyLen(4 * 1024 - 1));
  EXPECT_CALL(callbacks_, onBody(id1, _, _))
    .WillOnce(ExpectBodyLen(1));
  EXPECT_CALL(callbacks_, onMessageComplete(id1, _));
  parseOutput(*clientCodec_);

  // open the window
  clientCodec_->generateWindowUpdate(requests_, 0, 100);
  handler2->expectDetachTransaction();
  flushRequestsAndLoopN(2);

  EXPECT_CALL(callbacks_, onBody(id2, _, _))
    .WillOnce(ExpectBodyLen(1));
  EXPECT_CALL(callbacks_, onMessageComplete(id2, _));
  parseOutput(*clientCodec_);

  httpSession_->closeWhenIdle();
  expectDetachSession();
  this->eventBase_.loop();
}

TEST_F(HTTP2DownstreamSessionTest, TestShortContentLength) {
  auto req = getPostRequest(10);
  auto streamID = sendRequest(req, false);
  clientCodec_->generateBody(requests_, streamID, makeBuf(20),
                             HTTPCodec::NoPadding, true);
  auto handler1 = addSimpleStrictHandler();

  InSequence enforceOrder;
  handler1->expectHeaders();
  handler1->expectError([&handler1] (const HTTPException& ex) {
      EXPECT_EQ(ex.getProxygenError(), kErrorParseBody);
      handler1->txn_->sendAbort();
    });
  handler1->expectDetachTransaction();
  flushRequestsAndLoop();

  gracefulShutdown();
}

TEST_F(HTTP2DownstreamSessionTest, TestBadContentLengthUntieHandler) {
  auto req = getPostRequest(10);
  auto streamID = sendRequest(req, false);
  clientCodec_->generateBody(
      requests_,
      streamID,
      makeBuf(20),
      HTTPCodec::NoPadding,
      true);
  auto handler1 = addSimpleStrictHandler();

  InSequence enforceOrder;
  handler1->expectHeaders();
  handler1->expectError([&] (const HTTPException&) {
      if (handler1->txn_) {
        handler1->txn_->setHandler(nullptr);
      }
      handler1->txn_ = nullptr;
    });
  flushRequestsAndLoop();

  gracefulShutdown();
}

TEST_F(HTTP2DownstreamSessionTest, TestLongContentLength) {
  auto req = getPostRequest(30);
  auto streamID = sendRequest(req, false);
  clientCodec_->generateBody(requests_, streamID, makeBuf(20),
                             HTTPCodec::NoPadding, true);
  auto handler1 = addSimpleStrictHandler();

  InSequence enforceOrder;
  handler1->expectHeaders();
  handler1->expectBody();
  handler1->expectError([&handler1] (const HTTPException& ex) {
      EXPECT_EQ(ex.getProxygenError(), kErrorParseBody);
      handler1->txn_->sendAbort();
    });
  handler1->expectDetachTransaction();
  flushRequestsAndLoop();

  gracefulShutdown();
}

TEST_F(HTTP2DownstreamSessionTest, TestMalformedContentLength) {
  auto req = getPostRequest();
  req.getHeaders().set(HTTP_HEADER_CONTENT_LENGTH, "malformed");
  auto streamID = sendRequest(req, false);
  clientCodec_->generateBody(requests_, streamID, makeBuf(20),
                             HTTPCodec::NoPadding, true);
  auto handler1 = addSimpleStrictHandler();

  InSequence enforceOrder;
  handler1->expectHeaders();
  handler1->expectBody();
  handler1->expectEOM([&handler1] {
      handler1->sendReplyWithBody(200, 100);
    });
  handler1->expectDetachTransaction();
  flushRequestsAndLoop();

  gracefulShutdown();
}

TEST_F(HTTP2DownstreamSessionTest, TestHeadContentLength) {
  InSequence enforceOrder;
  auto req = getGetRequest();
  req.setMethod(HTTPMethod::HEAD);
  sendRequest(req);
  auto handler1 = addSimpleStrictHandler();

  handler1->expectHeaders();
  handler1->expectEOM([&handler1] {
      handler1->sendHeaders(200, 100);
      // no body for head
      handler1->txn_->sendEOM();
    });
  handler1->expectDetachTransaction();
  flushRequestsAndLoop();

  gracefulShutdown();
}

TEST_F(HTTP2DownstreamSessionTest, Test304ContentLength) {
  InSequence enforceOrder;
  auto req = getGetRequest();
  req.setMethod(HTTPMethod::HEAD);
  sendRequest(req);
  auto handler1 = addSimpleStrictHandler();

  handler1->expectHeaders();
  handler1->expectEOM([&handler1] {
      handler1->sendHeaders(304, 100);
      handler1->txn_->sendEOM();
    });
  handler1->expectDetachTransaction();
  flushRequestsAndLoop();

  gracefulShutdown();
}

// chunked with wrong content-length
TEST_F(HTTPDownstreamSessionTest, HttpShortContentLength) {
  InSequence enforceOrder;
  auto req = getPostRequest(10);
  req.setIsChunked(true);
  req.getHeaders().add(HTTP_HEADER_TRANSFER_ENCODING, "chunked");
  auto streamID = sendRequest(req, false);
  clientCodec_->generateChunkHeader(requests_, streamID, 20);
  clientCodec_->generateBody(requests_, streamID, makeBuf(20),
                             HTTPCodec::NoPadding, false);
  clientCodec_->generateChunkTerminator(requests_, streamID);
  clientCodec_->generateEOM(requests_, streamID);
  auto handler1 = addSimpleStrictHandler();

  handler1->expectHeaders();
  EXPECT_CALL(*handler1, onChunkHeader(20));

  handler1->expectError([&handler1] (const HTTPException& ex) {
      EXPECT_EQ(ex.getProxygenError(), kErrorParseBody);
      handler1->txn_->sendAbort();
    });
  handler1->expectDetachTransaction();
  expectDetachSession();
  flushRequestsAndLoop();

}

TEST_F(HTTP2DownstreamSessionTest, TestSessionStallByFlowControl) {
  NiceMock<MockHTTPSessionStats> stats;
  // By default the send and receive windows are 64K each.
  // If we use only a single transaction, that transaction
  // will be paused on reaching 64K. Therefore, to pause the session,
  // it is used 2 transactions each sending 32K.

  // Make write buffer limit exceding 64K, for example 128K
  httpSession_->setWriteBufferLimit(128 * 1024);
  httpSession_->setSessionStats(&stats);

  InSequence enforceOrder;
  sendRequest();
  sendRequest();

  auto handler1 = addSimpleStrictHandler();

  handler1->expectHeaders();
  handler1->expectEOM([&] {
      handler1->sendReplyWithBody(200, 32 * 1024);
    });

  auto handler2 = addSimpleStrictHandler();
  handler2->expectHeaders();
  handler2->expectEOM([&] {
      handler2->sendReplyWithBody(200, 32 * 1024);
    });

  EXPECT_CALL(stats, recordSessionStalled()).Times(1);

  handler1->expectDetachTransaction();

  // twice- once to send and once to receive
  flushRequestsAndLoopN(2);

  // open the window
  clientCodec_->generateWindowUpdate(requests_, 0, 100);
  handler2->expectDetachTransaction();
  flushRequestsAndLoopN(2);

  httpSession_->closeWhenIdle();
  expectDetachSession();
  flushRequestsAndLoop();
}

TEST_F(HTTP2DownstreamSessionTest, TestTransactionStallByFlowControl) {
  StrictMock<MockHTTPSessionStats> stats;

  httpSession_->setSessionStats(&stats);

  // Set the client side stream level flow control wind to 500 bytes,
  // and try to send 1000 bytes through it.
  // Then the flow control kicks in and stalls the transaction.
  clientCodec_->getEgressSettings()->setSetting(SettingsId::INITIAL_WINDOW_SIZE,
                                                500);
  clientCodec_->generateSettings(requests_);

  auto streamID = sendRequest();

  EXPECT_CALL(stats, recordTransactionOpened());

  InSequence handlerSequence;
  auto handler = addSimpleStrictHandler();
  handler->expectHeaders();
  handler->expectEOM([&] {
      handler->sendReplyWithBody(200, 1000);
    });

  EXPECT_CALL(stats, recordTransactionStalled());
  handler->expectEgressPaused();

  handler->expectError([&] (const HTTPException& ex) {
      ASSERT_EQ(ex.getProxygenError(), kErrorWriteTimeout);
      ASSERT_EQ(
        folly::to<std::string>("ingress timeout, streamID=", streamID),
        std::string(ex.what()));
      handler->terminate();
    });

  handler->expectDetachTransaction();

  EXPECT_CALL(stats, recordTransactionClosed());

  flushRequestsAndLoop();
  gracefulShutdown();
}

TEST_F(HTTP2DownstreamSessionTest, TestTransactionNotStallByFlowControl) {
  StrictMock<MockHTTPSessionStats> stats;

  httpSession_->setSessionStats(&stats);

  clientCodec_->getEgressSettings()->setSetting(SettingsId::INITIAL_WINDOW_SIZE,
                                                500);
  clientCodec_->generateSettings(requests_);

  sendRequest();

  EXPECT_CALL(stats, recordTransactionOpened());

  InSequence handlerSequence;
  auto handler = addSimpleStrictHandler();
  handler->expectHeaders();
  handler->expectEOM([&] {
      handler->sendReplyWithBody(200, 500);
    });

  // The egtress paused is notified due to existing logics,
  // but egress transaction should not be counted as stalled by flow control,
  // because there is nore more bytes to send
  handler->expectEgressPaused();

  handler->expectDetachTransaction();

  EXPECT_CALL(stats, recordTransactionClosed());

  flushRequestsAndLoop();
  gracefulShutdown();
}

TEST_F(HTTP2DownstreamSessionTest, TestSetEgressSettings) {
  SettingsList settings = {{ SettingsId::HEADER_TABLE_SIZE, 5555 },
                           { SettingsId::MAX_FRAME_SIZE, 16384 },
                           { SettingsId::ENABLE_PUSH, 1 }};

  const HTTPSettings* codecSettings = rawCodec_->getEgressSettings();
  for (const auto& setting: settings) {
    const HTTPSetting* currSetting = codecSettings->getSetting(setting.id);
    if (currSetting) {
      EXPECT_EQ(setting.value, currSetting->value);
    }
  }

  flushRequestsAndLoop();
  gracefulShutdown();
}