EIC code displayed by LXR master/​include/​boost/​redis/​detail/​connection_base.hpp	Source navigation Diff markup Identifier search General search
	Version: master test Revert   Change

File indexing completed on 2025-07-15 08:49:04

 /* Copyright (c) 2018-2023 Marcelo Zimbres Silva (mzimbres@gmail.com)
  *
  * Distributed under the Boost Software License, Version 1.0. (See
  * accompanying file LICENSE.txt)
  */
 
 #ifndef BOOST_REDIS_CONNECTION_BASE_HPP
 #define BOOST_REDIS_CONNECTION_BASE_HPP
 
 #include <boost/redis/adapter/adapt.hpp>
 #include <boost/redis/detail/helper.hpp>
 #include <boost/redis/error.hpp>
 #include <boost/redis/operation.hpp>
 #include <boost/redis/request.hpp>
 #include <boost/redis/resp3/type.hpp>
 #include <boost/redis/config.hpp>
 #include <boost/redis/detail/runner.hpp>
 #include <boost/redis/usage.hpp>
 
 #include <boost/system.hpp>
 #include <boost/asio/basic_stream_socket.hpp>
 #include <boost/asio/bind_executor.hpp>
 #include <boost/asio/experimental/parallel_group.hpp>
 #include <boost/asio/ip/tcp.hpp>
 #include <boost/asio/steady_timer.hpp>
 #include <boost/asio/write.hpp>
 #include <boost/assert.hpp>
 #include <boost/core/ignore_unused.hpp>
 #include <boost/asio/ssl/stream.hpp>
 #include <boost/asio/read_until.hpp>
 #include <boost/asio/buffer.hpp>
 #include <boost/asio/experimental/channel.hpp>
 
 #include <algorithm>
 #include <array>
 #include <chrono>
 #include <deque>
 #include <memory>
 #include <string_view>
 #include <type_traits>
 #include <functional>
 
 namespace boost::redis::detail
 {
 
 template <class DynamicBuffer>
 std::string_view buffer_view(DynamicBuffer buf) noexcept
 {
    char const* start = static_cast<char const*>(buf.data(0, buf.size()).data());
    return std::string_view{start, std::size(buf)};
 }
 
 template <class AsyncReadStream, class DynamicBuffer>
 class append_some_op {
 private:
    AsyncReadStream& stream_;
    DynamicBuffer buf_;
    std::size_t size_ = 0;
    std::size_t tmp_ = 0;
    asio::coroutine coro_{};
 
 public:
    append_some_op(AsyncReadStream& stream, DynamicBuffer buf, std::size_t size)
    : stream_ {stream}
    , buf_ {std::move(buf)}
    , size_{size}
    { }
 
    template <class Self>
    void operator()( Self& self
                   , system::error_code ec = {}
                   , std::size_t n = 0)
    {
       BOOST_ASIO_CORO_REENTER (coro_)
       {
          tmp_ = buf_.size();
          buf_.grow(size_);
 
          BOOST_ASIO_CORO_YIELD
          stream_.async_read_some(buf_.data(tmp_, size_), std::move(self));
          if (ec) {
             self.complete(ec, 0);
             return;
          }
 
          buf_.shrink(buf_.size() - tmp_ - n);
          self.complete({}, n);
       }
    }
 };
 
 template <class AsyncReadStream, class DynamicBuffer, class CompletionToken>
 auto
 async_append_some(
    AsyncReadStream& stream,
    DynamicBuffer buffer,
    std::size_t size,
    CompletionToken&& token)
 {
    return asio::async_compose
       < CompletionToken
       , void(system::error_code, std::size_t)
       >(append_some_op<AsyncReadStream, DynamicBuffer> {stream, buffer, size}, token, stream);
 }
 
 template <class Conn>
 struct exec_op {
    using req_info_type = typename Conn::req_info;
    using adapter_type = typename Conn::adapter_type;
 
    Conn* conn_ = nullptr;
    std::shared_ptr<req_info_type> info_ = nullptr;
    asio::coroutine coro{};
 
    template <class Self>
    void operator()(Self& self , system::error_code = {}, std::size_t = 0)
    {
       BOOST_ASIO_CORO_REENTER (coro)
       {
          // Check whether the user wants to wait for the connection to
          // be stablished.
          if (info_->req_->get_config().cancel_if_not_connected && !conn_->is_open()) {
             BOOST_ASIO_CORO_YIELD
             asio::post(std::move(self));
             return self.complete(error::not_connected, 0);
          }
 
          conn_->add_request_info(info_);
 
 EXEC_OP_WAIT:
          BOOST_ASIO_CORO_YIELD
          info_->async_wait(std::move(self));
 
          if (info_->ec_) {
             self.complete(info_->ec_, 0);
             return;
          }
 
          if (info_->stop_requested()) {
             // Don't have to call remove_request as it has already
             // been by cancel(exec).
             return self.complete(asio::error::operation_aborted, 0);
          }
 
          if (is_cancelled(self)) {
             if (!info_->is_waiting()) {
                using c_t = asio::cancellation_type;
                auto const c = self.get_cancellation_state().cancelled();
                if ((c & c_t::terminal) != c_t::none) {
                   // Cancellation requires closing the connection
                   // otherwise it stays in inconsistent state.
                   conn_->cancel(operation::run);
                   return self.complete(asio::error::operation_aborted, 0);
                } else {
                   // Can't implement other cancelation types, ignoring.
                   self.get_cancellation_state().clear();
 
                   // TODO: Find out a better way to ignore
                   // cancelation.
                   goto EXEC_OP_WAIT;
                }
             } else {
                // Cancelation can be honored.
                conn_->remove_request(info_);
                self.complete(asio::error::operation_aborted, 0);
                return;
             }
          }
 
          self.complete(info_->ec_, info_->read_size_);
       }
    }
 };
 
 template <class Conn, class Logger>
 struct run_op {
    Conn* conn = nullptr;
    Logger logger_;
    asio::coroutine coro{};
 
    template <class Self>
    void operator()( Self& self
                   , std::array<std::size_t, 2> order = {}
                   , system::error_code ec0 = {}
                   , system::error_code ec1 = {})
    {
       BOOST_ASIO_CORO_REENTER (coro)
       {
          conn->reset();
 
          BOOST_ASIO_CORO_YIELD
          asio::experimental::make_parallel_group(
             [this](auto token) { return conn->reader(logger_, token);},
             [this](auto token) { return conn->writer(logger_, token);}
          ).async_wait(
             asio::experimental::wait_for_one(),
             std::move(self));
 
          if (is_cancelled(self)) {
             logger_.trace("run-op: canceled. Exiting ...");
             self.complete(asio::error::operation_aborted);
             return;
          }
 
          logger_.on_run(ec0, ec1);
 
          switch (order[0]) {
            case 0: self.complete(ec0); break;
            case 1: self.complete(ec1); break;
            default: BOOST_ASSERT(false);
          }
       }
    }
 };
 
 template <class Conn, class Logger>
 struct writer_op {
    Conn* conn_;
    Logger logger_;
    asio::coroutine coro{};
 
    template <class Self>
    void operator()( Self& self
                   , system::error_code ec = {}
                   , std::size_t n = 0)
    {
       ignore_unused(n);
 
       BOOST_ASIO_CORO_REENTER (coro) for (;;)
       {
          while (conn_->coalesce_requests()) {
             if (conn_->use_ssl())
                BOOST_ASIO_CORO_YIELD asio::async_write(conn_->next_layer(), asio::buffer(conn_->write_buffer_), std::move(self));
             else
                BOOST_ASIO_CORO_YIELD asio::async_write(conn_->next_layer().next_layer(), asio::buffer(conn_->write_buffer_), std::move(self));
 
             logger_.on_write(ec, conn_->write_buffer_);
 
             if (ec) {
                logger_.trace("writer-op: error. Exiting ...");
                conn_->cancel(operation::run);
                self.complete(ec);
                return;
             }
 
             if (is_cancelled(self)) {
                logger_.trace("writer-op: canceled. Exiting ...");
                self.complete(asio::error::operation_aborted);
                return;
             }
 
             conn_->on_write();
 
             // A socket.close() may have been called while a
             // successful write might had already been queued, so we
             // have to check here before proceeding.
             if (!conn_->is_open()) {
                logger_.trace("writer-op: canceled (2). Exiting ...");
                self.complete({});
                return;
             }
          }
 
          BOOST_ASIO_CORO_YIELD
          conn_->writer_timer_.async_wait(std::move(self));
          if (!conn_->is_open() || is_cancelled(self)) {
             logger_.trace("writer-op: canceled (3). Exiting ...");
             // Notice this is not an error of the op, stoping was
             // requested from the outside, so we complete with
             // success.
             self.complete({});
             return;
          }
       }
    }
 };
 
 template <class Conn, class Logger>
 struct reader_op {
    using parse_result = typename Conn::parse_result;
    using parse_ret_type = typename Conn::parse_ret_type;
    Conn* conn_;
    Logger logger_;
    parse_ret_type res_{parse_result::resp, 0};
    asio::coroutine coro{};
 
    template <class Self>
    void operator()( Self& self
                   , system::error_code ec = {}
                   , std::size_t n = 0)
    {
       ignore_unused(n);
 
       BOOST_ASIO_CORO_REENTER (coro) for (;;)
       {
          // Appends some data to the buffer if necessary.
          if ((res_.first == parse_result::needs_more) || std::empty(conn_->read_buffer_)) {
             if (conn_->use_ssl()) {
                BOOST_ASIO_CORO_YIELD
                async_append_some(
                   conn_->next_layer(),
                   conn_->dbuf_,
                   conn_->get_suggested_buffer_growth(),
                   std::move(self));
             } else {
                BOOST_ASIO_CORO_YIELD
                async_append_some(
                   conn_->next_layer().next_layer(),
                   conn_->dbuf_,
                   conn_->get_suggested_buffer_growth(),
                   std::move(self));
             }
 
             logger_.on_read(ec, n);
 
             // EOF is not treated as error.
             if (ec == asio::error::eof) {
                logger_.trace("reader-op: EOF received. Exiting ...");
                conn_->cancel(operation::run);
                return self.complete({}); // EOFINAE: EOF is not an error.
             }
 
             // The connection is not viable after an error.
             if (ec) {
                logger_.trace("reader-op: error. Exiting ...");
                conn_->cancel(operation::run);
                self.complete(ec);
                return;
             }
 
             // Somebody might have canceled implicitly or explicitly
             // while we were suspended and after queueing so we have to
             // check.
             if (!conn_->is_open() || is_cancelled(self)) {
                logger_.trace("reader-op: canceled. Exiting ...");
                self.complete(ec);
                return;
             }
          }
 
          res_ = conn_->on_read(buffer_view(conn_->dbuf_), ec);
          if (ec) {
             logger_.trace("reader-op: parse error. Exiting ...");
             conn_->cancel(operation::run);
             self.complete(ec);
             return;
          }
 
          if (res_.first == parse_result::push) {
             if (!conn_->receive_channel_.try_send(ec, res_.second)) {
                BOOST_ASIO_CORO_YIELD
                conn_->receive_channel_.async_send(ec, res_.second, std::move(self));
             }
 
             if (ec) {
                logger_.trace("reader-op: error. Exiting ...");
                conn_->cancel(operation::run);
                self.complete(ec);
                return;
             }
 
             if (!conn_->is_open() || is_cancelled(self)) {
                logger_.trace("reader-op: canceled (2). Exiting ...");
                self.complete(asio::error::operation_aborted);
                return;
             }
 
          }
       }
    }
 };
 
 /** @brief Base class for high level Redis asynchronous connections.
  *  @ingroup high-level-api
  *
  *  @tparam Executor The executor type.
  *
  */
 template <class Executor>
 class connection_base {
 public:
    /// Executor type
    using executor_type = Executor;
 
    /// Type of the next layer
    using next_layer_type = asio::ssl::stream<asio::basic_stream_socket<asio::ip::tcp, Executor>>;
 
    using clock_type = std::chrono::steady_clock;
    using clock_traits_type = asio::wait_traits<clock_type>;
    using timer_type = asio::basic_waitable_timer<clock_type, clock_traits_type, executor_type>;
 
    using this_type = connection_base<Executor>;
 
    /// Constructs from an executor.
    connection_base(
       executor_type ex,
       asio::ssl::context ctx,
       std::size_t max_read_size)
    : ctx_{std::move(ctx)}
    , stream_{std::make_unique<next_layer_type>(ex, ctx_)}
    , writer_timer_{ex}
    , receive_channel_{ex, 256}
    , runner_{ex, {}}
    , dbuf_{read_buffer_, max_read_size}
    {
       set_receive_response(ignore);
       writer_timer_.expires_at((std::chrono::steady_clock::time_point::max)());
    }
 
    /// Returns the ssl context.
    auto const& get_ssl_context() const noexcept
       { return ctx_;}
 
    /// Resets the underlying stream.
    void reset_stream()
    {
       stream_ = std::make_unique<next_layer_type>(writer_timer_.get_executor(), ctx_);
    }
 
    /// Returns a reference to the next layer.
    auto& next_layer() noexcept { return *stream_; }
 
    /// Returns a const reference to the next layer.
    auto const& next_layer() const noexcept { return *stream_; }
 
    /// Returns the associated executor.
    auto get_executor() {return writer_timer_.get_executor();}
 
    /// Cancels specific operations.
    void cancel(operation op)
    {
       runner_.cancel(op);
       if (op == operation::all) {
          cancel_impl(operation::run);
          cancel_impl(operation::receive);
          cancel_impl(operation::exec);
          return;
       } 
 
       cancel_impl(op);
    }
 
    template <class Response, class CompletionToken>
    auto async_exec(request const& req, Response& resp, CompletionToken token)
    {
       using namespace boost::redis::adapter;
       auto f = boost_redis_adapt(resp);
       BOOST_ASSERT_MSG(req.get_expected_responses() <= f.get_supported_response_size(), "Request and response have incompatible sizes.");
 
       auto info = std::make_shared<req_info>(req, f, get_executor());
 
       return asio::async_compose
          < CompletionToken
          , void(system::error_code, std::size_t)
          >(exec_op<this_type>{this, info}, token, writer_timer_);
    }
 
    template <class Response, class CompletionToken>
    [[deprecated("Set the response with set_receive_response and use the other overload.")]]
    auto async_receive(Response& response, CompletionToken token)
    {
       set_receive_response(response);
       return receive_channel_.async_receive(std::move(token));
    }
 
    template <class CompletionToken>
    auto async_receive(CompletionToken token)
       { return receive_channel_.async_receive(std::move(token)); }
 
    std::size_t receive(system::error_code& ec)
    {
       std::size_t size = 0;
 
       auto f = [&](system::error_code const& ec2, std::size_t n)
       {
          ec = ec2;
          size = n;
       };
 
       auto const res = receive_channel_.try_receive(f);
       if (ec)
          return 0;
 
       if (!res)
          ec = error::sync_receive_push_failed;
 
       return size;
    }
 
    template <class Logger, class CompletionToken>
    auto async_run(config const& cfg, Logger l, CompletionToken token)
    {
       runner_.set_config(cfg);
       l.set_prefix(runner_.get_config().log_prefix);
       return runner_.async_run(*this, l, std::move(token));
    }
 
    template <class Response>
    void set_receive_response(Response& response)
    {
       using namespace boost::redis::adapter;
       auto g = boost_redis_adapt(response);
       receive_adapter_ = adapter::detail::make_adapter_wrapper(g);
    }
 
    usage get_usage() const noexcept
       { return usage_; }
 
    auto run_is_canceled() const noexcept
       { return cancel_run_called_; }
 
 private:
    using receive_channel_type = asio::experimental::channel<executor_type, void(system::error_code, std::size_t)>;
    using runner_type = runner<executor_type>;
    using adapter_type = std::function<void(std::size_t, resp3::basic_node<std::string_view> const&, system::error_code&)>;
    using receiver_adapter_type = std::function<void(resp3::basic_node<std::string_view> const&, system::error_code&)>;
    using exec_notifier_type = receive_channel_type;
 
    auto use_ssl() const noexcept
       { return runner_.get_config().use_ssl;}
 
    auto cancel_on_conn_lost() -> std::size_t
    {
       // Must return false if the request should be removed.
       auto cond = [](auto const& ptr)
       {
          BOOST_ASSERT(ptr != nullptr);
 
          if (ptr->is_waiting()) {
             return !ptr->req_->get_config().cancel_on_connection_lost;
          } else {
             return !ptr->req_->get_config().cancel_if_unresponded;
          }
       };
 
       auto point = std::stable_partition(std::begin(reqs_), std::end(reqs_), cond);
 
       auto const ret = std::distance(point, std::end(reqs_));
 
       std::for_each(point, std::end(reqs_), [](auto const& ptr) {
          ptr->stop();
       });
 
       reqs_.erase(point, std::end(reqs_));
 
       std::for_each(std::begin(reqs_), std::end(reqs_), [](auto const& ptr) {
          return ptr->mark_waiting();
       });
 
       return ret;
    }
 
    auto cancel_unwritten_requests() -> std::size_t
    {
       auto f = [](auto const& ptr)
       {
          BOOST_ASSERT(ptr != nullptr);
          return !ptr->is_waiting();
       };
 
       auto point = std::stable_partition(std::begin(reqs_), std::end(reqs_), f);
 
       auto const ret = std::distance(point, std::end(reqs_));
 
       std::for_each(point, std::end(reqs_), [](auto const& ptr) {
          ptr->stop();
       });
 
       reqs_.erase(point, std::end(reqs_));
       return ret;
    }
 
    void cancel_impl(operation op)
    {
       switch (op) {
          case operation::exec:
          {
             cancel_unwritten_requests();
          } break;
          case operation::run:
          {
             // Protects the code below from being called more than
             // once, see https://github.com/boostorg/redis/issues/181
             if (std::exchange(cancel_run_called_, true)) {
                return;
             }
 
             close();
             writer_timer_.cancel();
             receive_channel_.cancel();
             cancel_on_conn_lost();
          } break;
          case operation::receive:
          {
             receive_channel_.cancel();
          } break;
          default: /* ignore */;
       }
    }
 
    void on_write()
    {
       // We have to clear the payload right after writing it to use it
       // as a flag that informs there is no ongoing write.
       write_buffer_.clear();
 
       // Notice this must come before the for-each below.
       cancel_push_requests();
 
       // There is small optimization possible here: traverse only the
       // partition of unwritten requests instead of them all.
       std::for_each(std::begin(reqs_), std::end(reqs_), [](auto const& ptr) {
          BOOST_ASSERT_MSG(ptr != nullptr, "Expects non-null pointer.");
          if (ptr->is_staged()) {
             ptr->mark_written();
          }
       });
    }
 
    struct req_info {
    public:
       using node_type = resp3::basic_node<std::string_view>;
       using wrapped_adapter_type = std::function<void(node_type const&, system::error_code&)>;
 
       explicit req_info(request const& req, adapter_type adapter, executor_type ex)
       : notifier_{ex, 1}
       , req_{&req}
       , adapter_{}
       , expected_responses_{req.get_expected_responses()}
       , status_{status::waiting}
       , ec_{{}}
       , read_size_{0}
       {
          adapter_ = [this, adapter](node_type const& nd, system::error_code& ec)
          {
             auto const i = req_->get_expected_responses() - expected_responses_;
             adapter(i, nd, ec);
          };
       }
 
       auto proceed()
       {
          notifier_.try_send(std::error_code{}, 0);
       }
 
       void stop()
       {
          notifier_.close();
       }
 
       [[nodiscard]] auto is_waiting() const noexcept
          { return status_ == status::waiting; }
 
       [[nodiscard]] auto is_written() const noexcept
          { return status_ == status::written; }
 
       [[nodiscard]] auto is_staged() const noexcept
          { return status_ == status::staged; }
 
       void mark_written() noexcept
          { status_ = status::written; }
 
       void mark_staged() noexcept
          { status_ = status::staged; }
 
       void mark_waiting() noexcept
          { status_ = status::waiting; }
 
       [[nodiscard]] auto stop_requested() const noexcept
          { return !notifier_.is_open();}
 
       template <class CompletionToken>
       auto async_wait(CompletionToken token)
       {
          return notifier_.async_receive(std::move(token));
       }
 
    //private:
       enum class status
       { waiting
       , staged
       , written
       };
 
       exec_notifier_type notifier_;
       request const* req_;
       wrapped_adapter_type adapter_;
 
       // Contains the number of commands that haven't been read yet.
       std::size_t expected_responses_;
       status status_;
 
       system::error_code ec_;
       std::size_t read_size_;
    };
 
    void remove_request(std::shared_ptr<req_info> const& info)
    {
       reqs_.erase(std::remove(std::begin(reqs_), std::end(reqs_), info));
    }
 
    using reqs_type = std::deque<std::shared_ptr<req_info>>;
 
    template <class, class> friend struct reader_op;
    template <class, class> friend struct writer_op;
    template <class, class> friend struct run_op;
    template <class> friend struct exec_op;
    template <class, class, class> friend struct run_all_op;
 
    void cancel_push_requests()
    {
       auto point = std::stable_partition(std::begin(reqs_), std::end(reqs_), [](auto const& ptr) {
             return !(ptr->is_staged() && ptr->req_->get_expected_responses() == 0);
       });
 
       std::for_each(point, std::end(reqs_), [](auto const& ptr) {
          ptr->proceed();
       });
 
       reqs_.erase(point, std::end(reqs_));
    }
 
    [[nodiscard]] bool is_writing() const noexcept
    {
       return !write_buffer_.empty();
    }
 
    void add_request_info(std::shared_ptr<req_info> const& info)
    {
       reqs_.push_back(info);
 
       if (info->req_->has_hello_priority()) {
          auto rend = std::partition_point(std::rbegin(reqs_), std::rend(reqs_), [](auto const& e) {
                return e->is_waiting();
          });
 
          std::rotate(std::rbegin(reqs_), std::rbegin(reqs_) + 1, rend);
       }
 
       if (is_open() && !is_writing())
          writer_timer_.cancel();
    }
 
    template <class CompletionToken, class Logger>
    auto reader(Logger l, CompletionToken&& token)
    {
       return asio::async_compose
          < CompletionToken
          , void(system::error_code)
          >(reader_op<this_type, Logger>{this, l}, token, writer_timer_);
    }
 
    template <class CompletionToken, class Logger>
    auto writer(Logger l, CompletionToken&& token)
    {
       return asio::async_compose
          < CompletionToken
          , void(system::error_code)
          >(writer_op<this_type, Logger>{this, l}, token, writer_timer_);
    }
 
    template <class Logger, class CompletionToken>
    auto async_run_lean(config const& cfg, Logger l, CompletionToken token)
    {
       runner_.set_config(cfg);
       l.set_prefix(runner_.get_config().log_prefix);
       return asio::async_compose
          < CompletionToken
          , void(system::error_code)
          >(run_op<this_type, Logger>{this, l}, token, writer_timer_);
    }
 
    [[nodiscard]] bool coalesce_requests()
    {
       // Coalesces the requests and marks them staged. After a
       // successful write staged requests will be marked as written.
       auto const point = std::partition_point(std::cbegin(reqs_), std::cend(reqs_), [](auto const& ri) {
             return !ri->is_waiting();
       });
 
       std::for_each(point, std::cend(reqs_), [this](auto const& ri) {
          // Stage the request.
          write_buffer_ += ri->req_->payload();
          ri->mark_staged();
          usage_.commands_sent += ri->expected_responses_;
       });
 
       usage_.bytes_sent += std::size(write_buffer_);
 
       return point != std::cend(reqs_);
    }
 
    bool is_waiting_response() const noexcept
    {
       if (std::empty(reqs_))
          return false;
 
       // Under load and on low-latency networks we might start
       // receiving responses before the write operation completed and
       // the request is still maked as staged and not written.  See
       // https://github.com/boostorg/redis/issues/170
       return !reqs_.front()->is_waiting();
    }
 
    void close()
    {
       if (stream_->next_layer().is_open()) {
          system::error_code ec;
          stream_->next_layer().close(ec);
       }
    }
 
    auto is_open() const noexcept { return stream_->next_layer().is_open(); }
    auto& lowest_layer() noexcept { return stream_->lowest_layer(); }
 
    auto is_next_push()
    {
       BOOST_ASSERT(!read_buffer_.empty());
 
       // Useful links to understand the heuristics below.
       //
       // - https://github.com/redis/redis/issues/11784
       // - https://github.com/redis/redis/issues/6426
       // - https://github.com/boostorg/redis/issues/170
 
       // The message's resp3 type is a push.
       if (resp3::to_type(read_buffer_.front()) == resp3::type::push)
          return true;
 
       // This is non-push type and the requests queue is empty. I have
       // noticed this is possible, for example with -MISCONF. I don't
       // know why they are not sent with a push type so we can
       // distinguish them from responses to commands. If we are lucky
       // enough to receive them when the command queue is empty they
       // can be treated as server pushes, otherwise it is impossible
       // to handle them properly
       if (reqs_.empty())
          return true;
 
       // The request does not expect any response but we got one. This
       // may happen if for example, subscribe with wrong syntax.
       if (reqs_.front()->expected_responses_ == 0)
          return true;
 
       // Added to deal with MONITOR and also to fix PR170 which
       // happens under load and on low-latency networks, where we
       // might start receiving responses before the write operation
       // completed and the request is still maked as staged and not
       // written.
       return reqs_.front()->is_waiting();
    }
 
    auto get_suggested_buffer_growth() const noexcept
    {
       return parser_.get_suggested_buffer_growth(4096);
    }
 
    enum class parse_result { needs_more, push, resp };
 
    using parse_ret_type = std::pair<parse_result, std::size_t>;
 
    parse_ret_type on_finish_parsing(parse_result t)
    {
       if (t == parse_result::push) {
          usage_.pushes_received += 1;
          usage_.push_bytes_received += parser_.get_consumed();
       } else {
          usage_.responses_received += 1;
          usage_.response_bytes_received += parser_.get_consumed();
       }
 
       on_push_ = false;
       dbuf_.consume(parser_.get_consumed());
       auto const res = std::make_pair(t, parser_.get_consumed());
       parser_.reset();
       return res;
    }
 
    parse_ret_type on_read(std::string_view data, system::error_code& ec)
    {
       // We arrive here in two states:
       //
       //    1. While we are parsing a message. In this case we
       //       don't want to determine the type of the message in the
       //       buffer (i.e. response vs push) but leave it untouched
       //       until the parsing of a complete message ends.
       //
       //    2. On a new message, in which case we have to determine
       //       whether the next messag is a push or a response.
       //
       if (!on_push_) // Prepare for new message.
          on_push_ = is_next_push();
 
       if (on_push_) {
          if (!resp3::parse(parser_, data, receive_adapter_, ec))
             return std::make_pair(parse_result::needs_more, 0);
 
          if (ec)
             return std::make_pair(parse_result::push, 0);
 
          return on_finish_parsing(parse_result::push);
       }
 
       BOOST_ASSERT_MSG(is_waiting_response(), "Not waiting for a response (using MONITOR command perhaps?)");
       BOOST_ASSERT(!reqs_.empty());
       BOOST_ASSERT(reqs_.front() != nullptr);
       BOOST_ASSERT(reqs_.front()->expected_responses_ != 0);
 
       if (!resp3::parse(parser_, data, reqs_.front()->adapter_, ec))
          return std::make_pair(parse_result::needs_more, 0);
 
       if (ec) {
          reqs_.front()->ec_ = ec;
          reqs_.front()->proceed();
          return std::make_pair(parse_result::resp, 0);
       }
 
       reqs_.front()->read_size_ += parser_.get_consumed();
 
       if (--reqs_.front()->expected_responses_ == 0) {
          // Done with this request.
          reqs_.front()->proceed();
          reqs_.pop_front();
       }
 
       return on_finish_parsing(parse_result::resp);
    }
 
    void reset()
    {
       write_buffer_.clear();
       read_buffer_.clear();
       parser_.reset();
       on_push_ = false;
       cancel_run_called_ = false;
    }
 
    asio::ssl::context ctx_;
    std::unique_ptr<next_layer_type> stream_;
 
    // Notice we use a timer to simulate a condition-variable. It is
    // also more suitable than a channel and the notify operation does
    // not suspend.
    timer_type writer_timer_;
    receive_channel_type receive_channel_;
    runner_type runner_;
    receiver_adapter_type receive_adapter_;
 
    using dyn_buffer_type = asio::dynamic_string_buffer<char, std::char_traits<char>, std::allocator<char>>;
 
    std::string read_buffer_;
    dyn_buffer_type dbuf_;
    std::string write_buffer_;
    reqs_type reqs_;
    resp3::parser parser_{};
    bool on_push_ = false;
    bool cancel_run_called_ = false;
 
    usage usage_;
 };
 
 } // boost::redis::detail
 
 #endif // BOOST_REDIS_CONNECTION_BASE_HPP

This page was automatically generated by the 2.3.7 LXR engine. The LXR team