src/asio_client_session_impl.cc

   1 /*
   2  * nghttp2 - HTTP/2 C Library
   3  *
   4  * Copyright (c) 2015 Tatsuhiro Tsujikawa
   5  *
   6  * Permission is hereby granted, free of charge, to any person obtaining
   7  * a copy of this software and associated documentation files (the
   8  * "Software"), to deal in the Software without restriction, including
   9  * without limitation the rights to use, copy, modify, merge, publish,
  10  * distribute, sublicense, and/or sell copies of the Software, and to
  11  * permit persons to whom the Software is furnished to do so, subject to
  12  * the following conditions:
  13  *
  14  * The above copyright notice and this permission notice shall be
  15  * included in all copies or substantial portions of the Software.
  16  *
  17  * THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND,
  18  * EXPRESS OR IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF
  19  * MERCHANTABILITY, FITNESS FOR A PARTICULAR PURPOSE AND
  20  * NONINFRINGEMENT. IN NO EVENT SHALL THE AUTHORS OR COPYRIGHT HOLDERS BE
  21  * LIABLE FOR ANY CLAIM, DAMAGES OR OTHER LIABILITY, WHETHER IN AN ACTION
  22  * OF CONTRACT, TORT OR OTHERWISE, ARISING FROM, OUT OF OR IN CONNECTION
  23  * WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE SOFTWARE.
  24  */
  25 #include "asio_client_session_impl.h"
  26
  27 #include <iostream>
  28
  29 #include "asio_client_stream.h"
  30 #include "asio_client_request_impl.h"
  31 #include "asio_client_response_impl.h"
  32 #include "asio_common.h"
  33 #include "template.h"
  34 #include "util.h"
  35 #include "http2.h"
  36
  37 namespace nghttp2 {
  38 namespace asio_http2 {
  39 namespace client {
  40
  41 session_impl::session_impl(boost::asio::io_service &io_service)
  42     : wblen_(0), io_service_(io_service), resolver_(io_service),
  43       session_(nullptr), data_pending_(nullptr), data_pendinglen_(0),
  44       writing_(false), inside_callback_(false) {}
  45
  46 session_impl::~session_impl() {
  47   // finish up all active stream
  48   for (auto &p : streams_) {
  49     auto &strm = p.second;
  50     auto &req = strm->request().impl();
  51     req.call_on_close(NGHTTP2_INTERNAL_ERROR);
  52   }
  53
  54   nghttp2_session_del(session_);
  55 }
  56
  57 void session_impl::start_resolve(const std::string &host,
  58                                  const std::string &service) {
  59   resolver_.async_resolve({host, service},
  60                           [this](const boost::system::error_code &ec,
  61                                  tcp::resolver::iterator endpoint_it) {
  62     if (ec) {
  63       not_connected(ec);
  64       return;
  65     }
  66
  67     start_connect(endpoint_it);
  68   });
  69 }
  70
  71 void session_impl::connected(tcp::resolver::iterator endpoint_it) {
  72   if (!setup_session()) {
  73     return;
  74   }
  75
  76   socket().set_option(boost::asio::ip::tcp::no_delay(true));
  77
  78   do_write();
  79   do_read();
  80
  81   auto &connect_cb = on_connect();
  82   if (connect_cb) {
  83     connect_cb(endpoint_it);
  84   }
  85 }
  86
  87 void session_impl::not_connected(const boost::system::error_code &ec) {
  88   call_error_cb(ec);
  89 }
  90
  91 void session_impl::on_connect(connect_cb cb) { connect_cb_ = std::move(cb); }
  92
  93 void session_impl::on_error(error_cb cb) { error_cb_ = std::move(cb); }
  94
  95 const connect_cb &session_impl::on_connect() const { return connect_cb_; }
  96
  97 const error_cb &session_impl::on_error() const { return error_cb_; }
  98
  99 void session_impl::call_error_cb(const boost::system::error_code &ec) {
 100   auto &error_cb = on_error();
 101   if (!error_cb) {
 102     return;
 103   }
 104   error_cb(ec);
 105 }
 106
 107 namespace {
 108 int on_begin_headers_callback(nghttp2_session *session,
 109                               const nghttp2_frame *frame, void *user_data) {
 110   if (frame->hd.type != NGHTTP2_PUSH_PROMISE) {
 111     return 0;
 112   }
 113
 114   auto sess = static_cast<session_impl *>(user_data);
 115   sess->create_push_stream(frame->push_promise.promised_stream_id);
 116
 117   return 0;
 118 }
 119 } // namespace
 120
 121 namespace {
 122 int on_header_callback(nghttp2_session *session, const nghttp2_frame *frame,
 123                        const uint8_t *name, size_t namelen,
 124                        const uint8_t *value, size_t valuelen, uint8_t flags,
 125                        void *user_data) {
 126   auto sess = static_cast<session_impl *>(user_data);
 127   stream *strm;
 128
 129   switch (frame->hd.type) {
 130   case NGHTTP2_HEADERS: {
 131     strm = sess->find_stream(frame->hd.stream_id);
 132     if (!strm) {
 133       return 0;
 134     }
 135
 136     // ignore trailers
 137     if (frame->headers.cat == NGHTTP2_HCAT_HEADERS &&
 138         !strm->expect_final_response()) {
 139       return 0;
 140     }
 141
 142     auto token = http2::lookup_token(name, namelen);
 143
 144     auto &res = strm->response().impl();
 145     if (token == http2::HD__STATUS) {
 146       res.status_code(util::parse_uint(value, valuelen));
 147     } else {
 148
 149       if (token == http2::HD_CONTENT_LENGTH) {
 150         res.content_length(util::parse_uint(value, valuelen));
 151       }
 152
 153       res.header().emplace(
 154           std::string(name, name + namelen),
 155           header_value{std::string(value, value + valuelen),
 156                        (flags & NGHTTP2_NV_FLAG_NO_INDEX) != 0});
 157     }
 158     break;
 159   }
 160   case NGHTTP2_PUSH_PROMISE: {
 161     strm = sess->find_stream(frame->push_promise.promised_stream_id);
 162     if (!strm) {
 163       return 0;
 164     }
 165
 166     auto &req = strm->request().impl();
 167     auto &uri = req.uri();
 168
 169     switch (http2::lookup_token(name, namelen)) {
 170     case http2::HD__METHOD:
 171       req.method(std::string(value, value + valuelen));
 172       break;
 173     case http2::HD__SCHEME:
 174       uri.scheme.assign(value, value + valuelen);
 175       break;
 176     case http2::HD__PATH:
 177       split_path(uri, value, value + valuelen);
 178       break;
 179     case http2::HD__AUTHORITY:
 180       uri.host.assign(value, value + valuelen);
 181       break;
 182     case http2::HD_HOST:
 183       if (uri.host.empty()) {
 184         uri.host.assign(value, value + valuelen);
 185       }
 186     // fall through
 187     default:
 188       req.header().emplace(
 189           std::string(name, name + namelen),
 190           header_value{std::string(value, value + valuelen),
 191                        (flags & NGHTTP2_NV_FLAG_NO_INDEX) != 0});
 192     }
 193
 194     break;
 195   }
 196   default:
 197     return 0;
 198   }
 199
 200   return 0;
 201 }
 202 } // namespace
 203
 204 namespace {
 205 int on_frame_recv_callback(nghttp2_session *session, const nghttp2_frame *frame,
 206                            void *user_data) {
 207   auto sess = static_cast<session_impl *>(user_data);
 208   auto strm = sess->find_stream(frame->hd.stream_id);
 209
 210   switch (frame->hd.type) {
 211   case NGHTTP2_DATA: {
 212     if (!strm) {
 213       return 0;
 214     }
 215     if (frame->hd.flags & NGHTTP2_FLAG_END_STREAM) {
 216       strm->response().impl().call_on_data(nullptr, 0);
 217     }
 218     break;
 219   }
 220   case NGHTTP2_HEADERS: {
 221     if (!strm) {
 222       return 0;
 223     }
 224
 225     // ignore trailers
 226     if (frame->headers.cat == NGHTTP2_HCAT_HEADERS &&
 227         !strm->expect_final_response()) {
 228       return 0;
 229     }
 230
 231     if (strm->expect_final_response()) {
 232       // wait for final response
 233       return 0;
 234     }
 235
 236     auto &req = strm->request().impl();
 237     req.call_on_response(strm->response());
 238     if (frame->hd.flags & NGHTTP2_FLAG_END_STREAM) {
 239       strm->response().impl().call_on_data(nullptr, 0);
 240     }
 241     break;
 242   }
 243   case NGHTTP2_PUSH_PROMISE: {
 244     if (!strm) {
 245       return 0;
 246     }
 247
 248     auto push_strm = sess->find_stream(frame->push_promise.promised_stream_id);
 249     if (!push_strm) {
 250       return 0;
 251     }
 252
 253     strm->request().impl().call_on_push(push_strm->request());
 254
 255     break;
 256   }
 257   }
 258   return 0;
 259 }
 260 } // namespace
 261
 262 namespace {
 263 int on_data_chunk_recv_callback(nghttp2_session *session, uint8_t flags,
 264                                 int32_t stream_id, const uint8_t *data,
 265                                 size_t len, void *user_data) {
 266   auto sess = static_cast<session_impl *>(user_data);
 267   auto strm = sess->find_stream(stream_id);
 268   if (!strm) {
 269     return 0;
 270   }
 271
 272   auto &res = strm->response().impl();
 273   res.call_on_data(data, len);
 274
 275   return 0;
 276 }
 277 } // namespace
 278
 279 namespace {
 280 int on_stream_close_callback(nghttp2_session *session, int32_t stream_id,
 281                              uint32_t error_code, void *user_data) {
 282   auto sess = static_cast<session_impl *>(user_data);
 283   auto strm = sess->pop_stream(stream_id);
 284   if (!strm) {
 285     return 0;
 286   }
 287
 288   strm->request().impl().call_on_close(error_code);
 289
 290   return 0;
 291 }
 292 } // namespace
 293
 294 bool session_impl::setup_session() {
 295   nghttp2_session_callbacks *callbacks;
 296   nghttp2_session_callbacks_new(&callbacks);
 297   auto cb_del = defer(nghttp2_session_callbacks_del, callbacks);
 298
 299   nghttp2_session_callbacks_set_on_begin_headers_callback(
 300       callbacks, on_begin_headers_callback);
 301   nghttp2_session_callbacks_set_on_header_callback(callbacks,
 302                                                    on_header_callback);
 303   nghttp2_session_callbacks_set_on_frame_recv_callback(callbacks,
 304                                                        on_frame_recv_callback);
 305   nghttp2_session_callbacks_set_on_data_chunk_recv_callback(
 306       callbacks, on_data_chunk_recv_callback);
 307   nghttp2_session_callbacks_set_on_stream_close_callback(
 308       callbacks, on_stream_close_callback);
 309
 310   auto rv = nghttp2_session_client_new(&session_, callbacks, this);
 311   if (rv != 0) {
 312     call_error_cb(make_error_code(static_cast<nghttp2_error>(rv)));
 313     return false;
 314   }
 315
 316   const uint32_t window_size = 256 * 1024 * 1024;
 317
 318   std::array<nghttp2_settings_entry, 2> iv{
 319       {{NGHTTP2_SETTINGS_MAX_CONCURRENT_STREAMS, 100},
 320        // typically client is just a *sink* and just process data as
 321        // much as possible.  Use large window size by default.
 322        {NGHTTP2_SETTINGS_INITIAL_WINDOW_SIZE, window_size}}};
 323   nghttp2_submit_settings(session_, NGHTTP2_FLAG_NONE, iv.data(), iv.size());
 324   // increase connection window size up to window_size
 325   nghttp2_submit_window_update(session_, NGHTTP2_FLAG_NONE, 0,
 326                                window_size -
 327                                    NGHTTP2_INITIAL_CONNECTION_WINDOW_SIZE);
 328   return true;
 329 }
 330
 331 int session_impl::write_trailer(stream &strm, header_map h) {
 332   int rv;
 333   auto nva = std::vector<nghttp2_nv>();
 334   nva.reserve(h.size());
 335   for (auto &hd : h) {
 336     nva.push_back(nghttp2::http2::make_nv(hd.first, hd.second.value,
 337                                           hd.second.sensitive));
 338   }
 339
 340   rv = nghttp2_submit_trailer(session_, strm.stream_id(), nva.data(),
 341                               nva.size());
 342
 343   if (rv != 0) {
 344     return -1;
 345   }
 346
 347   signal_write();
 348
 349   return 0;
 350 }
 351
 352 void session_impl::cancel(stream &strm, uint32_t error_code) {
 353   nghttp2_submit_rst_stream(session_, NGHTTP2_FLAG_NONE, strm.stream_id(),
 354                             error_code);
 355   signal_write();
 356 }
 357
 358 void session_impl::resume(stream &strm) {
 359   nghttp2_session_resume_data(session_, strm.stream_id());
 360   signal_write();
 361 }
 362
 363 stream *session_impl::find_stream(int32_t stream_id) {
 364   auto it = streams_.find(stream_id);
 365   if (it == std::end(streams_)) {
 366     return nullptr;
 367   }
 368   return (*it).second.get();
 369 }
 370
 371 std::unique_ptr<stream> session_impl::pop_stream(int32_t stream_id) {
 372   auto it = streams_.find(stream_id);
 373   if (it == std::end(streams_)) {
 374     return nullptr;
 375   }
 376   auto strm = std::move((*it).second);
 377   streams_.erase(it);
 378   return strm;
 379 }
 380
 381 stream *session_impl::create_push_stream(int32_t stream_id) {
 382   auto strm = create_stream();
 383   strm->stream_id(stream_id);
 384   auto p = streams_.emplace(stream_id, std::move(strm));
 385   assert(p.second);
 386   return (*p.first).second.get();
 387 }
 388
 389 std::unique_ptr<stream> session_impl::create_stream() {
 390   return make_unique<stream>(this);
 391 }
 392
 393 const request *session_impl::submit(boost::system::error_code &ec,
 394                                     const std::string &method,
 395                                     const std::string &uri, generator_cb cb,
 396                                     header_map h) {
 397   ec.clear();
 398
 399   http_parser_url u{};
 400   // TODO Handle CONNECT method
 401   if (http_parser_parse_url(uri.c_str(), uri.size(), 0, &u) != 0) {
 402     ec = make_error_code(boost::system::errc::invalid_argument);
 403     return nullptr;
 404   }
 405
 406   if ((u.field_set & (1 << UF_SCHEMA)) == 0 ||
 407       (u.field_set & (1 << UF_HOST)) == 0) {
 408     ec = make_error_code(boost::system::errc::invalid_argument);
 409     return nullptr;
 410   }
 411
 412   auto strm = create_stream();
 413   auto &req = strm->request().impl();
 414   auto &uref = req.uri();
 415
 416   http2::copy_url_component(uref.scheme, &u, UF_SCHEMA, uri.c_str());
 417   http2::copy_url_component(uref.host, &u, UF_HOST, uri.c_str());
 418   http2::copy_url_component(uref.raw_path, &u, UF_PATH, uri.c_str());
 419   http2::copy_url_component(uref.raw_query, &u, UF_QUERY, uri.c_str());
 420
 421   if (util::ipv6_numeric_addr(uref.host.c_str())) {
 422     uref.host = "[" + uref.host;
 423     uref.host += "]";
 424   }
 425   if (u.field_set & (1 << UF_PORT)) {
 426     uref.host += ":";
 427     uref.host += util::utos(u.port);
 428   }
 429
 430   if (uref.raw_path.empty()) {
 431     uref.raw_path = "/";
 432   }
 433
 434   uref.path = percent_decode(uref.raw_path);
 435
 436   auto path = uref.raw_path;
 437   if (u.field_set & (1 << UF_QUERY)) {
 438     path += "?";
 439     path += uref.raw_query;
 440   }
 441
 442   auto nva = std::vector<nghttp2_nv>();
 443   nva.reserve(3 + h.size());
 444   nva.push_back(http2::make_nv_ls(":method", method));
 445   nva.push_back(http2::make_nv_ls(":scheme", uref.scheme));
 446   nva.push_back(http2::make_nv_ls(":path", path));
 447   nva.push_back(http2::make_nv_ls(":authority", uref.host));
 448   for (auto &kv : h) {
 449     nva.push_back(
 450         http2::make_nv(kv.first, kv.second.value, kv.second.sensitive));
 451   }
 452
 453   req.header(std::move(h));
 454
 455   nghttp2_data_provider *prdptr = nullptr;
 456   nghttp2_data_provider prd;
 457
 458   if (cb) {
 459     strm->request().impl().on_read(std::move(cb));
 460     prd.source.ptr = strm.get();
 461     prd.read_callback =
 462         [](nghttp2_session *session, int32_t stream_id, uint8_t *buf,
 463            size_t length, uint32_t *data_flags, nghttp2_data_source *source,
 464            void *user_data) -> ssize_t {
 465       auto strm = static_cast<stream *>(source->ptr);
 466       return strm->request().impl().call_on_read(buf, length, data_flags);
 467     };
 468     prdptr = &prd;
 469   }
 470
 471   auto stream_id = nghttp2_submit_request(session_, nullptr, nva.data(),
 472                                           nva.size(), prdptr, strm.get());
 473   if (stream_id < 0) {
 474     ec = make_error_code(static_cast<nghttp2_error>(stream_id));
 475     return nullptr;
 476   }
 477
 478   signal_write();
 479
 480   strm->stream_id(stream_id);
 481
 482   auto p = streams_.emplace(stream_id, std::move(strm));
 483   assert(p.second);
 484   return &(*p.first).second->request();
 485 }
 486
 487 void session_impl::shutdown() {
 488   nghttp2_session_terminate_session(session_, NGHTTP2_NO_ERROR);
 489   signal_write();
 490 }
 491
 492 boost::asio::io_service &session_impl::io_service() { return io_service_; }
 493
 494 void session_impl::signal_write() {
 495   if (!inside_callback_) {
 496     do_write();
 497   }
 498 }
 499
 500 bool session_impl::should_stop() const {
 501   return !writing_ && !nghttp2_session_want_read(session_) &&
 502          !nghttp2_session_want_write(session_);
 503 }
 504
 505 namespace {
 506 struct callback_guard {
 507   callback_guard(session_impl &sess) : sess(sess) { sess.enter_callback(); }
 508   ~callback_guard() { sess.leave_callback(); }
 509
 510   session_impl &sess;
 511 };
 512 } // namespace
 513
 514 void session_impl::enter_callback() {
 515   assert(!inside_callback_);
 516   inside_callback_ = true;
 517 }
 518
 519 void session_impl::leave_callback() {
 520   assert(inside_callback_);
 521   inside_callback_ = false;
 522 }
 523
 524 void session_impl::do_read() {
 525   read_socket([this](const boost::system::error_code &ec,
 526                      std::size_t bytes_transferred) {
 527     if (ec) {
 528       if (ec.value() == boost::asio::error::operation_aborted) {
 529         call_error_cb(ec);
 530         shutdown_socket();
 531       }
 532       return;
 533     }
 534
 535     {
 536       callback_guard cg(*this);
 537
 538       auto rv =
 539           nghttp2_session_mem_recv(session_, rb_.data(), bytes_transferred);
 540
 541       if (rv != static_cast<ssize_t>(bytes_transferred)) {
 542         call_error_cb(make_error_code(
 543             static_cast<nghttp2_error>(rv < 0 ? rv : NGHTTP2_ERR_PROTO)));
 544         shutdown_socket();
 545         return;
 546       }
 547     }
 548
 549     do_write();
 550
 551     if (should_stop()) {
 552       shutdown_socket();
 553       return;
 554     }
 555
 556     do_read();
 557   });
 558 }
 559
 560 void session_impl::do_write() {
 561   if (writing_) {
 562     return;
 563   }
 564
 565   if (data_pending_) {
 566     std::copy_n(data_pending_, data_pendinglen_, std::begin(wb_) + wblen_);
 567
 568     wblen_ += data_pendinglen_;
 569
 570     data_pending_ = nullptr;
 571     data_pendinglen_ = 0;
 572   }
 573
 574   {
 575     callback_guard cg(*this);
 576
 577     for (;;) {
 578       const uint8_t *data;
 579       auto n = nghttp2_session_mem_send(session_, &data);
 580       if (n < 0) {
 581         call_error_cb(make_error_code(static_cast<nghttp2_error>(n)));
 582         shutdown_socket();
 583         return;
 584       }
 585
 586       if (n == 0) {
 587         break;
 588       }
 589
 590       if (wblen_ + n > wb_.size()) {
 591         data_pending_ = data;
 592         data_pendinglen_ = n;
 593
 594         break;
 595       }
 596
 597       std::copy_n(data, n, std::begin(wb_) + wblen_);
 598
 599       wblen_ += n;
 600     }
 601   }
 602
 603   if (wblen_ == 0) {
 604     return;
 605   }
 606
 607   writing_ = true;
 608
 609   write_socket([this](const boost::system::error_code &ec, std::size_t n) {
 610     if (ec) {
 611       call_error_cb(ec);
 612       shutdown_socket();
 613       return;
 614     }
 615
 616     wblen_ = 0;
 617     writing_ = false;
 618
 619     do_write();
 620   });
 621 }
 622
 623 } // namespace client
 624 } // namespace asio_http2
 625 } // nghttp2