f [skip-ci]

Author: mzimbres

example/cpp20_chat_room.cpp

@@ -32,7 +32,7 @@ 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::generic_response;
 using boost::redis::ignore;
 using boost::redis::request;
 using boost::system::error_code;
@@ -46,7 +46,7 @@ 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()) {

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

@@ -60,24 +60,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 reading 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 a push
+         co_await conn->async_receive(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/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

@@ -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,20 +178,20 @@ 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();
       }
    }
 
@@ -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/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

@@ -859,6 +859,31 @@ class basic_connection {
       return impl_->receive_channel_.async_receive(std::forward<CompletionToken>(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.
+                     };
+
+                     // Drain the channel.
+                     while (conn.impl_->receive_channel_.try_receive(f));
+                  }
+
+                  return asio::deferred.values(ec);
+               }
+            )
+         )(std::forward<CompletionToken>(token));
+   }
+
    /** @brief Receives server pushes synchronously without blocking.
     *
     *  Receives a server push synchronously by calling `try_receive` on
@@ -1264,6 +1289,13 @@ class connection {
       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
    std::size_t receive(system::error_code& ec) { return impl_.receive(ec); }
 

include/boost/redis/detail/writer_fsm.hpp

@@ -1,6 +1,6 @@
 //
 // Copyright (c) 2025 Marcelo Zimbres Silva ([email protected]),
-// Ruben Perez Hidalgo (rubenperez038 at gmail dot com)
+// Nikolai Vladimirov (TODO)
 //
 // Distributed under the Boost Software License, Version 1.0. (See accompanying
 // file LICENSE_1_0.txt or copy at http://www.boost.org/LICENSE_1_0.txt)

include/boost/redis/generic_flat_response_value.hpp

@@ -0,0 +1,102 @@
+//
+// Copyright (c) 2025 Marcelo Zimbres Silva ([email protected]),
+// Nikolai Vladimirov (TODO)
+//
+// Distributed under the Boost Software License, Version 1.0. (See accompanying
+// file LICENSE_1_0.txt or copy at http://www.boost.org/LICENSE_1_0.txt)
+//
+
+#ifndef BOOST_REDIS_FLAT_RESPONSE_HPP
+#define BOOST_REDIS_FLAT_RESPONSE_HPP
+
+#include <boost/redis/adapter/result.hpp>
+#include <boost/redis/resp3/node.hpp>
+
+#include <boost/system/error_code.hpp>
+
+#include <string>
+#include <tuple>
+#include <vector>
+
+namespace boost::redis {
+
+// Similar to the generic_response but its data is stored flat
+// internally.
+struct generic_flat_response_value {
+public:
+   /** @brief Reserve capacity
+    *
+    *  Reserve memory for incoming data.
+    *
+    *  @param bytes Number of bytes to reserve for data.
+    *  @param nodes Number of nodes to reserver.
+    */
+   void reserve(std::size_t bytes, std::size_t nodes);
+
+   /** @brief Clear both the data and the node buffers
+    *  
+    *  @Note: A `boost::redis:.generic_flat_response` can contain the
+    *  response to multiple Redis commands and server pushes. Calling
+    *  this function will erase everything contianed in it. To clear
+    *  only one response/push use `boost::redis::consume_one`.
+    */
+   void clear();
+
+   /// Returns the size of the data buffer
+   auto data_size() const noexcept -> std::size_t
+      { return data_.size(); }
+
+   /** @brief Return the RESP3 response
+    *  
+    *  The data member in each `boost::redis::offset_string` are all
+    *  set and therefore safe to use.
+    */
+   auto get_view() const -> resp3::offset_response const&
+      { return view_; }
+
+   /** @brief Return the RESP3 response
+    *  
+    *  The data member in each `boost::redis::offset_string` are all
+    *  set and therefore safe to use.
+    */
+   auto get_view() -> resp3::offset_response&
+      { return view_; }
+
+   /// Push a new node to the response
+   void push(resp3::node_view const& nd);
+
+   /** @brief Returns the number of times reallocations
+    *
+    *  Each call to the push might result in a memory reallocation.
+    *  This number function returns how many reallocations were
+    *  detected and can be useful to determine how much memory to
+    *  reserve upfront.
+    */
+   auto get_reallocs() const noexcept
+      { return reallocs_; }
+
+   /** @brief Notify the object that all nodes were pushed.
+    *
+    *  This function is called automativally by the library.
+    */
+   void notify_done();
+
+   /// Returns the number of complete RESP3 messages contained in this object.
+   std::size_t get_total_msgs() const noexcept
+      { return total_msgs_; }
+
+private:
+   void add_node_impl(resp3::node_view const& nd);
+
+   std::string data_;
+   resp3::offset_response view_;
+   std::size_t pos_ = 0u;
+   std::size_t reallocs_ = 0u;
+
+   // The number of messages contained in this object.
+   std::size_t total_msgs_ = 0u;
+};
+
+}  // namespace boost::redis
+
+#endif  // BOOST_REDIS_FLAT_RESPONSE_HPP

include/boost/redis/impl/generic_flat_response_value.ipp

@@ -0,0 +1,66 @@
+//
+// Copyright (c) 2025 Marcelo Zimbres Silva ([email protected]),
+// Nikolai Vladimirov (TODO)
+//
+// Distributed under the Boost Software License, Version 1.0. (See accompanying
+// file LICENSE_1_0.txt or copy at http://www.boost.org/LICENSE_1_0.txt)
+//
+
+#include <boost/redis/error.hpp>
+#include <boost/redis/generic_flat_response_value.hpp>
+
+#include <boost/assert.hpp>
+
+namespace boost::redis {
+
+void generic_flat_response_value::reserve(std::size_t bytes, std::size_t nodes)
+{
+   data_.reserve(bytes);
+   view_.reserve(nodes);
+}
+
+void generic_flat_response_value::clear()
+{
+   pos_ = 0u;
+   total_msgs_ = 0u;
+   data_.clear();
+   view_.clear();
+}
+
+void generic_flat_response_value::notify_done()
+{
+   total_msgs_ += 1;
+
+   for (; pos_ < view_.size(); ++pos_) {
+      auto& v = view_.at(pos_).value;
+      v.data = std::string_view{data_.data() + v.offset, v.size};
+   }
+}
+
+void generic_flat_response_value::push(resp3::node_view const& nd)
+{
+   auto data_before = data_.data();
+   add_node_impl(nd);
+   auto data_after = data_.data();
+
+   if (data_after != data_before) {
+      pos_ = 0;
+      reallocs_ += 1;
+   }
+}
+
+void generic_flat_response_value::add_node_impl(resp3::node_view const& nd)
+{
+   resp3::offset_node ond;
+   ond.data_type = nd.data_type;
+   ond.aggregate_size = nd.aggregate_size;
+   ond.depth = nd.depth;
+   ond.value.offset = data_.size();
+   ond.value.size = nd.value.size();
+
+   // This must come after setting the offset above.
+   data_.append(nd.value.data(), nd.value.size());
+
+   view_.push_back(std::move(ond));
+}
+}  // namespace boost::redis