Line data    Source code

       1              : #include "http_stream.hpp"
       2              : #include "request.hpp"
       3              : #include "../common/http_request.hpp"
       4              : #include "../../asio/io_context_of.hpp"
       5              : #include "../../util/logger.hpp"
       6              : #include <boost/asio/dispatch.hpp>
       7              : #include <boost/asio/post.hpp>
       8              : #include <string_view>
       9              : 
      10              : namespace thinger::http {
      11              : 
      12          150 :     void detail::continue_frame::to_buffer(std::vector<boost::asio::const_buffer>& buffer) const {
      13              :         static constexpr std::string_view head = "HTTP/1.1 100 Continue\r\n\r\n";
      14          150 :         buffer.emplace_back(boost::asio::buffer(head.data(), head.size()));
      15          150 :     }
      16              : 
      17          150 :     void detail::continue_frame::log(const char* scope, const body_logging&) const {
      18          150 :         LOG_DEBUG("[{}] HTTP/1.1 100 Continue", scope);
      19          150 :     }
      20              : 
      21        27679 :     http_stream::http_stream(stream_id stream_id, const http_request& request, boost::asio::any_io_executor executor)
      22        27679 :         : stream_id_(stream_id), executor_(std::move(executor)), cancelled_(cancel_reason::none) {
      23              :         // A client that does not keep the connection: by its Connection header, or by default
      24              :         // before HTTP/1.1
      25        27679 :         if (!request.keep_alive()) {
      26         7334 :             bool http11 = request.get_http_version_major() > 1 ||
      27         3667 :                           (request.get_http_version_major() == 1 && request.get_http_version_minor() >= 1);
      28         3667 :             must_close(http11 ? detail::close_reason::client_asked : detail::close_reason::http10);
      29              :         }
      30        27679 :     }
      31              : 
      32          387 :     http_stream::http_stream(stream_id stream_id, detail::close_reason reason, boost::asio::any_io_executor executor)
      33          387 :         : stream_id_(stream_id), close_reason_(reason), executor_(std::move(executor)),
      34          774 :           cancelled_(cancel_reason::none) {}
      35              : 
      36         3667 :     void http_stream::cancel(cancel_reason reason) {
      37         4259 :         if (reason == cancel_reason::none) return;
      38              :         // The first reason wins, whatever thread gives it
      39         3664 :         auto none = cancel_reason::none;
      40         3664 :         if (!cancelled_.compare_exchange_strong(none, reason, std::memory_order_acq_rel)) {
      41              :             // The scope stays as it is, but a shutdown still interrupts a handler that the first
      42              :             // reason let go on (once: what it does about an interruption is up to it)
      43          592 :             if (reason != cancel_reason::shutdown) return;
      44          488 :             boost::asio::dispatch(executor_, [self = shared_from_this()] {
      45          488 :                 if (!self->handler_running_ || self->handler_interrupted_) return;
      46          101 :                 self->observer_signal_.emit(boost::asio::cancellation_type::terminal);
      47          101 :                 self->interrupt_handler();
      48              :             });
      49          488 :             return;
      50              :         }
      51         3072 :         boost::asio::dispatch(executor_, [self = shared_from_this(), reason] {
      52         3069 :             if (self->cancel_event_) self->cancel_event_->set();
      53         3069 :             if (!self->handler_running_) return;
      54         1856 :             self->observer_signal_.emit(boost::asio::cancellation_type::terminal);
      55              :             // A drain deadline interrupts every handler; another reason, those declaring it
      56         1856 :             if (reason == cancel_reason::shutdown || (self->cancel_on_ & (1u << static_cast<unsigned>(reason)))) {
      57         1360 :                 self->interrupt_handler();
      58              :             }
      59              :         });
      60              :     }
      61              : 
      62         1461 :     void http_stream::interrupt_handler() {
      63         1461 :         handler_interrupted_ = true;
      64         1461 :         auto& signal = bound_signal_ ? *bound_signal_ : handler_signal_;
      65         1461 :         signal.emit(boost::asio::cancellation_type::terminal);
      66         1461 :     }
      67              : 
      68         2040 :     void http_stream::end_input(bool follows) {
      69         2040 :         if (asio::running_in_this_thread(executor_)) [[likely]] {
      70         1986 :             set_input_pending(false, follows);
      71         1986 :             return;
      72              :         }
      73          108 :         boost::asio::post(executor_, [self = shared_from_this(), follows] { self->set_input_pending(false, follows); });
      74              :     }
      75              : 
      76          678 :     awaitable<cancel_reason> http_stream::wait_cancelled() {
      77              :         // Created cancelled if the reason is there: cancel() may have come from another thread,
      78              :         // with its dispatched half (which sets the event, if any) still to run here
      79              :         if (!cancel_event_) {
      80              :             cancel_event_.emplace(executor_);
      81              :             if (cancelled() != cancel_reason::none) cancel_event_->set();
      82              :         }
      83              :         if (!co_await cancel_event_->wait()) co_return cancel_reason::none;
      84              :         co_return cancelled();
      85         1356 :     }
      86              : 
      87            0 :     size_t http_stream::get_queue_size() const {
      88            0 :         return queue_.size();
      89              :     }
      90              : 
      91        64594 :     bool http_stream::empty_queue() const {
      92        64594 :         return queue_.empty();
      93              :     }
      94              : 
      95        39117 :     std::shared_ptr<http_frame> http_stream::current_frame() const {
      96        39117 :         return queue_.front();
      97              :     }
      98              : 
      99        39120 :     void http_stream::pop_frame() {
     100        39120 :         queue_.pop();
     101        39120 :     }
     102              : 
     103        39105 :     void http_stream::add_frame(std::shared_ptr<http_frame> frame) {
     104        39105 :         queue_.push(std::move(frame));
     105        39105 :         responded_ = true;
     106        39105 :     }
     107              : 
     108          150 :     void http_stream::add_interim_frame(std::shared_ptr<http_frame> frame) {
     109          150 :         queue_.push(std::move(frame));
     110          150 :     }
     111              : 
     112            0 :     size_t http_stream::get_queued_frames() const {
     113            0 :         return queue_.size();
     114              :     }
     115              : 
     116        78204 :     stream_id http_stream::id() const {
     117        78204 :         return stream_id_;
     118              :     }
     119              : }