asio_http2_client.cpp

asio_http2_client.cpp

// g++ -std=c++23 foo.cpp -lssl -lcrypto -lnghttp2 -lpthread
#include <boost/assert.hpp>
#include <boost/asio.hpp>
#include <boost/asio/ssl.hpp>
#include <nghttp2/nghttp2.h>

#include <atomic>
#include <cstring>
#include <deque>
#include <iostream>
#include <memory>
#include <string>
#include <vector>

namespace net = boost::asio;
using tcp = net::ip::tcp;

class http2_client : public std::enable_shared_from_this<http2_client> {
public:
  http2_client(net::io_context &ioc, net::ssl::context &ssl_ctx,
               std::string host, std::string port, std::string path)
      : ioc_(ioc), host_(std::move(host)), port_(std::move(port)),
        path_(std::move(path)), ssl_ctx_(ssl_ctx), stream_(ioc_, ssl_ctx_) {
    stream_.set_verify_mode(net::ssl::verify_peer);
  }

  ~http2_client() {
    if (session_)
      nghttp2_session_del(session_);
    if (cbs_)
      nghttp2_session_callbacks_del(cbs_);
  }

  void run() {
    net::co_spawn(
        ioc_,
        [this, self = shared_from_this()]() mutable -> net::awaitable<void> {
          try {
            tcp::resolver resolver(ioc_);
            auto results = co_await resolver.async_resolve(host_, port_,
                                                           net::use_awaitable);
            co_await net::async_connect(stream_.next_layer(), results,
                                        net::use_awaitable);
            stream_.next_layer().set_option(net::ip::tcp::no_delay(true));
            co_await start_tls();
          } catch (const std::exception &e) {
            fprintf(stderr, "[ERROR] resolve: %s\n", e.what());
          }
          co_return;
        },
        net::detached);
  }

private:
  auto start_tls() -> net::awaitable<void> {
    SSL_set_min_proto_version(stream_.native_handle(), TLS1_2_VERSION);

    boost::system::error_code ec;
    ssl_ctx_.set_verify_callback(
        boost::asio::ssl::host_name_verification(host_), ec);
    if (ec) {
      fprintf(stderr, "[ERROR] ssl_ctx_.set_verify_callback: %s\n",
              ec.message().c_str());
      co_return;
    }

    try {
      // SNI
      SSL_set_tlsext_host_name(stream_.native_handle(), host_.c_str());

      co_await stream_.async_handshake(net::ssl::stream_base::client,
                                       net::use_awaitable);

      // check ALPN selected
      const unsigned char *alpn = nullptr;
      unsigned int alpn_len = 0;

      SSL_get0_alpn_selected(stream_.native_handle(), &alpn, &alpn_len);
      std::string proto((const char *)alpn, (size_t)alpn_len);
      if (proto != "h2") {
        fprintf(stderr,
                "[INFO] ALPN negotiated: '%s' (not h2), len: %d, Server may not "
                "support HTTP/2.\n",
                proto.c_str(), alpn_len);
        co_return;
      }
      fprintf(stderr, "[INFO] ALPN negotiated: '%s'\n", proto.c_str());
    } catch (const std::exception &e) {
      fprintf(stderr, "[ERROR] async_handshake: %s\n", e.what());
      co_return;
    }

    init_nghttp2();

    submit_settings();
    submit_get();

    net::co_spawn(
        ioc_,
        [this, self = shared_from_this()]() mutable -> net::awaitable<void> {
          fprintf(stderr, "[INFO] pump_out start\n");
          try {
            co_await pump_out();
          } catch (const std::exception &e) {
            fprintf(stderr, "[ERROR] pump_out: %s\n", e.what());
            co_return;
          }
          fprintf(stderr, "[INFO] pump_out end\n");
          co_return;
        },
        net::detached);

    net::co_spawn(
        ioc_,
        [this, self = shared_from_this()]() mutable -> net::awaitable<void> {
          fprintf(stderr, "[INFO] pump_in start\n");
          try {
            co_await pump_in();
          } catch (const std::exception &e) {
            fprintf(stderr, "[ERROR] pump_in: %s\n", e.what());
          }
          done_ = true;
          fprintf(stderr, "[INFO] pump_in end\n");
          co_return;
        },
        net::detached);

    co_return;
  }

  void setup_nghttp2_callbacks(nghttp2_session_callbacks *callbacks) {
    nghttp2_session_callbacks_set_on_frame_send_callback(
        callbacks, on_frame_send_callback);
    nghttp2_session_callbacks_set_on_frame_recv_callback(
        callbacks, on_frame_recv_callback);

    nghttp2_session_callbacks_set_on_stream_close_callback(
        callbacks, on_stream_close_callback);
    nghttp2_session_callbacks_set_on_data_chunk_recv_callback(
        callbacks, on_data_chunk_recv_callback);

    nghttp2_session_callbacks_set_on_header_callback(callbacks,
                                                     on_header_callback);
  }

  void init_nghttp2() {
    nghttp2_session_callbacks_new(&cbs_);
    setup_nghttp2_callbacks(cbs_);

    int rv = nghttp2_session_client_new(&session_, cbs_, this);
    if (rv != 0) {
      throw std::runtime_error("nghttp2_session_client_new failed");
    }
  }

  void submit_settings() {
    // 提交初始 SETTINGS（会导致输出连接前言 + SETTINGS 等）
    nghttp2_submit_settings(session_, NGHTTP2_FLAG_NONE, nullptr, 0);
  }

  void submit_get() {
    fprintf(stderr, "[INFO] submit_get\n");

    std::vector<nghttp2_nv> nva;
    nva.push_back(make_nv(":method", "GET"));
    nva.push_back(make_nv(":scheme", "https"));
    nva.push_back(make_nv(":authority", host_.c_str()));
    nva.push_back(make_nv(":path", path_.c_str()));
    nva.push_back(make_nv("user-agent", "asio-nghttp2-min/0.1"));
    nva.push_back(make_nv("accept", "*/*"));

    stream_id_ = nghttp2_submit_request2(session_, nullptr, nva.data(),
                                         (size_t)nva.size(), nullptr, nullptr);

    if (stream_id_ < 0) {
      throw std::runtime_error("nghttp2_submit_request failed");
    }

    printf("[INFO] Stream ID = %d\n", stream_id_);
  }

  net::awaitable<void> pump_out() {
    net::steady_timer timer(ioc_);
    size_t total_written = 0;

    for (;;) {
      if (!nghttp2_session_want_write(session_)) {
        if (done_) {
          co_return;
        }
        fprintf(stderr, "[INFO] nghttp2_session_mem_send2 waiting data...\n");
        timer.expires_after(net::chrono::milliseconds(500));
        co_await timer.async_wait(net::use_awaitable);
        continue;
      }

      const uint8_t *data = nullptr;
      ssize_t len = nghttp2_session_mem_send2(session_, &data);
      if (len < 0) {
        fprintf(stderr, "[ERROR] nghttp2_session_mem_send2 error: %s\n",
                nghttp2_strerror(len));
        co_return;
      }

      if (len == 0) {
        continue;
      }

      total_written += len;

      co_await async_write(stream_, net::buffer(data, len), net::use_awaitable);
    }

    co_return;
  }

  net::awaitable<void> pump_in() {
    size_t total_read = 0;
    char read_buf[16 * 1024];

    for (;;) {
      // 从 socket 读入数据
      auto nread = co_await stream_.async_read_some(net::buffer(read_buf),
                                                    net::use_awaitable);
      if (nread == 0) {
        fprintf(stderr, "[ERROR] pump_in error: %d\n", total_read);
        break;
      }
      total_read += nread;

      // 把收到的数据喂给 nghttp2
      ssize_t rv =
          nghttp2_session_mem_recv2(session_, (const uint8_t *)read_buf, nread);
      if (rv < 0) {
        fprintf(stderr, "[ERROR] nghttp2_session_mem_recv2 error: %s\n",
                nghttp2_strerror(rv));
        break;
      }
      BOOST_ASSERT("nghttp2_session_mem_recv2 must return the same number of bytes read" && rv == nread);
    }

    co_return;
  }

  ////////////////////////////////////////////////////////////////////////////////////
  // ---- nghttp2 callbacks ----

  static void print_header(FILE *f, const uint8_t *name, size_t namelen,
                           const uint8_t *value, size_t valuelen) {
    fwrite(name, 1, namelen, f);
    fprintf(f, ": ");
    fwrite(value, 1, valuelen, f);
    fprintf(f, "\n");
  }

  static int on_header_callback(nghttp2_session *session,
                                const nghttp2_frame *frame, const uint8_t *name,
                                size_t namelen, const uint8_t *value,
                                size_t valuelen, uint8_t flags,
                                void *user_data) {
    http2_client *client = (http2_client *)user_data;
    (void)session;
    (void)flags;

    switch (frame->hd.type) {
    case NGHTTP2_HEADERS:
      if (frame->headers.cat == NGHTTP2_HCAT_RESPONSE &&
          client->stream_id_ == frame->hd.stream_id) {
        /* Print response headers for the initiated request. */
        print_header(stderr, name, namelen, value, valuelen);
        break;
      }
    }
    return 0;
  }

  static int on_frame_send_callback(nghttp2_session *session,
                                    const nghttp2_frame *frame,
                                    void *user_data) {
    size_t i;
    (void)user_data;

    switch (frame->hd.type) {
    case NGHTTP2_HEADERS: {
      const nghttp2_nv *nva = frame->headers.nva;
      for (i = 0; i < frame->headers.nvlen; ++i) {
        fwrite(nva[i].name, 1, nva[i].namelen, stdout);
        printf(": ");
        fwrite(nva[i].value, 1, nva[i].valuelen, stdout);
        printf("\n");
      }
      printf("[INFO] C ----------------------------> S (HEADERS)\n");
    } break;
    case NGHTTP2_RST_STREAM:
      printf("[INFO] C ----------------------------> S (RST_STREAM)\n");
      break;
    case NGHTTP2_GOAWAY:
      printf("[INFO] C ----------------------------> S (GOAWAY)\n");
      break;
    }
    return 0;
  }

  static int on_frame_recv_callback(nghttp2_session *session,
                                    const nghttp2_frame *frame,
                                    void *user_data) {
    size_t i;
    (void)user_data;

    switch (frame->hd.type) {
    case NGHTTP2_HEADERS:
      if (frame->headers.cat == NGHTTP2_HCAT_RESPONSE) {
        printf("[INFO] C <---------------------------- S (HEADERS)\n");
      }
      break;
    case NGHTTP2_RST_STREAM:
      printf("[INFO] C <---------------------------- S (RST_STREAM)\n");
      break;
    case NGHTTP2_GOAWAY:
      printf("[INFO] C <---------------------------- S (GOAWAY)\n");
      break;
    }
    return 0;
  }

  static int on_stream_close_callback(nghttp2_session *session,
                                      int32_t stream_id, uint32_t error_code,
                                      void *user_data) {
    (void)error_code;
    (void)user_data;

    printf("[INFO] C <---------------------------- S (STREAM_CLOSE)\n");

    int rv;
    rv = nghttp2_session_terminate_session(session, NGHTTP2_NO_ERROR);
    if (rv != 0) {
      fprintf(stderr, "nghttp2_session_terminate_session error: %s\n",
              nghttp2_strerror(rv));
    }

    return 0;
  }

  static int on_data_chunk_recv_callback(nghttp2_session *session,
                                         uint8_t flags, int32_t stream_id,
                                         const uint8_t *data, size_t len,
                                         void *user_data) {
    (void)session;
    (void)flags;
    (void)stream_id;
    (void)user_data;

    printf("[INFO] C <---------------------------- S (DATA chunk)\n"
           "%lu bytes\n",
           (unsigned long int)len);
    fwrite(data, 1, len, stdout);
    printf("\n");

    return 0;
  }

  nghttp2_nv make_nv(const char *name, const char *value) {
    nghttp2_nv nv;
    nv.name = (uint8_t *)name;
    nv.value = (uint8_t *)value;
    nv.namelen = strlen(name);
    nv.valuelen = strlen(value);
    nv.flags = NGHTTP2_NV_FLAG_NONE; // 让库拷贝
    return nv;
  }

private:
  net::io_context &ioc_;
  net::ssl::context &ssl_ctx_;
  std::string host_, port_, path_;

  net::ssl::stream<tcp::socket> stream_;

  nghttp2_session_callbacks *cbs_ = nullptr;
  nghttp2_session *session_ = nullptr;

  int32_t stream_id_ = -1;
  std::atomic_bool done_{false};
};

int main() {
  net::io_context ioc;
  net::ssl::context ssl_ctx(net::ssl::context::tls_client);

  printf("OpenSSL runtime version: %s\n", OpenSSL_version(OPENSSL_VERSION));

  ssl_ctx.set_options(
      net::ssl::context::default_workarounds | net::ssl::context::no_sslv2 |
      net::ssl::context::no_sslv3 | net::ssl::context::no_tlsv1 |
      net::ssl::context::no_tlsv1_1);

  SSL_CTX_set_mode(ssl_ctx.native_handle(), SSL_MODE_AUTO_RETRY);
  SSL_CTX_set_mode(ssl_ctx.native_handle(), SSL_MODE_RELEASE_BUFFERS);

  // 证书校验（示例：用系统 CA）
  ssl_ctx.set_default_verify_paths();

  // ALPN: advertise "h2"
  // OpenSSL API：SSL_CTX_set_alpn_protos(ctx, "\x02h2", 3)
  int rc = SSL_CTX_set_alpn_protos(ssl_ctx.native_handle(),
                                   (const unsigned char *)"\x02h2", 3);
  if (rc != 0) {
    fprintf(stderr, "SSL_set_alpn_protos failed, rc=%d\n", rc);
    return 1;
  }

  // GET https://nghttp2.org/
  auto c =
      std::make_shared<http2_client>(ioc, ssl_ctx, "google.com", "443", "/");
  c->run();

  ioc.run();
  return 0;
}

// g++ -std=c++23 foo.cpp -lssl -lcrypto -lnghttp2 -lpthread
#include <boost/asio.hpp>
#include <boost/asio/ssl.hpp>
#include <boost/assert.hpp>
#include <nghttp2/nghttp2.h>

// #include <iostream>
#include <atomic>
#include <cstring>
#include <map>
#include <memory>
#include <string>
#include <vector>

namespace net = boost::asio;
using tcp = net::ip::tcp;

template <typename Stream> class async_http2_session {
  //////////////////////////////////////////////////////////////////////////
public:
  async_http2_session(Stream &&stream) : stream_(std::move(stream)) {
    init_nghttp2();
  }

  net::awaitable<bool>
  async_request(std::string_view host, std::string_view port,
                std::string_view path,
                std::map<std::string, std::string> headers) noexcept {
    if (settings_entries_.empty()) {
      nghttp2_submit_settings(session_, NGHTTP2_FLAG_NONE, nullptr, 0);
    } else {
      nghttp2_submit_settings(session_, NGHTTP2_FLAG_NONE,
                              settings_entries_.data(),
                              settings_entries_.size());
    }

    std::vector<nghttp2_nv> nva;
    std::string authority = std::string(host);
    if (port.empty() || port == "443") {
      authority += ":" + std::string(port);
    }

    nva.push_back(make_nv(":method", "GET"));
    nva.push_back(make_nv(":scheme", "https"));
    nva.push_back(make_nv(":authority", authority.data()));
    nva.push_back(make_nv(":path", path.data()));

    for (auto &[k, v] : headers) {
      nva.push_back(make_nv(k.data(), v.data()));
    }

    stream_id_ = nghttp2_submit_request2(session_, nullptr, nva.data(),
                                         (size_t)nva.size(), nullptr, nullptr);
    if (stream_id_ < 0) {
      co_return false;
    }

    printf("[INFO] Stream ID = %d\n", stream_id_);

    // 下面需要循环异步交换数据，直到 headers
    // 接收完成才返回，通过回调函数中设置的 状态变量来判断.

    while (!headers_done_) {
      auto need_more = co_await pump_io();
      if (!need_more) {
        break;
      }
    }

    // 重置 headers_done_ 为 false，准备下一次请求时状态不被影响.
    headers_done_ = false;

    co_return true;
  }

  net::awaitable<size_t> async_read_some(net::mutable_buffer buffer) noexcept {
    if (stream_buf_.size() > 0) {
      co_return stream_buf_.sgetn((char *)buffer.data(), buffer.size());
    }

    if (data_done_) {
      co_return 0;
    }

    while (!data_done_) {
      auto need_more = co_await pump_io();
      if (!need_more) {
        break;
      }

      if (stream_buf_.size() > 0) {
        break;
      }
    }

    co_return stream_buf_.sgetn((char *)buffer.data(), buffer.size());
  }

  //////////////////////////////////////////////////////////////////////////

private:
  net::awaitable<bool> pump_io() {
    bool need_more = false;

    try {
      while (nghttp2_session_want_write(session_)) {
        need_more = true;

        const uint8_t *data = nullptr;
        ssize_t len = nghttp2_session_mem_send2(session_, &data);
        if (len < 0) {
          fprintf(stderr, "[ERROR] nghttp2_session_mem_send2 error: %s\n",
                  nghttp2_strerror(len));
          co_return false;
        }

        if (len != 0) {
          co_await async_write(stream_, net::buffer(data, len),
                               net::use_awaitable);
        }
      }

      if (nghttp2_session_want_read(session_)) {
        need_more = true;

        char read_buf[16 * 1024];

        // 从 socket 读入数据
        auto nread = co_await stream_.async_read_some(net::buffer(read_buf),
                                                      net::use_awaitable);
        if (nread == 0) {
          co_return false;
        }

        // 把收到的数据喂给 nghttp2
        ssize_t rv = nghttp2_session_mem_recv2(
            session_, (const uint8_t *)read_buf, nread);
        if (rv < 0) {
          fprintf(stderr, "[ERROR] nghttp2_session_mem_recv2 error: %s\n",
                  nghttp2_strerror(rv));
          co_return false;
        }
        BOOST_ASSERT("nghttp2_session_mem_recv2 must return the same number of "
                     "bytes read" &&
                     rv == nread);
      }
    } catch (const std::exception &e) {
      fprintf(stderr, "[ERROR] pump_io: %s\n", e.what());
      co_return false;
    }

    co_return need_more;
  }

  void init_nghttp2() {
    nghttp2_session_callbacks_new(&cbs_);

    nghttp2_session_callbacks_set_on_frame_send_callback(
        cbs_, on_frame_send_callback);
    nghttp2_session_callbacks_set_on_frame_recv_callback(
        cbs_, on_frame_recv_callback);

    nghttp2_session_callbacks_set_on_stream_close_callback(
        cbs_, on_stream_close_callback);
    nghttp2_session_callbacks_set_on_data_chunk_recv_callback(
        cbs_, on_data_chunk_recv_callback);

    nghttp2_session_callbacks_set_on_header_callback(cbs_, on_header_callback);

    int rv = nghttp2_session_client_new(&session_, cbs_, this);
    if (rv != 0) {
      throw std::runtime_error("nghttp2_session_client_new failed");
    }
  }

  int on_header_callback(nghttp2_session *session, const nghttp2_frame *frame,
                         const uint8_t *name, size_t namelen,
                         const uint8_t *value, size_t valuelen, uint8_t flags) {
    (void)session;
    (void)flags;

    switch (frame->hd.type) {
    case NGHTTP2_HEADERS:
      if (frame->headers.cat == NGHTTP2_HCAT_RESPONSE &&
          frame->hd.stream_id == stream_id_) {
        /* Print response headers for the initiated request. */
        print_header(stderr, name, namelen, value, valuelen);
        break;
      }
    }

    return 0;
  }

  int on_frame_send_callback(nghttp2_session *session,
                             const nghttp2_frame *frame) {
    (void)session;

    size_t i;

    switch (frame->hd.type) {
    case NGHTTP2_HEADERS: {
      const nghttp2_nv *nva = frame->headers.nva;
      for (i = 0; i < frame->headers.nvlen; ++i) {
        fwrite(nva[i].name, 1, nva[i].namelen, stdout);
        printf(": ");
        fwrite(nva[i].value, 1, nva[i].valuelen, stdout);
        printf("\n");
      }
      printf("[INFO] C ----------------------------> S (HEADERS)\n");
    } break;
    case NGHTTP2_RST_STREAM:
      printf("[INFO] C ----------------------------> S (RST_STREAM)\n");
      break;
    case NGHTTP2_GOAWAY:
      data_done_ = true;
      printf("[INFO] C ----------------------------> S (GOAWAY)\n");
      break;
    }
    return 0;
  }

  int on_frame_recv_callback(nghttp2_session *session,
                             const nghttp2_frame *frame) {
    size_t i;

    switch (frame->hd.type) {
    case NGHTTP2_HEADERS:
      if (frame->headers.cat == NGHTTP2_HCAT_RESPONSE) {
        headers_done_ = true;
        printf("[INFO] C <---------------------------- S (HEADERS)\n");
      }
      break;
    case NGHTTP2_RST_STREAM:
      printf("[INFO] C <---------------------------- S (RST_STREAM)\n");
      break;
    case NGHTTP2_GOAWAY:
      data_done_ = true;
      printf("[INFO] C <---------------------------- S (GOAWAY)\n");
      break;
    }
    return 0;
  }

  int on_stream_close_callback(nghttp2_session *session, int32_t stream_id,
                               uint32_t error_code) {
    (void)error_code;

    data_done_ = true;
    printf("[INFO] C <---------------------------- S (STREAM_CLOSE)\n");

    int rv;
    rv = nghttp2_session_terminate_session(session, NGHTTP2_NO_ERROR);
    if (rv != 0) {
      fprintf(stderr, "nghttp2_session_terminate_session error: %s\n",
              nghttp2_strerror(rv));
    }

    return 0;
  }

  int on_data_chunk_recv_callback(nghttp2_session *session, uint8_t flags,
                                  int32_t stream_id, const uint8_t *data,
                                  size_t len) {
    (void)session;
    (void)flags;
    (void)stream_id;

#if 0
    printf("[INFO] C <---------------------------- S (DATA chunk)\n"
           "%lu bytes\n",
           (unsigned long int)len);
    fwrite(data, 1, len, stdout);
    printf("\n");
#endif

    auto bufs = stream_buf_.prepare(len);
    memcpy(bufs.data(), data, len);
    stream_buf_.commit(len);

    return 0;
  }

  nghttp2_nv make_nv(const char *name, const char *value) {
    nghttp2_nv nv;
    nv.name = (uint8_t *)name;
    nv.value = (uint8_t *)value;
    nv.namelen = strlen(name);
    nv.valuelen = strlen(value);
    nv.flags = NGHTTP2_NV_FLAG_NONE; // 让库拷贝
    return nv;
  }

  //////////////////////////////////////////////////////////////////////////
  // nghttp2 callbacks
  void print_header(FILE *f, const uint8_t *name, size_t namelen,
                    const uint8_t *value, size_t valuelen) {
    fwrite(name, 1, namelen, f);
    fprintf(f, ": ");
    fwrite(value, 1, valuelen, f);
    fprintf(f, "\n");
  }

  static int on_header_callback(nghttp2_session *session,
                                const nghttp2_frame *frame, const uint8_t *name,
                                size_t namelen, const uint8_t *value,
                                size_t valuelen, uint8_t flags,
                                void *user_data) {
    async_http2_session<Stream> *client =
        (async_http2_session<Stream> *)user_data;
    return client->on_header_callback(session, frame, name, namelen, value,
                                      valuelen, flags);
  }

  static int on_frame_send_callback(nghttp2_session *session,
                                    const nghttp2_frame *frame,
                                    void *user_data) {
    async_http2_session<Stream> *client =
        (async_http2_session<Stream> *)user_data;
    return client->on_frame_send_callback(session, frame);
  }

  static int on_frame_recv_callback(nghttp2_session *session,
                                    const nghttp2_frame *frame,
                                    void *user_data) {
    async_http2_session<Stream> *client =
        (async_http2_session<Stream> *)user_data;

    return client->on_frame_recv_callback(session, frame);
  }

  static int on_stream_close_callback(nghttp2_session *session,
                                      int32_t stream_id, uint32_t error_code,
                                      void *user_data) {
    async_http2_session<Stream> *client =
        (async_http2_session<Stream> *)user_data;
    return client->on_stream_close_callback(session, stream_id, error_code);
  }

  static int on_data_chunk_recv_callback(nghttp2_session *session,
                                         uint8_t flags, int32_t stream_id,
                                         const uint8_t *data, size_t len,
                                         void *user_data) {
    async_http2_session<Stream> *client =
        (async_http2_session<Stream> *)user_data;
    return client->on_data_chunk_recv_callback(session, flags, stream_id, data,
                                               len);
  }

private:
  Stream stream_;
  nghttp2_session *session_ = nullptr;
  nghttp2_session_callbacks *cbs_ = nullptr;
  std::vector<nghttp2_settings_entry> settings_entries_;

  bool headers_done_ = false;
  bool data_done_ = false;
  net::streambuf stream_buf_;

  int32_t stream_id_ = -1;
};