Added example cpp20_streams, which reproduces an assertion.

Author: bram

CMakeLists.txt

@@ -92,6 +92,15 @@ if (MSVC)
     target_compile_definitions(cpp20_intro PRIVATE _WIN32_WINNT=0x0601)
 endif()
 
+add_executable(cpp20_streams examples/cpp20_streams.cpp)
+target_link_libraries(cpp20_streams common)
+target_compile_features(cpp20_streams PUBLIC cxx_std_20)
+add_test(cpp20_streams cpp20_streams)
+if (MSVC)
+    target_compile_options(cpp20_streams PRIVATE /bigobj)
+    target_compile_definitions(cpp20_streams PRIVATE _WIN32_WINNT=0x0601)
+endif()
+
 add_executable(cpp17_intro examples/cpp17_intro.cpp)
 target_compile_features(cpp17_intro PUBLIC cxx_std_17)
 add_test(cpp17_intro cpp17_intro)

examples/cpp20_streams.cpp

@@ -0,0 +1,91 @@
+/* Copyright (c) 2018-2022 Marcelo Zimbres Silva ([email protected])
+ *
+ * Distributed under the Boost Software License, Version 1.0. (See
+ * accompanying file LICENSE.txt)
+ */
+
+#include <boost/redis/connection.hpp>
+#include <boost/asio/deferred.hpp>
+#include <boost/asio/co_spawn.hpp>
+#include <boost/asio/detached.hpp>
+#include <boost/asio/consign.hpp>
+#include <boost/asio/signal_set.hpp>
+#include <boost/asio/awaitable.hpp>
+#include <iostream>
+
+#if defined(BOOST_ASIO_HAS_CO_AWAIT)
+
+namespace net = boost::asio;
+using boost::redis::config;
+using boost::redis::generic_response;
+using boost::redis::operation;
+using boost::redis::request;
+using connection = net::deferred_t::as_default_on_t<boost::redis::connection>;
+using signal_set = net::deferred_t::as_default_on_t<net::signal_set>;
+
+auto stream_reader(std::shared_ptr<connection> conn) -> net::awaitable<void>
+{
+    std::string redisStreamKey_;
+    request req;
+    generic_response resp;
+
+    std::string stream_id{"$"};
+    std::string const field = "myfield";
+
+    for (;;) {
+        req.push("XREAD", "BLOCK", "0", "STREAMS", "test-topic", stream_id);
+        co_await conn->async_exec(req, resp);
+
+        // std::cout << "Response: ";
+        // for (int i = 0; i < resp->value().size(); ++i) {
+        //     std::cout << resp->value().at(i).value << ", ";
+        // }
+        // std::cout << std::endl;
+
+        // 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_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.
+
+        std::size_t item_index = 0;
+        while (item_index < std::size(resp.value())) {
+           auto const& val = resp.value().at(item_index).value;
+
+           if (field.compare(val) == 0) {
+               // We've hit a myfield field.
+               // The streamId is located at item_index - 2
+               // The payload is located at item_index + 1
+               stream_id = resp.value().at(item_index - 2).value;
+               std::cout
+                  << "StreamId: " << stream_id << ", "
+                  << "MyField: " << resp.value().at(item_index + 1).value
+                  << std::endl;
+               ++item_index; // We can increase so we don't read this again
+           }
+
+           ++item_index;
+        }
+
+        req.clear();
+        resp.value().clear();
+    }
+
+}
+
+// Run this in another terminal:
+// redis-cli -r 100000 -i 0.0001 XADD "test-topic" "*" "myfield" "myfieldvalue1"
+auto co_main(config cfg) -> net::awaitable<void>
+{
+   auto ex = co_await net::this_coro::executor;
+   auto conn = std::make_shared<connection>(ex);
+   net::co_spawn(ex, stream_reader(conn), net::detached);
+   conn->async_run(cfg, {}, net::consign(net::detached, conn));
+
+   signal_set sig_set(ex, SIGINT, SIGTERM);
+   co_await sig_set.async_wait();
+   conn->cancel();
+}
+#endif // defined(BOOST_ASIO_HAS_CO_AWAIT)

examples/cpp20_subscriber.cpp

@@ -19,14 +19,14 @@
 #if defined(BOOST_ASIO_HAS_CO_AWAIT)
 
 namespace net = boost::asio;
+using namespace std::chrono_literals;
 using boost::redis::request;
 using boost::redis::generic_response;
 using boost::redis::logger;
 using boost::redis::config;
 using boost::system::error_code;
 using connection = net::deferred_t::as_default_on_t<boost::redis::connection>;
 using signal_set = net::deferred_t::as_default_on_t<net::signal_set>;
-using namespace std::chrono_literals;
 
 /* This example will subscribe and read pushes indefinitely.
  *