Generic flat response implementation

example/cpp20_subscriber.cpp

@@ -23,6 +23,7 @@ namespace asio = boost::asio;
 using namespace std::chrono_literals;
 using boost::redis::request;
 using boost::redis::generic_response;
+using boost::redis::generic_flat_response;
 using boost::redis::consume_one;
 using boost::redis::logger;
 using boost::redis::config;
@@ -54,6 +55,8 @@ auto receiver(std::shared_ptr<connection> conn) -> asio::awaitable<void>
    request req;
    req.push("SUBSCRIBE", "channel");
 
+   // Alternatively, you can use generic_flat_response here, but keep in mind
+   // that to access elements you need to call .view() on resp.value()
    generic_response resp;
    conn->set_receive_response(resp);
 

include/boost/redis/adapter/any_adapter.hpp

@@ -34,10 +34,13 @@ namespace boost::redis {
  */
 class any_adapter {
 public:
-   using fn_type = std::function<void(std::size_t, resp3::node_view const&, system::error_code&)>;
+   using adapt_fn_type = std::function<
+      void(std::size_t, resp3::node_view const&, system::error_code&)>;
+   using done_fn_type = adapter::detail::done_fn_type;
 
    struct impl_t {
-      fn_type adapt_fn;
+      adapt_fn_type adapt_fn;
+      done_fn_type prepare_done_fn;
       std::size_t supported_response_size;
    } impl_;
 
@@ -47,7 +50,7 @@ class any_adapter {
       using namespace boost::redis::adapter;
       auto adapter = boost_redis_adapt(resp);
       std::size_t size = adapter.get_supported_response_size();
-      return {std::move(adapter), size};
+      return {std::move(adapter), detail::prepare_done(resp), size};
    }
 
    template <class Executor>

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

@@ -12,6 +12,7 @@
 #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/assert.hpp>
 
@@ -138,6 +139,26 @@ void boost_redis_from_bulk(T& t, resp3::basic_node<String> const& node, system::
 
 //================================================
 
+using done_fn_type = std::function<void()>;
+
+template <typename T>
+auto prepare_done(T&) noexcept -> done_fn_type
+{
+   return [] { };
+}
+
+template <typename T>
+auto prepare_done(generic_flat_response& resp) noexcept -> done_fn_type
+{
+   return [resp]() mutable {
+      if (resp.has_value()) {
+         resp.value().set_view();
+      }
+   };
+}
+
+//================================================
+
 template <class Result>
 class general_aggregate {
 private:
@@ -170,6 +191,32 @@ class general_aggregate {
    }
 };
 
+template <>
+class general_aggregate<result<flat_response_value>> {
+private:
+   result<flat_response_value>* result_;
+
+public:
+   explicit general_aggregate(result<flat_response_value>* c = nullptr)
+   : result_(c)
+   { }
+   template <class String>
+   void operator()(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().push_back(nd);
+      }
+   }
+};
+
 template <class Node>
 class general_simple {
 private:

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

@@ -133,6 +133,14 @@ struct response_traits<response<Ts...>> {
    static auto adapt(response_type& r) noexcept { return adapter_type{r}; }
 };
 
+template <>
+struct response_traits<generic_flat_response> {
+   using response_type = generic_flat_response;
+   using adapter_type = vector_adapter<response_type>;
+
+   static auto adapt(response_type& v) noexcept { return adapter_type{v}; }
+};
+
 template <class Adapter>
 class wrapper {
 public:

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

@@ -13,6 +13,7 @@
 #include <boost/redis/error.hpp>
 #include <boost/redis/ignore.hpp>
 #include <boost/redis/resp3/type.hpp>
+#include <boost/redis/response.hpp>
 
 #include <boost/mp11.hpp>
 
@@ -62,6 +63,13 @@ struct result_traits<result<std::vector<resp3::basic_node<String>, Allocator>>>
    static auto adapt(response_type& v) noexcept { return adapter_type{&v}; }
 };
 
+template <>
+struct result_traits<generic_flat_response> {
+   using response_type = generic_flat_response;
+   using adapter_type = adapter::detail::general_aggregate<response_type>;
+   static auto adapt(response_type& v) noexcept { return adapter_type{&v}; }
+};
+
 template <class T>
 using adapter_t = typename result_traits<std::decay_t<T>>::adapter_type;
 

include/boost/redis/connection.hpp

@@ -764,7 +764,8 @@ class basic_connection {
          1);
       auto info = detail::make_elem(req, std::move(adapter_impl.adapt_fn));
 
-      info->set_done_callback([notifier]() {
+      info->set_done_callback([notifier, prepare_done = std::move(adapter_impl.prepare_done_fn)]() {
+         prepare_done();
          notifier->try_send(std::error_code{}, 0);
       });
 

include/boost/redis/impl/response.ipp

@@ -11,15 +11,30 @@
 
 namespace boost::redis {
 
-void consume_one(generic_response& r, system::error_code& ec)
+namespace {
+template <typename Container>
+auto& get_value(Container& c)
+{
+   return c;
+}
+
+template <>
+auto& get_value(flat_response_value& c)
+{
+   return c.view();
+}
+
+template <typename Response>
+void consume_one_impl(Response& r, system::error_code& ec)
 {
    if (r.has_error())
       return;  // Nothing to consume.
 
-   if (std::empty(r.value()))
+   auto& value = get_value(r.value());
+   if (std::empty(value))
       return;  // Nothing to consume.
 
-   auto const depth = r.value().front().depth;
+   auto const depth = value.front().depth;
 
    // To simplify we will refuse to consume any data-type that is not
    // a root node. I think there is no use for that and it is complex
@@ -33,17 +48,15 @@ void consume_one(generic_response& r, system::error_code& ec)
       return e.depth == depth;
    };
 
-   auto match = std::find_if(std::next(std::cbegin(r.value())), std::cend(r.value()), f);
+   auto match = std::find_if(std::next(std::cbegin(value)), std::cend(value), f);
 
-   r.value().erase(std::cbegin(r.value()), match);
+   value.erase(std::cbegin(value), match);
 }
 
-void consume_one(generic_response& r)
-{
-   system::error_code ec;
-   consume_one(r, ec);
-   if (ec)
-      throw system::system_error(ec);
-}
+}  // namespace
+
+void consume_one(generic_response& r, system::error_code& ec) { consume_one_impl(r, ec); }
+
+void consume_one(generic_flat_response& r, system::error_code& ec) { consume_one_impl(r, ec); }
 
 }  // namespace boost::redis

include/boost/redis/resp3/node.hpp

@@ -59,6 +59,14 @@ using node = basic_node<std::string>;
 /// A node in the response tree that does not own its data.
 using node_view = basic_node<std::string_view>;
 
+struct offset_string {
+   std::string_view data;
+   std::size_t offset{};
+   std::size_t size{};
+};
+
+using offset_node = basic_node<offset_string>;
+
 }  // namespace boost::redis::resp3
 
 #endif  // BOOST_REDIS_RESP3_NODE_HPP

include/boost/redis/response.hpp

@@ -31,6 +31,58 @@ using response = std::tuple<adapter::result<Ts>...>;
  */
 using generic_response = adapter::result<std::vector<resp3::node>>;
 
+struct flat_response_value {
+public:
+   /// Reserve capacity for nodes and data storage.
+   void reserve(std::size_t num_nodes, std::size_t string_size)
+   {
+      data_.reserve(num_nodes * string_size);
+      view_.reserve(num_nodes);
+   }
+
+   std::vector<resp3::offset_node> const& view() const { return view_; }
+   std::vector<resp3::offset_node>& view() { return view_; }
+
+   template <typename String>
+   void push_back(const resp3::basic_node<String>& nd)
+   {
+      resp3::offset_string offset_string;
+      offset_string.offset = data_.size();
+      offset_string.size = nd.value.size();
+
+      data_.append(nd.value.data(), nd.value.size());
+
+      resp3::offset_node new_node;
+      new_node.data_type = nd.data_type;
+      new_node.aggregate_size = nd.aggregate_size;
+      new_node.depth = nd.depth;
+      new_node.value = std::move(offset_string);
+
+      view_.push_back(std::move(new_node));
+   }
+
+   void set_view()
+   {
+      for (auto& node : view_) {
+         auto& offset_string = node.value;
+         offset_string.data = std::string_view{
+            data_.data() + offset_string.offset,
+            offset_string.size};
+      }
+   }
+
+private:
+   std::string data_;
+   std::vector<resp3::offset_node> view_;
+};
+
+/** @brief A memory-efficient generic response to a request.
+ *  @ingroup high-level-api
+ * 
+ *  Uses a compact buffer to store RESP3 data with reduced allocations.
+ */
+using generic_flat_response = adapter::result<flat_response_value>;
+
 /** @brief Consume on response from a generic response
  *
  *  This function rotates the elements so that the start of the next
@@ -72,12 +124,22 @@ using generic_response = adapter::result<std::vector<resp3::node>>;
  */
 void consume_one(generic_response& r, system::error_code& ec);
 
+/// Consume on response from a generic flat response
+void consume_one(generic_flat_response& r, system::error_code& ec);
+
 /**
  * @brief Throwing overload of `consume_one`.
  *
  * @param r The response to modify.
  */
-void consume_one(generic_response& r);
+template <typename Response>
+void consume_one(Response& r)
+{
+   system::error_code ec;
+   consume_one(r, ec);
+   if (ec)
+      throw system::system_error(ec);
+}
 
 }  // namespace boost::redis