MockCodecDownstreamTest.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 <wangle/acceptor/ConnectionManager.h>
#include <folly/io/Cursor.h>
#include <folly/io/async/EventBase.h>
#include <folly/io/async/TimeoutManager.h>
#include <folly/portability/GTest.h>
#include <proxygen/lib/http/codec/test/MockHTTPCodec.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/MockSecondaryAuthManager.h>
#include <proxygen/lib/http/session/test/TestUtils.h>
#include <proxygen/lib/test/TestAsyncTransport.h>
#include <sstream>
#include <string>
#include <folly/io/async/test/MockAsyncTransport.h>
#include <vector>
#include <boost/optional/optional_io.hpp>
#include <fizz/record/Extensions.h>
#include <fizz/record/Types.h>

using folly::test::MockAsyncTransport;

using namespace wangle;
using namespace fizz;
using namespace folly;
using namespace proxygen;
using namespace std;
using namespace testing;

const HTTPSettings kDefaultIngressSettings{
  { SettingsId::INITIAL_WINDOW_SIZE, 65536 }
};
const HTTPSettings kIngressCertAuthSettings{
  { SettingsId::SETTINGS_HTTP_CERT_AUTH, 128}
};
HTTPSettings kEgressCertAuthSettings{
  { SettingsId::SETTINGS_HTTP_CERT_AUTH, 128}
};

class MockCodecDownstreamTest: public testing::Test {
 public:
  MockCodecDownstreamTest()
    : eventBase_(),
      codec_(new StrictMock<MockHTTPCodec>()),
      transport_(new NiceMock<MockAsyncTransport>()),
      transactionTimeouts_(makeInternalTimeoutSet(&eventBase_)) {

    EXPECT_CALL(*transport_, writeChain(_, _, _))
      .WillRepeatedly(Invoke(this, &MockCodecDownstreamTest::onWriteChain));
    EXPECT_CALL(*transport_, good())
      .WillRepeatedly(ReturnPointee(&transportGood_));
    EXPECT_CALL(*transport_, closeNow())
      .WillRepeatedly(Assign(&transportGood_, false));
    EXPECT_CALL(*transport_, getEventBase())
      .WillRepeatedly(Return(&eventBase_));
    EXPECT_CALL(*transport_, setReadCB(_))
      .WillRepeatedly(SaveArg<0>(&transportCb_));
    EXPECT_CALL(mockController_, getGracefulShutdownTimeout())
      .WillRepeatedly(Return(std::chrono::milliseconds(0)));
    EXPECT_CALL(mockController_, attachSession(_));
    EXPECT_CALL(*codec_, setCallback(_))
      .WillRepeatedly(SaveArg<0>(&codecCallback_));
    EXPECT_CALL(*codec_, supportsParallelRequests())
      .WillRepeatedly(Return(true));
    EXPECT_CALL(*codec_, supportsPushTransactions())
      .WillRepeatedly(Return(true));
    EXPECT_CALL(*codec_, getTransportDirection())
      .WillRepeatedly(Return(TransportDirection::DOWNSTREAM));
    EXPECT_CALL(*codec_, getEgressSettings());
    EXPECT_CALL(*codec_, supportsStreamFlowControl())
      .WillRepeatedly(Return(true));
    EXPECT_CALL(*codec_, getProtocol())
      .WillRepeatedly(Return(CodecProtocol::SPDY_3_1));
    EXPECT_CALL(*codec_, getUserAgent())
      .WillRepeatedly(ReturnRef(userAgent_));
    EXPECT_CALL(*codec_, setParserPaused(_))
      .WillRepeatedly(Return());
    EXPECT_CALL(*codec_, supportsSessionFlowControl())
      .WillRepeatedly(Return(true)); // simulate spdy 3.1
    EXPECT_CALL(*codec_, getIngressSettings())
      .WillRepeatedly(Return(&kDefaultIngressSettings));
    EXPECT_CALL(*codec_, isReusable())
      .WillRepeatedly(ReturnPointee(&reusable_));
    EXPECT_CALL(*codec_, isWaitingToDrain())
      .WillRepeatedly(ReturnPointee(&drainPending_));
    EXPECT_CALL(*codec_, generateSettings(_));
    EXPECT_CALL(*codec_, getDefaultWindowSize())
      .WillRepeatedly(Return(65536));
    EXPECT_CALL(*codec_, createStream())
      .WillRepeatedly(InvokeWithoutArgs([&] {
            return pushStreamID_ += 2;
          }));
    EXPECT_CALL(*codec_, enableDoubleGoawayDrain())
      .WillRepeatedly(Invoke([&] { doubleGoaway_ = true; }));
    EXPECT_CALL(*codec_, generateGoaway(_, _, _, _))
        .WillRepeatedly(Invoke([this](IOBufQueue& writeBuf,
                                      HTTPCodec::StreamID /*lastStream*/,
                                      ErrorCode,
                                      std::shared_ptr<folly::IOBuf>) {
          if (reusable_) {
            reusable_ = false;
            drainPending_ = doubleGoaway_;
          } else if (!drainPending_) {
            return 0;
          } else {
            drainPending_ = false;
          }
          if (liveGoaways_) {
            writeBuf.append(string("x"));
          }
          return 1;
        }));
    EXPECT_CALL(*codec_, generateRstStream(_, _, _))
      .WillRepeatedly(Return(1));
    EXPECT_CALL(*codec_, addPriorityNodes(_, _, _))
      .WillOnce(Return(0));
    EXPECT_CALL(*codec_, mapPriorityToDependency(_))
      .WillRepeatedly(Return(0));

    HTTPSession::setDefaultReadBufferLimit(65536);
    httpSession_ = new HTTPDownstreamSession(
        transactionTimeouts_.get(),
        AsyncTransportWrapper::UniquePtr(transport_), localAddr, peerAddr,
        &mockController_, std::unique_ptr<HTTPCodec>(codec_),
        mockTransportInfo, nullptr);
    httpSession_->startNow();
    eventBase_.loop();
  }

  void onWriteChain(folly::AsyncTransportWrapper::WriteCallback* callback,
                    std::shared_ptr<IOBuf> /*iob*/,
                    WriteFlags) {
    writeCount_++;
    if (invokeWriteSuccess_) {
      callback->writeSuccess();
    } else {
      cbs_.push_back(callback);
    }
  }

  ~MockCodecDownstreamTest() override {
    AsyncSocketException ex(AsyncSocketException::UNKNOWN, "");
    for (auto& cb : cbs_) {
      cb->writeErr(0, ex);
    }
  }

  void SetUp() override {
    HTTPSession::setDefaultWriteBufferLimit(65536);
  }

  // Pass a function to execute inside HTTPCodec::onIngress(). This
  // function also takes care of passing an empty ingress buffer to the codec.
  template<class T>
  void onIngressImpl(T f) {
    EXPECT_CALL(*codec_, onIngress(_))
      .WillOnce(Invoke([&f] (const IOBuf& buf) {
            CHECK_GT(buf.computeChainDataLength(), 0);
            // The test should be independent of the dummy buffer,
            // so don't pass it in.
            f();
            return buf.computeChainDataLength();
          }));

    void* buf;
    size_t bufSize;
    transportCb_->getReadBuffer(&buf, &bufSize);
    transportCb_->readDataAvailable(bufSize);
  }

  void testGoaway(bool doubleGoaway, bool dropConnection);

  void testConnFlowControlBlocked(bool timeout);

 protected:

  EventBase eventBase_;
  // invalid once httpSession_ is destroyed
  StrictMock<MockHTTPCodec>* codec_;
  std::string userAgent_{"MockCodec"};
  HTTPCodec::Callback* codecCallback_{nullptr};
  NiceMock<MockAsyncTransport>* transport_;
  folly::AsyncTransportWrapper::ReadCallback* transportCb_;
  folly::HHWheelTimer::UniquePtr transactionTimeouts_;
  StrictMock<MockController> mockController_;
  HTTPDownstreamSession* httpSession_;
  HTTPCodec::StreamID pushStreamID_{0};
  bool reusable_{true};
  bool transportGood_{true};
  bool drainPending_{false};
  bool doubleGoaway_{false};
  bool liveGoaways_{false};
  bool invokeWriteSuccess_{false};
  uint32_t writeCount_{0};
  std::vector<folly::AsyncTransportWrapper::WriteCallback*> cbs_;
};

TEST_F(MockCodecDownstreamTest, OnAbortThenTimeouts) {
  // Test what happens when txn1 (out of many transactions) gets an abort
  // followed by a transaction timeout followed by a write timeout
  MockHTTPHandler handler1;
  MockHTTPHandler handler2;
  auto req1 = makeGetRequest();
  auto req2 = makeGetRequest();

  fakeMockCodec(*codec_);

  EXPECT_CALL(mockController_, getRequestHandler(_, _))
    .WillOnce(Return(&handler1))
    .WillOnce(Return(&handler2));

  EXPECT_CALL(handler1, setTransaction(_))
    .WillOnce(Invoke([&handler1] (HTTPTransaction* txn) {
          handler1.txn_ = txn; }));
  EXPECT_CALL(handler1, onHeadersComplete(_))
    .WillOnce(Invoke([&handler1] (std::shared_ptr<HTTPMessage>) {
          handler1.sendHeaders(200, 100);
          handler1.sendBody(100);
        }));
  EXPECT_CALL(handler1, onError(_));
  EXPECT_CALL(handler1, detachTransaction());
  EXPECT_CALL(handler2, setTransaction(_))
    .WillOnce(Invoke([&handler2] (HTTPTransaction* txn) {
          handler2.txn_ = txn; }));
  EXPECT_CALL(handler2, onHeadersComplete(_))
    .WillOnce(Invoke([&handler2] (std::shared_ptr<HTTPMessage>) {
          handler2.sendHeaders(200, 100);
          handler2.sendBody(100);
        }));
  EXPECT_CALL(handler2, onBody(_));
  EXPECT_CALL(handler2, onError(_))
    .WillOnce(Invoke([&] (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()));
        }));
  EXPECT_CALL(handler2, detachTransaction());

  EXPECT_CALL(mockController_, detachSession(_));

  codecCallback_->onMessageBegin(HTTPCodec::StreamID(1), req1.get());
  codecCallback_->onHeadersComplete(HTTPCodec::StreamID(1), std::move(req1));
  codecCallback_->onMessageBegin(HTTPCodec::StreamID(3), req2.get());
  codecCallback_->onHeadersComplete(HTTPCodec::StreamID(3), std::move(req2));
  // do the write, enqeue byte event
  eventBase_.loop();

  // recv an abort, detach the handler from txn1 (txn1 stays around due to the
  // enqueued byte event)
  codecCallback_->onAbort(HTTPCodec::StreamID(1), ErrorCode::PROTOCOL_ERROR);
  // recv a transaction timeout on txn1 (used to erroneously create a direct
  // response handler)
  handler1.txn_->timeoutExpired();

  // have a write timeout expire (used to cause the direct response handler to
  // write out data, messing up the state machine)
  eventBase_.runAfterDelay(
    [this] {
      // refresh ingress timeout
      codecCallback_->onBody(HTTPCodec::StreamID(3), makeBuf(10), 0);
    }, transactionTimeouts_->getDefaultTimeout().count() / 2);
  // hold evb open long enough to fire write timeout (since transactionTimeouts_
  // is internal
  eventBase_.runAfterDelay(
    [] {}, transactionTimeouts_->getDefaultTimeout().count() + 100);
  eventBase_.loop();
}

TEST_F(MockCodecDownstreamTest, ServerPush) {
  MockHTTPHandler handler;
  MockHTTPPushHandler pushHandler;
  auto req = makeGetRequest();
  HTTPTransaction* pushTxn = nullptr;

  InSequence enforceOrder;

  EXPECT_CALL(mockController_, getRequestHandler(_, _))
    .WillOnce(Return(&handler));
  EXPECT_CALL(handler, setTransaction(_))
    .WillOnce(SaveArg<0>(&handler.txn_));

  EXPECT_CALL(handler, onHeadersComplete(_))
    .WillOnce(Invoke([&] (std::shared_ptr<HTTPMessage>) {
          pushTxn = handler.txn_->newPushedTransaction(&pushHandler);
          pushHandler.sendPushHeaders("/foo", "www.foo.com", 100,
                                      handler.txn_->getPriority());
          pushHandler.sendBody(100);
          pushTxn->sendEOM();
          eventBase_.loop(); // flush the push txn's body
        }));
  EXPECT_CALL(pushHandler, setTransaction(_))
    .WillOnce(Invoke([&pushHandler] (HTTPTransaction* txn) {
          pushHandler.txn_ = txn; }));

  EXPECT_CALL(*codec_, generatePushPromise(_, 2, _, _, _, _));
  EXPECT_CALL(*codec_, generateBody(_, 2, PtrBufHasLen(uint64_t(100)),
                                    _, true));
  EXPECT_CALL(pushHandler, detachTransaction());

  EXPECT_CALL(handler, onEOM())
    .WillOnce(Invoke([&] {
          handler.sendReplyWithBody(200, 100);
          eventBase_.loop(); // flush the response to the normal request
        }));

  EXPECT_CALL(*codec_, generateHeader(_, 1, _, _, _));
  EXPECT_CALL(*codec_, generateBody(_, 1, PtrBufHasLen(uint64_t(100)),
                                    _, true));
  EXPECT_CALL(handler, detachTransaction());

  codecCallback_->onMessageBegin(HTTPCodec::StreamID(1), req.get());
  codecCallback_->onHeadersComplete(HTTPCodec::StreamID(1), std::move(req));
  codecCallback_->onMessageComplete(HTTPCodec::StreamID(1), false);

  EXPECT_CALL(*codec_, onIngressEOF());
  EXPECT_CALL(mockController_, detachSession(_));
  httpSession_->dropConnection();
}

TEST_F(MockCodecDownstreamTest, ServerPushAfterGoaway) {
  // Tests if goaway
  //   - drains acknowledged server push transactions
  //   - aborts server pushed transactions not created at the client
  //   - prevents new transactions from being created.
  MockHTTPHandler handler;
  MockHTTPPushHandler pushHandler1;
  MockHTTPPushHandler pushHandler2;
  HTTPTransaction* pushTxn = nullptr;

  fakeMockCodec(*codec_);

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

  EXPECT_CALL(handler, setTransaction(_))
    .WillOnce(Invoke([&handler] (HTTPTransaction* txn) {
          handler.txn_ = txn; }));
  EXPECT_CALL(handler, onHeadersComplete(_))
    .WillOnce(Invoke([&] (std::shared_ptr<HTTPMessage>) {
          // Initiate server push transactions.
          pushTxn = handler.txn_->newPushedTransaction(&pushHandler1);
          CHECK_EQ(pushTxn->getID(), HTTPCodec::StreamID(2));
          pushHandler1.sendPushHeaders("/foo", "www.foo.com", 100,
                                       handler.txn_->getPriority());
          pushHandler1.sendBody(100);
          pushTxn->sendEOM();
          // Initiate the second push transaction which will be aborted
          pushTxn = handler.txn_->newPushedTransaction(&pushHandler2);
          CHECK_EQ(pushTxn->getID(), HTTPCodec::StreamID(4));
          pushHandler2.sendPushHeaders("/foo", "www.foo.com", 100,
                                       handler.txn_->getPriority());
          pushHandler2.sendBody(100);
          pushTxn->sendEOM();
        }));
  // Push transaction 1 - drained
  EXPECT_CALL(pushHandler1, setTransaction(_))
    .WillOnce(Invoke([&pushHandler1] (HTTPTransaction* txn) {
          pushHandler1.txn_ = txn; }));
  EXPECT_CALL(pushHandler1, detachTransaction());
  // Push transaction 2 - aborted by onError after goaway
  EXPECT_CALL(pushHandler2, setTransaction(_))
    .WillOnce(Invoke([&pushHandler2] (HTTPTransaction* txn) {
          pushHandler2.txn_ = txn; }));
  EXPECT_CALL(pushHandler2, onError(_))
    .WillOnce(Invoke([&] (const HTTPException& err) {
          EXPECT_TRUE(err.hasProxygenError());
          EXPECT_EQ(err.getProxygenError(), kErrorStreamUnacknowledged);
          ASSERT_EQ(
            folly::to<std::string>("StreamUnacknowledged on transaction id: ",
              pushHandler2.txn_->getID()),
            std::string(err.what()));
        }));
  EXPECT_CALL(pushHandler2, detachTransaction());

  EXPECT_CALL(handler, onEOM());
  EXPECT_CALL(handler, detachTransaction());

  // Receive client request
  auto req = makeGetRequest();
  codecCallback_->onMessageBegin(HTTPCodec::StreamID(1), req.get());
  codecCallback_->onHeadersComplete(HTTPCodec::StreamID(1), std::move(req));
  codecCallback_->onMessageComplete(HTTPCodec::StreamID(1), false);

  // Receive goaway acknowledging only the first pushed transactions with id 2.
  codecCallback_->onGoaway(2, ErrorCode::NO_ERROR);

  // New server pushed transaction cannot be created after goaway
  MockHTTPPushHandler pushHandler3;
  EXPECT_EQ(handler.txn_->newPushedTransaction(&pushHandler3), nullptr);

  // Send response to the initial client request and this destroys the session
  handler.sendReplyWithBody(200, 100);

  eventBase_.loop();

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

TEST_F(MockCodecDownstreamTest, ServerPushAbort) {
  // Test that assoc txn and other push txns are not affected when client aborts
  // a push txn
  MockHTTPHandler handler;
  MockHTTPPushHandler pushHandler1;
  MockHTTPPushHandler pushHandler2;
  HTTPTransaction* pushTxn1 = nullptr;
  HTTPTransaction* pushTxn2 = nullptr;

  fakeMockCodec(*codec_);

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

  EXPECT_CALL(handler, setTransaction(_))
    .WillOnce(Invoke([&handler] (HTTPTransaction* txn) {
          handler.txn_ = txn; }));
  EXPECT_CALL(handler, onHeadersComplete(_))
    .WillOnce(Invoke([&] (std::shared_ptr<HTTPMessage>) {
          // Initiate server push transactions
          pushTxn1 = handler.txn_->newPushedTransaction(&pushHandler1);
          CHECK_EQ(pushTxn1->getID(), HTTPCodec::StreamID(2));
          pushHandler1.sendPushHeaders("/foo", "www.foo.com", 100,
                                      handler.txn_->getPriority());
          pushHandler1.sendBody(100);

          pushTxn2 = handler.txn_->newPushedTransaction(&pushHandler2);
          CHECK_EQ(pushTxn2->getID(), HTTPCodec::StreamID(4));
          pushHandler2.sendPushHeaders("/bar", "www.bar.com", 200,
                                       handler.txn_->getPriority());
          pushHandler2.sendBody(200);
          pushTxn2->sendEOM();
        }));

  // pushTxn1 should be aborted
  EXPECT_CALL(pushHandler1, setTransaction(_))
    .WillOnce(Invoke([&pushHandler1] (HTTPTransaction* txn) {
          pushHandler1.txn_ = txn; }));
  EXPECT_CALL(pushHandler1, onError(_))
    .WillOnce(Invoke([&] (const HTTPException& err) {
          EXPECT_TRUE(err.hasProxygenError());
          EXPECT_EQ(err.getProxygenError(), kErrorStreamAbort);
          ASSERT_EQ(
            "Stream aborted, streamID=2, code=CANCEL",
            std::string(err.what()));
        }));
  EXPECT_CALL(pushHandler1, detachTransaction());

  EXPECT_CALL(pushHandler2, setTransaction(_))
    .WillOnce(Invoke([&pushHandler2] (HTTPTransaction* txn) {
          pushHandler2.txn_ = txn; }));
  EXPECT_CALL(pushHandler2, detachTransaction());

  EXPECT_CALL(handler, onEOM());
  EXPECT_CALL(handler, detachTransaction());

  // Receive client request
  auto req = makeGetRequest();
  codecCallback_->onMessageBegin(HTTPCodec::StreamID(1), req.get());
  codecCallback_->onHeadersComplete(HTTPCodec::StreamID(1), std::move(req));
  codecCallback_->onMessageComplete(HTTPCodec::StreamID(1), false);

  // Send client abort on one push txn
  codecCallback_->onAbort(HTTPCodec::StreamID(2), ErrorCode::CANCEL);

  handler.sendReplyWithBody(200, 100);

  eventBase_.loop();

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

TEST_F(MockCodecDownstreamTest, ServerPushAbortAssoc) {
  // Test that all associated push transactions are aborted when client aborts
  // the assoc stream
  MockHTTPHandler handler;
  MockHTTPPushHandler pushHandler1;
  MockHTTPPushHandler pushHandler2;

  fakeMockCodec(*codec_);

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

  EXPECT_CALL(handler, setTransaction(_))
    .WillOnce(Invoke([&handler] (HTTPTransaction* txn) {
          handler.txn_ = txn; }));
  EXPECT_CALL(handler, onHeadersComplete(_))
    .WillOnce(Invoke([&] (std::shared_ptr<HTTPMessage>) {
          // Initiate server push transactions
          auto pushTxn = handler.txn_->newPushedTransaction(&pushHandler1);
          CHECK_EQ(pushTxn->getID(), HTTPCodec::StreamID(2));
          pushHandler1.sendPushHeaders("/foo", "www.foo.com", 100,
                                       handler.txn_->getPriority());
          pushHandler1.sendBody(100);
          eventBase_.loop();

          pushTxn = handler.txn_->newPushedTransaction(&pushHandler2);
          CHECK_EQ(pushTxn->getID(), HTTPCodec::StreamID(4));
          pushHandler2.sendPushHeaders("/foo", "www.foo.com", 100,
                                       handler.txn_->getPriority());
          pushHandler2.sendBody(100);
          eventBase_.loop();
        }));

  // Both push txns and the assoc txn should be aborted
  EXPECT_CALL(pushHandler1, setTransaction(_))
    .WillOnce(Invoke([&pushHandler1] (HTTPTransaction* txn) {
          pushHandler1.txn_ = txn; }));
  EXPECT_CALL(pushHandler1, onError(_))
    .WillOnce(Invoke([&] (const HTTPException& err) {
          EXPECT_TRUE(err.hasProxygenError());
          EXPECT_EQ(err.getProxygenError(), kErrorStreamAbort);
          ASSERT_EQ(
            "Stream aborted, streamID=1, code=CANCEL",
            std::string(err.what()));
        }));
  EXPECT_CALL(pushHandler1, detachTransaction());

  EXPECT_CALL(pushHandler2, setTransaction(_))
    .WillOnce(Invoke([&pushHandler2] (HTTPTransaction* txn) {
          pushHandler2.txn_ = txn; }));
  EXPECT_CALL(pushHandler2, onError(_))
    .WillOnce(Invoke([&] (const HTTPException& err) {
          EXPECT_TRUE(err.hasProxygenError());
          EXPECT_EQ(err.getProxygenError(), kErrorStreamAbort);
          ASSERT_EQ(
            "Stream aborted, streamID=1, code=CANCEL",
            std::string(err.what()));
        }));
  EXPECT_CALL(pushHandler2, detachTransaction());

  EXPECT_CALL(handler, onError(_))
    .WillOnce(Invoke([&] (const HTTPException& err) {
          EXPECT_TRUE(err.hasProxygenError());
          EXPECT_EQ(err.getProxygenError(), kErrorStreamAbort);
          ASSERT_EQ(
            "Stream aborted, streamID=1, code=CANCEL",
            std::string(err.what()));
        }));
  EXPECT_CALL(handler, detachTransaction());

  // Receive client request
  auto req = makeGetRequest();
  codecCallback_->onMessageBegin(HTTPCodec::StreamID(1), req.get());
  codecCallback_->onHeadersComplete(HTTPCodec::StreamID(1), std::move(req));

  // Send client abort on assoc stream
  codecCallback_->onAbort(HTTPCodec::StreamID(1), ErrorCode::CANCEL);

  eventBase_.loop();

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

TEST_F(MockCodecDownstreamTest, ServerPushClientMessage) {
  // Test that error is generated when client sends data on a pushed stream
  MockHTTPHandler handler;
  MockHTTPPushHandler pushHandler;
  auto req = makeGetRequest();
  HTTPTransaction* pushTxn = nullptr;

  InSequence enforceOrder;

  EXPECT_CALL(mockController_, getRequestHandler(_, _))
    .WillOnce(Return(&handler));
  EXPECT_CALL(handler, setTransaction(_))
    .WillOnce(SaveArg<0>(&handler.txn_));

  EXPECT_CALL(handler, onHeadersComplete(_))
    .WillOnce(Invoke([&] (std::shared_ptr<HTTPMessage> msg) {
          pushTxn = handler.txn_->newPushedTransaction(&pushHandler);
          auto pri = handler.txn_->getPriority();
          msg->setHTTP2Priority(std::make_tuple(pri.streamDependency,
                                                pri.exclusive, pri.weight));
        }));
  EXPECT_CALL(pushHandler, setTransaction(_))
    .WillOnce(Invoke([&pushHandler] (HTTPTransaction* txn) {
          pushHandler.txn_ = txn; }));

  codecCallback_->onMessageBegin(HTTPCodec::StreamID(1), req.get());
  codecCallback_->onHeadersComplete(HTTPCodec::StreamID(1), std::move(req));

  EXPECT_CALL(*codec_, generateRstStream(_, 2, ErrorCode::STREAM_CLOSED))
    .WillRepeatedly(Return(1));
  EXPECT_CALL(pushHandler, onError(_))
    .WillOnce(Invoke([&] (const HTTPException& ex) {
          EXPECT_TRUE(ex.hasCodecStatusCode());
          EXPECT_EQ(ex.getCodecStatusCode(), ErrorCode::STREAM_CLOSED);
          ASSERT_EQ(
            "Downstream attempts to send ingress, abort.",
            std::string(ex.what()));
        }));
  EXPECT_CALL(pushHandler, detachTransaction());

  // While the assoc stream is open and pushHandler has been initialized, send
  // an upstream message on the push stream causing a RST_STREAM.
  req = makeGetRequest();
  codecCallback_->onMessageBegin(HTTPCodec::StreamID(2), req.get());

  EXPECT_CALL(handler, onEOM())
    .WillOnce(InvokeWithoutArgs([&] {
          handler.sendReplyWithBody(200, 100);
          eventBase_.loop(); // flush the response to the assoc request
        }));
  EXPECT_CALL(*codec_, generateHeader(_, 1, _, _, _));
  EXPECT_CALL(*codec_, generateBody(_, 1, PtrBufHasLen(uint64_t(100)),
                                    _, true));
  EXPECT_CALL(handler, detachTransaction());

  // Complete the assoc request/response
  codecCallback_->onMessageComplete(HTTPCodec::StreamID(1), false);

  eventBase_.loop();

  EXPECT_CALL(*codec_, onIngressEOF());
  EXPECT_CALL(mockController_, detachSession(_));
  httpSession_->dropConnection();
}

TEST_F(MockCodecDownstreamTest, ReadTimeout) {
  // Test read timeout path
  MockHTTPHandler handler1;
  auto req1 = makeGetRequest();

  fakeMockCodec(*codec_);
  EXPECT_CALL(*codec_, onIngressEOF())
    .WillRepeatedly(Return());

  EXPECT_CALL(mockController_, getRequestHandler(_, _))
    .WillOnce(Return(&handler1));

  EXPECT_CALL(handler1, setTransaction(_))
    .WillOnce(Invoke([&handler1] (HTTPTransaction* txn) {
          handler1.txn_ = txn; }));
  EXPECT_CALL(handler1, onHeadersComplete(_));

  codecCallback_->onMessageBegin(HTTPCodec::StreamID(1), req1.get());
  codecCallback_->onHeadersComplete(HTTPCodec::StreamID(1), std::move(req1));
  // force the read timeout to expire, should be a no-op because the txn is
  // still expecting EOM and has its own timer.
  httpSession_->timeoutExpired();
  EXPECT_EQ(httpSession_->getConnectionCloseReason(),
            ConnectionCloseReason::kMAX_REASON);

  EXPECT_CALL(handler1, onEOM())
    .WillOnce(Invoke([&handler1] () {
          handler1.txn_->pauseIngress();
        }));

  // send the EOM, then another timeout.  Still no-op since it's waiting
  // upstream
  codecCallback_->onMessageComplete(HTTPCodec::StreamID(1), false);
  httpSession_->timeoutExpired();
  EXPECT_EQ(httpSession_->getConnectionCloseReason(),
            ConnectionCloseReason::kMAX_REASON);

  EXPECT_CALL(*transport_, writeChain(_, _, _))
    .WillRepeatedly(Invoke([] (
            folly::AsyncTransportWrapper::WriteCallback* callback,
            std::shared_ptr<folly::IOBuf>,
            folly::WriteFlags) {
          callback->writeSuccess();
          }));

  EXPECT_CALL(handler1, detachTransaction());

  // Send the response, timeout.  Now it's idle and should close.
  handler1.txn_->resumeIngress();
  handler1.sendReplyWithBody(200, 100);
  eventBase_.loop();

  httpSession_->timeoutExpired();
  EXPECT_EQ(httpSession_->getConnectionCloseReason(),
            ConnectionCloseReason::TIMEOUT);

  // tear down the test
  EXPECT_CALL(mockController_, detachSession(_));
  httpSession_->dropConnection();
}

TEST_F(MockCodecDownstreamTest, Ping) {
  // Test ping mechanism and that we prioritize the ping reply
  MockHTTPHandler handler1;
  auto req1 = makeGetRequest();

  InSequence enforceOrder;

  EXPECT_CALL(mockController_, getRequestHandler(_, _))
    .WillOnce(Return(&handler1));

  EXPECT_CALL(handler1, setTransaction(_))
    .WillOnce(Invoke([&handler1] (HTTPTransaction* txn) {
          handler1.txn_ = txn; }));
  EXPECT_CALL(handler1, onHeadersComplete(_));
  EXPECT_CALL(handler1, onEOM())
    .WillOnce(InvokeWithoutArgs([&handler1] () {
          handler1.sendReplyWithBody(200, 100);
        }));

  // Header egresses immediately
  EXPECT_CALL(*codec_, generateHeader(_, _, _, _, _));
  // Ping jumps ahead of queued body in the loop callback
  EXPECT_CALL(*codec_, generatePingReply(_, _));
  EXPECT_CALL(*codec_, generateBody(_, _, _, _, true));
  EXPECT_CALL(handler1, detachTransaction());

  codecCallback_->onMessageBegin(HTTPCodec::StreamID(1), req1.get());
  codecCallback_->onHeadersComplete(HTTPCodec::StreamID(1), std::move(req1));
  codecCallback_->onMessageComplete(HTTPCodec::StreamID(1), false);
  codecCallback_->onPingRequest(1);

  eventBase_.loop();

  EXPECT_CALL(*codec_, onIngressEOF());
  EXPECT_CALL(mockController_, detachSession(_));
  httpSession_->dropConnection();
}

TEST_F(MockCodecDownstreamTest, FlowControlAbort) {
  MockHTTPHandler handler1;
  auto req1 = makePostRequest();

  InSequence enforceOrder;

  EXPECT_CALL(mockController_, getRequestHandler(_, _))
    .WillOnce(Return(&handler1));

  EXPECT_CALL(handler1, setTransaction(_))
    .WillOnce(Invoke([&handler1] (HTTPTransaction* txn) {
          handler1.txn_ = txn; }));
  EXPECT_CALL(handler1, onHeadersComplete(_))
    .WillOnce(InvokeWithoutArgs([&handler1] () {
          handler1.txn_->sendAbort();
        }));

  // Header egresses immediately
  EXPECT_CALL(handler1, detachTransaction());

  codecCallback_->onMessageBegin(HTTPCodec::StreamID(1), req1.get());
  codecCallback_->onHeadersComplete(HTTPCodec::StreamID(1), std::move(req1));
  EXPECT_CALL(*codec_, generateWindowUpdate(_, 0, spdy::kInitialWindow));
  codecCallback_->onBody(HTTPCodec::StreamID(1),
                         makeBuf(spdy::kInitialWindow), 0);
  EXPECT_CALL(*codec_, generateWindowUpdate(_, 0, spdy::kInitialWindow));
  codecCallback_->onBody(HTTPCodec::StreamID(1),
                         makeBuf(spdy::kInitialWindow), 0);

  eventBase_.loop();

  EXPECT_CALL(*codec_, onIngressEOF());
  EXPECT_CALL(mockController_, detachSession(_));
  httpSession_->dropConnection();
}

TEST_F(MockCodecDownstreamTest, Buffering) {
  StrictMock<MockHTTPHandler> handler;
  auto req1 = makePostRequest(20);
  auto chunk = makeBuf(10);
  auto chunkStr = chunk->clone()->moveToFbString();

  fakeMockCodec(*codec_);

  httpSession_->setDefaultReadBufferLimit(10);

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

  EXPECT_CALL(handler, setTransaction(_))
    .WillOnce(Invoke([&handler] (HTTPTransaction* txn) {
          handler.txn_ = txn; }));
  EXPECT_CALL(handler, onHeadersComplete(_))
    .WillOnce(InvokeWithoutArgs([&handler] () {
          handler.txn_->pauseIngress();
        }));

  EXPECT_CALL(*transport_, writeChain(_, _, _))
    .WillRepeatedly(Invoke([&] (
            folly::AsyncTransportWrapper::WriteCallback* callback,
            const shared_ptr<IOBuf>&,
            WriteFlags) {
          callback->writeSuccess();
          }));

  codecCallback_->onMessageBegin(HTTPCodec::StreamID(1), req1.get());
  codecCallback_->onHeadersComplete(HTTPCodec::StreamID(1), std::move(req1));
  for (int i = 0; i < 2; i++) {
    codecCallback_->onBody(HTTPCodec::StreamID(1), chunk->clone(), 0);
  }
  codecCallback_->onMessageComplete(HTTPCodec::StreamID(1), false);

  EXPECT_CALL(handler, onBody(_))
    .WillOnce(ExpectString(chunkStr))
    .WillOnce(ExpectString(chunkStr));

  EXPECT_CALL(handler, onEOM());

  EXPECT_CALL(handler, detachTransaction());

  eventBase_.tryRunAfterDelay([&handler, this] {
      handler.txn_->resumeIngress();
      handler.sendReplyWithBody(200, 100);
      eventBase_.runInLoop([this] { httpSession_->dropConnection(); });
    }, 30);

  EXPECT_CALL(*codec_, onIngressEOF());
  EXPECT_CALL(mockController_, detachSession(_));
  eventBase_.loop();
}

TEST_F(MockCodecDownstreamTest, SpdyWindow) {
  // Test window updates
  MockHTTPHandler handler1;
  auto req1 = makeGetRequest();

  fakeMockCodec(*codec_);

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

    EXPECT_CALL(handler1, setTransaction(_))
      .WillOnce(Invoke([&handler1] (HTTPTransaction* txn) {
            handler1.txn_ = txn; }));
    EXPECT_CALL(handler1, onHeadersComplete(_))
      .WillOnce(InvokeWithoutArgs([this] () {
            codecCallback_->onSettings(
              {{SettingsId::INITIAL_WINDOW_SIZE, 4000}});
          }));
    EXPECT_CALL(*codec_, generateSettingsAck(_));
    EXPECT_CALL(handler1, onEOM())
      .WillOnce(InvokeWithoutArgs([&handler1] () {
            handler1.sendHeaders(200, 16000);
            handler1.sendBody(12000);
            // 12kb buffered -> pause upstream
          }));
    EXPECT_CALL(handler1, onEgressPaused())
      .WillOnce(InvokeWithoutArgs([this] () {
            eventBase_.runInLoop([this] {
                // triggers 4k send, 8kb buffered, handler still paused
                codecCallback_->onWindowUpdate(1, 4000);
              });
            eventBase_.runAfterDelay([this] {
                // triggers 6k send, 2kb buffered, handler still paused
                codecCallback_->onWindowUpdate(1, 6000);
              }, 10);
            eventBase_.runAfterDelay([this] {
                // triggers 2kb send, 0 buffered, 2k window => resume
                codecCallback_->onWindowUpdate(1, 4000);
              }, 20);
          }));
    EXPECT_CALL(handler1, onEgressResumed())
      .WillOnce(InvokeWithoutArgs([&handler1] () {
            handler1.sendBody(4000);
            // 2kb send, 2kb buffered => pause upstream
          }));
    EXPECT_CALL(handler1, onEgressPaused())
      .WillOnce(InvokeWithoutArgs([this] () {
            eventBase_.runInLoop([this] {
                // triggers 2kb send, resume
                codecCallback_->onWindowUpdate(1, 4000);
              });
          }));
    EXPECT_CALL(handler1, onEgressResumed())
      .WillOnce(InvokeWithoutArgs([&handler1] () {
            handler1.txn_->sendEOM();
          }));

    EXPECT_CALL(handler1, detachTransaction());

    codecCallback_->onMessageBegin(HTTPCodec::StreamID(1), req1.get());
    codecCallback_->onHeadersComplete(HTTPCodec::StreamID(1), std::move(req1));
    codecCallback_->onMessageComplete(HTTPCodec::StreamID(1), false);
    // Pad coverage numbers
    std::ostringstream stream;
    stream << *handler1.txn_ << *httpSession_
           << httpSession_->getLocalAddress()
           << httpSession_->getPeerAddress();
    EXPECT_TRUE(httpSession_->isBusy());

    EXPECT_CALL(*codec_, onIngressEOF());
    EXPECT_CALL(mockController_, detachSession(_));
  }

  EXPECT_CALL(*transport_, writeChain(_, _, _))
    .WillRepeatedly(Invoke([] (
            folly::AsyncTransportWrapper::WriteCallback* callback,
            std::shared_ptr<folly::IOBuf>,
            folly::WriteFlags) {
          callback->writeSuccess();
          }));
  eventBase_.loop();
  httpSession_->dropConnection();
}

TEST_F(MockCodecDownstreamTest, DoubleResume) {
  // Test spdy ping mechanism and egress re-ordering
  MockHTTPHandler handler1;
  auto req1 = makePostRequest(5);
  auto buf = makeBuf(5);
  auto bufStr = buf->clone()->moveToFbString();

  fakeMockCodec(*codec_);

  EXPECT_CALL(mockController_, getRequestHandler(_, _))
    .WillOnce(Return(&handler1));

  EXPECT_CALL(handler1, setTransaction(_))
    .WillOnce(Invoke([&handler1] (HTTPTransaction* txn) {
          handler1.txn_ = txn; }));
  EXPECT_CALL(handler1, onHeadersComplete(_))
    .WillOnce(InvokeWithoutArgs([&handler1, this] {
          handler1.txn_->pauseIngress();
          eventBase_.tryRunAfterDelay([&handler1] {
              handler1.txn_->resumeIngress();
            }, 50);
        }));
  EXPECT_CALL(handler1, onBody(_))
    .WillOnce(Invoke([&handler1, &bufStr] (
                       std::shared_ptr<folly::IOBuf> chain) {
          EXPECT_EQ(bufStr, chain->moveToFbString());
          handler1.txn_->pauseIngress();
          handler1.txn_->resumeIngress();
        }));

  EXPECT_CALL(handler1, onEOM())
    .WillOnce(InvokeWithoutArgs([&handler1] () {
          handler1.sendReplyWithBody(200, 100, false);
        }));
  EXPECT_CALL(handler1, detachTransaction());

  codecCallback_->onMessageBegin(HTTPCodec::StreamID(1), req1.get());
  codecCallback_->onHeadersComplete(HTTPCodec::StreamID(1), std::move(req1));
  codecCallback_->onBody(HTTPCodec::StreamID(1), std::move(buf), 0);
  codecCallback_->onMessageComplete(HTTPCodec::StreamID(1), false);

  EXPECT_CALL(*codec_, onIngressEOF());
  EXPECT_CALL(mockController_, detachSession(_));

  EXPECT_CALL(*transport_, writeChain(_, _, _))
    .WillRepeatedly(Invoke([] (
            folly::AsyncTransportWrapper::WriteCallback* callback,
            std::shared_ptr<folly::IOBuf>,
            folly::WriteFlags) {
          callback->writeSuccess();
          }));

  eventBase_.loop();
  httpSession_->dropConnection();
}

void MockCodecDownstreamTest::testConnFlowControlBlocked(bool timeout) {
  // Let the connection level flow control window fill and then make sure
  // control frames still can be processed
  InSequence enforceOrder;
  NiceMock<MockHTTPHandler> handler1;
  NiceMock<MockHTTPHandler> handler2;
  auto wantToWrite = spdy::kInitialWindow + 50000;
  auto wantToWriteStr = folly::to<string>(wantToWrite);
  auto req1 = makeGetRequest();
  auto req2 = makeGetRequest();
  auto resp1 = makeResponse(200);
  resp1->getHeaders().set(HTTP_HEADER_CONTENT_LENGTH, wantToWriteStr);
  auto resp2 = makeResponse(200);
  resp2->getHeaders().set(HTTP_HEADER_CONTENT_LENGTH, wantToWriteStr);
  invokeWriteSuccess_ = true;

  EXPECT_CALL(mockController_, getRequestHandler(_, _))
    .WillOnce(Return(&handler1));
  EXPECT_CALL(handler1, setTransaction(_))
    .WillOnce(SaveArg<0>(&handler1.txn_));
  EXPECT_CALL(handler1, onHeadersComplete(_));
  EXPECT_CALL(*codec_, generateHeader(_, 1, _, _, _))
    .WillOnce(Invoke([] (folly::IOBufQueue& writeBuf,
                         HTTPCodec::StreamID,
                         const HTTPMessage&,
                         bool,
                         HTTPHeaderSize*) {
                       writeBuf.append("", 1);
                     }));
  unsigned bodyLen = 0;
  EXPECT_CALL(*codec_, generateBody(_, 1, _, _, false))
      .WillRepeatedly(Invoke([&](folly::IOBufQueue& /*writeBuf*/,
                                 HTTPCodec::StreamID,
                                 std::shared_ptr<folly::IOBuf> chain,
                                 folly::Optional<uint8_t>,
                                 bool /*eom*/) {
        bodyLen += chain->computeChainDataLength();
        return 0; // don't want byte events
      }));

  codecCallback_->onMessageBegin(1, req1.get());
  codecCallback_->onHeadersComplete(1, std::move(req1));
  codecCallback_->onWindowUpdate(1, wantToWrite); // ensure the per-stream
                                                  // window doesn't block
  handler1.txn_->sendHeaders(*resp1);
  handler1.txn_->sendBody(makeBuf(wantToWrite)); // conn blocked, stream open
  handler1.txn_->sendEOM();
  eventBase_.loop(); // actually send (most of) the body
  CHECK_EQ(bodyLen, spdy::kInitialWindow); // should have written a full window

  EXPECT_CALL(mockController_, getRequestHandler(_, _))
    .WillOnce(Return(&handler2));
  EXPECT_CALL(handler2, setTransaction(_))
    .WillOnce(SaveArg<0>(&handler2.txn_));
  EXPECT_CALL(handler2, onHeadersComplete(_));
  EXPECT_CALL(*codec_, generateHeader(_, 3, _, _, _))
    .WillOnce(Invoke([] (folly::IOBufQueue& writeBuf,
                         HTTPCodec::StreamID,
                         const HTTPMessage&,
                         bool,
                         HTTPHeaderSize*) {
                       writeBuf.append("", 1);
                     }));

  auto writeCount = writeCount_;

  // Make sure we can send headers of response to a second request
  codecCallback_->onMessageBegin(3, req2.get());
  codecCallback_->onHeadersComplete(3, std::move(req2));
  handler2.txn_->sendHeaders(*resp2);

  eventBase_.loop();

  EXPECT_EQ(writeCount + 1, writeCount_);

  if (timeout) {
    // don't send a window update, the handlers will get timeouts
    EXPECT_CALL(handler1, onError(_))
      .WillOnce(Invoke([] (const HTTPException& ex) {
            EXPECT_EQ(ex.getProxygenError(), kErrorTimeout);
          }));
    EXPECT_CALL(handler2, onError(_))
      .WillOnce(Invoke([] (const HTTPException& ex) {
            EXPECT_EQ(ex.getProxygenError(), kErrorTimeout);
          }));
    EXPECT_CALL(mockController_, detachSession(_));
    // send a window update to refresh the stream level timeout
    codecCallback_->onWindowUpdate(1, 1);
    // silly, the timeout set is internal and there's no fd, so hold the
    // eventBase open until the timeout can fire
    eventBase_.runAfterDelay([] {}, 500);

    transactionTimeouts_->cancelAll();
  } else {
    // Give a connection level window update of 10 bytes -- this
    // should allow 10 bytes of the txn1 response to be written
    codecCallback_->onWindowUpdate(0, 10);
    EXPECT_CALL(*codec_, generateBody(_, 1, PtrBufHasLen(uint64_t(10)),
                                      _, false));
    eventBase_.loop();

    // Just tear everything down now.
    EXPECT_CALL(handler1, detachTransaction());
    codecCallback_->onAbort(handler1.txn_->getID(),
                            ErrorCode::INTERNAL_ERROR);
    eventBase_.loop();

    EXPECT_CALL(handler2, detachTransaction());
    EXPECT_CALL(mockController_, detachSession(_));
    httpSession_->dropConnection();
  }

  eventBase_.loop();
}

TEST_F(MockCodecDownstreamTest, ConnFlowControlBlocked) {
  testConnFlowControlBlocked(false);
}

TEST_F(MockCodecDownstreamTest, ConnFlowControlTimeout) {
  testConnFlowControlBlocked(true);
}

TEST_F(MockCodecDownstreamTest, UnpausedLargePost) {
  // Make sure that a large POST that streams into the handler generates
  // connection level flow control so that the entire POST can be received.
  InSequence enforceOrder;
  NiceMock<MockHTTPHandler> handler1;
  unsigned kNumChunks = 10;
  auto wantToWrite = spdy::kInitialWindow * kNumChunks;
  auto wantToWriteStr = folly::to<string>(wantToWrite);
  auto req1 = makePostRequest(wantToWrite);
  auto req1Body = makeBuf(wantToWrite);

  EXPECT_CALL(mockController_, getRequestHandler(_, _))
    .WillOnce(Return(&handler1));
  EXPECT_CALL(handler1, setTransaction(_))
    .WillOnce(SaveArg<0>(&handler1.txn_));

  EXPECT_CALL(handler1, onHeadersComplete(_));
  for (unsigned i = 0; i < kNumChunks; ++i) {
    EXPECT_CALL(*codec_, generateWindowUpdate(_, 0, spdy::kInitialWindow));
    EXPECT_CALL(handler1, onBody(PtrBufHasLen(spdy::kInitialWindow)));
    EXPECT_CALL(*codec_, generateWindowUpdate(_, 1, spdy::kInitialWindow));
  }
  EXPECT_CALL(handler1, onEOM());

  codecCallback_->onMessageBegin(1, req1.get());
  codecCallback_->onHeadersComplete(1, std::move(req1));
  // Give kNumChunks chunks, each of the maximum window size. We should generate
  // window update for each chunk
  for (unsigned i = 0; i < kNumChunks; ++i) {
    codecCallback_->onBody(1, makeBuf(spdy::kInitialWindow), 0);
  }
  codecCallback_->onMessageComplete(1, false);

  // Just tear everything down now.
  EXPECT_CALL(mockController_, detachSession(_));
  httpSession_->dropConnection();
}

TEST_F(MockCodecDownstreamTest, IngressPausedWindowUpdate) {
  // Test sending a large response body while the handler has ingress paused. We
  // should process the ingress window_updates and deliver the full body
  InSequence enforceOrder;
  NiceMock<MockHTTPHandler> handler1;
  auto req = makeGetRequest();
  size_t respSize = spdy::kInitialWindow * 10;
  unique_ptr<HTTPMessage> resp;
  unique_ptr<folly::IOBuf> respBody;
  tie(resp, respBody) = makeResponse(200, respSize);
  size_t written = 0;

  EXPECT_CALL(mockController_, getRequestHandler(_, _))
    .WillOnce(Return(&handler1));
  EXPECT_CALL(handler1, setTransaction(_))
    .WillOnce(SaveArg<0>(&handler1.txn_));

  EXPECT_CALL(handler1, onHeadersComplete(_))
    .WillOnce(InvokeWithoutArgs([&] () {
          // Pause ingress. Make sure we process the window updates anyway
          handler1.txn_->pauseIngress();
        }));
  EXPECT_CALL(*codec_, generateHeader(_, _, _, _, _));
  EXPECT_CALL(*codec_, generateBody(_, _, _, _, _))
      .WillRepeatedly(Invoke([&](folly::IOBufQueue&,
                                 HTTPCodec::StreamID,
                                 std::shared_ptr<folly::IOBuf> chain,
                                 folly::Optional<uint8_t>,
                                 bool /*eom*/) {
        auto len = chain->computeChainDataLength();
        written += len;
        return len;
      }));

  codecCallback_->onWindowUpdate(0, respSize); // open conn-level window
  codecCallback_->onMessageBegin(1, req.get());
  codecCallback_->onHeadersComplete(1, std::move(req));
  EXPECT_TRUE(handler1.txn_->isIngressPaused());

  // Unblock txn-level flow control and try to egress the body
  codecCallback_->onWindowUpdate(1, respSize);
  handler1.txn_->sendHeaders(*resp);
  handler1.txn_->sendBody(std::move(respBody));

  eventBase_.loop();
  EXPECT_EQ(written, respSize);

  // Just tear everything down now.
  EXPECT_CALL(mockController_, detachSession(_));
  httpSession_->dropConnection();
}

TEST_F(MockCodecDownstreamTest, ShutdownThenSendPushHeaders) {
  // Test that notifying session of shutdown before sendHeaders() called on a
  // pushed txn lets that push txn finish.
  EXPECT_CALL(*codec_, supportsPushTransactions())
    .WillRepeatedly(Return(true));

  InSequence enforceOrder;
  NiceMock<MockHTTPHandler> handler;
  MockHTTPPushHandler pushHandler;
  auto req = makeGetRequest();

  EXPECT_CALL(mockController_, getRequestHandler(_, _))
    .WillOnce(Return(&handler));
  EXPECT_CALL(handler, setTransaction(_))
    .WillOnce(SaveArg<0>(&handler.txn_));

  EXPECT_CALL(handler, onHeadersComplete(_))
    .WillOnce(Invoke([&] (std::shared_ptr<HTTPMessage>) {
          auto pushTxn = handler.txn_->newPushedTransaction(&pushHandler);
          // start shutdown process
          httpSession_->notifyPendingShutdown();
          // we should be able to process new requests
          EXPECT_TRUE(codec_->isReusable());
          pushHandler.sendPushHeaders("/foo", "www.foo.com", 0,
                                      handler.txn_->getPriority());
          // we should* still* be able to process new requests
          EXPECT_TRUE(codec_->isReusable());
          pushTxn->sendEOM();
        }));
  EXPECT_CALL(pushHandler, setTransaction(_))
    .WillOnce(SaveArg<0>(&pushHandler.txn_));
  EXPECT_CALL(*codec_, generatePushPromise(_, 2, _, _, _, _));
  EXPECT_CALL(*codec_, generateEOM(_, 2));
  EXPECT_CALL(pushHandler, detachTransaction());
  EXPECT_CALL(handler, onEOM())
    .WillOnce(Invoke([&] {
          handler.sendReply();
        }));
  EXPECT_CALL(*codec_, generateHeader(_, 1, _, _, _));
  EXPECT_CALL(*codec_, generateEOM(_, 1));
  EXPECT_CALL(handler, detachTransaction());

  codecCallback_->onMessageBegin(1, req.get());
  codecCallback_->onHeadersComplete(1, std::move(req));
  codecCallback_->onMessageComplete(1, false);

  // finish shutdown
  EXPECT_CALL(*codec_, onIngressEOF());
  EXPECT_CALL(mockController_, detachSession(_));
  httpSession_->dropConnection();

  eventBase_.loop();
}

TEST_F(MockCodecDownstreamTest, ReadIobufChainShutdown) {
  // Given an ingress IOBuf chain of 2 parts, if we shutdown after reading the
  // first part of the chain, we shouldn't read the second part.  One way to
  // simulate a 2 part chain is to put more ingress in readBuf while we are
  // inside HTTPCodec::onIngress()

  InSequence enforceOrder;

  auto f = [&] () {
    void* buf;
    size_t bufSize;
    transportCb_->getReadBuffer(&buf, &bufSize);
    transportCb_->readDataAvailable(bufSize);
  };

  EXPECT_CALL(*codec_, onIngress(_))
    .WillOnce(Invoke([&] (const IOBuf& buf) {
          // This first time, don't process any data. This will cause the
          // ingress chain to grow in size later.
          EXPECT_FALSE(buf.isChained());
          return 0;
        }))
    .WillOnce(Invoke([&] (const IOBuf& buf) {
          // Now there should be a second buffer in the chain.
          EXPECT_TRUE(buf.isChained());
          // Shutdown writes. This enough to destroy the session.
          httpSession_->closeWhenIdle();
          return buf.length();
        }));
  // We shouldn't get a third onIngress() callback. This will be enforced by the
  // test framework since the codec is a strict mock.
  EXPECT_CALL(*codec_, isBusy());
  EXPECT_CALL(*codec_, onIngressEOF());
  EXPECT_CALL(mockController_, detachSession(_));

  f();
  f(); // The first time wasn't processed, so this should make a len=2 chain.
  eventBase_.loop();
}

void MockCodecDownstreamTest::testGoaway(bool doubleGoaway,
                                         bool dropConnection) {
  NiceMock<MockHTTPHandler> handler;
  MockHTTPHandler pushHandler;

  liveGoaways_ = true;
  if (doubleGoaway) {
    EXPECT_CALL(mockController_, getRequestHandler(_, _))
      .WillOnce(Return(&handler));
    EXPECT_CALL(handler, setTransaction(_))
      .WillOnce(SaveArg<0>(&handler.txn_));

    EXPECT_CALL(handler, onHeadersComplete(_));
    EXPECT_CALL(handler, onEOM())
      .WillOnce(Invoke([&] {
            handler.sendReply();
          }));
    EXPECT_CALL(*codec_, generateHeader(_, 1, _, _, _));
    EXPECT_CALL(*codec_, generateEOM(_, 1));
    EXPECT_CALL(handler, detachTransaction());

    // Turn on double GOAWAY drain
    codec_->enableDoubleGoawayDrain();
  }

  // Send a GOAWAY acking uninitiated transactions
  EXPECT_FALSE(drainPending_);
  httpSession_->notifyPendingShutdown();
  EXPECT_EQ(drainPending_, doubleGoaway);
  EXPECT_FALSE(reusable_);

  if (doubleGoaway) {
    // Should be able to process new requests
    auto req1 = makeGetRequest();
    codecCallback_->onMessageBegin(1, req1.get());
    codecCallback_->onHeadersComplete(1, std::move(req1));
    codecCallback_->onMessageComplete(1, false);
  }

  folly::AsyncTransportWrapper::WriteCallback* cb = nullptr;
  EXPECT_CALL(*transport_, writeChain(_, _, _))
    .WillOnce(Invoke([&] (folly::AsyncTransportWrapper::WriteCallback* callback,
                          const shared_ptr<IOBuf>,
                          WriteFlags) {
                       // don't immediately flush the goaway
                       cb = callback;
                     }));
  if (doubleGoaway || !dropConnection) {
    // single goaway, drop connection doesn't get onIngressEOF
    EXPECT_CALL(*codec_, onIngressEOF());
  }
  eventBase_.loopOnce();

  EXPECT_CALL(mockController_, detachSession(_));
  if (dropConnection) {
    EXPECT_CALL(*transport_, closeWithReset())
      .Times(AtLeast(1))
      .WillOnce(DoAll(Assign(&transportGood_, false),
                      Invoke([cb] {
                          AsyncSocketException ex(
                            AsyncSocketException::UNKNOWN, "");
                          cb->writeErr(0, ex);
                        })));

    httpSession_->dropConnection();
  } else {
    EXPECT_CALL(*codec_, isBusy());
    httpSession_->closeWhenIdle();
    cb->writeSuccess();
  }
  EXPECT_FALSE(drainPending_);
  EXPECT_FALSE(reusable_);
}

TEST_F(MockCodecDownstreamTest, SendDoubleGoawayTimeout) {
  testGoaway(true, true);
}
TEST_F(MockCodecDownstreamTest, SendDoubleGoawayIdle) {
  testGoaway(true, false);
}
TEST_F(MockCodecDownstreamTest, SendGoawayTimeout) {
  testGoaway(false, true);
}
TEST_F(MockCodecDownstreamTest, SendGoawayIdle) {
  testGoaway(false, false);
}

TEST_F(MockCodecDownstreamTest, DropConnection) {
  NiceMock<MockHTTPHandler> handler;
  MockHTTPHandler pushHandler;

  liveGoaways_ = true;

  EXPECT_CALL(*codec_, onIngressEOF());
  EXPECT_CALL(mockController_, detachSession(_));
  EXPECT_CALL(*transport_, closeWithReset())
    .Times(AtLeast(1))
    .WillOnce(Assign(&transportGood_, false));
  httpSession_->dropConnection();
}

TEST_F(MockCodecDownstreamTest, DropConnectionNogoaway) {
  NiceMock<MockHTTPHandler> handler;
  MockHTTPHandler pushHandler;

  liveGoaways_ = false;

  EXPECT_CALL(*codec_, onIngressEOF());
  EXPECT_CALL(mockController_, detachSession(_));
  EXPECT_CALL(*transport_, closeNow())
    .Times(AtLeast(1))
    .WillOnce(Assign(&transportGood_, false));
  httpSession_->dropConnection();
}

TEST_F(MockCodecDownstreamTest, ShutdownThenError) {
  // Test that we ignore any errors after we shutdown the socket in HTTPSession.
  onIngressImpl([&] {
      // This executes as the implementation of HTTPCodec::onIngress()
      InSequence dummy;

      HTTPException err(HTTPException::Direction::INGRESS, "foo");
      err.setHttpStatusCode(400);
      HTTPMessage req = getGetRequest();
      EXPECT_CALL(mockController_, getParseErrorHandler(_, _, _))
        .WillOnce(Return(nullptr));

      // Creates and adds a txn to the session
      codecCallback_->onMessageBegin(1, &req);

      httpSession_->closeWhenIdle();

      codecCallback_->onError(1, err, false);
    });
  // flush the shutdown callback
  EXPECT_CALL(mockController_, detachSession(_));
  eventBase_.loopOnce();
}

TEST_F(MockCodecDownstreamTest, PingDuringShutdown) {
  onIngressImpl([&] {
      InSequence dummy;

      // Shutdown writes only. Since the session is empty, this normally
      // causes the session to close, but it is held open since we are in
      // the middle of parsing ingress.
      EXPECT_CALL(*codec_, isBusy());
      EXPECT_CALL(*codec_, onIngressEOF());
      httpSession_->closeWhenIdle();

      // We read a ping off the wire, which makes us enqueue a ping reply
      EXPECT_CALL(*codec_, generatePingReply(_, _))
        .WillOnce(Return(10));
      codecCallback_->onPingRequest(1);

      // When this function returns, the controller gets detachSession()
      EXPECT_CALL(mockController_, detachSession(_));
    });
}

TEST_F(MockCodecDownstreamTest, SettingsAck) {
  EXPECT_CALL(*codec_, generateSettingsAck(_));
  codecCallback_->onSettings(
    {{SettingsId::INITIAL_WINDOW_SIZE, 4000}});
  EXPECT_CALL(*codec_, onIngressEOF());
  EXPECT_CALL(mockController_, detachSession(_));
  httpSession_->dropConnection();
}

TEST_F(MockCodecDownstreamTest, TestSendCertificateRequest) {
  auto certRequestContext = folly::IOBuf::copyBuffer("0123456789abcdef");
  fizz::SignatureAlgorithms sigAlgs;
  sigAlgs.supported_signature_algorithms.push_back(
      SignatureScheme::ecdsa_secp256r1_sha256);
  std::vector<fizz::Extension> extensions;
  extensions.push_back(encodeExtension(std::move(sigAlgs)));

  std::unique_ptr<StrictMock<MockSecondaryAuthManager>> secondAuthManager_(
      new StrictMock<MockSecondaryAuthManager>());
  httpSession_->setSecondAuthManager(std::move(secondAuthManager_));
  auto authManager = dynamic_cast<MockSecondaryAuthManager*>(
      httpSession_->getSecondAuthManager());
  EXPECT_CALL(*codec_, getIngressSettings())
      .WillOnce(Return(&kIngressCertAuthSettings));
  EXPECT_CALL(*codec_, getEgressSettings())
      .WillOnce(Return(&kEgressCertAuthSettings));
  EXPECT_CALL(*authManager, createAuthRequest(_, _))
      .WillOnce(InvokeWithoutArgs([]() {
        return std::make_pair(120, IOBuf::copyBuffer("authenticatorrequest"));
      }));
  EXPECT_CALL(*codec_, generateCertificateRequest(_, _, _))
      .WillOnce(Return(20));
  auto encodedSize = httpSession_->sendCertificateRequest(
      std::move(certRequestContext), std::move(extensions));
  EXPECT_EQ(encodedSize, 20);

  EXPECT_CALL(*codec_, onIngressEOF());
  EXPECT_CALL(mockController_, detachSession(_));
  httpSession_->dropConnection();
}