Work on AsyncClient and add execute<T> method

Redis/Makefile

@@ -10,7 +10,7 @@ include $(POCO_BASE)/build/rules/global
 
 INCLUDE += -I $(POCO_BASE)/Redis/include/Poco/Redis
 
-objects = Array Client Error Exception RedisSocket Type
+objects = AsyncClient Array Client Error Exception RedisSocket Type
 
 target         = PocoRedis
 target_version = $(LIBVERSION)

Redis/include/Poco/Redis/AsyncClient.h

@@ -0,0 +1,98 @@
+//
+// AsyncClient.h
+//
+// $Id$
+//
+// Library: Redis
+// Package: Redis
+// Module:  AsyncClient
+//
+// Definition of the AsyncClient class.
+//
+// Copyright (c) 2012, Applied Informatics Software Engineering GmbH.
+// and Contributors.
+//
+// SPDX-License-Identifier:	BSL-1.0
+//
+
+
+#ifndef Redis_AsyncClient_INCLUDED
+#define Redis_AsyncClient_INCLUDED
+
+
+#include "Poco/Redis/Redis.h"
+#include "Poco/Redis/Client.h"
+#include "Poco/Activity.h"
+
+namespace Poco {
+namespace Redis {
+
+
+class Redis_API AsyncClient : public Client
+{
+public:
+
+
+	BasicEvent<RedisType::Ptr> redisResponse;
+
+
+	AsyncClient();
+		/// Default constructor. Use this when you want to
+		/// connect later on.
+
+	AsyncClient(const std::string& hostAndPort);
+		/// Constructor which connects to the given Redis host/port.
+		/// The host and port must be separated with a colon.
+
+	AsyncClient(const std::string& host, int port);
+		/// Constructor which connects to the given Redis host/port.
+
+	AsyncClient(const Net::SocketAddress& addrs);
+		/// Constructor which connects to the given Redis host/port.
+
+	virtual ~AsyncClient();
+		/// Destructor
+
+	bool isStopped();
+		/// Returns true if the activity is not running, false when it is.
+
+	void start();
+		/// Starts the activity to read replies from the Redis server.
+
+	void stop();
+		/// Stops the read activity.
+
+protected:
+
+	void runActivity();
+
+private:
+
+	AsyncClient(const AsyncClient&);
+	AsyncClient& operator = (const AsyncClient&);
+
+
+	Activity<AsyncClient> _activity;
+
+};
+
+
+inline bool AsyncClient::isStopped()
+{
+	return _activity.isStopped();
+}
+
+inline void AsyncClient::start()
+{
+	_activity.start();
+}
+
+inline void AsyncClient::stop()
+{
+	_activity.stop();
+}
+
+
+} } // namespace Poco::Redis
+
+#endif //Redis_Client_INCLUDED

Redis/include/Poco/Redis/Client.h

@@ -16,11 +16,11 @@
 //
 
 
-#ifndef Redis_Connection_INCLUDED
+#ifndef Redis_Client_INCLUDED
-#define Redis_Connection_INCLUDED
+#define Redis_Client_INCLUDED
-
 
 #include "Poco/Net/SocketAddress.h"
+#include "Poco/Timespan.h"
 
 #include "Poco/Redis/Redis.h"
 #include "Poco/Redis/Array.h"
@@ -41,7 +41,7 @@ class Redis_API Client
 	/// implemented as a typedef for Poco::Nullable<std::string>. This is
 	/// because a bulk string can represent a Null value.
 	///
-	///   BulkString bs = client.sendCommand(...);
+	///   BulkString bs = client.execute<BulkString>(...);
 	///   if ( bs.isNull() )
 	///   {
 	///      // We have a Null value
@@ -84,24 +84,41 @@ public:
 	void connect(const Net::SocketAddress& addrs);
 		/// Connects to the given Redis server.
 
+	void connect(const std::string& hostAndPort, const Timespan& timeout);
+		/// Connects to the given Redis server. The host and port must be
+		/// separated with a colon.
+
+	void connect(const std::string& host, int port, const Timespan& timeout);
+		/// Connects to the given Redis server.
+
+	void connect(const Net::SocketAddress& addrs, const Timespan& timeout);
+		/// Connects to the given Redis server.
+
 	void disconnect();
 		/// Disconnects from the Redis server.
 
-	RedisType::Ptr sendCommand(const Array& command);
-		/// Sends a Redis command to the server and returns the reply
-
 	template<typename T>
-	void sendCommand(const Array& command, T& result)
+	T execute(const Array& command)
-		/// Sends a Redis command to the server, gets the reply and tries
+		/// Sends the Redis Command to the server. It gets the reply
-		/// to convert that reply to the template type. When
+		/// and tries to convert it to the given template type.
-		/// the reply is a Redis error, it wil throw a RedisException.
+		/// A specialization exists for type void, which doesn't read
-		/// A BadCastException will be thrown, when the reply is not of the
+		/// the reply. If the server sends a reply, it is your
-		/// given type.
+		/// responsibility to read it ... (Use this for pipelining)
+		/// A BadCastException will be thrown when the reply couldn't be 
+		/// converted. Supported types are Int64, std::string, BulkString, 
+		/// Array and void. When the reply is an Error, it will throw 
+		/// a RedisException.
 	{
+		T result;
 		writeCommand(command);
 		readReply(result);
+		return result;
 	}
 
+	RedisType::Ptr sendCommand(const Array& command);
+		/// Sends a Redis command to the server and returns the reply.
+		/// Use this when the type of the reply isn't known.
+
 	RedisType::Ptr readReply();
 		/// Read a reply from the Redis server.
 
@@ -122,16 +139,10 @@ public:
 		else throw BadCastException();
 	}
 
-	void sendCommands(const std::vector<Array>& commands, Array& results);
+	Array sendCommands(const std::vector<Array>& commands);
 		/// Sends all commands (pipelining) to the Redis server before
 		/// getting all replies.
 
-	void writeCommand(const Array& command);
-		/// Sends a request to the Redis server. Use readReply to get the
-		/// answer. Can also be used for pipelining commands. Make sure you
-		/// call readReply as many times as you called writeCommand, even when
-		/// an error occurred on a command.
-
 private:
 
 	Client(const Client&);
@@ -142,6 +153,15 @@ private:
 
 	void connect();
 		/// Connects to the Redis server
+	void connect(const Timespan& timeout);
+		/// Connects to the Redis server and sets a timeout.
+
+	void writeCommand(const Array& command);
+		/// Sends a request to the Redis server. Use readReply to get the
+		/// answer. Can also be used for pipelining commands. Make sure you
+		/// call readReply as many times as you called writeCommand, even when
+		/// an error occurred on a command.
+		
 };
 
 
@@ -150,8 +170,14 @@ inline Net::SocketAddress Client::address() const
 	return _address;
 }
 
+template<> inline
+void Client::execute<void>(const Array& command)
+{
+	writeCommand(command);
+}
+
 
 } } // namespace Poco::Redis
 
 
-#endif //Redis_Connection_INCLUDED
+#endif //Redis_Client_INCLUDED

Redis/include/Poco/Redis/RedisSocket.h

@@ -18,6 +18,7 @@
 #ifndef Redis_RedisSocket_INCLUDED
 #define Redis_RedisSocket_INCLUDED
 
+#include "Poco/Timespan.h"
 #include "Poco/Net/SocketAddress.h"
 #include "Poco/Net/StreamSocket.h"
 
@@ -37,6 +38,8 @@ public:
 
 	void connect(const Net::SocketAddress& addrs);
 
+	void connect(const Net::SocketAddress& addrs, const Timespan& timeout);
+
 	int get();
 
 	int peek();
@@ -45,8 +48,6 @@ public:
 
 	int write(const char* buffer, std::streamsize length);
 
-	int buffered();
-
 	void refill();
 
 	void readLine(std::string& line);
@@ -63,6 +64,7 @@ private:
 	char* _end;
 };
 
+
 } }
 
 #endif // Redis_RedisSocket_INCLUDED

Redis/src/AsyncClient.cpp

@@ -0,0 +1,75 @@
+//
+// AsyncClient.cpp
+//
+// $Id$
+//
+// Library: Redis
+// Package: Redis
+// Module:  AsyncClient
+//
+// Implementation of the AsyncClient class.
+//
+// Copyright (c) 2012, Applied Informatics Software Engineering GmbH.
+// and Contributors.
+//
+// SPDX-License-Identifier:	BSL-1.0
+//
+
+#include "Poco/Redis/AsyncClient.h"
+
+namespace Poco {
+namespace Redis {
+
+
+AsyncClient::AsyncClient() : Client(),
+	_activity(this, &AsyncClient::runActivity)
+{
+}
+
+
+AsyncClient::AsyncClient(const std::string& hostAndPort) : Client(hostAndPort),
+	_activity(this, &AsyncClient::runActivity)
+{
+}
+
+
+AsyncClient::AsyncClient(const std::string& host, int port) : Client(host, port),
+	_activity(this, &AsyncClient::runActivity)
+{
+}
+
+
+AsyncClient::AsyncClient(const Net::SocketAddress& addrs) : Client(addrs),
+	_activity(this, &AsyncClient::runActivity)
+{
+}
+
+
+AsyncClient::~AsyncClient()
+{
+	stop();
+}
+
+
+void AsyncClient::runActivity()
+{
+	while(!_activity.isStopped())
+	{
+		try
+		{
+			RedisType::Ptr reply = readReply();
+			redisResponse.notify(this, reply);
+		}
+		catch(TimeoutException&)
+		{
+			continue;
+		}
+		catch(Exception &)
+		{
+			stop();
+		}
+	}
+}
+
+
+} } // Poco::Redis

Redis/src/Client.cpp

@@ -55,7 +55,6 @@ void Client::connect()
 	_socket.connect(_address);
 }
 
-
 void Client::connect(const std::string& hostAndPort)
 {
 	_address = Net::SocketAddress(hostAndPort);
@@ -76,6 +75,30 @@ void Client::connect(const Net::SocketAddress& addrs)
 	connect();
 }
 
+void Client::connect(const Timespan& timeout)
+{
+	_socket.connect(_address, timeout);
+}
+
+void Client::connect(const std::string& hostAndPort, const Timespan& timeout)
+{
+	_address = Net::SocketAddress(hostAndPort);
+	connect(timeout);
+}
+
+
+void Client::connect(const std::string& host, int port, const Timespan& timeout)
+{
+	_address = Net::SocketAddress(host, port);
+	connect(timeout);
+}
+
+
+void Client::connect(const Net::SocketAddress& addrs, const Timespan& timeout)
+{
+	_address = addrs;
+	connect(timeout);
+}
 
 void Client::disconnect()
 {
@@ -107,8 +130,10 @@ RedisType::Ptr Client::sendCommand(const Array& command)
 	return readReply();
 }
 
-void Client::sendCommands(const std::vector<Array>& commands, Array& results)
+Array Client::sendCommands(const std::vector<Array>& commands)
 {
+	Array results;
+
 	for(std::vector<Array>::const_iterator it = commands.begin(); it != commands.end(); ++it)
 	{
 		writeCommand(*it);
@@ -118,7 +143,8 @@ void Client::sendCommands(const std::vector<Array>& commands, Array& results)
 	{
 		results.add(readReply());
 	}
+
+	return results;
 }
 
-
 } } // Poco::Redis

Redis/src/RedisSocket.cpp

@@ -46,6 +46,12 @@ void RedisSocket::connect(const Net::SocketAddress& addrs)
 	_socket.connect(addrs);
 }
 
+void RedisSocket::connect(const Net::SocketAddress& addrs, const Timespan& timeout)
+{
+	_socket.connect(addrs, timeout);
+}
+
+
 int RedisSocket::get()
 {
 	if ( _current == _end ) refill();

Redis/testsuite/src/RedisTest.cpp

@@ -10,13 +10,11 @@
 //
 #include <iostream>
 
-#include "Poco/DateTime.h"
-#include "Poco/ObjectPool.h"
-
 #include "Poco/Net/NetException.h"
+#include "Poco/Delegate.h"
+#include "Poco/Thread.h"
 
 #include "RedisTest.h"
-#include "Poco/Redis/Client.h"
 
 #include "CppUnit/TestCaller.h"
 #include "CppUnit/TestSuite.h"
@@ -25,7 +23,7 @@ using namespace Poco::Redis;
 
 
 bool RedisTest::_connected = false;
-Poco::Redis::Client RedisTest::_redis;
+Poco::Redis::AsyncClient RedisTest::_redis;
 
 
 RedisTest::RedisTest(const std::string& name):
@@ -37,7 +35,8 @@ RedisTest::RedisTest(const std::string& name):
 	{
 		try
 		{
-			_redis.connect(_host, _port);
+			Poco::Timespan t(30, 0); // 30 seconds
+			_redis.connect(_host, _port, t);
 			_connected = true;
 			std::cout << "Connected to [" << _host << ':' << _port << ']' << std::endl;
 		}
@@ -84,8 +83,7 @@ void RedisTest::testAppend()
 		.add("mykey");
 	try
 	{
-		Poco::Int64 result;
+		_redis.execute<Poco::Int64>(delCommand);
-		_redis.sendCommand(delCommand, result);
 	}
 	catch(RedisException& e)
 	{
@@ -102,8 +100,7 @@ void RedisTest::testAppend()
 		.add("Hello");
 	try
 	{
-		std::string result;
+		std::string result = _redis.execute<std::string>(setCommand);
-		_redis.sendCommand(setCommand, result);
 		assert(result.compare("OK") == 0);
 	}
 	catch(RedisException& e)
@@ -121,9 +118,7 @@ void RedisTest::testAppend()
 		.add(" World");
 	try
 	{
-		Poco::Int64 result;
+		Poco::Int64 result = _redis.execute<Poco::Int64>(appendCommand);
-		_redis.sendCommand(appendCommand, result);
-
 		assert(result == 11);
 	}
 	catch(RedisException& e)
@@ -140,9 +135,7 @@ void RedisTest::testAppend()
 		.add("mykey");
 	try
 	{
-		BulkString result;
+		BulkString result = _redis.execute<BulkString>(getCommand);
-		_redis.sendCommand(getCommand, result);
-
 		assert(result.value().compare("Hello World") == 0);
 	}
 	catch(RedisException& e)
@@ -169,9 +162,7 @@ void RedisTest::testEcho()
 
 	try
 	{
-		BulkString result;
+		BulkString result = _redis.execute<BulkString>(command);
-		_redis.sendCommand(command, result);
-
 		assert(!result.isNull());
 		assert(result.value().compare("Hello World") == 0);
 	}
@@ -197,9 +188,7 @@ void RedisTest::testIncr()
 	// A set responds with a simple OK string
 	try
 	{
-		std::string result;
+		std::string result = _redis.execute<std::string>(command);
-		_redis.sendCommand(command, result);
-
 		assert(result.compare("OK") == 0);
 	}
 	catch(RedisException &e)
@@ -213,9 +202,7 @@ void RedisTest::testIncr()
 
 	try
 	{
-		Poco::Int64 value;
+		Poco::Int64 value = _redis.execute<Poco::Int64>(command);
-		_redis.sendCommand(command, value);
-
 		assert(value == 11);
 	}
 	catch(RedisException &e)
@@ -240,9 +227,7 @@ void RedisTest::testIncrBy()
 	// A set responds with a simple OK string
 	try
 	{
-		std::string result;
+		std::string result = _redis.execute<std::string>(command);
-		_redis.sendCommand(command, result);
-
 		assert(result.compare("OK") == 0);
 	}
 	catch(RedisException &e)
@@ -257,9 +242,7 @@ void RedisTest::testIncrBy()
 
 	try
 	{
-		Poco::Int64 value;
+		Poco::Int64 value = _redis.execute<Poco::Int64>(command);
-		_redis.sendCommand(command, value);
-
 		assert(value == 15);
 	}
 	catch(RedisException &e)
@@ -282,9 +265,7 @@ void RedisTest::testPing()
 	// A PING without a custom strings, responds with a simple "PONG" string
 	try
 	{
-		std::string result;
+		std::string result = _redis.execute<std::string>(command);
-		_redis.sendCommand(command, result);
-
 		assert(result.compare("PONG") == 0);
 	}
 	catch(RedisException &e)
@@ -296,9 +277,7 @@ void RedisTest::testPing()
 	command.add("Hello");
 	try
 	{
-		BulkString result;
+		BulkString result = _redis.execute<BulkString>(command);
-		_redis.sendCommand(command, result);
-
 		assert(!result.isNull());
 		assert(result.value().compare("Hello") == 0);
 	}
@@ -325,9 +304,7 @@ void RedisTest::testSet()
 	// A set responds with a simple OK string
 	try
 	{
-		std::string result;
+		std::string result = _redis.execute<std::string>(command);
-		_redis.sendCommand(command, result);
-
 		assert(result.compare("OK") == 0);
 	}
 	catch(RedisException &e)
@@ -340,9 +317,7 @@ void RedisTest::testSet()
 	// when the key is already set
 	try
 	{
-		BulkString result;
+		BulkString result = _redis.execute<BulkString>(command);
-		_redis.sendCommand(command, result);
-
 		assert(result.isNull());
 	}
 	catch(RedisException &e)
@@ -369,9 +344,7 @@ void RedisTest::testMSet()
 	// A MSET responds with a simple OK string
 	try
 	{
-		std::string result;
+		std::string result = _redis.execute<std::string>(command);
-		_redis.sendCommand(command, result);
-
 		assert(result.compare("OK") == 0);
 	}
 	catch(RedisException &e)
@@ -386,8 +359,7 @@ void RedisTest::testMSet()
 		.add("nonexisting");
 	try
 	{
-		Array result;
+		Array result = _redis.execute<Array>(command);
-		_redis.sendCommand(command, result);
 
 		assert(result.size() == 3);
 		BulkString value = result.get<BulkString>(0);
@@ -425,9 +397,7 @@ void RedisTest::testStrlen()
 	// A set responds with a simple OK string
 	try
 	{
-		std::string result;
+		std::string result = _redis.execute<std::string>(command);
-		_redis.sendCommand(command, result);
-
 		assert(result.compare("OK") == 0);
 	}
 	catch(RedisException &e)
@@ -441,8 +411,7 @@ void RedisTest::testStrlen()
 
 	try
 	{
-		Poco::Int64 result;
+		Poco::Int64 result = _redis.execute<Poco::Int64>(command);
-		_redis.sendCommand(command, result);
 
 		assert(result == 11);
 	}
@@ -466,8 +435,7 @@ void RedisTest::testRPush()
 		.add("mylist");
 	try
 	{
-		Poco::Int64 result;
+		_redis.execute<Poco::Int64>(delCommand);
-		_redis.sendCommand(delCommand, result);
 	}
 	catch(RedisException& e)
 	{
@@ -490,9 +458,7 @@ void RedisTest::testRPush()
 		// A RPUSH responds with an integer
 		try
 		{
-			Poco::Int64 result;
+			Poco::Int64 result = _redis.execute<Poco::Int64>(command);
-			_redis.sendCommand(command, result);
-
 			assert(result == (i + 1));
 		}
 		catch(RedisException &e)
@@ -509,8 +475,7 @@ void RedisTest::testRPush()
 
 	try
 	{
-		Array result;
+		Array result = _redis.execute<Array>(command);
-		_redis.sendCommand(command, result);
 
 		assert(result.size() == 2);
 		BulkString value = result.get<BulkString>(0);
@@ -539,8 +504,7 @@ void RedisTest::testLIndex()
 		.add("mylist");
 	try
 	{
-		Poco::Int64 result;
+		_redis.execute<Poco::Int64>(delCommand);
-		_redis.sendCommand(delCommand, result);
 	}
 	catch(RedisException& e)
 	{
@@ -563,9 +527,7 @@ void RedisTest::testLIndex()
 		// A RPUSH responds with an integer
 		try
 		{
-			Poco::Int64 result;
+			Poco::Int64 result = _redis.execute<Poco::Int64>(command);
-			_redis.sendCommand(command, result);
-
 			assert(result == (i + 1));
 		}
 		catch(RedisException &e)
@@ -581,9 +543,7 @@ void RedisTest::testLIndex()
 
 	try
 	{
-		BulkString result;
+		BulkString result = _redis.execute<BulkString>(command);
-		_redis.sendCommand(command, result);
-
 		assert(result.value().compare("Hello") == 0);
 	}
 	catch(RedisException &e)
@@ -607,8 +567,7 @@ void RedisTest::testMulti()
 		.add("bar");
 	try
 	{
-		Poco::Int64 result;
+		_redis.execute<Poco::Int64>(delCommand);
-		_redis.sendCommand(delCommand, result);
 	}
 	catch(RedisException& e)
 	{
@@ -623,8 +582,7 @@ void RedisTest::testMulti()
 	command.add("MULTI");
 	try
 	{
-		std::string result;
+		std::string result = _redis.execute<std::string>(command);
-		_redis.sendCommand(command, result);
 		assert(result.compare("OK") == 0);
 	}
 	catch(RedisException& e)
@@ -641,8 +599,7 @@ void RedisTest::testMulti()
 		.add("foo");
 	try
 	{
-		std::string result;
+		std::string result = _redis.execute<std::string>(command);
-		_redis.sendCommand(command, result);
 		assert(result.compare("QUEUED") == 0);
 	}
 	catch(RedisException& e)
@@ -659,8 +616,7 @@ void RedisTest::testMulti()
 		.add("bar");
 	try
 	{
-		std::string result;
+		std::string result = _redis.execute<std::string>(command);
-		_redis.sendCommand(command, result);
 		assert(result.compare("QUEUED") == 0);
 	}
 	catch(RedisException& e)
@@ -676,8 +632,7 @@ void RedisTest::testMulti()
 	command.add("EXEC");
 	try
 	{
-		Array result;
+		Array result = _redis.execute<Array>(command);
-		_redis.sendCommand(command, result);
 		assert(result.size() == 2);
 
 		Poco::Int64 v = result.get<Poco::Int64>(0);
@@ -710,8 +665,7 @@ void RedisTest::testPipeliningWithSendCommands()
 	commands.push_back(ping);
 	commands.push_back(ping);
 
-	Array result;
+	Array result = _redis.sendCommands(commands); 
-	_redis.sendCommands(commands, result);
 
 	// We expect 2 results
 	assert(result.size() == 2);
@@ -719,7 +673,7 @@ void RedisTest::testPipeliningWithSendCommands()
 	// The 2 results must be simple PONG strings
 	for(size_t i = 0; i < 2; ++i)
 	{
-		try
+		try 
 		{
 			std::string pong = result.get<std::string>(i);
 			assert(pong.compare("PONG") == 0);
@@ -742,14 +696,14 @@ void RedisTest::testPipeliningWithWriteCommand()
 	Array ping;
 	ping.add("PING");
 
-	_redis.writeCommand(ping);
+	_redis.execute<void>(ping);
-	_redis.writeCommand(ping);
+	_redis.execute<void>(ping);
 
 	// We expect 2 results with simple "PONG" strings
 	for(int i = 0; i < 2; ++i)
 	{
 		std::string pong;
-		try
+		try 
 		{
 			_redis.readReply<std::string>(pong);
 			assert(pong.compare("PONG") == 0);
@@ -761,6 +715,41 @@ void RedisTest::testPipeliningWithWriteCommand()
 	}
 }
 
+class RedisSubscriber
+{
+public:
+
+	void onMessage(const void* pSender, RedisType::Ptr& message)
+	{
+		std::cout << message->toString() << std::endl;
+	}
+
+};
+
+void RedisTest::testPubSub()
+{
+	RedisSubscriber subscriber;
+
+	Array subscribe;
+	subscribe.add("SUBSCRIBE")
+		.add("test");
+
+	Array subscribeReply = _redis.execute<Array>(subscribe);
+
+	_redis.redisResponse += Poco::delegate(&subscriber, &RedisSubscriber::onMessage);
+	_redis.start();
+
+	Poco::Thread::sleep(30000);
+
+	Array unsubscribe;
+	unsubscribe.add("UNSUBSCRIBE");
+
+	Array unsubscribeReply = _redis.execute<Array>(unsubscribe);
+	std::cout << "SUBS: " << unsubscribeReply.toString() << std::endl;
+
+	_redis.stop();
+}
+
 CppUnit::Test* RedisTest::suite()
 {
 	CppUnit::TestSuite* pSuite = new CppUnit::TestSuite("RedisTest");
@@ -777,6 +766,8 @@ CppUnit::Test* RedisTest::suite()
 	CppUnit_addTest(pSuite, RedisTest, testLIndex);
 	CppUnit_addTest(pSuite, RedisTest, testMulti);
 
+	CppUnit_addTest(pSuite, RedisTest, testPubSub);
+
 	CppUnit_addTest(pSuite, RedisTest, testPipeliningWithSendCommands);
 	CppUnit_addTest(pSuite, RedisTest, testPipeliningWithWriteCommand);
 

Redis/testsuite/src/RedisTest.h

@@ -17,7 +17,7 @@
 
 
 #include "Poco/Redis/Redis.h"
-#include "Poco/Redis/Client.h"
+#include "Poco/Redis/AsyncClient.h"
 
 #include "CppUnit/TestCase.h"
 
@@ -41,6 +41,9 @@ public:
 	void testRPush();
 	void testLIndex();
 	void testMulti();
+
+	void testPubSub();
+
 	void testPipeliningWithSendCommands();
 	void testPipeliningWithWriteCommand();
 
@@ -54,7 +57,7 @@ private:
 	std::string _host;
 	unsigned    _port;
 	static bool _connected;
-	static Poco::Redis::Client _redis;
+	static Poco::Redis::AsyncClient _redis;
 
 };