support AsyncRedisCluster::redis, so that we can send command that ha…

…s no key to cluster in async mode

README.md

@@ -2597,8 +2597,15 @@ auto mget_res = async_cluster.mget<std::vector<OptionalString>>({"{hashtag}key1"
 
 unordered_map<string, string> m = {{"a", "b"}, {"c", "d"}};
 Future<void> hmset_res = async_redis.hmset("hash", m.begin(), m.end());
+
+// Create an AsyncRedis object with hash-tag, so that we can send commands that has no key.
+// It connects to Redis instance that holds the given key, i.e. hash-tag.
+auto r = async_cluster.redis("hash-tag");
+Future<string> ping_res = r.command<string>("ping");
 ```
 
+**NOTE**: By default, when you use `AsyncRedisCluster::redis(const StringView &hash_tag, bool new_connection = true)` to create an `AsyncRedis` object, instead of picking a connection from the underlying connection pool, it creates a new connection to the corresponding Redis server. So this is NOT a cheap operation, and you should try to reuse this newly created `AsyncRedis` object as much as possible. If you pass `false` as the second parameter, you can create a `AsyncRedis` object without creating a new connection. However, in this case, you should be very careful, otherwise, you might get bad performance or even dead lock. Please carefully check the related [pipeline section](#very-important-notes) before using this feature. Also the returned `AsyncRedis` object is NOT thread-safe, and if it throws exception, you need to destroy it, and create a new one with the `AsyncRedisCluster::redis` method.
+
 #### Async Subscriber
 
 **NOTE**: I'm not quite satisfied with the interface of `AsyncSubscriber`. If you have a better idea, feel free to open an issue for discussion.

src/sw/redis++/async_connection.h

@@ -470,6 +470,8 @@ class GuardedAsyncConnection {
     std::shared_ptr<AsyncConnection> _connection;
 };
 
+using GuardedAsyncConnectionSPtr = std::shared_ptr<GuardedAsyncConnection>;
+
 namespace detail {
 
 // We seperate this function from ClusterEvent to avoid

src/sw/redis++/async_connection_pool.cpp

@@ -110,7 +110,10 @@ AsyncConnectionPool& AsyncConnectionPool::operator=(AsyncConnectionPool &&that)
 }
 
 AsyncConnectionPool::~AsyncConnectionPool() {
-    assert(_loop);
+    if (!_loop) {
+        // This pool has been moved.
+        return;
+    }
 
     // TODO: what if the connection has been borrowed but not returned?
     // Or we dont' need to worry about that, since it's destructing and

src/sw/redis++/async_redis.cpp

@@ -50,6 +50,10 @@ AsyncRedis::AsyncRedis(const std::shared_ptr<AsyncSentinel> &sentinel,
                                                     opts);
 }
 
+AsyncRedis::AsyncRedis(const GuardedAsyncConnectionSPtr &connection) : _connection(connection) {
+    assert(_connection);
+}
+
 AsyncSubscriber AsyncRedis::subscriber() {
     // TODO: maybe we don't need to check this,
     // since there's no Transaction or Pipeline for AsyncRedis

src/sw/redis++/async_redis.h

@@ -1017,6 +1017,10 @@ class AsyncRedis {
     }
 
 private:
+    friend class AsyncRedisCluster;
+
+    explicit AsyncRedis(const GuardedAsyncConnectionSPtr &connection);
+
     explicit AsyncRedis(const Uri &uri);
 
     template <typename Result, typename Formatter, typename ...Args>
@@ -1029,10 +1033,20 @@ class AsyncRedis {
     Future<Result> _command_with_parser(Formatter formatter, Args &&...args) {
         auto formatted_cmd = formatter(std::forward<Args>(args)...);
 
-        assert(_pool);
-        SafeAsyncConnection connection(*_pool);
+        if (_connection) {
+            // Single connection mode.
+            auto &connection = _connection->connection();
+            if (connection.broken()) {
+                throw Error("connection is broken");
+            }
+
+            return connection.send<Result, ResultParser>(std::move(formatted_cmd));
+        } else {
+            assert(_pool);
+            SafeAsyncConnection connection(*_pool);
 
-        return connection.connection().send<Result, ResultParser>(std::move(formatted_cmd));
+            return connection.connection().send<Result, ResultParser>(std::move(formatted_cmd));
+        }
     }
 
     template <typename Result, typename Callback, std::size_t ...Is, typename ...Args>
@@ -1065,16 +1079,29 @@ class AsyncRedis {
     void _callback_command_with_parser(Callback &&cb, Formatter formatter, Args &&...args) {
         auto formatted_cmd = formatter(std::forward<Args>(args)...);
 
-        assert(_pool);
-        SafeAsyncConnection connection(*_pool);
+        if (_connection) {
+            // Single connection mode.
+            auto &connection = _connection->connection();
+            if (connection.broken()) {
+                throw Error("connection is broken");
+            }
 
-        connection.connection().send<Result, ResultParser, Callback>(
-                std::move(formatted_cmd), std::forward<Callback>(cb));
+            connection.send<Result, ResultParser, Callback>(
+                    std::move(formatted_cmd), std::forward<Callback>(cb));
+        } else {
+            assert(_pool);
+            SafeAsyncConnection connection(*_pool);
+
+            connection.connection().send<Result, ResultParser, Callback>(
+                    std::move(formatted_cmd), std::forward<Callback>(cb));
+        }
     }
 
     EventLoopSPtr _loop;
 
     AsyncConnectionPoolSPtr _pool;
+
+    GuardedAsyncConnectionSPtr _connection;
 };
 
 }

src/sw/redis++/async_redis_cluster.cpp

@@ -35,6 +35,18 @@ AsyncRedisCluster::AsyncRedisCluster(const ConnectionOptions &opts,
     _pool = std::make_shared<AsyncShardsPool>(_loop, pool_opts, opts, role);
 }
 
+AsyncRedis AsyncRedisCluster::redis(const StringView &hash_tag, bool new_connection) {
+    assert(_pool);
+
+    auto pool = _pool->fetch(hash_tag);
+    if (new_connection) {
+        // Create a new pool.
+        pool = std::make_shared<AsyncConnectionPool>(pool->clone());
+    }
+
+    return AsyncRedis(std::make_shared<GuardedAsyncConnection>(pool));
+}
+
 AsyncSubscriber AsyncRedisCluster::subscriber() {
     assert(_pool);
 

src/sw/redis++/async_redis_cluster.h

@@ -21,6 +21,7 @@
 #include "sw/redis++/utils.h"
 #include "sw/redis++/async_connection.h"
 #include "sw/redis++/async_connection_pool.h"
+#include "sw/redis++/async_redis.h"
 #include "sw/redis++/async_shards_pool.h"
 #include "sw/redis++/async_subscriber.h"
 #include "sw/redis++/event_loop.h"
@@ -48,6 +49,8 @@ class AsyncRedisCluster {
 
     ~AsyncRedisCluster() = default;
 
+    AsyncRedis redis(const StringView &hash_tag, bool new_connection = true);
+
     AsyncSubscriber subscriber();
 
     template <typename Result, typename ...Args>