#include using namespace inaf::oasbo::Providers; RedisProvider::RedisProvider() { setIp("127.0.0.1"); setPort(6379); setKey("DAQ_key"); } RedisProvider::RedisProvider(std::string ip, int port, std::string key) { setIp(ip); setPort(port); setKey(key); } int RedisProvider::write(PacketLib::BasePacket &packet) { uint size = packet.getHeaderSize() + packet.getPayloadSize() + packet.getTailSize(); redisReply *r = (redisReply *) redisCommand(context, "LPUSH %s %b", this->key.c_str(), (char*) packet.getBinaryPointer(), size); if (r == NULL) { // Error executing Redis command return -1; } // Check the type of the Redis reply if (r->type == REDIS_REPLY_ERROR) { // Error executing Redis command std::cout << r->str << std::endl; freeReplyObject(r); return -1; } freeReplyObject(r); return size; } int RedisProvider::write(PacketLib::BasePacket &packet, std::string key) { uint size = packet.getHeaderSize() + packet.getPayloadSize() + packet.getTailSize(); redisReply *r = (redisReply *) redisCommand(context, "LPUSH %s %b", key.c_str(), (char*) packet.getBinaryPointer(), size); if (r == NULL) { // Error executing Redis command return -1; } // Check the type of the Redis reply if (r->type == REDIS_REPLY_ERROR) { // Error executing Redis command std::cout << r->str << std::endl; freeReplyObject(r); return -1; } freeReplyObject(r); return size; } int RedisProvider::open() { context = redisConnect(getIp().c_str(), getPort()); if (context == NULL || context->err) { if (context) { printf("Redis Error: %s\n", context->errstr); } else { printf("Can't allocate Redis context\n"); } } return 1; } int RedisProvider::close() { if (!(context == NULL || context->err)) redisFree(this->context); return 1; } RedisProvider::~RedisProvider(){ close(); }