Concludes the work started by Nikolai Vladimirov on the generic_flat_response

Author: mzimbres

README.md

@@ -87,9 +87,9 @@ them are:
 * [Client-side caching](https://redis.io/docs/manual/client-side-caching/).
 
 The connection class supports server pushes by means of the
-`connection::async_receive` function, which can be
+`connection::async_receive2` function, which can be
 called in the same connection that is being used to execute commands.
-The coroutine below shows how to use it:
+The coroutine below shows how to use it
 
 
 ```cpp
@@ -99,31 +99,30 @@ receiver(std::shared_ptr<connection> conn) -> net::awaitable<void>
    request req;
    req.push("SUBSCRIBE", "channel");
 
-   generic_response resp;
+   flat_tree resp;
    conn->set_receive_response(resp);
 
    // Loop while reconnection is enabled
    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.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

@@ -97,9 +97,9 @@ them are:
 * https://redis.io/docs/manual/client-side-caching/[Client-side caching].
 
 The connection class supports server pushes by means of the
-xref:reference:boost/redis/basic_connection/async_receive.adoc[`connection::async_receive`] function, which can be
+xref:reference:boost/redis/basic_connection/async_receive.adoc[`connection::async_receive2`] function, which can be
 called in the same connection that is being used to execute commands.
-The coroutine below shows how to use it:
+The coroutine below shows how to use it
 
 
 [source,cpp]
@@ -110,26 +110,25 @@ receiver(std::shared_ptr<connection> conn) -> net::awaitable<void>
    request req;
    req.push("SUBSCRIBE", "channel");
 
-   generic_response resp;
+   flat_tree resp;
    conn->set_receive_response(resp);
 
    // Loop while reconnection is enabled
    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.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::resp3::flat_tree;
 using boost::redis::request;
 using boost::system::error_code;
 using namespace std::chrono_literals;
@@ -47,21 +45,25 @@ auto receiver(std::shared_ptr<connection> conn) -> awaitable<void>
    request req;
    req.push("SUBSCRIBE", "channel");
 
-   generic_flat_response resp;
+   flat_tree 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 << " "
-                   << resp.value().at(3).value << std::endl;
-         resp.value().clear();
+
+         for (auto const& elem: resp.get_view())
+            std::cout << elem.value << "\n";
+
+         std::cout << std::endl;
+
+         resp.clear();
       }
    }
 }
@@ -74,7 +76,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>
@@ -22,12 +21,8 @@
 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::resp3::flat_tree;
 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;
@@ -54,30 +49,29 @@ auto receiver(std::shared_ptr<connection> conn) -> asio::awaitable<void>
    request req;
    req.push("SUBSCRIBE", "channel");
 
-   generic_flat_response resp;
+   flat_tree resp;
    conn->set_receive_response(resp);
 
    // 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.get_view())
+            std::cout << elem.value << "\n";
+
+         std::cout << std::endl;
 
-         consume_one(resp);
+         resp.clear();
       }
    }
 }

include/boost/redis/adapter/any_adapter.hpp

@@ -53,7 +53,6 @@ class any_adapter {
    static auto create_impl(T& resp) -> impl_t
    {
       using namespace boost::redis::adapter;
-
       return [adapter2 = boost_redis_adapt(resp)](
                   any_adapter::parse_event ev,
                   resp3::node_view const& nd,

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

@@ -1,4 +1,4 @@
-/* Copyright (c) 2018-2024 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)
@@ -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/resp3/flat_tree.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,37 +178,54 @@ class general_aggregate {
 };
 
 template <>
-class general_aggregate<result<flat_response_value>> {
+class general_aggregate<resp3::tree> {
 private:
-   result<flat_response_value>* result_;
+   resp3::tree* tree_ = nullptr;
 
 public:
-   explicit general_aggregate(result<flat_response_value>* c = nullptr)
-   : result_(c)
+   explicit general_aggregate(resp3::tree* c = nullptr)
+   : tree_(c)
+   { }
+
+   void on_init() { }
+   void on_done() { }
+
+   template <class String>
+   void on_node(resp3::basic_node<String> const& nd, system::error_code&)
+   {
+      BOOST_ASSERT_MSG(!!tree_, "Unexpected null pointer");
+
+      resp3::node tmp;
+      tmp.data_type = nd.data_type;
+      tmp.aggregate_size = nd.aggregate_size;
+      tmp.depth = nd.depth;
+      tmp.value = std::string{std::cbegin(nd.value), std::cend(nd.value)};
+
+      tree_->push_back(std::move(tmp));
+   }
+};
+
+template <>
+class general_aggregate<resp3::flat_tree> {
+private:
+   resp3::flat_tree* tree_ = nullptr;
+
+public:
+   explicit general_aggregate(resp3::flat_tree* c = nullptr)
+   : tree_(c)
    { }
 
    void on_init() { }
    void on_done()
    {
-      if (result_->has_value()) {
-         result_->value().set_view();
-      }
+      tree_->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) {
-         case resp3::type::blob_error:
-         case resp3::type::simple_error:
-            *result_ = error{
-               nd.data_type,
-               std::string{std::cbegin(nd.value), std::cend(nd.value)}
-            };
-            break;
-         default: result_->value().add_node(nd);
-      }
+      BOOST_ASSERT_MSG(!!tree_, "Unexpected null pointer");
+      tree_->push(nd);
    }
 };