RequestDecoder& ConnectionManagerImpl::newStream(ResponseEncoder& response_encoder,
                                                 bool is_internally_created) {
  TRACE_EVENT("core", "ConnectionManagerImpl::newStream");
  if (connection_idle_timer_) {
    connection_idle_timer_->disableTimer();
  }

  ENVOY_CONN_LOG(debug, "new stream", read_callbacks_->connection());

  // Create account, wiring the stream to use it for tracking bytes.
  // If tracking is disabled, the wiring becomes a NOP.
  auto& buffer_factory = read_callbacks_->connection().dispatcher().getWatermarkFactory();
  Buffer::BufferMemoryAccountSharedPtr downstream_stream_account =
      buffer_factory.createAccount(response_encoder.getStream());
  response_encoder.getStream().setAccount(downstream_stream_account);
  ActiveStreamPtr new_stream(new ActiveStream(*this, response_encoder.getStream().bufferLimit(),
                                              std::move(downstream_stream_account)));

  accumulated_requests_++;
  if (config_.maxRequestsPerConnection() > 0 &&
      accumulated_requests_ >= config_.maxRequestsPerConnection()) {
    if (codec_->protocol() < Protocol::Http2) {
      new_stream->state_.saw_connection_close_ = true;
      // Prevent erroneous debug log of closing due to incoming connection close header.
      drain_state_ = DrainState::Closing;
    } else if (drain_state_ == DrainState::NotDraining) {
      startDrainSequence();
    }
    ENVOY_CONN_LOG(debug, "max requests per connection reached", read_callbacks_->connection());
    stats_.named_.downstream_cx_max_requests_reached_.inc();
  }

  new_stream->state_.is_internally_created_ = is_internally_created;
  new_stream->response_encoder_ = &response_encoder;
  new_stream->response_encoder_->getStream().addCallbacks(*new_stream);
  new_stream->response_encoder_->getStream().setFlushTimeout(new_stream->idle_timeout_ms_);
  new_stream->streamInfo().setDownstreamBytesMeter(response_encoder.getStream().bytesMeter());
  // If the network connection is backed up, the stream should be made aware of it on creation.
  // Both HTTP/1.x and HTTP/2 codecs handle this in StreamCallbackHelper::addCallbacksHelper.
  ASSERT(read_callbacks_->connection().aboveHighWatermark() == false ||
         new_stream->filter_manager_.aboveHighWatermark());
  LinkedList::moveIntoList(std::move(new_stream), streams_);
  return **streams_.begin();
}
RequestDecoder& ConnectionManagerImpl::newStream(ResponseEncoder& response_encoder,
                                                 bool is_internally_created) {
  TRACE_EVENT("core", "ConnectionManagerImpl::newStream");
  if (connection_idle_timer_) {
    connection_idle_timer_->disableTimer();
  }

  ENVOY_CONN_LOG(debug, "new stream", read_callbacks_->connection());

   
   
  auto& buffer_factory = read_callbacks_->connection().dispatcher().getWatermarkFactory();
  Buffer::BufferMemoryAccountSharedPtr downstream_stream_account =
      buffer_factory.createAccount(response_encoder.getStream());
  response_encoder.getStream().setAccount(downstream_stream_account);
  ActiveStreamPtr new_stream(new ActiveStream(*this, response_encoder.getStream().bufferLimit(),
                                              std::move(downstream_stream_account)));

  accumulated_requests_++;
  if (config_.maxRequestsPerConnection() > 0 &&
      accumulated_requests_ >= config_.maxRequestsPerConnection()) {
    if (codec_->protocol() < Protocol::Http2) {
      new_stream->state_.saw_connection_close_ = true;
       
      drain_state_ = DrainState::Closing;
    } else if (drain_state_ == DrainState::NotDraining) {
      startDrainSequence();
    }
    ENVOY_CONN_LOG(debug, "max requests per connection reached", read_callbacks_->connection());
    stats_.named_.downstream_cx_max_requests_reached_.inc();
  }

  new_stream->state_.is_internally_created_ = is_internally_created;
  new_stream->response_encoder_ = &response_encoder;
  new_stream->response_encoder_->getStream().addCallbacks(*new_stream);
  new_stream->response_encoder_->getStream().setFlushTimeout(new_stream->idle_timeout_ms_);
  new_stream->streamInfo().setDownstreamBytesMeter(response_encoder.getStream().bytesMeter());
   
   
  ASSERT(read_callbacks_->connection().aboveHighWatermark() == false ||
         new_stream->filter_manager_.aboveHighWatermark());
  LinkedList::moveIntoList(std::move(new_stream), streams_);
  return **streams_.begin();
}