Concludes the work started by Nikolai Vladimirov on the generic_flat_response

Author: mzimbres

README.md

@@ -106,24 +106,23 @@ receiver(std::shared_ptr<connection> conn) -> net::awaitable<void>
    while (conn->will_reconnect()) {
 
       // Reconnect to channels.
-      co_await conn->async_exec(req, ignore);
+      co_await conn->async_exec(req);
 
       // Loop reading Redis pushes.
-      for (;;) {
-         error_code ec;
-         co_await conn->async_receive(resp, net::redirect_error(net::use_awaitable, ec));
+      for (error_code ec;;) {
+         co_await conn->async_receive2(resp, redirect_error(ec));
          if (ec)
             break; // Connection lost, break so we can reconnect to channels.
 
          // Use the response resp in some way and then clear it.
          ...
 
-         consume_one(resp);
+         resp.value().clear();
       }
    }
 }
 ```
 
 ## Further reading
 
-Full documentation is [here](https://www.boost.org/doc/libs/master/libs/redis/index.html).
+Full documentation is [here](https://www.boost.org/doc/libs/master/libs/redis/index.html).

doc/modules/ROOT/pages/index.adoc

@@ -117,19 +117,18 @@ receiver(std::shared_ptr<connection> conn) -> net::awaitable<void>
    while (conn->will_reconnect()) {
 
       // Reconnect to channels.
-      co_await conn->async_exec(req, ignore);
+      co_await conn->async_exec(req);
 
       // Loop reading Redis pushes.
-      for (;;) {
-         error_code ec;
-         co_await conn->async_receive(resp, net::redirect_error(net::use_awaitable, ec));
+      for (error_code ec;;) {
+         co_await conn->async_receive2(resp, redirect_error(ec));
          if (ec)
             break; // Connection lost, break so we can reconnect to channels.
 
-         // Use the response resp in some way and then clear it.
+         // Use the response here and then clear it.
          ...
 
-         consume_one(resp);
+         resp.value().clear();
       }
    }
 }

example/cpp20_chat_room.cpp

@@ -30,11 +30,9 @@ using boost::asio::consign;
 using boost::asio::detached;
 using boost::asio::dynamic_buffer;
 using boost::asio::redirect_error;
-using boost::asio::use_awaitable;
 using boost::redis::config;
 using boost::redis::connection;
-using boost::redis::generic_flat_response;
-using boost::redis::ignore;
+using boost::redis::generic_response;
 using boost::redis::request;
 using boost::system::error_code;
 using namespace std::chrono_literals;
@@ -47,16 +45,16 @@ auto receiver(std::shared_ptr<connection> conn) -> awaitable<void>
    request req;
    req.push("SUBSCRIBE", "channel");
 
-   generic_flat_response resp;
+   generic_response resp;
    conn->set_receive_response(resp);
 
    while (conn->will_reconnect()) {
       // Subscribe to channels.
-      co_await conn->async_exec(req, ignore);
+      co_await conn->async_exec(req);
 
       // Loop reading Redis push messages.
       for (error_code ec;;) {
-         co_await conn->async_receive(redirect_error(use_awaitable, ec));
+         co_await conn->async_receive2(redirect_error(ec));
          if (ec)
             break;  // Connection lost, break so we can reconnect to channels.
          std::cout << resp.value().at(1).value << " " << resp.value().at(2).value << " "
@@ -74,7 +72,7 @@ auto publisher(std::shared_ptr<stream_descriptor> in, std::shared_ptr<connection
       auto n = co_await async_read_until(*in, dynamic_buffer(msg, 1024), "\n");
       request req;
       req.push("PUBLISH", "channel", msg);
-      co_await conn->async_exec(req, ignore);
+      co_await conn->async_exec(req);
       msg.erase(0, n);
    }
 }

example/cpp20_streams.cpp

@@ -23,7 +23,7 @@
 
 namespace net = boost::asio;
 using boost::redis::config;
-using boost::redis::generic_flat_response;
+using boost::redis::generic_response;
 using boost::redis::operation;
 using boost::redis::request;
 using boost::redis::connection;
@@ -33,7 +33,7 @@ auto stream_reader(std::shared_ptr<connection> conn) -> net::awaitable<void>
 {
    std::string redisStreamKey_;
    request req;
-   generic_flat_response resp;
+   generic_response resp;
 
    std::string stream_id{"$"};
    std::string const field = "myfield";
@@ -51,7 +51,7 @@ auto stream_reader(std::shared_ptr<connection> conn) -> net::awaitable<void>
       // The following approach was taken in order to be able to
       // deal with the responses, as generated by redis in the case
       // that there are multiple stream 'records' within a single
-      // generic_flat_response.  The nesting and number of values in
+      // generic_response.  The nesting and number of values in
       // resp.value() are different, depending on the contents
       // of the stream in redis.  Uncomment the above commented-out
       // code for examples while running the XADD command.

example/cpp20_subscriber.cpp

@@ -1,11 +1,10 @@
-/* Copyright (c) 2018-2022 Marcelo Zimbres Silva (mzimbres@gmail.com)
+/* Copyright (c) 2018-2025 Marcelo Zimbres Silva (mzimbres@gmail.com)
  *
  * Distributed under the Boost Software License, Version 1.0. (See
  * accompanying file LICENSE.txt)
  */
 
 #include <boost/redis/connection.hpp>
-#include <boost/redis/logger.hpp>
 
 #include <boost/asio/awaitable.hpp>
 #include <boost/asio/co_spawn.hpp>
@@ -23,11 +22,7 @@ namespace asio = boost::asio;
 using namespace std::chrono_literals;
 using boost::redis::request;
 using boost::redis::generic_flat_response;
-using boost::redis::consume_one;
-using boost::redis::logger;
 using boost::redis::config;
-using boost::redis::ignore;
-using boost::redis::error;
 using boost::system::error_code;
 using boost::redis::connection;
 using asio::signal_set;
@@ -60,24 +55,23 @@ auto receiver(std::shared_ptr<connection> conn) -> asio::awaitable<void>
    // Loop while reconnection is enabled
    while (conn->will_reconnect()) {
       // Reconnect to the channels.
-      co_await conn->async_exec(req, ignore);
+      co_await conn->async_exec(req);
 
-      // Loop reading Redis pushs messages.
+      // Loop to read Redis push messages.
       for (error_code ec;;) {
-         // First tries to read any buffered pushes.
-         conn->receive(ec);
-         if (ec == error::sync_receive_push_failed) {
-            ec = {};
-            co_await conn->async_receive(asio::redirect_error(asio::use_awaitable, ec));
-         }
-
+         // Wait for pushes
+         co_await conn->async_receive2(asio::redirect_error(ec));
          if (ec)
             break;  // Connection lost, break so we can reconnect to channels.
 
-         std::cout << resp.value().at(1).value << " " << resp.value().at(2).value << " "
-                   << resp.value().at(3).value << std::endl;
+         // The response must be consumed without suspending the
+         // coroutine i.e. without the use of async operations.
+         for (auto const& elem: resp.value().get_view())
+            std::cout << elem.value.data << "\n";
+
+         std::cout << std::endl;
 
-         consume_one(resp);
+         resp.value().clear();
       }
    }
 }

include/boost/redis/adapter/detail/adapters.hpp

@@ -12,15 +12,14 @@
 #include <boost/redis/resp3/node.hpp>
 #include <boost/redis/resp3/serialization.hpp>
 #include <boost/redis/resp3/type.hpp>
-#include <boost/redis/response.hpp>
+#include <boost/redis/generic_flat_response_value.hpp>
 
 #include <boost/assert.hpp>
 
 #include <array>
 #include <charconv>
 #include <deque>
 #include <forward_list>
-#include <iostream>
 #include <list>
 #include <map>
 #include <optional>
@@ -138,6 +137,8 @@ void boost_redis_from_bulk(T& t, resp3::basic_node<String> const& node, system::
    from_bulk_impl<T>::apply(t, node, ec);
 }
 
+//================================================
+
 template <class Result>
 class general_aggregate {
 private:
@@ -177,25 +178,25 @@ class general_aggregate {
 };
 
 template <>
-class general_aggregate<result<flat_response_value>> {
+class general_aggregate<result<generic_flat_response_value>> {
 private:
-   result<flat_response_value>* result_;
+   result<generic_flat_response_value>* result_ = nullptr;
 
 public:
-   explicit general_aggregate(result<flat_response_value>* c = nullptr)
+   explicit general_aggregate(result<generic_flat_response_value>* c = nullptr)
    : result_(c)
    { }
 
    void on_init() { }
    void on_done()
    {
       if (result_->has_value()) {
-         result_->value().set_view();
+         result_->value().notify_done();
       }
    }
 
    template <class String>
-   void operator()(resp3::basic_node<String> const& nd, system::error_code&)
+   void on_node(resp3::basic_node<String> const& nd, system::error_code&)
    {
       BOOST_ASSERT_MSG(!!result_, "Unexpected null pointer");
       switch (nd.data_type) {
@@ -206,7 +207,7 @@ class general_aggregate<result<flat_response_value>> {
                std::string{std::cbegin(nd.value), std::cend(nd.value)}
             };
             break;
-         default: result_->value().add_node(nd);
+         default: result_->value().push(nd);
       }
    }
 };

include/boost/redis/adapter/detail/response_traits.hpp

@@ -110,9 +110,9 @@ struct response_traits<response<Ts...>> {
 template <>
 struct response_traits<generic_flat_response> {
    using response_type = generic_flat_response;
-   using adapter_type = vector_adapter<response_type>;
+   using adapter_type = general_aggregate<response_type>;
 
-   static auto adapt(response_type& v) noexcept { return adapter_type{v}; }
+   static auto adapt(response_type& v) noexcept { return adapter_type{&v}; }
 };
 }  // namespace boost::redis::adapter::detail
 

include/boost/redis/adapter/detail/result_traits.hpp

@@ -66,7 +66,7 @@ struct result_traits<result<std::vector<resp3::basic_node<String>, Allocator>>>
 template <>
 struct result_traits<generic_flat_response> {
    using response_type = generic_flat_response;
-   using adapter_type = adapter::detail::general_aggregate<response_type>;
+   using adapter_type = general_aggregate<response_type>;
    static auto adapt(response_type& v) noexcept { return adapter_type{&v}; }
 };
 

include/boost/redis/connection.hpp

@@ -593,7 +593,7 @@ class basic_connection {
       return async_run(config{}, std::forward<CompletionToken>(token));
    }
 
-   /** @brief Receives server side pushes asynchronously.
+   /** @brief (Deprecated) Receives server side pushes asynchronously.
     *
     * When pushes arrive and there is no `async_receive` operation in
     * progress, pushed data, requests, and responses will be paused
@@ -623,12 +623,76 @@ class basic_connection {
     * @param token Completion token.
     */
    template <class CompletionToken = asio::default_completion_token_t<executor_type>>
+   BOOST_DEPRECATED("Please use async_receive2 instead.")
    auto async_receive(CompletionToken&& token = {})
    {
       return impl_->receive_channel_.async_receive(std::forward<CompletionToken>(token));
    }
 
-   /** @brief Receives server pushes synchronously without blocking.
+   /** @brief Wait for server pushes asynchronously
+    *
+    * This function suspends until a server push is received by the
+    * connection. On completion an unspecified number of pushes will
+    * have been added to the response object set with @ref
+    * boost::redis::connection::set_receive_response.
+    *
+    * To prevent receiving an unbound number of pushes the connection
+    * blocks further read operations on the socket when 256 pushes
+    * accumulate internally (we don't make any commitment to this
+    * exact number). When that happens ongoing `async_exec`s and
+    * health-checks won't make any progress and the connection will
+    * eventually timeout. To avoid that Apps should call
+    * `async_receive2` continuously in a loop.
+    *
+    * @Note To avoid deadlocks the task (e.g. coroutine) calling
+    * `async_receive2` should not call `async_exec` in a way where
+    * they could block each other.
+    *
+    * For an example see cpp20_subscriber.cpp. The completion token
+    * must have the following signature
+    *
+    * @code
+    * void f(system::error_code);
+    * @endcode
+    *
+    * @par Per-operation cancellation
+    * This operation supports the following cancellation types:
+    *
+    *   @li `asio::cancellation_type_t::terminal`.
+    *   @li `asio::cancellation_type_t::partial`.
+    *   @li `asio::cancellation_type_t::total`.
+    *
+    * Calling `basic_connection::cancel(operation::receive)` will
+    * also cancel any ongoing receive operations.
+    * 
+    * @param token Completion token.
+    */
+   template <class CompletionToken = asio::default_completion_token_t<executor_type>>
+   auto async_receive2(CompletionToken&& token = {})
+   {
+      return
+         impl_->receive_channel_.async_receive(
+            asio::deferred(
+               [&conn = *this](system::error_code ec, std::size_t)
+               {
+                  if (!ec) {
+                     auto f = [](system::error_code, std::size_t) {
+                       // There is no point in checking for errors
+                       // here since async_receive just completed
+                       // without errors.
+                     };
+
+                     // We just want to drain the channel.
+                     while (conn.impl_->receive_channel_.try_receive(f));
+                  }
+
+                  return asio::deferred.values(ec);
+               }
+            )
+         )(std::forward<CompletionToken>(token));
+   }
+
+   /** @brief (Deprecated) Receives server pushes synchronously without blocking.
     *
     *  Receives a server push synchronously by calling `try_receive` on
     *  the underlying channel. If the operation fails because
@@ -638,6 +702,7 @@ class basic_connection {
     *  @param ec Contains the error if any occurred.
     *  @returns The number of bytes read from the socket.
     */
+   BOOST_DEPRECATED("Please, use async_receive2 instead.")
    std::size_t receive(system::error_code& ec)
    {
       std::size_t size = 0;
@@ -837,7 +902,7 @@ class basic_connection {
       "the other member functions to interact with the connection.")
    auto const& next_layer() const noexcept { return impl_->stream_.next_layer(); }
 
-   /// Sets the response object of @ref async_receive operations.
+   /// Sets the response object of @ref async_receive2 operations.
    template <class Response>
    void set_receive_response(Response& resp)
    {
@@ -1028,12 +1093,21 @@ class connection {
 
    /// @copydoc basic_connection::async_receive
    template <class CompletionToken = asio::deferred_t>
+   BOOST_DEPRECATED("Please use async_receive2 instead.")
    auto async_receive(CompletionToken&& token = {})
    {
       return impl_.async_receive(std::forward<CompletionToken>(token));
    }
 
+   /// @copydoc basic_connection::async_receive2
+   template <class CompletionToken = asio::deferred_t>
+   auto async_receive2(CompletionToken&& token = {})
+   {
+      return impl_.async_receive2(std::forward<CompletionToken>(token));
+   }
+
    /// @copydoc basic_connection::receive
+   BOOST_DEPRECATED("Please use async_receive2 instead.")
    std::size_t receive(system::error_code& ec) { return impl_.receive(ec); }
 
    /**