HTTPSession::onMessageBegin(HTTPCodec::StreamID streamID, HTTPMessage* msg) {
  VLOG(4) << "processing new msg streamID=" << streamID << " " << *this;
  if (infoCallback_) {
    infoCallback_->onRequestBegin(*this);
  }

  HTTPTransaction* txn = findTransaction(streamID);
  if (txn) {
    if (isDownstream() && txn->isPushed()) {
      // Push streams are unidirectional (half-closed). If the downstream
      // attempts to send ingress, abort with STREAM_CLOSED error.
      HTTPException ex(HTTPException::Direction::INGRESS_AND_EGRESS,
                       "Downstream attempts to send ingress, abort.");
      ex.setCodecStatusCode(ErrorCode::STREAM_CLOSED);
      txn->onError(ex);
    }
    return;  // If this transaction is already registered, no need to add it now
  }

  http2::PriorityUpdate messagePriority = getMessagePriority(msg);
  txn = createTransaction(streamID, HTTPCodec::NoStream,
                          HTTPCodec::NoExAttributes, messagePriority);
  if (!txn) {
    return;  // This could happen if the socket is bad.
  }

  if (!codec_->supportsParallelRequests() && getPipelineStreamCount() > 1) {
    // The previous transaction hasn't completed yet. Pause reads until
    // it completes; this requires pausing both transactions.

    // HTTP/1.1 pipeline is detected, and which is incompactible with
    // ByteEventTracker. Drain all the ByteEvents
    CHECK(byteEventTracker_);
    byteEventTracker_->drainByteEvents();

    // drainByteEvents() may detach txn(s). Don't pause read if one txn left
    if (getPipelineStreamCount() < 2) {
      DCHECK(readsUnpaused());
      return;
    }

    // There must be at least two transactions (we just checked). The previous
    // txns haven't completed yet. Pause reads until they complete
    DCHECK_GE(transactions_.size(), 2);
    for (auto it = ++transactions_.rbegin(); it != transactions_.rend(); ++it) {
      DCHECK(it->second.isIngressEOMSeen());
      it->second.pauseIngress();
    }
    transactions_.rbegin()->second.pauseIngress();
    DCHECK_EQ(liveTransactions_, 0);
    DCHECK(readsPaused());
  }
}
HTTPSession::onMessageBegin(HTTPCodec::StreamID streamID, HTTPMessage* msg) {
  VLOG(4) << "processing new msg streamID=" << streamID << " " << *this;
  if (infoCallback_) {
    infoCallback_->onRequestBegin(*this);
  }

  HTTPTransaction* txn = findTransaction(streamID);
  if (txn) {
    if (isDownstream() && txn->isPushed()) {
       
       
      HTTPException ex(HTTPException::Direction::INGRESS_AND_EGRESS,
                       "Downstream attempts to send ingress, abort.");
      ex.setCodecStatusCode(ErrorCode::STREAM_CLOSED);
      txn->onError(ex);
    }
    return;   
  }

  http2::PriorityUpdate messagePriority = getMessagePriority(msg);
  txn = createTransaction(streamID, HTTPCodec::NoStream,
                          HTTPCodec::NoExAttributes, messagePriority);
  if (!txn) {
    return;   
  }

  if (!codec_->supportsParallelRequests() && getPipelineStreamCount() > 1) {
     
     

     
     
    CHECK(byteEventTracker_);
    byteEventTracker_->drainByteEvents();

     
    if (getPipelineStreamCount() < 2) {
      DCHECK(readsUnpaused());
      return;
    }

     
     
    DCHECK_GE(transactions_.size(), 2);
    for (auto it = ++transactions_.rbegin(); it != transactions_.rend(); ++it) {
      DCHECK(it->second.isIngressEOMSeen());
      it->second.pauseIngress();
    }
    transactions_.rbegin()->second.pauseIngress();
    DCHECK_EQ(liveTransactions_, 0);
    DCHECK(readsPaused());
  }
}