/* imhiredis.c
* Copyright 2021 aDvens
*
* This file is contrib for rsyslog.
* This input plugin is a log consumer from REDIS
* See README for doc
*
*
* This program is free software: you can redistribute it and/or
* modify it under the terms of the GNU Lesser General Public License
* as published by the Free Software Foundation, either version 3 of
* the License, or (at your option) any later version.
*
* This program is distributed in the hope that it will be useful, but
* WITHOUT ANY WARRANTY; without even the implied warranty of
* MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the GNU
* Lesser General Public License for more details.
*
* You should have received a copy of the GNU Lesser General Public
* License along with this program. If not, see
* <http://www.gnu.org/licenses/>.
*
* Author: Jérémie Jourdin
* <jeremie.jourdin@advens.fr>
*/

#include "config.h"
#include <stdio.h>
#include <stdarg.h>
#include <stdlib.h>
#include <string.h>
#include <assert.h>
#include <errno.h>
#include <fcntl.h>
#include <pthread.h>
#include <sys/uio.h>
#include <hiredis/hiredis.h>
#include <hiredis/async.h>
#include <hiredis/adapters/libevent.h>
#include <event2/thread.h>

#include "rsyslog.h"
#include "conf.h"
#include "syslogd-types.h"
#include "srUtils.h"
#include "template.h"
#include "module-template.h"
#include "errmsg.h"
#include "atomic.h"
#include "statsobj.h"
#include "unicode-helper.h"
#include "prop.h"
#include "ruleset.h"
#include "glbl.h"
#include "cfsysline.h"
#include "msg.h"
#include "dirty.h"

MODULE_TYPE_INPUT
MODULE_TYPE_NOKEEP
MODULE_CNFNAME("imhiredis")

/* static data */
DEF_IMOD_STATIC_DATA
#define QUEUE_BATCH_SIZE 10
#define IMHIREDIS_MODE_QUEUE 1
#define IMHIREDIS_MODE_SUBSCRIBE 2
DEFobjCurrIf(prop)
DEFobjCurrIf(ruleset)
DEFobjCurrIf(glbl)
DEFobjCurrIf(statsobj)


typedef struct redisNode_s {
	sbool isMaster;
	sbool usesSocket;
	uchar *socketPath;
	uchar *server;
	int port;
	struct redisNode_s *next;
} redisNode;


struct instanceConf_s {
	uchar *password;
	uchar *key;
	uchar *modeDescription;
	int mode;
	sbool useLPop;
	ruleset_t *pBindRuleset;	/* ruleset to bind listener to (use system default if unspecified) */
	uchar *pszBindRuleset;		/* default name of Ruleset to bind to */

	redisContext *conn;
	redisAsyncContext *aconn;
	struct event_base *evtBase;

	redisNode *currentNode; /* currently used redis node, can be any of the nodes in the redisNodesList list */
	/* the list of seen nodes
	 * the preferred node (the one from configuration) will always be the first element
	 * second one is a master (if preferred one is unavailable/replica) or a replica, others are replicas
	 * the preferred node may appear twice, but it is accepted
	 */
	redisNode *redisNodesList;

	struct instanceConf_s *next;
};


struct modConfData_s {
	rsconf_t *pConf;		/* our overall config object */
	instanceConf_t *root, *tail;
};

/* The following structure controls the worker threads. Global data is
 * needed for their access.
 */
static struct imhiredisWrkrInfo_s {
	pthread_t tid;		/* the worker's thread ID */
	instanceConf_t *inst;	/* Pointer to imhiredis instance */
} *imhiredisWrkrInfo;


/* GLOBAL DATA */
pthread_attr_t wrkrThrdAttr;	/* Attribute for worker threads ; read only after startup */

static int activeHiredisworkers = 0;
static char *redis_replies[] = {"unknown", "string", "array", "integer", "nil", "status", "error"};

static modConfData_t *loadModConf = NULL;/* modConf ptr to use for the current load process */
static modConfData_t *runModConf = NULL;/* modConf ptr to use for the current load process */

static prop_t *pInputName = NULL;
/* there is only one global inputName for all messages generated by this input */


/* module-global parameters */
static struct cnfparamdescr modpdescr[] = {};
static struct cnfparamblk modpblk =
	{ CNFPARAMBLK_VERSION,
	  sizeof(modpdescr)/sizeof(struct cnfparamdescr),
	  modpdescr
	};

/* input instance parameters */
static struct cnfparamdescr inppdescr[] = {
	{ "socketPath", eCmdHdlrGetWord, 0 },
	{ "server", eCmdHdlrGetWord, 0 },
	{ "port", eCmdHdlrInt, 0 },
	{ "password", eCmdHdlrGetWord, 0 },
	{ "mode", eCmdHdlrGetWord, 0 },
	{ "key", eCmdHdlrGetWord, 0 },
	{ "uselpop", eCmdHdlrBinary, 0 },
	{ "ruleset", eCmdHdlrString, 0 },
};
static struct cnfparamblk inppblk =
	{ CNFPARAMBLK_VERSION,
	  sizeof(inppdescr)/sizeof(struct cnfparamdescr),
	  inppdescr
	};

struct timeval glblRedisConnectTimeout = { 3, 0 }; /* 3 seconds */


#include "im-helper.h" /* must be included AFTER the type definitions! */


/* forward references */
static void redisAsyncRecvCallback (redisAsyncContext __attribute__((unused)) *c, void *reply, void *inst_obj);
static void redisAsyncConnectCallback (const redisAsyncContext *c, int status);
static void redisAsyncDisconnectCallback (const redisAsyncContext *c, int status);
static rsRetVal enqMsg(instanceConf_t *const inst, const char *message);
rsRetVal redisAuthentSynchronous(redisContext *conn, uchar *password);
rsRetVal redisAuthentAsynchronous(redisAsyncContext *aconn, uchar *password);
rsRetVal redisActualizeCurrentNode(instanceConf_t *inst);
rsRetVal redisGetServersList(redisNode *node, uchar *password, redisNode **result);
rsRetVal redisAuthenticate(instanceConf_t *inst);
rsRetVal redisConnectSync(redisContext **conn, redisNode *node);
rsRetVal redisConnectAsync(redisAsyncContext **aconn, redisNode *node);
rsRetVal redisDequeue(instanceConf_t *inst);
void workerLoopSubscribe(struct imhiredisWrkrInfo_s *me);
void workerLoopQueue(struct imhiredisWrkrInfo_s *me);
static void *imhirediswrkr(void *myself);
static rsRetVal createRedisNode(redisNode **root);
rsRetVal copyNode(redisNode *src, redisNode **dst);
redisNode *freeNode(redisNode *node);
void insertNodeAfter(redisNode *root, redisNode *elem);
void dbgPrintNode(redisNode *node);


/* create input instance, set default parameters, and
 * add it to the list of instances.
 */
static rsRetVal
createInstance(instanceConf_t **pinst)
{
	DEFiRet;
	instanceConf_t *inst;
	CHKmalloc(inst = malloc(sizeof(instanceConf_t)));

	inst->next = NULL;
	inst->password = NULL;
	inst->key = NULL;
	inst->mode = 0;
	inst->useLPop = 0;
	inst->pszBindRuleset = NULL;
	inst->pBindRuleset = NULL;

	/* Redis objects */
	inst->conn = NULL;
	inst->aconn = NULL;

	/* redis nodes list */
	CHKiRet(createRedisNode(&(inst->redisNodesList)));
	inst->currentNode = inst->redisNodesList;

	/* libevent base for async connection */
	inst->evtBase = NULL;

	/* node created, let's add to config */
	if(loadModConf->tail == NULL) {
		loadModConf->tail = loadModConf->root = inst;
	} else {
		loadModConf->tail->next = inst;
		loadModConf->tail = inst;
	}

	*pinst = inst;
finalize_it:
	RETiRet;
}

/* this function checks instance parameters and does some required pre-processing
 */
static rsRetVal ATTR_NONNULL()
checkInstance(instanceConf_t *const inst)
{
	DEFiRet;
	/* first node should be created from configuration */
	assert(inst->redisNodesList != NULL);

	/* check and print redis connection settings */
	if (inst->redisNodesList->server != NULL && inst->redisNodesList->socketPath != NULL) {
		LogMsg(0, RS_RET_CONFIG_ERROR, LOG_WARNING,"imhiredis: both 'server' and 'socketPath' are given, "
								"ignoring 'socketPath'.");
		free(inst->redisNodesList->socketPath);
		inst->redisNodesList->socketPath = NULL;
	}

	if(inst->redisNodesList->server != NULL && inst->redisNodesList->server[0] != '\0') {
		if (inst->redisNodesList->port == 0) {
			LogMsg(0, RS_RET_OK_WARN, LOG_WARNING, "imhiredis: port not set, setting default 6379");
			inst->redisNodesList->port = 6379;
		}
		DBGPRINTF("imhiredis: preferred server is %s (%d)\n",
			inst->redisNodesList->server,
			inst->redisNodesList->port);
		inst->redisNodesList->usesSocket = 0;
	}
	else if(inst->redisNodesList->socketPath != NULL && inst->redisNodesList->socketPath[0] != '\0') {
		DBGPRINTF("imhiredis: preferred server is %s\n",
			inst->redisNodesList->socketPath);
		inst->redisNodesList->usesSocket = 1;
	} else {
		LogError(0, RS_RET_CONFIG_ERROR, "imhiredis: neither 'server' nor 'socketPath' are defined!");
		ABORT_FINALIZE(RS_RET_CONFIG_ERROR);
	}

	if (inst->key != NULL) {
		DBGPRINTF("imhiredis: key/channel is '%s'\n", inst->key);
	} else {
		LogError(0, RS_RET_CONFIG_ERROR, "imhiredis: no key defined !");
		ABORT_FINALIZE(RS_RET_CONFIG_ERROR);
	}

	if (inst->mode != IMHIREDIS_MODE_QUEUE && inst->mode != IMHIREDIS_MODE_SUBSCRIBE) {
		LogError(0, RS_RET_CONFIG_ERROR, "imhiredis: invalid mode, please choose 'subscribe' or 'queue' mode.");
		ABORT_FINALIZE(RS_RET_CONFIG_ERROR);
	}

	if (inst->mode == IMHIREDIS_MODE_SUBSCRIBE && inst->useLPop) {
		LogMsg(0, RS_RET_CONFIG_ERROR, LOG_WARNING,"imhiredis: 'uselpop' set with mode = subscribe : ignored.");

	}

	if (inst->password != NULL) {
		DBGPRINTF("imhiredis: password is '%s'\n", inst->password);
	}

	// set default current node as first node in list (preferred node)
	inst->currentNode = inst->redisNodesList;
	// search master node (should be either preferred node or its master)
	if (RS_RET_OK != redisActualizeCurrentNode(inst) || inst->currentNode == NULL) {
		LogMsg(0, RS_RET_REDIS_ERROR, LOG_WARNING, "imhiredis: could not connect to a valid master!");
	}

finalize_it:
	RETiRet;
}

/* function to generate an error message if the ruleset cannot be found */
static inline void
std_checkRuleset_genErrMsg(__attribute__((unused)) modConfData_t *modConf, instanceConf_t *inst)
{
	LogError(0, NO_ERRCODE, "imhiredis: ruleset '%s' not found - "
		"using default ruleset instead",
		inst->pszBindRuleset);
}


BEGINnewInpInst
	struct cnfparamvals *pvals;
	instanceConf_t *inst;
	int i;
CODESTARTnewInpInst
	DBGPRINTF("newInpInst (imhiredis)\n");

	if((pvals = nvlstGetParams(lst, &inppblk, NULL)) == NULL) {
		ABORT_FINALIZE(RS_RET_MISSING_CNFPARAMS);
	}

	if(Debug) {
		dbgprintf("input param blk in imhiredis:\n");
		cnfparamsPrint(&inppblk, pvals);
	}

	CHKiRet(createInstance(&inst));
	for(i = 0 ; i < inppblk.nParams ; ++i) {
		if(!pvals[i].bUsed)
			continue;

		if(!strcmp(inppblk.descr[i].name, "server")) {
			inst->redisNodesList->server = (uchar*)es_str2cstr(pvals[i].val.d.estr, NULL);
		} else if(!strcmp(inppblk.descr[i].name, "socketPath")) {
			inst->redisNodesList->socketPath = (uchar*)es_str2cstr(pvals[i].val.d.estr, NULL);
		} else if(!strcmp(inppblk.descr[i].name, "ruleset")) {
			inst->pszBindRuleset = (uchar*)es_str2cstr(pvals[i].val.d.estr, NULL);
		} else if(!strcmp(inppblk.descr[i].name, "port")) {
			inst->redisNodesList->port = (int) pvals[i].val.d.n;
		} else if(!strcmp(inppblk.descr[i].name, "password")) {
			inst->password = (uchar*)es_str2cstr(pvals[i].val.d.estr, NULL);
		} else if(!strcmp(inppblk.descr[i].name, "uselpop")) {
			inst->useLPop = pvals[i].val.d.n;
		} else if(!strcmp(inppblk.descr[i].name, "mode")) {
			inst->modeDescription = (uchar*)es_str2cstr(pvals[i].val.d.estr, NULL);
			if (!strcmp((const char*)inst->modeDescription, "queue")) {
				inst->mode = IMHIREDIS_MODE_QUEUE;
			} else if (!strcmp((const char*)inst->modeDescription, "subscribe")) {
				inst->mode = IMHIREDIS_MODE_SUBSCRIBE;
			} else {
				LogMsg(0, RS_RET_PARAM_ERROR, LOG_ERR, "imhiredis: unsupported mode "
					"'%s'", inppblk.descr[i].name);
				ABORT_FINALIZE(RS_RET_PARAM_ERROR);
			}
		} else if(!strcmp(inppblk.descr[i].name, "key")) {
			inst->key = (uchar*)es_str2cstr(pvals[i].val.d.estr, NULL);
		} else {
			dbgprintf("imhiredis: program error, non-handled "
				"param '%s'\n", inppblk.descr[i].name);
		}
	}

	DBGPRINTF("imhiredis: checking config sanity\n");
	if (inst->modeDescription == NULL) {
		CHKmalloc(inst->modeDescription = (uchar*)strdup("subscribe"));
		inst->mode = IMHIREDIS_MODE_SUBSCRIBE;
		LogMsg(0, RS_RET_OK_WARN, LOG_WARNING, "imhiredis: \"mode\" parameter not specified "
			"using default redis 'subscribe' mode -- this may not be what you want!");
	}
	if (inst->key == NULL) {
		LogMsg(0, RS_RET_PARAM_ERROR, LOG_ERR, "imhiredis: \"key\" required parameter not specified!");
		ABORT_FINALIZE(RS_RET_PARAM_ERROR);
	}
	if(inst->redisNodesList->server == NULL && inst->redisNodesList->socketPath == NULL) {
		CHKmalloc(inst->redisNodesList->server = (uchar *)strdup("127.0.0.1"));
		inst->redisNodesList->port = 6379;
		LogMsg(0, RS_RET_OK_WARN, LOG_WARNING, "imhiredis: no server parameter specified "
			"using default 127.0.0.1:6379 -- this may not be what you want!");
	}
	if (inst->password == NULL) {
		LogMsg(0, RS_RET_OK, LOG_INFO, "imhiredis: no password specified");
	}

	DBGPRINTF("imhiredis: newInpInst key=%s, mode=%s, uselpop=%d\n",
		inst->key, inst->modeDescription, inst->useLPop);

finalize_it:
CODE_STD_FINALIZERnewInpInst
	cnfparamvalsDestruct(pvals, &inppblk);
ENDnewInpInst


BEGINbeginCnfLoad
CODESTARTbeginCnfLoad
	loadModConf = pModConf;
	pModConf->pConf = pConf;
ENDbeginCnfLoad


BEGINsetModCnf
	struct cnfparamvals *pvals = NULL;
	int i;
CODESTARTsetModCnf
	pvals = nvlstGetParams(lst, &modpblk, NULL);
	if(pvals == NULL) {
		LogError(0, RS_RET_MISSING_CNFPARAMS, "imhiredis: error processing module "
			"config parameters [module(...)]");
		ABORT_FINALIZE(RS_RET_MISSING_CNFPARAMS);
	}

	if(Debug) {
		dbgprintf("module (global) param blk for imhiredis:\n");
		cnfparamsPrint(&modpblk, pvals);
	}

	for(i = 0 ; i < modpblk.nParams ; ++i) {
		if(!pvals[i].bUsed) {
			continue;
		} else {
			dbgprintf("imhiredis: program error, non-handled "
			  "param '%s' in beginCnfLoad\n", modpblk.descr[i].name);
		}
	}
finalize_it:
	if(pvals != NULL)
		cnfparamvalsDestruct(pvals, &modpblk);
ENDsetModCnf

BEGINendCnfLoad
CODESTARTendCnfLoad
ENDendCnfLoad

BEGINcheckCnf
	instanceConf_t *inst;
CODESTARTcheckCnf
	for(inst = pModConf->root ; inst != NULL ; inst = inst->next) {
		std_checkRuleset(pModConf, inst);
	}
ENDcheckCnf


BEGINactivateCnfPrePrivDrop
CODESTARTactivateCnfPrePrivDrop
	runModConf = pModConf;
ENDactivateCnfPrePrivDrop

BEGINactivateCnf
CODESTARTactivateCnf
	for(instanceConf_t *inst = pModConf->root ; inst != NULL ; inst = inst->next) {
		iRet = checkInstance(inst);
		if (inst->mode == IMHIREDIS_MODE_SUBSCRIBE)
			inst->evtBase = event_base_new();
	}
ENDactivateCnf


BEGINfreeCnf
	instanceConf_t *inst, *del;
	redisNode *node;
CODESTARTfreeCnf
	for(inst = pModConf->root ; inst != NULL ; ) {
		if (inst->evtBase)
			event_base_free(inst->evtBase);
		if (inst->password != NULL)
			free(inst->password);
		free(inst->modeDescription);
		free(inst->key);
		free(inst->pszBindRuleset);
		if(inst->conn != NULL) {
			redisFree(inst->conn);
			inst->conn = NULL;
		}
		if(inst->aconn != NULL) {
			redisAsyncFree(inst->aconn);
			inst->aconn = NULL;
		}

		for (node = inst->redisNodesList; node != NULL; node = freeNode(node)) {;}

		del = inst;
		inst = inst->next;
		free(del);
	}
ENDfreeCnf


/* Cleanup imhiredis worker threads */
static void
shutdownImhiredisWorkers(void)
{
	int i;
	instanceConf_t *inst;

	assert(imhiredisWrkrInfo != NULL);

	for(inst = runModConf->root ; inst != NULL ; inst = inst->next) {
		if (inst->mode == IMHIREDIS_MODE_SUBSCRIBE && inst->aconn) {
			DBGPRINTF("imhiredis: disconnecting async worker\n");
			redisAsyncDisconnect(inst->aconn);
		}
	}

	// event_base_loopbreak(runModConf->evtBase);

	DBGPRINTF("imhiredis: waiting on imhiredis workerthread termination\n");
	for(i = 0 ; i < activeHiredisworkers ; ++i) {
		pthread_join(imhiredisWrkrInfo[i].tid, NULL);
		DBGPRINTF("imhiredis: Stopped worker %d\n", i);
	}
	free(imhiredisWrkrInfo);
	imhiredisWrkrInfo = NULL;

	return;
}


/* This function is called to gather input.  */
BEGINrunInput
	int i;
	instanceConf_t *inst;
CODESTARTrunInput
	DBGPRINTF("imhiredis: runInput loop started ...\n");
	activeHiredisworkers = 0;
	for(inst = runModConf->root ; inst != NULL ; inst = inst->next) {
		++activeHiredisworkers;
	}

	if(activeHiredisworkers == 0) {
		LogError(0, RS_RET_ERR, "imhiredis: no active inputs, input does "
			"not run - there should have been additional error "
			"messages given previously");
		ABORT_FINALIZE(RS_RET_ERR);
	}


	DBGPRINTF("imhiredis: Starting %d imhiredis workerthreads\n", activeHiredisworkers);
	imhiredisWrkrInfo = calloc(activeHiredisworkers, sizeof(struct imhiredisWrkrInfo_s));
	if (imhiredisWrkrInfo == NULL) {
		LogError(errno, RS_RET_OUT_OF_MEMORY, "imhiredis: worker-info array allocation failed.");
		ABORT_FINALIZE(RS_RET_OUT_OF_MEMORY);
	}

	/* Start worker threads for each imhiredis input source
	*/
	i = 0;
	for(inst = runModConf->root ; inst != NULL ; inst = inst->next) {
		/* init worker info structure! */
		imhiredisWrkrInfo[i].inst = inst; /* Set reference pointer */
		pthread_create(&imhiredisWrkrInfo[i].tid, &wrkrThrdAttr, imhirediswrkr, &(imhiredisWrkrInfo[i]));
		i++;
	}

	// This thread simply runs the various threads, then waits for Rsyslog to stop
	while(glbl.GetGlobalInputTermState() == 0) {
		if(glbl.GetGlobalInputTermState() == 0)
			/* Check termination state every 100ms
			 * should be sufficient to grant fast response to shutdown while not hogging CPU
			 */
			srSleep(0, 100000);
	}
	DBGPRINTF("imhiredis: terminating upon request of rsyslog core\n");

	shutdownImhiredisWorkers();
finalize_it:
ENDrunInput


BEGINwillRun
CODESTARTwillRun
	/* we need to create the inputName property (only once during our lifetime) */
	CHKiRet(prop.Construct(&pInputName));
	CHKiRet(prop.SetString(pInputName, UCHAR_CONSTANT("imhiredis"), sizeof("imhiredis") - 1));
	CHKiRet(prop.ConstructFinalize(pInputName));
finalize_it:
ENDwillRun


BEGINafterRun
CODESTARTafterRun
	if(pInputName != NULL)
		prop.Destruct(&pInputName);

ENDafterRun


BEGINmodExit
CODESTARTmodExit
	pthread_attr_destroy(&wrkrThrdAttr);

	/* force cleaning of all libevent-related structures
	 * (clean shutdowns are not always guaranteed without it)
	 */
	libevent_global_shutdown();

	/* release objects we used */
	objRelease(statsobj, CORE_COMPONENT);
	objRelease(ruleset, CORE_COMPONENT);
	objRelease(glbl, CORE_COMPONENT);
	objRelease(prop, CORE_COMPONENT);
ENDmodExit


BEGINisCompatibleWithFeature
CODESTARTisCompatibleWithFeature
	if(eFeat == sFEATURENonCancelInputTermination)
		iRet = RS_RET_OK;
ENDisCompatibleWithFeature


BEGINqueryEtryPt
CODESTARTqueryEtryPt
CODEqueryEtryPt_STD_IMOD_QUERIES
CODEqueryEtryPt_STD_CONF2_QUERIES
CODEqueryEtryPt_STD_CONF2_PREPRIVDROP_QUERIES
CODEqueryEtryPt_STD_CONF2_IMOD_QUERIES
CODEqueryEtryPt_STD_CONF2_setModCnf_QUERIES
CODEqueryEtryPt_IsCompatibleWithFeature_IF_OMOD_QUERIES
ENDqueryEtryPt


BEGINmodInit()
CODESTARTmodInit
	*ipIFVersProvided = CURR_MOD_IF_VERSION;
CODEmodInit_QueryRegCFSLineHdlr
	/* request objects we use */
	CHKiRet(objUse(glbl, CORE_COMPONENT));
	CHKiRet(objUse(prop, CORE_COMPONENT));
	CHKiRet(objUse(ruleset, CORE_COMPONENT));
	CHKiRet(objUse(statsobj, CORE_COMPONENT));

	/* initialize "read-only" thread attributes */
	pthread_attr_init(&wrkrThrdAttr);
	pthread_attr_setstacksize(&wrkrThrdAttr, 4096*1024);

	/* activate libevent for (p)threads support */
	evthread_use_pthreads();

ENDmodInit


/* ------------------------------ callbacks ------------------------------ */


/**
 *	Asynchronous subscribe callback handler
 */
static void redisAsyncRecvCallback (redisAsyncContext *aconn, void *reply, void __attribute__((unused)) *unused) {
	/*
		redisReply is supposed to be an array of three elements: [''message', <channel>, <message>]


		JJO: For future reference (https://github.com/redis/hiredis/blob/master/README.md)

		Important: the current version of hiredis (1.0.0) frees replies when the asynchronous API is used.
		This means you should not call freeReplyObject when you use this API.
		The reply is cleaned up by hiredis after the callback returns.
		TODO We may have to change this function in the future to free replies.
	*/
	instanceConf_t *const inst = (instanceConf_t *) aconn->data;
	redisReply * r = (redisReply *) reply;
	if (r == NULL) return;

	if (r->elements < 3 || r->element[2]->str == NULL) {
		return;
	}
	enqMsg(inst, r->element[2]->str);

	return;
}


/**
 *	Asynchronous connection callback handler
 */
static void redisAsyncConnectCallback (const redisAsyncContext *c, int status) {
	if (status != REDIS_OK) {
		LogMsg(0, RS_RET_REDIS_ERROR, LOG_ERR, "imhiredis (async): could not connect to redis: "
			"%s", c->errstr);
		// remove async context from instance config object, still contained in context's 'data' field
		instanceConf_t *inst = (instanceConf_t *) c->data;
		assert(inst != NULL);
		inst->aconn = NULL;
		return;
	}
	DBGPRINTF("imhiredis (async): successfully connected!\n");

	return;
}


/**
 *	Asynchronous disconnection callback handler
 */
static void redisAsyncDisconnectCallback (const redisAsyncContext *c, int status) {

	// remove context from instance config object (which is stored in the context 'data' field by us)
	// context will be freed by the library, so it's only set to NULL here
	instanceConf_t *inst = (instanceConf_t *) c->data;
	assert(inst != NULL);
	inst->aconn = NULL;
	inst->currentNode = NULL;

	if (status != REDIS_OK) {
		LogMsg(0, RS_RET_REDIS_ERROR, LOG_ERR, "imhiredis (async): got disconnected from redis: "
			"%s", c->errstr);
		return;
	}
	DBGPRINTF("imhiredis (async): successfully disconnected!\n");

	return;
}


/* ------------------------------ end callbacks ------------------------------ */

/*
 *	sends a ROLE command to the redis pointed by context
 *	context should be a valid redis context
 *	returns a valid redisReply pointer if an array reply was received, NULL otherwise
 */
redisReply *getRole(redisContext *c) {
	redisReply *reply;

	assert(c != NULL);

	reply = redisCommand(c, "ROLE");
	if (reply == NULL) {
		DBGPRINTF("imhiredis: could not get reply from ROLE command\n");
	}
	else if (reply->type == REDIS_REPLY_ERROR) {
		LogMsg(0, RS_RET_REDIS_ERROR, LOG_WARNING, "imhiredis got an error while querying role -> "
			"%s\n", reply->str);
		freeReplyObject(reply);
		reply = NULL;
	}
	else if (reply->type != REDIS_REPLY_ARRAY) {
		LogMsg(0, RS_RET_REDIS_ERROR, LOG_ERR, "imhiredis: did not get an array from ROLE command");
		freeReplyObject(reply);
		reply = NULL;
	}

	return reply;
}


/*
 *	enqueue the hiredis message. The provided string is
 *	not freed - this must be done by the caller.
 */
static rsRetVal enqMsg(instanceConf_t *const inst, const char *message) {
	DEFiRet;
	smsg_t *pMsg;

	if (message == NULL || message[0] == '\0') {
		/* we do not process empty lines */
		FINALIZE;
	}

	DBGPRINTF("imhiredis: enqMsg: Msg -> '%s'\n", message);

	CHKiRet(msgConstruct(&pMsg));
	MsgSetInputName(pMsg, pInputName);
	MsgSetRawMsg(pMsg, message, strlen(message));
	MsgSetFlowControlType(pMsg, eFLOWCTL_LIGHT_DELAY);
	MsgSetRuleset(pMsg, inst->pBindRuleset);
	MsgSetMSGoffs(pMsg, 0);	/* we do not have a header... */
	CHKiRet(submitMsg2(pMsg));

finalize_it:
	RETiRet;
}


/*
 *	execute a synchronous authentication using the context conn
 *	conn and password should be non-NULL
 *	conn should be a valid context
 */
rsRetVal redisAuthentSynchronous(redisContext *conn, uchar *password) {
	DEFiRet;
	redisReply *reply = NULL;

	assert(conn != NULL);
	assert(password != NULL);
	assert(password[0] != '\0');

	reply = (redisReply *) redisCommand(conn, "AUTH %s", password);
	if (reply == NULL) {
		LogError(0, RS_RET_REDIS_ERROR, "imhiredis: Could not authenticate!\n");
		ABORT_FINALIZE(RS_RET_REDIS_ERROR);
	} else if (strncmp(reply->str, "OK", 2)) {
		LogError(0, RS_RET_REDIS_AUTH_FAILED, "imhiredis: Authentication failure -> %s\n", reply->str);
		ABORT_FINALIZE(RS_RET_REDIS_AUTH_FAILED);
	}

finalize_it:
	if(reply)
		freeReplyObject(reply);
	RETiRet;
}


/*
 *	execute an asynchronous authentication using the context aconn
 *	aconn and password should be non-NULL
 *	aconn should be a valid (async) context
 */
rsRetVal redisAuthentAsynchronous(redisAsyncContext *aconn, uchar *password) {
	DEFiRet;

	assert(aconn != NULL);
	assert(password != NULL);
	assert(password[0] != '\0');

	if (REDIS_OK != redisAsyncCommand(aconn, NULL, NULL, "AUTH %s", password)) {
		LogError(0, RS_RET_REDIS_ERROR, "imhiredis: error while authenticating asynchronously -> %s\n",
			aconn->errstr);
		ABORT_FINALIZE(RS_RET_REDIS_ERROR);
	}

finalize_it:
	RETiRet;
}


/*
 *	connect to node, authenticate (if necessary), get role, then get all node information provided by ROLE
 *	node should be a non-NULL valid redisNode pointer
 *	password can be NULL, meaning no authentication will be done
 *	result will hold the result of the ROLE command executed on the node:
 *		- NULL if the node was a single master instance
 *		- a single (master) node if the provided node was a replica
 *		- a list of (replica) nodes if the provided node was a master
 */
rsRetVal redisGetServersList(redisNode *node, uchar *password, redisNode **result) {
	DEFiRet;
	redisContext *context;
	redisReply *reply = NULL, *replica;
	unsigned int i;

	assert(node != NULL);

	CHKiRet(redisConnectSync(&context, node));

	if(password != NULL && password[0] != '\0') {
		CHKiRet(redisAuthentSynchronous(context, password));
	}

	reply = getRole(context);

	if(reply == NULL) {
		LogMsg(0, RS_RET_REDIS_ERROR, LOG_WARNING, "imhiredis: did not get the role of the server");
		ABORT_FINALIZE(RS_RET_REDIS_ERROR);
	}

	/*
	* string comparisons for ROLE could be skipped
	* as each role returns a different number of elements,
	* but lets keep it as a security...
	*/
	if (	reply->elements == 5 &&
		strncmp(reply->element[0]->str, "slave", 5) == 0) {

		CHKiRet(createRedisNode(result));
		(*result)->server = (uchar *) strdup((const char *)reply->element[1]->str);
		(*result)->port = reply->element[2]->integer;
		(*result)->isMaster = 1;
	}
	else if (	reply->elements == 3 &&
			reply->element[2]->type == REDIS_REPLY_ARRAY &&
			strncmp(reply->element[0]->str, "master", 6) == 0) {

		// iterate on all replicas given in the reply (if any)
		for (i = 0; i < reply->element[2]->elements; i++) {
			replica = reply->element[2]->element[i];

			if (replica->type == REDIS_REPLY_ARRAY && replica->elements == 3) {
				/* node will be a new node every time
				* with old ones shifted in the list
				*/
				CHKiRet(createRedisNode(result));
				(*result)->server = (uchar *) strdup((const char *)replica->element[0]->str);
				// yes, the value in that case is a string and NOT an integer!
				(*result)->port = atoi(replica->element[1]->str);
			}
		}
	} else {
		// we have a sentinel, or a problem
		ABORT_FINALIZE(RS_RET_NOT_FOUND);
	}

finalize_it:
	if (reply != NULL)
		freeReplyObject(reply);
	if (context != NULL)
		redisFree(context);
	RETiRet;
}



/*
 *	actualize the current master node to use during connection for instance inst
 *	inst should be a valid, non-NULL instanceConf object
 *	inst should also possess at least a single node in inst->redisNodeList
 *	if the function returns RS_RET_OK, inst->currentNode and inst->redisNodeList have been both updated
 *	to reflect new master and potential replicas
 *	the first configured node (called preferred node) is always kept as the first entry in redisNodeList
 */
rsRetVal redisActualizeCurrentNode(instanceConf_t *inst) {
	DEFiRet;
	redisReply *reply = NULL;
	redisNode *node, *tmp, *newList = NULL;

	assert(inst != NULL);
	assert(inst->redisNodesList != NULL);

	inst->currentNode = NULL;
	// keep first node in list = preferred node (comes from configuration)
	copyNode(inst->redisNodesList, &newList);
	newList->next = NULL;

	for (node = inst->redisNodesList; node != NULL; node = node->next) {
		tmp = NULL;

		DBGPRINTF("imhiredis: trying to connect to node to get info...\n");
		dbgPrintNode(node);

		if (RS_RET_OK == redisGetServersList(node, inst->password, &tmp)) {
			// server replied

			if (tmp && tmp->isMaster) {
				DBGPRINTF("imhiredis: node replied with a master node, is a replica\n");
				// master node, keep it as potential new active node
				inst->currentNode = tmp;
				tmp = NULL;

				// try to connect to the master and get replicas
				if(RS_RET_OK != redisGetServersList(inst->currentNode, inst->password, &tmp)) {

					/* had a master, but cannot connect
					* save suspected master in new list but keep searching with other nodes
					*/
					DBGPRINTF("imhiredis: had a master but cannot connect, keeping in list\n");
					dbgPrintNode(inst->currentNode);
					insertNodeAfter(newList, inst->currentNode);
					inst->currentNode = NULL;
					continue;
				}
			} else {
				DBGPRINTF("imhiredis: node replied with a list of replicas, is a master\n");
				// copy the node to the new currentNode, list owning node will be freed
				node->isMaster = 1;
				copyNode(node, &(inst->currentNode));
				inst->currentNode->next = NULL;
			}

			/*
			 * here, tmp is a list of replicas or NULL (single node)
			 * inst->currentNode is the new active master
			 */

			// add the replicas to the list
			if (tmp) {
				insertNodeAfter(newList, tmp);
				DBGPRINTF("imhiredis: inserting replicas to list\n");
				for (tmp = newList->next; tmp != NULL; tmp = tmp->next) {
					dbgPrintNode(tmp);
				}
			}
			// insert the master after the preferred node (configuration)
			DBGPRINTF("imhiredis: inserting new master node in list\n");
			dbgPrintNode(inst->currentNode);
			insertNodeAfter(newList, inst->currentNode);

			// swap newList and redisNodesList to free old list at the end of the function
			tmp = newList;
			newList = inst->redisNodesList;
			inst->redisNodesList = tmp;
			FINALIZE;
		}
	}

	DBGPRINTF("imhiredis: did not find a valid master");
	iRet = RS_RET_NOT_FOUND;
	inst->currentNode = NULL;

finalize_it:
	if (reply != NULL)
		freeReplyObject(reply);
	// newList is always completely freed
	for (node = newList; node != NULL; ) {
		node = freeNode(node);
	}

	RETiRet;
}


/*
 *	authentication function, for both synchronous and asynchronous modes (queue or subscribe)
 *	inst, inst->curentMode and inst->password should not be NULL
 */
rsRetVal redisAuthenticate(instanceConf_t *inst) {
	DEFiRet;
	redisContext *usedContext = NULL;
	redisReply *reply = NULL;

	assert(inst != NULL);
	assert(inst->currentNode != NULL);
	assert(inst->password != NULL);
	assert(inst->password[0] != '\0');

	DBGPRINTF("imhiredis: authenticating...\n");

	// Create a temporary context for synchronous connection, used to validate AUTH command in asynchronous contexts
	if (inst->mode == IMHIREDIS_MODE_SUBSCRIBE) {
		if (RS_RET_OK != redisConnectSync(&usedContext, inst->currentNode)) {
			LogMsg(0, RS_RET_REDIS_ERROR, LOG_WARNING, "imhiredis: could not connect to current "
				"active node synchronously to validate authentication");
			ABORT_FINALIZE(RS_RET_REDIS_ERROR);
		}
	} else {
		usedContext = inst->conn;
	}

	/*
	 * Try synchronous connection, whatever the method for the instance
	 * This is also done for the asynchronous mode, to validate the successful authentication
	 */
	CHKiRet(redisAuthentSynchronous(usedContext, inst->password));

	if (inst->mode == IMHIREDIS_MODE_SUBSCRIBE) {
		CHKiRet(redisAuthentAsynchronous(inst->aconn, inst->password));
	}

	DBGPRINTF("imhiredis: authentication successful\n");

finalize_it:
	if(inst->mode == IMHIREDIS_MODE_SUBSCRIBE && usedContext)
		redisFree(usedContext);
	if(reply)
		freeReplyObject(reply);
	RETiRet;
}


/*
 *	connection function for synchronous (queue) mode
 *	node should not be NULL
 */
rsRetVal redisConnectSync(redisContext **conn, redisNode *node) {
	DEFiRet;

	assert(node != NULL);

	if (node->usesSocket)
		*conn = redisConnectUnixWithTimeout((const char *)node->socketPath, glblRedisConnectTimeout);
	else
		*conn = redisConnectWithTimeout((const char *)node->server, node->port, glblRedisConnectTimeout);

	if (*conn == NULL) {
		if (node->usesSocket) {
			LogError(0, RS_RET_REDIS_ERROR, "imhiredis: can not connect to redis server '%s' "
				"-> could not allocate context!\n", node->socketPath);
		} else {
			LogError(0, RS_RET_REDIS_ERROR, "imhiredis: can not connect to redis server '%s', "
				"port %d -> could not allocate context!\n", node->server, node->port);
		}
		ABORT_FINALIZE(RS_RET_REDIS_ERROR);
	}
	else if ((*conn)->err) {
		if (node->usesSocket) {
			LogError(0, RS_RET_REDIS_ERROR, "imhiredis: can not connect to redis server '%s' "
				"-> %s\n", node->socketPath, (*conn)->errstr);
		} else {
			LogError(0, RS_RET_REDIS_ERROR, "imhiredis: can not connect to redis server '%s', "
				"port %d -> %s\n", node->server, node->port, (*conn)->errstr);
		}
		ABORT_FINALIZE(RS_RET_REDIS_ERROR);
	}

finalize_it:
	if (iRet != RS_RET_OK) {
		if (*conn)
			redisFree(*conn);
		*conn = NULL;
	}
	RETiRet;
}


/*
 *	connection function for asynchronous (subscribe) mode
 *	node should not be NULL
 */
rsRetVal redisConnectAsync(redisAsyncContext **aconn, redisNode *node) {
	DEFiRet;

	assert(node != NULL);

	if (node->usesSocket)
		*aconn = redisAsyncConnectUnix((const char*)node->socketPath);
	else
		*aconn = redisAsyncConnect((const char *)node->server, node->port);

	if(*aconn == NULL) {
		LogError(0, RS_RET_REDIS_ERROR, "imhiredis (async): could not allocate context!\n");
		ABORT_FINALIZE(RS_RET_REDIS_ERROR);
	} else if ((*aconn)->err) {
		if (node->usesSocket) {
			LogError(0, RS_RET_REDIS_ERROR, "imhiredis (async): cannot connect to server '%s' "
				"-> %s\n", node->socketPath, (*aconn)->errstr);
		} else {
			LogError(0, RS_RET_REDIS_ERROR, "imhiredis (async): cannot connect to server '%s', port '%d' "
				"-> %s\n", node->server, node->port, (*aconn)->errstr);
		}
		ABORT_FINALIZE(RS_RET_REDIS_ERROR);
	}

finalize_it:
	if (iRet != RS_RET_OK) {
		if(*aconn)
			redisAsyncFree(*aconn);
		*aconn = NULL;
	}
	RETiRet;
}

/*
 *	dequeue all entries in the redis list, using batches of 10 commands
 */
rsRetVal redisDequeue(instanceConf_t *inst) {
	DEFiRet;
	redisReply *reply = NULL;
	uint replyType = 0, i;

	assert(inst != NULL);

	DBGPRINTF("imhiredis: beginning to dequeue key '%s'\n", inst->key);

	do {
		// append a batch of QUEUE_BATCH_SIZE POP commands (either LPOP or RPOP depending on conf)
		if (inst->useLPop == 1) {
			DBGPRINTF("imhiredis: Queuing #%d LPOP commands on key '%s' \n",
					QUEUE_BATCH_SIZE,
					inst->key);
			for (i=0; i<QUEUE_BATCH_SIZE; ++i ) {
				if (REDIS_OK != redisAppendCommand(inst->conn, "LPOP %s", inst->key))
					break;
			}
		} else {
			DBGPRINTF("imhiredis: Queuing #%d RPOP commands on key '%s' \n",
					QUEUE_BATCH_SIZE,
					inst->key);
			for (i=0; i<QUEUE_BATCH_SIZE; i++) {
				if (REDIS_OK != redisAppendCommand(inst->conn, "RPOP %s", inst->key))
					break;
			}
		}

		// parse responses from appended commands
		do {
			if (REDIS_OK != redisGetReply(inst->conn, (void **) &reply)) {
				// error getting reply, must stop
				LogError(0, RS_RET_REDIS_ERROR, "imhiredis: Error reading reply after POP #%d on key "
								"'%s'", (QUEUE_BATCH_SIZE - i), inst->key);
				ABORT_FINALIZE(RS_RET_REDIS_ERROR);
			} else {
				if (reply != NULL) {
					replyType = reply->type;
					switch(replyType) {
						case REDIS_REPLY_STRING:
							enqMsg(inst, reply->str);
							break;
						case REDIS_REPLY_NIL:
							// replies are dequeued but are empty = end of list
							break;
						case REDIS_REPLY_ERROR:
							// There is a problem with the key or the Redis instance
							LogMsg(0, RS_RET_REDIS_ERROR, LOG_ERR, "imhiredis: error "
							"while POP'ing key '%s' -> %s", inst->key, reply->str);
							ABORT_FINALIZE(RS_RET_REDIS_ERROR);
						default:
							LogMsg(0, RS_RET_OK_WARN, LOG_WARNING, "imhiredis: unexpected "
							"reply type: %s", redis_replies[replyType%7]);
					}
					freeReplyObject(reply);
					reply = NULL;
				} else { /* reply == NULL */
					LogMsg(0, RS_RET_REDIS_ERROR, LOG_ERR, "imhiredis: unexpected empty reply "
						"for successful return");
					ABORT_FINALIZE(RS_RET_REDIS_ERROR);
				}
			}

		// while there are replies to unpack, continue
		} while (--i > 0);

	// while input can run and last reply was a string, continue with a new batch
	} while (replyType == REDIS_REPLY_STRING && glbl.GetGlobalInputTermState() == 0);

	DBGPRINTF("imhiredis: finished to dequeue key '%s'\n", inst->key);

finalize_it:
	if (reply)
		freeReplyObject(reply);
	RETiRet;
}


/*
 *	worker function for asynchronous (subscribe) mode
 */
void workerLoopSubscribe(struct imhiredisWrkrInfo_s *me) {
	uint i;
	DBGPRINTF("imhiredis (async): beginning of subscribe worker loop...\n");

	// Connect first time without delay
	if (me->inst->currentNode != NULL) {
		if(RS_RET_OK != redisConnectAsync(&(me->inst->aconn), me->inst->currentNode)) {
			me->inst->currentNode = NULL;
		}
		if(	me->inst->password != NULL &&
			me->inst->password[0] != '\0' &&
			RS_RET_OK != redisAuthenticate(me->inst)) {

			redisAsyncFree(me->inst->aconn);
			me->inst->aconn = NULL;
			me->inst->currentNode = NULL;
		}

		// finalize context creation
		me->inst->aconn->data = (void *)me->inst;
		redisAsyncSetConnectCallback(me->inst->aconn, redisAsyncConnectCallback);
		redisAsyncSetDisconnectCallback(me->inst->aconn, redisAsyncDisconnectCallback);
		redisLibeventAttach(me->inst->aconn, me->inst->evtBase);
	}

	while(glbl.GetGlobalInputTermState() == 0) {
		if (me->inst->aconn == NULL) {
			/*
			 * Sleep 10 seconds before attempting to resume a broken connexion
			 * (sleep small amounts to avoid missing termination status)
			 */
			DBGPRINTF("imhiredis: no valid connection, sleeping 10 seconds before retrying...\n");
			for(i = 0; i < 100; i++) {
				// Rsyslog asked for shutdown, thread should be stopped
				if (glbl.GetGlobalInputTermState() != 0)
					goto end_loop;
				// 100ms sleeps
				srSleep(0, 100000);
			}

			// search the current master node
			if (me->inst->currentNode == NULL) {
				if(RS_RET_OK != redisActualizeCurrentNode(me->inst))
					continue;
			}

			// connect to current master
			if(me->inst->currentNode != NULL) {
				if(RS_RET_OK != redisConnectAsync(&(me->inst->aconn), me->inst->currentNode)) {
					me->inst->currentNode = NULL;
					continue;
				}
				if(	me->inst->password != NULL &&
					me->inst->password[0] != '\0' &&
					RS_RET_OK != redisAuthenticate(me->inst)) {

					redisAsyncFree(me->inst->aconn);
					me->inst->aconn = NULL;
					me->inst->currentNode = NULL;
					continue;
				}
			}

			// finalize context creation
			me->inst->aconn->data = (void *)me->inst;
			redisAsyncSetConnectCallback(me->inst->aconn, redisAsyncConnectCallback);
			redisAsyncSetDisconnectCallback(me->inst->aconn, redisAsyncDisconnectCallback);
			redisLibeventAttach(me->inst->aconn, me->inst->evtBase);
		}
		if (me->inst->aconn != NULL) {
			DBGPRINTF("imhiredis (async): subscribing to channel '%s'\n", me->inst->key);
			redisAsyncCommand(
				me->inst->aconn,
				redisAsyncRecvCallback,
				NULL,
				"SUBSCRIBE %s",
				me->inst->key);
			event_base_dispatch(me->inst->evtBase);
		}
	}

end_loop:
	return;
}


/*
 *	worker function for synchronous (queue) mode
 */
void workerLoopQueue(struct imhiredisWrkrInfo_s *me) {
	uint i;
	DBGPRINTF("imhiredis: beginning of queue worker loop...\n");
	// Connect first time without delay
	if (me->inst->currentNode != NULL) {
		if(RS_RET_OK != redisConnectSync(&(me->inst->conn), me->inst->currentNode)) {
			me->inst->currentNode = NULL;
		}
		if(	me->inst->password != NULL &&
			me->inst->password[0] != '\0' &&
			RS_RET_OK != redisAuthenticate(me->inst)) {

			redisFree(me->inst->conn);
			me->inst->conn = NULL;
			me->inst->currentNode = NULL;
		}
	}

	while(glbl.GetGlobalInputTermState() == 0) {
		if (me->inst->conn == NULL) {
			/* Sleep 10 seconds before attempting to resume a broken connexion
			 * (sleep small amounts to avoid missing termination status)
			 */
			DBGPRINTF("imhiredis: no valid connection, sleeping 10 seconds before retrying...\n");
			for(i = 0; i < 100; i++) {
				// Time to stop the thread
				if (glbl.GetGlobalInputTermState() != 0)
					goto end_loop;
				// 100ms sleeps
				srSleep(0, 100000);
			}

			// search the current master node
			if (me->inst->currentNode == NULL) {
				if(RS_RET_OK != redisActualizeCurrentNode(me->inst))
					continue;
			}

			// connect to current master
			if(me->inst->currentNode != NULL) {
				if(RS_RET_OK != redisConnectSync(&(me->inst->conn), me->inst->currentNode)) {
					me->inst->currentNode = NULL;
					continue;
				}
				if(	me->inst->password != NULL &&
					me->inst->password[0] != '\0' &&
					RS_RET_OK != redisAuthenticate(me->inst)) {

					redisFree(me->inst->conn);
					me->inst->conn = NULL;
					me->inst->currentNode = NULL;
					continue;
				}
			}
		}
		if (me->inst->conn != NULL) {
			if (redisDequeue(me->inst) == RS_RET_REDIS_ERROR) {
				DBGPRINTF("imhiredis: current connection invalidated\n");
				redisFree(me->inst->conn);
				me->inst->currentNode = NULL;
				me->inst->conn = NULL;
			}
			if(glbl.GetGlobalInputTermState() == 0) {
				/* sleep 1s between 2 POP tries
				* this does NOT limit dequeing rate, but prevents the input from polling Redis too often
				*/
				for(i = 0; i < 10; i++) {
					// Time to stop the thread
					if (glbl.GetGlobalInputTermState() != 0)
						goto end_loop;
					// 100ms sleeps
					srSleep(0, 100000);
				}
			}
		}
	}

end_loop:
	return;
}


/*
 *	Workerthread function for a single hiredis consumer
 */
static void *
imhirediswrkr(void *myself)
{
	struct imhiredisWrkrInfo_s *me = (struct imhiredisWrkrInfo_s*) myself;
	DBGPRINTF("imhiredis: started hiredis consumer workerthread\n");
	dbgPrintNode(me->inst->currentNode);

	if(me->inst->mode == IMHIREDIS_MODE_QUEUE)
		workerLoopQueue(me);
	else if (me->inst->mode == IMHIREDIS_MODE_SUBSCRIBE)
		workerLoopSubscribe(me);

	DBGPRINTF("imhiredis: stopped hiredis consumer workerthread\n");
	return NULL;
}



// -------------------------- redisNode functions -----------------------------------

/*
 *	create a redisNode and set default values
 *	if a valid node is given as parameter, the new node is inserted as the new head of the linked list
 */
static rsRetVal
createRedisNode(redisNode **root) {
	redisNode *node;
	DEFiRet;
	CHKmalloc(node = malloc(sizeof(redisNode)));
	node->port = 0;
	node->server = NULL;
	node->socketPath = NULL;
	node->usesSocket = 0;
	node->isMaster = 0;
	node->next = NULL;

	if (!root) {
		*root = node;
	} else {
		node->next = (*root);
		*root = node;
	}

finalize_it:
	RETiRet;
}

/*
 *	make a complete copy of the src node into the newly-created node in dst
 *	if dst already contains a node, the new node will be added as the new head of the provided list
 *	src should not be NULL
 */
rsRetVal copyNode(redisNode *src, redisNode **dst) {
	DEFiRet;

	assert(src != NULL);

	CHKiRet(createRedisNode(dst));

	(*dst)->isMaster = src->isMaster;
	(*dst)->next = src->next;
	(*dst)->port = src->port;
	(*dst)->usesSocket = src->usesSocket;

	if (src->server)
		(*dst)->server = (uchar *) strdup((const char *)src->server);
	if (src->socketPath)
		(*dst)->socketPath = (uchar *) strdup((const char *)src->socketPath);

finalize_it:
	RETiRet;
}

/*
 *	free all ressources of the node
 *	will return next node if one is present, NULL otherwise
 */
redisNode *freeNode(redisNode *node) {
	redisNode *ret = NULL;
	if (node != NULL) {
		if (node->next != NULL)
			ret = node->next;

		if(node->server != NULL)
			free(node->server);
		if(node->socketPath != NULL)
			free(node->socketPath);
		free(node);
	}

	return ret;
}

/*
 *	insert node 'elem' after node 'root' in the linked list
 *	both root and elem should not be NULL
 */
void insertNodeAfter(redisNode *root, redisNode *elem) {
	assert(root != NULL);
	assert(elem != NULL);

	if(root->next != NULL) {
		elem->next = root->next;
	}
	root->next = elem;

	return;
}

void dbgPrintNode(redisNode *node) {
	if (node != NULL) {
		if (node->usesSocket) {
			if (node->isMaster) {
				DBGPRINTF("imhiredis: node is %s (master)\n", node->socketPath);
			} else {
				DBGPRINTF("imhiredis: node is %s (replica)\n", node->socketPath);
			}
		} else {
			if (node->isMaster) {
				DBGPRINTF("imhiredis: node is %s:%d (master)\n", node->server, node->port);
			} else {
				DBGPRINTF("imhiredis: node is %s:%d (replica)\n", node->server, node->port);
			}
		}
	}
	return;
}
