Added subscribe, unsubscribe and publish

eXo-OpenSource · Jul 27, 2019 · e99ac06 · e99ac06
1 parent a3a7838
commit e99ac06
Show file tree

Hide file tree

Showing 3 changed files with 59 additions and 4 deletions.
diff --git a/module/CFunctions.cpp b/module/CFunctions.cpp
@@ -141,8 +141,8 @@ int CFunctions::redis_test(lua_State* lua_vm)
 			lua_pushboolean(lua_vm, false);
 			return 1;
 		}
-		
-		lua_pushboolean(lua_vm, true);
+
+		lua_pushboolean(lua_vm, client->is_connected());
 		return 1;
 	} catch(std::exception& e)
 	{

diff --git a/module/RedisClient.cpp b/module/RedisClient.cpp
@@ -28,7 +28,7 @@ cpp_redis::reply redis_client::set(const std::string& key, const std::string& va
 
 void redis_client::set(const std::map<std::string, std::string>& pairs, const std::function<void(const std::string&, cpp_redis::reply &)>& callback) const
 {
-	for(auto& [key, value]: pairs)
+	for(auto& [key, value] : pairs)
 	{
 		_client->set(key, value, [key, callback](cpp_redis::reply& reply)
 		{
@@ -58,3 +58,45 @@ void redis_client::get(const std::list<std::string>& keys, const std::function<v
 	}
 	_client->commit();
 }
+
+void redis_client::subscribe(const std::list<std::string>& channels,
+	const cpp_redis::subscriber::subscribe_callback_t& callback) const
+{
+	for (auto& channel : channels)
+	{
+		_subscriber->subscribe(channel, callback);
+	}
+
+	_subscriber->commit();
+}
+
+void redis_client::unsubscribe(const std::list<std::string>& channels) const
+{
+	for (auto& channel : channels)
+	{
+		_subscriber->unsubscribe(channel);
+	}
+	_subscriber->commit();
+}
+
+cpp_redis::reply redis_client::publish(const std::string& channel, const std::string& message) const
+{
+	auto result = _client->publish(channel, message);
+	_client->commit();
+
+	result.wait();
+	return result.get();
+}
+
+void redis_client::publish(const std::map<std::string, std::string>& pairs,
+	const std::function<void(const std::string&, cpp_redis::reply&)>& callback) const
+{
+	for (auto& [channel, message] : pairs)
+	{
+		_client->publish(channel, message, [channel, callback](cpp_redis::reply& reply)
+		{
+			callback(channel, reply);
+		});
+	}
+	_client->commit();
+}
diff --git a/module/RedisClient.h b/module/RedisClient.h
@@ -12,13 +12,26 @@ class redis_client
 	void disconnect() const { _client->disconnect(); _subscriber->disconnect(); }
 	[[nodiscard]] bool is_connected() const { return _client->is_connected() && _subscriber->is_connected(); }
 
+	// Key, Value
 	[[nodiscard]] cpp_redis::reply set(const std::string& key, const std::string& value) const;
 	void set(const std::string& key, const std::string& value, const cpp_redis::client::reply_callback_t& callback) const { _client->set(key, value, callback); _client->commit(); }
-	void set(const std::map<std::string, std::string>& pairs, const std::function<void(const std::string&, cpp_redis::reply &)>& callback) const;
+	void set(const std::map<std::string, std::string>& pairs, const std::function<void(const std::string&, cpp_redis::reply&)>& callback) const;
 
 	[[nodiscard]] cpp_redis::reply get(const std::string& key) const;
 	void get(const std::string& key, const cpp_redis::client::reply_callback_t& callback) const { _client->get(key, callback); _client->commit(); };
 	void get(const std::list<std::string>& keys, const std::function<void(const std::string&, cpp_redis::reply &)>& callback) const;
+
+	// Channels
+	void subscribe(const std::string& channel, const cpp_redis::subscriber::subscribe_callback_t& callback) const { _subscriber->subscribe(channel, callback); _subscriber->commit(); }
+	void subscribe(const std::list<std::string>& channels, const cpp_redis::subscriber::subscribe_callback_t& callback) const;
+
+	void unsubscribe(const std::string& channel) const { _subscriber->unsubscribe(channel); _subscriber->commit(); }
+	void unsubscribe(const std::list<std::string>& channels) const;
+
+	[[nodiscard]] cpp_redis::reply publish(const std::string& channel, const std::string& message) const;
+	void publish(const std::string& channel, const std::string& message, cpp_redis::client::reply_callback_t& callback) const { _client->publish(channel, message, callback); _client->commit(); }
+	void publish(const std::map<std::string, std::string>& pairs, const std::function<void(const std::string&, cpp_redis::reply&)>& callback) const;
+
 private:
 	cpp_redis::client* _client;
 	cpp_redis::subscriber* _subscriber;