Added thirdparty: boost library

thirdparty/include/boost/redis/detail/connection_base.hpp

@@ -0,0 +1,958 @@
+/* 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 ec = {})
+   {
+      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));
+         BOOST_ASSERT(ec == asio::error::operation_aborted);
+
+         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(ec, 0);
+         }
+
+         if (is_cancelled(self)) {
+            if (info_->is_written()) {
+               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(ec, 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(ec, 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::method method,
+      std::size_t max_read_size)
+   : ctx_{method}
+   , 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_;}
+
+   /// Returns the ssl context.
+   auto& get_ssl_context() 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_; }
+
+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&)>;
+
+   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_written()) {
+            return !ptr->req_->get_config().cancel_if_unresponded;
+         } else {
+            return !ptr->req_->get_config().cancel_on_connection_lost;
+         }
+      };
+
+      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->reset_status();
+      });
+
+      return ret;
+   }
+
+   auto cancel_unwritten_requests() -> std::size_t
+   {
+      auto f = [](auto const& ptr)
+      {
+         BOOST_ASSERT(ptr != nullptr);
+         return ptr->is_written();
+      };
+
+      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:
+         {
+            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&)>;
+
+      enum class action
+      {
+         stop,
+         proceed,
+         none,
+      };
+
+      explicit req_info(request const& req, adapter_type adapter, executor_type ex)
+      : timer_{ex}
+      , action_{action::none}
+      , req_{&req}
+      , adapter_{}
+      , expected_responses_{req.get_expected_responses()}
+      , status_{status::none}
+      , ec_{{}}
+      , read_size_{0}
+      {
+         timer_.expires_at((std::chrono::steady_clock::time_point::max)());
+
+         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()
+      {
+         timer_.cancel();
+         action_ = action::proceed;
+      }
+
+      void stop()
+      {
+         timer_.cancel();
+         action_ = action::stop;
+      }
+
+      [[nodiscard]] auto is_waiting_write() const noexcept
+         { return !is_written() && !is_staged(); }
+
+      [[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 reset_status() noexcept
+         { status_ = status::none; }
+
+      [[nodiscard]] auto stop_requested() const noexcept
+         { return action_ == action::stop;}
+
+      template <class CompletionToken>
+      auto async_wait(CompletionToken token)
+      {
+         return timer_.async_wait(std::move(token));
+      }
+
+   //private:
+      enum class status
+      { none
+      , staged
+      , written
+      };
+
+      timer_type timer_;
+      action action_;
+      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_write();
+         });
+
+         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_write();
+      });
+
+      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
+   {
+      return !std::empty(reqs_) && reqs_.front()->is_written();
+   }
+
+   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()
+   {
+      // We handle unsolicited events in the following way
+      //
+      // 1. Its resp3 type is a push.
+      //
+      // 2. A non-push type is received with an empty requests
+      //    queue. I have noticed this is possible (e.g. -MISCONF).
+      //    I expect them to have type push so we can distinguish
+      //    them from responses to commands, but it is a
+      //    simple-error. If we are lucky enough to receive them
+      //    when the command queue is empty we can treat them as
+      //    server pushes, otherwise it is impossible to handle
+      //    them properly
+      //
+      // 3. The request does not expect any response but we got
+      //    one. This may happen if for example, subscribe with
+      //    wrong syntax.
+      //
+      // Useful links:
+      //
+      // - https://github.com/redis/redis/issues/11784
+      // - https://github.com/redis/redis/issues/6426
+      //
+
+      BOOST_ASSERT(!read_buffer_.empty());
+
+      return
+         (resp3::to_type(read_buffer_.front()) == resp3::type::push)
+          || reqs_.empty()
+          || (!reqs_.empty() && reqs_.front()->expected_responses_ == 0)
+          || !is_waiting_response(); // Added to deal with MONITOR.
+   }
+
+   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;
+   }
+
+   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;
+
+   usage usage_;
+};
+
+} // boost::redis::detail
+
+#endif // BOOST_REDIS_CONNECTION_BASE_HPP