add AsyncRedisCluster.for_each function

src/sw/redis++/async_redis_cluster.h

@@ -55,6 +55,20 @@ class AsyncRedisCluster {
 
     AsyncSubscriber subscriber(const StringView &hash_tag);
 
+    template <typename Callback>
+    void for_each(Callback &&cb) {
+        assert(_pool);
+        _pool->update();
+
+        auto pools = _pool->pools();
+        for (auto &pool : pools) {
+            auto connection = std::make_shared<GuardedAsyncConnection>(pool);
+
+            auto ar = AsyncRedis(connection);
+            cb(ar);
+        }
+    }
+
     template <typename Result, typename ...Args>
     auto command(const StringView &cmd_name, const StringView &key, Args &&...args)
         -> typename std::enable_if<!IsInvocable<typename LastType<Args...>::type,

src/sw/redis++/async_shards_pool.cpp

@@ -114,6 +114,18 @@ ConnectionOptions AsyncShardsPool::connection_options() {
     return _connection_options(slot);
 }
 
+std::vector<AsyncConnectionPoolSPtr> AsyncShardsPool::pools() {
+    std::lock_guard<std::mutex> lock(_mutex);
+
+    std::vector<AsyncConnectionPoolSPtr> nodes;
+    nodes.reserve(_pools.size());
+    for (const auto &pool : _pools) {
+        nodes.push_back(pool.second);
+    }
+
+    return nodes;
+}
+
 ConnectionOptions AsyncShardsPool::_connection_options(Slot slot) {
     std::lock_guard<std::mutex> lock(_mutex);
 

src/sw/redis++/async_shards_pool.h

@@ -59,6 +59,8 @@ class AsyncShardsPool {
 
     ConnectionOptions connection_options();
 
+    std::vector<AsyncConnectionPoolSPtr> pools();
+
 private:
     struct RedeliverEvent {
         std::string key;