redis connect and restore procedure - STUBS

daemon/main.c

@@ -41,7 +41,7 @@ static int port_min;
 static int port_max;
 static u_int32_t redis_ip;
 static u_int16_t redis_port;
-static char *redis_key;
+static int redis_db = -1;
 
 
 
@@ -124,7 +124,7 @@ static void options(int *argc, char ***argv) {
 		{ "port-min",	'm', 0, G_OPTION_ARG_INT,	&port_min,	"Lowest port to use for RTP",	"INT"		},
 		{ "port-max",	'M', 0, G_OPTION_ARG_INT,	&port_max,	"Highest port to use for RTP",	"INT"		},
 		{ "redis",	'r', 0, G_OPTION_ARG_STRING,	&redisps,	"Connect to Redis database",	"IP:PORT"	},
-		{ "redis-key",	'R', 0, G_OPTION_ARG_STRING,	&redis_key,	"Namespace key for Redis",	"STRING"	},
+		{ "redis-db",	'R', 0, G_OPTION_ARG_INT,	&redis_db,	"Which Redis DB to use",	"INT"	},
 		{ NULL, }
 	};
 
@@ -171,8 +171,8 @@ static void options(int *argc, char ***argv) {
 	if (redisps) {
 		if (parse_ip_port(&redis_ip, &redis_port, redisps) || !redis_ip)
 			die("Invalid IP or port (--redis)\n");
-		if (!redis_key)
-			die("Must specify Redis namespace key (--redis-key) when using Redis\n");
+		if (redis_db < 0)
+			die("Must specify Redis DB number (--redis-db) when using Redis\n");
 	}
 }
 
@@ -252,10 +252,9 @@ int main(int argc, char **argv) {
 	}
 
 	if (redis_ip) {
-		m->redis = redis_new(redis_ip, redis_port, redis_key);
+		m->redis = redis_new(redis_ip, redis_port, redis_db);
 		if (!m->redis)
-			die("Failed to connect to Redis database\n");
-		mylog(LOG_INFO, "Connected to Redis\n");
+			die("Cannot start up without Redis database\n");
 	}
 
 	mylog(LOG_INFO, "Startup complete");
@@ -264,6 +263,11 @@ int main(int argc, char **argv) {
 		daemonize();
 	wpidfile();
 
+	if (m->redis) {
+		if (redis_restore(m))
+			die("Refusing to continue without working Redis database\n");
+	}
+
 	for (;;) {
 		ret = poller_poll(p, 100);
 		if (ret == -1)

daemon/redis.c

@@ -2,27 +2,31 @@
 #include <hiredis.h>
 #include <sys/types.h>
 #include <sys/time.h>
+#include <unistd.h>
 
 #include "redis.h"
 #include "aux.h"
+#include "call.h"
 #include "log.h"
 
 
 
 
 
-struct redis *redis_new(u_int32_t ip, u_int16_t port, char *key) {
-	struct redis *r;
+#define redisCommandNR(a...) (int)({ void *__tmp; __tmp = redisCommand(a); if (__tmp) freeReplyObject(__tmp); __tmp ? 0 : -1;})
+
+
+
+
+
+static int redis_connect(struct redis *r, int wait) {
 	struct timeval tv;
 	redisReply *rp;
 	char *s;
 
-	r = malloc(sizeof(*r));
-	ZERO(*r);
-
-	sprintf(r->host, IPF, IPP(ip));
-	r->port = port;
-	r->key = key;
+	if (r->ctx)
+		redisFree(r->ctx);
+	r->ctx = NULL;
 
 	tv.tv_sec = 1;
 	tv.tv_usec = 0;
@@ -31,39 +35,123 @@ struct redis *redis_new(u_int32_t ip, u_int16_t port, char *key) {
 	if (!r->ctx)
 		goto err;
 	if (r->ctx->err)
-		goto err;
+		goto err2;
 
-	rp = redisCommand(r->ctx, "PING");
-	if (!rp)
-		goto err;
-	freeReplyObject(rp);
+	if (redisCommandNR(r->ctx, "PING"))
+		goto err2;
 
-	rp = redisCommand(r->ctx, "INFO");
-	if (!rp)
-		goto err;
-	s = strstr(rp->str, "role:");
-	if (!s) {
-		freeReplyObject(rp);
-		goto err;
-	}
-	if (!memcmp(s, "role:master", 9))
-		r->active = 1;
-	else if (!memcmp(s, "role:slave", 8))
-		;	/* it's already 0 */
-	else {
-		mylog(LOG_ERR, "Unable to determine Redis master/slave state\n");
+	if (redisCommandNR(r->ctx, "SELECT %i", r->db))
+		goto err2;
+
+	while (wait-- >= 0) {
+		mylog(LOG_INFO, "Asking Redis whether it's master or slave...\n");
+		rp = redisCommand(r->ctx, "INFO");
+		if (!rp)
+			goto err2;
+
+		s = strstr(rp->str, "role:");
+		if (!s)
+			goto err3;
+		if (!memcmp(s, "role:master", 9))
+			goto done;
+		else if (!memcmp(s, "role:slave", 8))
+			goto next;
+		else
+			goto err3;
+
+next:
 		freeReplyObject(rp);
-		goto err;
+		mylog(LOG_INFO, "Connected to Redis, but it's in slave mode\n");
+		sleep(1);
 	}
+
+	goto err2;
+
+done:
+	freeReplyObject(rp);
+	mylog(LOG_INFO, "Connected to Redis\n");
+	return 0;
+
+err3:
 	freeReplyObject(rp);
+err2:
+	if (r->ctx->err)
+		mylog(LOG_ERR, "Redis error: %s\n", r->ctx->errstr);
+	redisFree(r->ctx);
+	r->ctx = NULL;
+err:
+	mylog(LOG_ERR, "Failed to connect to master Redis database\n");
+	return -1;
+}
+
+
+
+
+struct redis *redis_new(u_int32_t ip, u_int16_t port, int db) {
+	struct redis *r;
+
+	r = malloc(sizeof(*r));
+	ZERO(*r);
+
+	sprintf(r->host, IPF, IPP(ip));
+	r->port = port;
+	r->db = db;
+
+	if (redis_connect(r, 10))
+		goto err;
 
 	return r;
 
 err:
-	if (r->ctx && r->ctx->err && r->ctx->errstr)
-		mylog(LOG_CRIT, "Redis error: %s\n", r->ctx->errstr);
-	if (r->ctx)
-		redisFree(r->ctx);
 	free(r);
 	return NULL;
 }
+
+
+
+
+int redis_restore(struct callmaster *m) {
+	struct redis *r = m->redis;
+	redisReply *rp, *rp2, *rp3;
+	int i;
+
+	rp = redisCommand(r->ctx, "SMEMBERS calls");
+	if (!rp || rp->type != REDIS_REPLY_ARRAY) {
+		mylog(LOG_ERR, "Could not retrieve call list from Redis: %s\n", r->ctx->errstr);
+		goto err;
+	}
+
+	for (i = 0; i < rp->elements; i++) {
+		rp2 = rp->element[i];
+		if (rp2->type != REDIS_REPLY_STRING)
+			continue;
+
+		rp3 = redisCommand(r->ctx, "HMGET %s callid created", rp2->str);
+
+		if (!rp3)
+			goto del;
+		if (rp3->type != REDIS_REPLY_ARRAY)
+			goto del2;
+		if (rp3->elements != 2)
+			goto del2;
+		if (rp3->element[0]->type != REDIS_REPLY_STRING)
+			goto del2;
+		if (rp3->element[1]->type != REDIS_REPLY_INTEGER)
+			goto del2;
+
+		continue;
+
+del2:
+		freeReplyObject(rp3);
+del:
+		redisCommandNR(r->ctx, "DEL %s", rp2->str);
+		redisCommandNR(r->ctx, "SREM calls %s", rp2->str);
+	}
+
+	freeReplyObject(rp);
+
+	return 0;
+
+err:
+	return -1;
+}

daemon/redis.h

@@ -10,19 +10,23 @@
 
 
 
+struct callmaster;
+
+
+
 struct redis {
 	char		host[32];
 	int		port;
 
 	redisContext	*ctx;
-	char		*key;
-	int		active:1;
+	int		db;
 };
 
 
 
 
-struct redis *redis_new(u_int32_t, u_int16_t, char *);
+struct redis *redis_new(u_int32_t, u_int16_t, int);
+int redis_restore(struct callmaster *);