From Redis List to Redis Stream

include/Redis_Receiver.h

@@ -20,6 +20,8 @@ class RedisReceiver: public BaseProtocol {
 protected:
 	std::string key;
 	redisContext *context;
+
+	bool checkRedisStream(redisReply &r);
 public:
 	RedisReceiver(std::string hs, int prt, std::string topic);
 	~RedisReceiver() {

src/Redis_Receiver.cpp

@@ -5,6 +5,7 @@
 #include <fstream>
 #include <map>
 #include <string.h>
+#include <sstream>
 
 using namespace inaf::oasbo::ConnectionProtocols;
 
@@ -12,6 +13,7 @@ RedisReceiver::RedisReceiver(std::string hs, int prt, std::string key) {
 	this->host = hs;
 	this->port = prt;
 	this->key = key;
+	this->context = NULL;
 }
 
 /*
@@ -31,9 +33,8 @@ int RedisReceiver::connectToClient() {
 	return 1;
 }
 
-int RedisReceiver::connectToServer() {
-
-	return 1;
+int RedisReceiver::connectToServer() { // no differences in implementation
+	return connectToClient();
 }
 
 /*
@@ -43,21 +44,34 @@ int RedisReceiver::connectToServer() {
  * @return number of bytes received [-1 error, 0 done]
  */
 int RedisReceiver::rcvPacketFromCli(PacketLib::BasePacket &pack) {
-	std::string cmd("RPOP ");
-	cmd.append(key);
-	redisReply *r = (redisReply *) redisCommand(context, cmd.c_str());
-	if(r->type == REDIS_REPLY_STRING){
-		int sizeRead = r->len;
-		pack.copyToBinaryPointer(reinterpret_cast<uint8_t*>(r->str), sizeRead);
+	std::stringstream cmd;
+	cmd << "XREAD BLOCK 0 STREAMS ";
+	cmd << key;
+	cmd << " $";
+	redisReply *r = (redisReply*) redisCommand(context, cmd.str().c_str());
+	if(checkRedisStream(*r)){
+		redisReply *binaryValue = r->element[0]->element[1]->element[0]->element[1]->element[1];
+		int size = binaryValue->len;
+		std::cout << "len: " << size << std::endl;
+		pack.copyToBinaryPointer(reinterpret_cast<uint8_t*>(binaryValue->str), size);
 		freeReplyObject(r);
-		return sizeRead;
+		return size;
+
 	}
+	std::cout << "FAIL" << std::endl;
+	freeReplyObject(r);
 	return -1;
 
 }
 
 int RedisReceiver::sendPacketToSrv(PacketLib::BasePacket &pack) {
 
-
 	return 0;
 }
+
+bool RedisReceiver::checkRedisStream(redisReply &r) {
+	if (r.type == REDIS_REPLY_ARRAY && r.element[0] != NULL)
+		return true;
+	return false;
+}
+