diff --git a/daemon/main.c b/daemon/main.c index 8becb4db7..ba499d38e 100644 --- a/daemon/main.c +++ b/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) diff --git a/daemon/redis.c b/daemon/redis.c index 875ab2fcf..cae5eb57f 100644 --- a/daemon/redis.c +++ b/daemon/redis.c @@ -2,27 +2,31 @@ #include #include #include +#include #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; +} diff --git a/daemon/redis.h b/daemon/redis.h index 887773046..f06c1f871 100644 --- a/daemon/redis.h +++ b/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 *);