Use http-parser instead of htparse

examples/Makefile.am

@@ -88,7 +88,7 @@ shrpx_SOURCES = ${HELPER_OBJECTS} ${HELPER_HFILES} \
 	shrpx_ssl.cc shrpx_ssl.h \
 	shrpx_thread_event_receiver.cc shrpx_thread_event_receiver.h \
 	shrpx_worker.cc shrpx_worker.h \
-	htparse/htparse.c htparse/htparse.h
+	http-parser/http_parser.c http-parser/http_parser.h
 endif # HAVE_LIBEVENT_OPENSSL
 
 noinst_PROGRAMS = spdycli

examples/shrpx_downstream.cc

@@ -50,18 +50,20 @@ Downstream::Downstream(Upstream *upstream, int stream_id, int priority)
     chunked_request_(false),
     request_connection_close_(false),
     request_expect_100_continue_(false),
+    request_header_key_prev_(false),
     response_state_(INITIAL),
     response_http_status_(0),
     response_major_(1),
     response_minor_(1),
     chunked_response_(false),
     response_connection_close_(false),
-    response_htp_(htparser_new()),
+    response_header_key_prev_(false),
+    response_htp_(new http_parser()),
     response_body_buf_(0),
     recv_window_size_(0)
 {
-  htparser_init(response_htp_, htp_type_response);
+  http_parser_init(response_htp_, HTTP_RESPONSE);
-  htparser_set_userdata(response_htp_, this);
+  response_htp_->data = this;
 }
 
 Downstream::~Downstream()
@@ -157,11 +159,13 @@ void check_connection_close(bool *connection_close,
 void Downstream::add_request_header(const std::string& name,
                                     const std::string& value)
 {
+  request_header_key_prev_ = true;
   request_headers_.push_back(std::make_pair(name, value));
 }
 
 void Downstream::set_last_request_header_value(const std::string& value)
 {
+  request_header_key_prev_ = false;
   Headers::value_type &item = request_headers_.back();
   item.second = value;
   check_transfer_encoding_chunked(&chunked_request_, item);
@@ -169,6 +173,25 @@ void Downstream::set_last_request_header_value(const std::string& value)
   //check_connection_close(&request_connection_close_, item);
 }
 
+bool Downstream::get_request_header_key_prev() const
+{
+  return request_header_key_prev_;
+}
+
+void Downstream::append_last_request_header_key(const char *data, size_t len)
+{
+  assert(request_header_key_prev_);
+  Headers::value_type &item = request_headers_.back();
+  item.first.append(data, len);
+}
+
+void Downstream::append_last_request_header_value(const char *data, size_t len)
+{
+  assert(!request_header_key_prev_);
+  Headers::value_type &item = request_headers_.back();
+  item.second.append(data, len);
+}
+
 void Downstream::set_request_method(const std::string& method)
 {
   request_method_ = method;
@@ -184,6 +207,11 @@ void Downstream::set_request_path(const std::string& path)
   request_path_ = path;
 }
 
+void Downstream::append_request_path(const char *data, size_t len)
+{
+  request_path_.append(data, len);
+}
+
 const std::string& Downstream::get_request_path() const
 {
   return request_path_;
@@ -393,17 +421,39 @@ const Headers& Downstream::get_response_headers() const
 void Downstream::add_response_header(const std::string& name,
                                      const std::string& value)
 {
+  response_header_key_prev_ = true;
   response_headers_.push_back(std::make_pair(name, value));
 }
 
 void Downstream::set_last_response_header_value(const std::string& value)
 {
+  response_header_key_prev_ = false;
   Headers::value_type &item = response_headers_.back();
   item.second = value;
   check_transfer_encoding_chunked(&chunked_response_, item);
   //check_connection_close(&response_connection_close_, item);
 }
 
+bool Downstream::get_response_header_key_prev() const
+{
+  return response_header_key_prev_;
+}
+
+void Downstream::append_last_response_header_key(const char *data, size_t len)
+{
+  assert(response_header_key_prev_);
+  Headers::value_type &item = response_headers_.back();
+  item.first.append(data, len);
+}
+
+void Downstream::append_last_response_header_value(const char *data,
+                                                   size_t len)
+{
+  assert(!response_header_key_prev_);
+  Headers::value_type &item = response_headers_.back();
+  item.second.append(data, len);
+}
+
 unsigned int Downstream::get_response_http_status() const
 {
   return response_http_status_;
@@ -455,14 +505,14 @@ void Downstream::set_response_connection_close(bool f)
 }
 
 namespace {
-int htp_hdrs_completecb(htparser *htp)
+int htp_hdrs_completecb(http_parser *htp)
 {
   Downstream *downstream;
-  downstream = reinterpret_cast<Downstream*>(htparser_get_userdata(htp));
+  downstream = reinterpret_cast<Downstream*>(htp->data);
-  downstream->set_response_http_status(htparser_get_status(htp));
+  downstream->set_response_http_status(htp->status_code);
-  downstream->set_response_major(htparser_get_major(htp));
+  downstream->set_response_major(htp->http_major);
-  downstream->set_response_minor(htparser_get_minor(htp));
+  downstream->set_response_minor(htp->http_minor);
-  downstream->set_response_connection_close(!htparser_should_keep_alive(htp));
+  downstream->set_response_connection_close(!http_should_keep_alive(htp));
   downstream->set_response_state(Downstream::HEADER_COMPLETE);
   downstream->get_upstream()->on_downstream_header_complete(downstream);
   return 0;
@@ -470,30 +520,38 @@ int htp_hdrs_completecb(htparser *htp)
 } // namespace
 
 namespace {
-int htp_hdr_keycb(htparser *htp, const char *data, size_t len)
+int htp_hdr_keycb(http_parser *htp, const char *data, size_t len)
 {
   Downstream *downstream;
-  downstream = reinterpret_cast<Downstream*>(htparser_get_userdata(htp));
+  downstream = reinterpret_cast<Downstream*>(htp->data);
-  downstream->add_response_header(std::string(data, len), "");
+  if(downstream->get_response_header_key_prev()) {
+    downstream->append_last_response_header_key(data, len);
+  } else {
+    downstream->add_response_header(std::string(data, len), "");
+  }
   return 0;
 }
 } // namespace
 
 namespace {
-int htp_hdr_valcb(htparser *htp, const char *data, size_t len)
+int htp_hdr_valcb(http_parser *htp, const char *data, size_t len)
 {
   Downstream *downstream;
-  downstream = reinterpret_cast<Downstream*>(htparser_get_userdata(htp));
+  downstream = reinterpret_cast<Downstream*>(htp->data);
-  downstream->set_last_response_header_value(std::string(data, len));
+  if(downstream->get_response_header_key_prev()) {
+    downstream->set_last_response_header_value(std::string(data, len));
+  } else {
+    downstream->append_last_response_header_value(data, len);
+  }
   return 0;
 }
 } // namespace
 
 namespace {
-int htp_bodycb(htparser *htp, const char *data, size_t len)
+int htp_bodycb(http_parser *htp, const char *data, size_t len)
 {
   Downstream *downstream;
-  downstream = reinterpret_cast<Downstream*>(htparser_get_userdata(htp));
+  downstream = reinterpret_cast<Downstream*>(htp->data);
 
   downstream->get_upstream()->on_downstream_body
     (downstream, reinterpret_cast<const uint8_t*>(data), len);
@@ -502,10 +560,10 @@ int htp_bodycb(htparser *htp, const char *data, size_t len)
 } // namespace
 
 namespace {
-int htp_body_completecb(htparser *htp)
+int htp_msg_completecb(http_parser *htp)
 {
   Downstream *downstream;
-  downstream = reinterpret_cast<Downstream*>(htparser_get_userdata(htp));
+  downstream = reinterpret_cast<Downstream*>(htp->data);
 
   if(downstream->tunnel_established()) {
     // For tunneling, we remove timeouts.
@@ -514,30 +572,26 @@ int htp_body_completecb(htparser *htp)
 
   downstream->set_response_state(Downstream::MSG_COMPLETE);
   downstream->get_upstream()->on_downstream_body_complete(downstream);
-  return 0;
+
+  if(downstream->get_request_method() == "HEAD") {
+    // Ignore the response body. HEAD response may contain
+    // Content-Length or Transfer-Encoding: chunked.
+    return 1;
+  } else {
+    return 0;
+  }
 }
 } // namespace
 
 namespace {
-htparse_hooks htp_hooks = {
+http_parser_settings htp_hooks = {
-  0, /*htparse_hook      on_msg_begin;*/
+  0, /*http_cb      on_message_begin;*/
-  0, /*htparse_data_hook method;*/
+  0, /*http_data_cb on_url;*/
-  0, /* htparse_data_hook scheme;*/
+  htp_hdr_keycb, /*http_data_cb on_header_field;*/
-  0, /* htparse_data_hook host; */
+  htp_hdr_valcb, /*http_data_cb on_header_value;*/
-  0, /* htparse_data_hook port; */
+  htp_hdrs_completecb, /*http_cb      on_headers_complete;*/
-  0, /* htparse_data_hook path; */
+  htp_bodycb, /*http_data_cb on_body;*/
-  0, /* htparse_data_hook args; */
+  htp_msg_completecb /*http_cb      on_message_complete;*/
-  0, /* htparse_data_hook uri; */
-  0, /* htparse_hook      on_hdrs_begin; */
-  htp_hdr_keycb, /* htparse_data_hook hdr_key; */
-  htp_hdr_valcb, /* htparse_data_hook hdr_val; */
-  0, /* htparse_data_hook hostname; */
-  htp_hdrs_completecb, /* htparse_hook      on_hdrs_complete; */
-  0, /*htparse_hook      on_new_chunk;*/
-  0, /*htparse_hook      on_chunk_complete;*/
-  0, /*htparse_hook      on_chunks_complete;*/
-  htp_bodycb, /* htparse_data_hook body; */
-  htp_body_completecb /* htparse_hook      on_msg_complete;*/
 };
 } // namespace
 
@@ -547,17 +601,19 @@ int Downstream::parse_http_response()
   evbuffer *input = bufferevent_get_input(bev);
   unsigned char *mem = evbuffer_pullup(input, -1);
 
-  size_t nread = htparser_run(response_htp_, &htp_hooks,
+  size_t nread = http_parser_execute(response_htp_, &htp_hooks,
-                              reinterpret_cast<const char*>(mem),
+                                     reinterpret_cast<const char*>(mem),
-                              evbuffer_get_length(input));
+                                     evbuffer_get_length(input));
 
   evbuffer_drain(input, nread);
-  if(htparser_get_error(response_htp_) == htparse_error_none) {
+  http_errno htperr = HTTP_PARSER_ERRNO(response_htp_);
+  if(htperr == HPE_OK) {
     return 0;
   } else {
     if(ENABLE_LOG) {
       LOG(INFO) << "Downstream HTTP parser failure: "
-                << htparser_get_strerror(response_htp_);
+                << "(" << http_errno_name(htperr) << ") "
+                << http_errno_description(htperr);
     }
     return SHRPX_ERR_HTTP_PARSE;
   }

examples/shrpx_downstream.h

@@ -36,7 +36,7 @@
 #include <event2/bufferevent.h>
 
 extern "C" {
-#include "htparse/htparse.h"
+#include "http-parser/http_parser.h"
 }
 
 #include "shrpx_io_control.h"
@@ -72,9 +72,15 @@ public:
   const Headers& get_request_headers() const;
   void add_request_header(const std::string& name, const std::string& value);
   void set_last_request_header_value(const std::string& value);
+
+  bool get_request_header_key_prev() const;
+  void append_last_request_header_key(const char *data, size_t len);
+  void append_last_request_header_value(const char *data, size_t len);
+
   void set_request_method(const std::string& method);
   const std::string& get_request_method() const;
   void set_request_path(const std::string& path);
+  void append_request_path(const char *data, size_t len);
   const std::string& get_request_path() const;
   void set_request_major(int major);
   void set_request_minor(int minor);
@@ -101,6 +107,11 @@ public:
   const Headers& get_response_headers() const;
   void add_response_header(const std::string& name, const std::string& value);
   void set_last_response_header_value(const std::string& value);
+
+  bool get_response_header_key_prev() const;
+  void append_last_response_header_key(const char *data, size_t len);
+  void append_last_response_header_value(const char *data, size_t len);
+
   unsigned int get_response_http_status() const;
   void set_response_http_status(unsigned int status);
   void set_response_major(int major);
@@ -131,6 +142,7 @@ private:
   bool request_connection_close_;
   bool request_expect_100_continue_;
   Headers request_headers_;
+  bool request_header_key_prev_;
 
   int response_state_;
   unsigned int response_http_status_;
@@ -139,7 +151,8 @@ private:
   bool chunked_response_;
   bool response_connection_close_;
   Headers response_headers_;
-  htparser *response_htp_;
+  bool response_header_key_prev_;
+  http_parser *response_htp_;
   // This buffer is used to temporarily store downstream response
   // body. Spdylay reads data from this in the callback.
   evbuffer *response_body_buf_;

examples/shrpx_https_upstream.cc

@@ -46,12 +46,12 @@ const size_t SHRPX_HTTPS_MAX_HEADER_LENGTH = 64*1024;
 
 HttpsUpstream::HttpsUpstream(ClientHandler *handler)
   : handler_(handler),
-    htp_(htparser_new()),
+    htp_(new http_parser()),
     current_header_length_(0),
     ioctrl_(handler->get_bev())
 {
-  htparser_init(htp_, htp_type_request);
+  http_parser_init(htp_, HTTP_REQUEST);
-  htparser_set_userdata(htp_, this);
+  htp_->data = this;
 }
 
 HttpsUpstream::~HttpsUpstream()
@@ -69,10 +69,10 @@ void HttpsUpstream::reset_current_header_length()
 }
 
 namespace {
-int htp_msg_begin(htparser *htp)
+int htp_msg_begin(http_parser *htp)
 {
   HttpsUpstream *upstream;
-  upstream = reinterpret_cast<HttpsUpstream*>(htparser_get_userdata(htp));
+  upstream = reinterpret_cast<HttpsUpstream*>(htp->data);
   if(ENABLE_LOG) {
     LOG(INFO) << "Upstream https request start " << upstream;
   }
@@ -84,73 +84,61 @@ int htp_msg_begin(htparser *htp)
 } // namespace
 
 namespace {
-int htp_methodcb(htparser *htp, const char *data, size_t len)
+int htp_uricb(http_parser *htp, const char *data, size_t len)
 {
   HttpsUpstream *upstream;
-  upstream = reinterpret_cast<HttpsUpstream*>(htparser_get_userdata(htp));
+  upstream = reinterpret_cast<HttpsUpstream*>(htp->data);
   Downstream *downstream = upstream->get_last_downstream();
-  downstream->set_request_method(std::string(data, len));
+  downstream->append_request_path(data, len);
   return 0;
 }
 } // namespace
 
 namespace {
-int htp_uricb(htparser *htp, const char *data, size_t len)
+int htp_hdr_keycb(http_parser *htp, const char *data, size_t len)
 {
   HttpsUpstream *upstream;
-  upstream = reinterpret_cast<HttpsUpstream*>(htparser_get_userdata(htp));
+  upstream = reinterpret_cast<HttpsUpstream*>(htp->data);
   Downstream *downstream = upstream->get_last_downstream();
-  downstream->set_request_path(std::string(data, len));
+  if(downstream->get_request_header_key_prev()) {
-  return 0;
+    downstream->append_last_request_header_key(data, len);
-}
+  } else {
-} // namespace
+    downstream->add_request_header(std::string(data, len), "");
-
-namespace {
-int htp_hdrs_begincb(htparser *htp)
-{
-  if(ENABLE_LOG) {
-    LOG(INFO) << "Upstream https request headers start";
   }
   return 0;
 }
 } // namespace
 
 namespace {
-int htp_hdr_keycb(htparser *htp, const char *data, size_t len)
+int htp_hdr_valcb(http_parser *htp, const char *data, size_t len)
 {
   HttpsUpstream *upstream;
-  upstream = reinterpret_cast<HttpsUpstream*>(htparser_get_userdata(htp));
+  upstream = reinterpret_cast<HttpsUpstream*>(htp->data);
   Downstream *downstream = upstream->get_last_downstream();
-  downstream->add_request_header(std::string(data, len), "");
+  if(downstream->get_request_header_key_prev()) {
+    downstream->set_last_request_header_value(std::string(data, len));
+  } else {
+    downstream->append_last_request_header_value(data, len);
+  }
   return 0;
 }
 } // namespace
 
 namespace {
-int htp_hdr_valcb(htparser *htp, const char *data, size_t len)
+int htp_hdrs_completecb(http_parser *htp)
 {
   HttpsUpstream *upstream;
-  upstream = reinterpret_cast<HttpsUpstream*>(htparser_get_userdata(htp));
+  upstream = reinterpret_cast<HttpsUpstream*>(htp->data);
-  Downstream *downstream = upstream->get_last_downstream();
-  downstream->set_last_request_header_value(std::string(data, len));
-  return 0;
-}
-} // namespace
-
-namespace {
-int htp_hdrs_completecb(htparser *htp)
-{
-  HttpsUpstream *upstream;
-  upstream = reinterpret_cast<HttpsUpstream*>(htparser_get_userdata(htp));
   if(ENABLE_LOG) {
     LOG(INFO) << "Upstream https request headers complete " << upstream;
   }
   Downstream *downstream = upstream->get_last_downstream();
 
-  downstream->set_request_major(htparser_get_major(htp));
+  downstream->set_request_method(http_method_str((enum http_method)htp->method));
-  downstream->set_request_minor(htparser_get_minor(htp));
+  downstream->set_request_major(htp->http_major);
+  downstream->set_request_minor(htp->http_minor);
 
-  downstream->set_request_connection_close(!htparser_should_keep_alive(htp));
+  downstream->set_request_connection_close(!http_should_keep_alive(htp));
 
   DownstreamConnection *dconn;
   dconn = upstream->get_client_handler()->get_downstream_connection();
@@ -176,10 +164,10 @@ int htp_hdrs_completecb(htparser *htp)
 } // namespace
 
 namespace {
-int htp_bodycb(htparser *htp, const char *data, size_t len)
+int htp_bodycb(http_parser *htp, const char *data, size_t len)
 {
   HttpsUpstream *upstream;
-  upstream = reinterpret_cast<HttpsUpstream*>(htparser_get_userdata(htp));
+  upstream = reinterpret_cast<HttpsUpstream*>(htp->data);
   Downstream *downstream = upstream->get_last_downstream();
   downstream->push_upload_data_chunk(reinterpret_cast<const uint8_t*>(data),
                                      len);
@@ -188,44 +176,35 @@ int htp_bodycb(htparser *htp, const char *data, size_t len)
 } // namespace
 
 namespace {
-int htp_msg_completecb(htparser *htp)
+int htp_msg_completecb(http_parser *htp)
 {
   if(ENABLE_LOG) {
     LOG(INFO) << "Upstream https request complete";
   }
   HttpsUpstream *upstream;
-  upstream = reinterpret_cast<HttpsUpstream*>(htparser_get_userdata(htp));
+  upstream = reinterpret_cast<HttpsUpstream*>(htp->data);
   Downstream *downstream = upstream->get_last_downstream();
   downstream->end_upload_data();
   downstream->set_request_state(Downstream::MSG_COMPLETE);
   // Stop further processing to complete this request
-  return 1;
+  http_parser_pause(htp, 1);
+  return 0;
 }
 } // namespace
 
 namespace {
-htparse_hooks htp_hooks = {
+http_parser_settings htp_hooks = {
-  htp_msg_begin, /*htparse_hook      on_msg_begin;*/
+  htp_msg_begin, /*http_cb      on_message_begin;*/
-  htp_methodcb, /*htparse_data_hook method;*/
+  htp_uricb, /*http_data_cb on_url;*/
-  0, /* htparse_data_hook scheme;*/
+  htp_hdr_keycb, /*http_data_cb on_header_field;*/
-  0, /* htparse_data_hook host; */
+  htp_hdr_valcb, /*http_data_cb on_header_value;*/
-  0, /* htparse_data_hook port; */
+  htp_hdrs_completecb, /*http_cb      on_headers_complete;*/
-  0, /* htparse_data_hook path; */
+  htp_bodycb, /*http_data_cb on_body;*/
-  0, /* htparse_data_hook args; */
+  htp_msg_completecb /*http_cb      on_message_complete;*/
-  htp_uricb, /* htparse_data_hook uri; */
-  htp_hdrs_begincb, /* htparse_hook      on_hdrs_begin; */
-  htp_hdr_keycb, /* htparse_data_hook hdr_key; */
-  htp_hdr_valcb, /* htparse_data_hook hdr_val; */
-  0, /* htparse_data_hook hostname; */
-  htp_hdrs_completecb, /* htparse_hook      on_hdrs_complete; */
-  0, /*htparse_hook      on_new_chunk;*/
-  0, /*htparse_hook      on_chunk_complete;*/
-  0, /*htparse_hook      on_chunks_complete;*/
-  htp_bodycb, /* htparse_data_hook body; */
-  htp_msg_completecb /* htparse_hook      on_msg_complete;*/
 };
 } // namespace
 
+
 // on_read() does not consume all available data in input buffer if
 // one http request is fully received.
 int HttpsUpstream::on_read()
@@ -235,15 +214,19 @@ int HttpsUpstream::on_read()
 
   unsigned char *mem = evbuffer_pullup(input, -1);
 
-  int nread = htparser_run(htp_, &htp_hooks,
+  if(evbuffer_get_length(input) == 0) {
-                           reinterpret_cast<const char*>(mem),
+    return 0;
-                           evbuffer_get_length(input));
+  }
+
+  size_t nread = http_parser_execute(htp_, &htp_hooks,
+                                     reinterpret_cast<const char*>(mem),
+                                     evbuffer_get_length(input));
   evbuffer_drain(input, nread);
   // Well, actually header length + some body bytes
   current_header_length_ += nread;
-  htpparse_error htperr = htparser_get_error(htp_);
   Downstream *downstream = get_top_downstream();
-  if(htperr == htparse_error_user) {
+  http_errno htperr = HTTP_PARSER_ERRNO(htp_);
+  if(htperr == HPE_PAUSED) {
     if(downstream->get_request_state() == Downstream::CONNECT_FAIL) {
       get_client_handler()->set_should_close_after_write(true);
       error_reply(503);
@@ -259,7 +242,7 @@ int HttpsUpstream::on_read()
         pause_read(SHRPX_MSG_BLOCK);
       }
     }
-  } else if(htperr == htparse_error_none) {
+  } else if(htperr == HPE_OK) {
     // downstream can be NULL here.
     if(downstream) {
       if(downstream->get_request_state() == Downstream::INITIAL &&
@@ -278,7 +261,8 @@ int HttpsUpstream::on_read()
   } else {
     if(ENABLE_LOG) {
       LOG(INFO) << "Upstream http parse failure: "
-                << htparser_get_strerror(htp_);
+                << "(" << http_errno_name(htperr) << ") "
+                << http_errno_description(htperr);
     }
     get_client_handler()->set_should_close_after_write(true);
     error_reply(400);
@@ -319,6 +303,7 @@ void HttpsUpstream::resume_read(IOCtrlReason reason)
   if(ioctrl_.resume_read(reason)) {
     // Process remaining data in input buffer here because these bytes
     // are not notified by readcb until new data arrive.
+    http_parser_pause(htp_, 0);
     on_read();
   }
 }

examples/shrpx_https_upstream.h

@@ -32,7 +32,7 @@
 #include <deque>
 
 extern "C" {
-#include "htparse/htparse.h"
+#include "http-parser/http_parser.h"
 }
 
 #include "shrpx_upstream.h"
@@ -71,7 +71,7 @@ public:
   void reset_current_header_length();
 private:
   ClientHandler *handler_;
-  htparser *htp_;
+  http_parser *htp_;
   size_t current_header_length_;
   std::deque<Downstream*> downstream_queue_;
   IOControl ioctrl_;

examples/shrpx_spdy_upstream.cc

@@ -351,9 +351,6 @@ ClientHandler* SpdyUpstream::get_client_handler() const
 namespace {
 void spdy_downstream_readcb(bufferevent *bev, void *ptr)
 {
-  if(ENABLE_LOG) {
-    LOG(INFO) << "spdy_downstream_readcb";
-  }
   DownstreamConnection *dconn = reinterpret_cast<DownstreamConnection*>(ptr);
   Downstream *downstream = dconn->get_downstream();
   SpdyUpstream *upstream;
@@ -674,9 +671,6 @@ int SpdyUpstream::on_downstream_header_complete(Downstream *downstream)
 int SpdyUpstream::on_downstream_body(Downstream *downstream,
                                      const uint8_t *data, size_t len)
 {
-  if(ENABLE_LOG) {
-    LOG(INFO) << "Downstream on_downstream_body";
-  }
   evbuffer *body = downstream->get_response_body_buf();
   evbuffer_add(body, data, len);
   spdylay_session_resume_data(session_, downstream->get_stream_id());