Uses boost.describe to simplify json serialization.

benchmarks/nodejs/README.txt

@@ -0,0 +1,4 @@
+
+$ npm install
+
+$ node echo_server_over_redis.js

benchmarks/nodejs/echo_server_over_redis/echo_server_over_redis.js

@@ -1,7 +1,7 @@
 import { createClient } from 'redis';
 import * as net from 'net';
 
-const client = createClient({url: 'redis://db.occase.de:6379' });
+const client = createClient({url: 'redis://aedis.occase.de:63799' });
 client.on('error', (err) => console.log('Redis Client Error', err));
 await client.connect();
 

examples/common/serialization.hpp

@@ -0,0 +1,105 @@
+/* Copyright (c) 2018-2022 Marcelo Zimbres Silva (mzimbres@gmail.com)
+ *
+ * Distributed under the Boost Software License, Version 1.0. (See
+ * accompanying file LICENSE.txt)
+ */
+
+#define BOOST_JSON_NO_LIB
+#define BOOST_CONTAINER_NO_LIB
+#include <boost/redis.hpp>
+#include <boost/describe.hpp>
+#include <boost/mp11.hpp>
+#include <boost/json.hpp>
+
+using namespace boost::describe;
+using namespace boost::json;
+
+template <
+      class T,
+      class Bd = describe_bases<T, mod_any_access>,
+      class Md = describe_members<T, mod_any_access>,
+      class En = std::enable_if_t<!std::is_union<T>::value>
+   >
+std::ostream& operator<<(std::ostream& os, T const & t)
+{
+    os << "{";
+
+    bool first = true;
+
+    boost::mp11::mp_for_each<Bd>([&](auto D){
+
+        if( !first ) { os << ", "; } first = false;
+
+        using B = typename decltype(D)::type;
+        os << (B const&)t;
+
+    });
+
+    boost::mp11::mp_for_each<Md>([&](auto D){
+
+        if( !first ) { os << ", "; } first = false;
+
+        os << "." << D.name << " = " << t.*D.pointer;
+
+    });
+
+    os << "}";
+    return os;
+}
+
+template <
+      class T,
+      class D1 = boost::describe::describe_members<T, boost::describe::mod_public | boost::describe::mod_protected>,
+      class D2 = boost::describe::describe_members<T, boost::describe::mod_private>,
+      class En = std::enable_if_t<boost::mp11::mp_empty<D2>::value && !std::is_union<T>::value>
+   >
+void tag_invoke(boost::json::value_from_tag const&, boost::json::value& v, T const & t)
+{
+    auto& obj = v.emplace_object();
+
+    boost::mp11::mp_for_each<D1>([&](auto D){
+
+        obj[ D.name ] = boost::json::value_from( t.*D.pointer );
+
+    });
+}
+
+template<class T> void extract( boost::json::object const & obj, char const * name, T & value )
+{
+    value = boost::json::value_to<T>( obj.at( name ) );
+}
+
+template<class T,
+    class D1 = boost::describe::describe_members<T,
+        boost::describe::mod_public | boost::describe::mod_protected>,
+    class D2 = boost::describe::describe_members<T, boost::describe::mod_private>,
+    class En = std::enable_if_t<boost::mp11::mp_empty<D2>::value && !std::is_union<T>::value> >
+
+    T tag_invoke( boost::json::value_to_tag<T> const&, boost::json::value const& v )
+{
+    auto const& obj = v.as_object();
+
+    T t{};
+
+    boost::mp11::mp_for_each<D1>([&](auto D){
+
+        extract( obj, D.name, t.*D.pointer );
+
+    });
+
+    return t;
+}
+
+// Serialization
+template <class T>
+void boost_redis_to_bulk(std::string& to, T const& u)
+{
+   boost::redis::resp3::boost_redis_to_bulk(to, serialize(value_from(u)));
+}
+
+template <class T>
+void boost_redis_from_bulk(T& u, std::string_view sv, boost::system::error_code&)
+{
+   value jv = parse(sv);
+   u = value_to<T>(jv);
+}

examples/cpp20_serialization.cpp

@@ -6,108 +6,82 @@
 
 #include <boost/asio.hpp>
 #if defined(BOOST_ASIO_HAS_CO_AWAIT)
-#include <boost/asio/experimental/awaitable_operators.hpp>
-#define BOOST_JSON_NO_LIB
-#define BOOST_CONTAINER_NO_LIB
-#include <boost/json.hpp>
 #include <boost/redis.hpp>
-#include <algorithm>
-#include <cstdint>
 #include <iostream>
 #include <set>
-#include <iterator>
 #include <string>
 #include "common/common.hpp"
+#include "common/serialization.hpp"
 
 // Include this in no more than one .cpp file.
 #include <boost/json/src.hpp>
 
 namespace net = boost::asio;
 namespace redis = boost::redis;
-using namespace net::experimental::awaitable_operators;
-using namespace boost::json;
 using boost::redis::request;
 using boost::redis::response;
 using boost::redis::ignore_t;
+using boost::redis::operation;
 
 struct user {
    std::string name;
    std::string age;
    std::string country;
 
-   friend auto operator<(user const& a, user const& b)
-   {
-      return std::tie(a.name, a.age, a.country) < std::tie(b.name, b.age, b.country);
-   }
-
-   friend auto operator<<(std::ostream& os, user const& u) -> std::ostream&
-   {
-      os << "Name: " << u.name << "\n"
-         << "Age: " << u.age << "\n"
-         << "Country: " << u.country;
-
-      return os;
-   }
+   friend
+   auto operator<(user const& a, user const& b)
+      { return std::tie(a.name, a.age, a.country) < std::tie(b.name, b.age, b.country); }
 };
 
-// Boost.Json serialization.
-void tag_invoke(value_from_tag, value& jv, user const& u)
+BOOST_DESCRIBE_STRUCT(user, (), (name, age, country))
+
+auto run(std::shared_ptr<connection> conn, std::string host, std::string port) -> net::awaitable<void>
 {
-   jv =
-   { {"name", u.name}
-   , {"age", u.age}
-   , {"country", u.country}
-   };
+   co_await connect(conn, host, port);
+   co_await conn->async_run();
 }
 
-template<class T>
-void extract(object const& obj, T& t, std::string_view key)
+auto hello(std::shared_ptr<connection> conn) -> net::awaitable<void>
 {
-   t = value_to<T>(obj.at(key));
+   request req;
+   req.push("HELLO", 3);
+
+   co_await conn->async_exec(req);
 }
 
-auto tag_invoke(value_to_tag<user>, value const& jv)
-{
-   user u;
-   object const& obj = jv.as_object();
-   extract(obj, u.name, "name");
-   extract(obj, u.age, "age");
-   extract(obj, u.country, "country");
-   return u;
-}
-
-// Serialization
-void boost_redis_to_bulk(std::string& to, user const& u)
-{
-   redis::resp3::boost_redis_to_bulk(to, serialize(value_from(u)));
-}
-
-void boost_redis_from_bulk(user& u, std::string_view sv, boost::system::error_code&)
-{
-   value jv = parse(sv);
-   u = value_to<user>(jv);
-}
-
-net::awaitable<void> co_main(std::string host, std::string port)
+auto sadd(std::shared_ptr<connection> conn) -> net::awaitable<void>
 {
    std::set<user> users
       {{"Joao", "58", "Brazil"} , {"Serge", "60", "France"}};
 
    request req;
-   req.push("HELLO", 3);
    req.push_range("SADD", "sadd-key", users); // Sends
-   req.push("SMEMBERS", "sadd-key"); // Retrieves
-   req.push("QUIT");
 
-   response<ignore_t, int, std::set<user>, std::string> resp;
+   co_await conn->async_exec(req);
+}
 
-   auto conn = std::make_shared<connection>(co_await net::this_coro::executor);
+auto smembers(std::shared_ptr<connection> conn) -> net::awaitable<void>
+{
+   request req;
+   req.push("SMEMBERS", "sadd-key");
 
-   co_await connect(conn, host, port);
-   co_await (conn->async_run() || conn->async_exec(req, resp));
+   response<std::set<user>> resp;
 
-   for (auto const& e: std::get<2>(resp).value())
+   co_await conn->async_exec(req, resp);
+
+   for (auto const& e: std::get<0>(resp).value())
       std::cout << e << "\n";
 }
 
+net::awaitable<void> co_main(std::string host, std::string port)
+{
+   auto ex = co_await net::this_coro::executor;
+   auto conn = std::make_shared<connection>(ex);
+   net::co_spawn(ex, run(conn, host, port), net::detached);
+   co_await hello(conn);
+   co_await sadd(conn);
+   co_await smembers(conn);
+   conn->cancel(operation::run);
+}
+
 #endif // defined(BOOST_ASIO_HAS_CO_AWAIT)

include/boost/redis/connection.hpp

@@ -139,9 +139,13 @@ public:
 
    /** @brief Receives server side pushes asynchronously.
     *
-    *  Users that expect server pushes should call this function in a
-    *  loop. If a push arrives and there is no reader, the connection
-    *  will hang.
+    *  When pushes arrive and there is no async_receive operation in
+    *  progress, pushed data, requests, and responses will be paused
+    *  until async_receive is called again.  Apps will usually want to
+    *  call `async_receive` in a loop. 
+    *
+    *  To cancel an ongoing receive operation apps should call
+    *  `connection::cancel(operation::receive)`.
     *
     *  @param response The response object.
     *  @param token The Asio completion token.