diff --git a/Makefile b/Makefile index 21f03ac..5f0d3e2 100644 --- a/Makefile +++ b/Makefile @@ -11,14 +11,14 @@ OTR_OBJS = json.o \ file.o \ safewrite.o \ base64.o \ - ghash.o \ misc.o \ util.o \ storage.o -ifneq ($(HAVE_REDIS),no) - CFLAGS += -DHAVE_REDIS=1 - LIBS += -lhiredis +ifeq ($(HAVE_LMDB),yes) + CFLAGS += -DHAVE_LMDB=1 -Imdb/ + OTR_OBJS += gcache.o + LIBS += mdb/liblmdb.a endif ifeq ($(HAVE_HTTP),yes) @@ -38,15 +38,15 @@ geo.o: geo.h geo.c udata.h Makefile config.mk config.h geohash.o: geohash.h geohash.c udata.h Makefile config.mk file.o: file.h file.c config.h misc.h Makefile config.mk base64.o: base64.h base64.c -ghash.o: ghash.h ghash.c config.h udata.h misc.h Makefile config.mk +gcache.o: gcache.c gcache.h json.h safewrite.o: safewrite.h safewrite.c jget.o: jget.c jget.h json.h Makefile config.mk misc.o: misc.c misc.h udata.h Makefile config.mk http.o: http.c mongoose.h util.h http.h storage.h util.o: util.c util.h Makefile config.mk -ocat: ocat.o storage.o json.o geohash.o ghash.o mkpath.o util.o - $(CC) $(CFLAGS) -o ocat ocat.o storage.o json.o geohash.o ghash.o mkpath.o util.o $(LIBS) +ocat: ocat.o storage.o json.o geohash.o mkpath.o util.o gcache.o + $(CC) $(CFLAGS) -o ocat ocat.o storage.o json.o geohash.o mkpath.o util.o gcache.o $(LIBS) ocat.o: ocat.c storage.h storage.o: storage.c storage.h config.h util.h diff --git a/README.md b/README.md index b8bdc19..d1b93da 100644 --- a/README.md +++ b/README.md @@ -26,7 +26,7 @@ Specifying `--fields lat,tid,lon` will request just those JSON elements from _st We took a number of decisions for the design of the recorder and its utilities: -* Flat files. The filesystem is the database. Period. That's were everything is stored. It makes incremental backups, purging old data, manipulation via the Unix toolset easy. (Admittedly, for fast lookups you can employ Redis as a cache, but the final word is in the filesystem.) We considered all manner of databases and decided to keep this as simple and lightweight as possible. +* Flat files. The filesystem is the database. Period. That's were everything is stored. It makes incremental backups, purging old data, manipulation via the Unix toolset easy. (Admittedly, for fast lookups you can employ LMDB as a cache, but the final word is in the filesystem.) We considered all manner of databases and decided to keep this as simple and lightweight as possible. * Storage format is typically JSON because it's extensible. If we add an attribute to the JSON published by our apps, you have it right there. There's one slight exception: the monthly logs have a leading timestamp and a relative topic; see below. * File names are lower case. A user called `JaNe` with a device named `myPHONe` will be found in a file named `jane/myphone`. * All times are UTC (a.k.a. Zulu or GMT). We got sick and tired of converting stuff back and forth. It is up to the consumer of the data to convert to localtime if need be. @@ -49,7 +49,7 @@ As mentioned earlier, data is stored in files, and these files are relative to ` ## Requirements -* [hiredis](https://github.com/redis/hiredis) unless `HAVE_REDIS` is false. +* [lmdb](http://symas.com/mdb) unless `HAVE_LMDB` is false. ## Installation @@ -60,7 +60,7 @@ As mentioned earlier, data is stored in files, and these files are relative to ` ## Reverse Geo -If not disabled with option `-G`, the _recorder_ will attempt to perform a reverse-geo lookup on the location coordinates it obtains. This is stored in Redis (ghash:xxx) if available or in files. If a lookup is not possible, for example because you're over quota, the service isn't available, etc., _recorder_ keeps tracks of the coordinates which could *not* be resolved in a `missing` file: +If not disabled with option `-G`, the _recorder_ will attempt to perform a reverse-geo lookup on the location coordinates it obtains. This is stored in LMDB if it can be obtained. If a lookup is not possible, for example because you're over quota, the service isn't available, etc., _recorder_ keeps tracks of the coordinates which could *not* be resolved in a `missing` file: ``` $ cat store/ghash/missing @@ -74,15 +74,7 @@ This can be used to subsequently obtain said geo lookups. ## Monitoring -In order to monitor the _recorder_, whenever an MQTT message is received, the _recorder_ will add an epoch timestamp and the last received topic a Redis key (if configured) or a file otherwise. The Redis key looks like this: - -``` -redis 127.0.0.1:6379> hgetall ot-recorder-monitor -1) "time" -2) "1439738692" -3) "topic" -4) "owntracks/jjolie/ipad" -``` +In order to monitor the _recorder_, whenever an MQTT message is received, the _recorder_ will add an epoch timestamp and the last received topic to a file. The `monitor` file is located relative to STORE and contains a single line, the epoch timestamp at the moment of message reception and the topic separated from eachother by a single space: diff --git a/config.mk b/config.mk index 0d95c50..c1d207a 100644 --- a/config.mk +++ b/config.mk @@ -1,3 +1,4 @@ # Select features HAVE_REDIS ?= no HAVE_HTTP ?= yes +HAVE_LMDB ?= yes diff --git a/gcache.c b/gcache.c new file mode 100644 index 0000000..d220e48 --- /dev/null +++ b/gcache.c @@ -0,0 +1,241 @@ +#include +#include +#include +#include +#include +#include +#include "gcache.h" + +static int is_directory(char *path) +{ + struct stat sb; + + if (stat(path, &sb) != 0) + return (0); + return (S_ISDIR(sb.st_mode)); +} + +struct gcache *gcache_open(char *path, int rdonly) +{ + MDB_txn *txn = NULL; + int rc; + unsigned int flags = 0; + struct gcache *gc; + + if (!is_directory(path)) { + fprintf(stderr, "gcache_open: %s is not a directory\n", path); + return (NULL); + } + + if ((gc = malloc(sizeof (struct gcache))) == NULL) + return (NULL); + + memset(gc, 0, sizeof(struct gcache)); + + if (rdonly) { + flags |= MDB_RDONLY; + } + + rc = mdb_env_create(&gc->env); + if (rc != 0) { + fprintf(stderr, "%s\n", mdb_strerror(rc)); + free(gc); + return (NULL); + } + + mdb_env_set_mapsize(gc->env, LMDB_DB_SIZE); + + rc = mdb_env_open(gc->env, path, flags, 0664); + if (rc != 0) { + fprintf(stderr, "%s\n", mdb_strerror(rc)); + free(gc); + return (NULL); + } + + /* Open a pseudo TX so that we can open DBI */ + + mdb_txn_begin(gc->env, NULL, flags, &txn); + if (rc != 0) { + fprintf(stderr, "%s\n", mdb_strerror(rc)); + mdb_env_close(gc->env); + free(gc); + return (NULL); + } + + rc = mdb_dbi_open(txn, NULL, 0, &gc->dbi); + if (rc != 0) { + write(1, "HELLO\n", 6); + fprintf(stderr, "%s\n", mdb_strerror(rc)); + mdb_txn_abort(txn); + mdb_env_close(gc->env); + free(gc); + return (NULL); + } + + rc = mdb_txn_commit(txn); + if (rc != 0) { + fprintf(stderr, "commit after open %s\n", mdb_strerror(rc)); + mdb_env_close(gc->env); + free(gc); + return (NULL); + } + + return (gc); +} + +void gcache_close(struct gcache *gc) +{ + if (gc == NULL) + return; + + mdb_env_close(gc->env); + free(gc); +} + +int gcache_put(struct gcache *gc, char *ghash, char *payload) +{ + int rc; + MDB_val key, data; + MDB_txn *txn; + + if (gc == NULL) + return (1); + + rc = mdb_txn_begin(gc->env, NULL, 0, &txn); + if (rc != 0) + puts(mdb_strerror(rc)); + + key.mv_data = ghash; + key.mv_size = strlen(ghash); + data.mv_data = payload; + data.mv_size = strlen(payload) + 1; /* including nul-byte so we can + * later decode string directly + * from this buffer */ + + rc = mdb_put(txn, gc->dbi, &key, &data, 0); + if (rc != 0) + puts(mdb_strerror(rc)); + + rc = mdb_txn_commit(txn); + if (rc) { + fprintf(stderr, "mdb_txn_commit: (%d) %s\n", rc, mdb_strerror(rc)); + mdb_txn_abort(txn); + } + return (rc); +} + +int gcache_json_put(struct gcache *gc, char *ghash, JsonNode *geo) +{ + int rc; + MDB_val key, data; + MDB_txn *txn; + char *js; + + if (gc == NULL) + return (1); + + rc = mdb_txn_begin(gc->env, NULL, 0, &txn); + if (rc != 0) + puts(mdb_strerror(rc)); + + if ((js = json_stringify(geo, NULL)) == NULL) { + puts("CANIT stringify"); + return (1); + } + + key.mv_data = ghash; + key.mv_size = strlen(ghash); + data.mv_data = js; + data.mv_size = strlen(js) + 1; /* including nul-byte so we can + * later decode string directly + * from this buffer */ + + rc = mdb_put(txn, gc->dbi, &key, &data, 0); + if (rc != 0) + puts(mdb_strerror(rc)); + + rc = mdb_txn_commit(txn); + if (rc) { + fprintf(stderr, "mdb_txn_commit: (%d) %s\n", rc, mdb_strerror(rc)); + mdb_txn_abort(txn); + } + free(js); + return (rc); +} + +int gcache_get(struct gcache *gc, char *k) +{ + MDB_val key, data; + MDB_txn *txn; + int rc; + + if (gc == NULL) + return (1); + + rc = mdb_txn_begin(gc->env, NULL, MDB_RDONLY, &txn); + if (rc) { + fprintf(stderr, "gcache_get: cannot txn_begin: (%d) %s\n", rc, mdb_strerror(rc)); + return (1); + } + + key.mv_data = k; + key.mv_size = strlen(k); + + rc = mdb_get(txn, gc->dbi, &key, &data); + if (rc != 0) { + if (rc != MDB_NOTFOUND) { + printf("get: %s\n", mdb_strerror(rc)); + } else { + printf(" [%s] not found\n", k); + } + } else { + printf("%s\n", (char *)data.mv_data); + } + mdb_txn_commit(txn); + return (0); +} + +/* + * Attempt to get key `k` (a geohash string) from LMDB. If + * found, decode the JSON string in it and return a JSON + * object, else NULL. + */ + +JsonNode *gcache_json_get(struct gcache *gc, char *k) +{ + MDB_val key, data; + MDB_txn *txn; + int rc; + JsonNode *geo = NULL; + + if (gc == NULL) + return (NULL); + + rc = mdb_txn_begin(gc->env, NULL, MDB_RDONLY, &txn); + if (rc) { + fprintf(stderr, "gcache_get: cannot txn_begin: (%d) %s\n", rc, mdb_strerror(rc)); + return (NULL); + } + + key.mv_data = k; + key.mv_size = strlen(k); + + rc = mdb_get(txn, gc->dbi, &key, &data); + if (rc != 0) { + if (rc != MDB_NOTFOUND) { + fprintf(stderr, "gcache_json_get(%s): %s\n", k, mdb_strerror(rc)); + } else { + // printf(" [%s] not found\n", k); + geo = NULL; + } + } else { + // printf("%s\n", (char *)data.mv_data); + if ((geo = json_decode((char *)data.mv_data)) == NULL) { + fprintf(stderr, "Cannot decode JSON from lmdb\n"); + } + } + + mdb_txn_commit(txn); + + return (geo); +} diff --git a/gcache.h b/gcache.h new file mode 100644 index 0000000..0117d6e --- /dev/null +++ b/gcache.h @@ -0,0 +1,24 @@ +#ifndef _GCACHE_H_INCLUDED_ +# define _GCACHE_H_INCLUDED_ + +#ifdef HAVE_LMDB + +#include "json.h" +#include "lmdb.h" + +#define LMDB_DB_SIZE (120 * 1024 * 1024) + +struct gcache { + MDB_env *env; + MDB_dbi dbi; +}; + +struct gcache *gcache_open(char *path, int rdonly); +void gcache_close(struct gcache *); +int gcache_put(struct gcache *, char *ghash, char *payload); +int gcache_json_put(struct gcache *, char *ghash, JsonNode *geo); +int gcache_get(struct gcache *, char *key); +JsonNode *gcache_json_get(struct gcache *, char *key); + +#endif +#endif diff --git a/ghash.c b/ghash.c deleted file mode 100644 index 23305d7..0000000 --- a/ghash.c +++ /dev/null @@ -1,200 +0,0 @@ -#include -#include -#include "ghash.h" -#include "misc.h" - -#ifdef HAVE_REDIS - -void redis_ping(redisContext **redis) -{ - struct timeval timeout = { 1, 500000 }; // 1.5 seconds - redisReply *r; - int i = 0; - - do { - if ((r = redisCommand(*redis,"PING")) != NULL) { - // printf("PING: %s\n", r->str); - freeReplyObject(r); - return; - } - fprintf(stderr, "REDIS: %d %s\n", (*redis)->err, (*redis)->errstr); - - *redis = redisConnectWithTimeout("localhost", 6379, timeout); - - fprintf(stderr, "Reconnecting to Redis...\n"); - sleep(5); - } while (i++ < 10); - -} - -void ghash_store_redis(redisContext **redis, char *ghash, char *addr, char *cc) -{ - redisReply *r; - - redis_ping(redis); - - r = redisCommand(*redis, "HMSET ghash:%s cc %s addr %s", ghash, cc, addr); - if (r) /* FIXME */ - return; -} - - -void last_storeredis(redisContext **redis, char *username, char *device, char *jsonstring) -{ - redisReply *r; - - redis_ping(redis); - - r = redisCommand(*redis, "SET lastpos:%s-%s %s", username, device, jsonstring); - if (r) /* FIXME */ - return; -} -int ghash_get_redis_cache(redisContext **redis, char *ghash, UT_string *addr, UT_string *cc) -{ - redisReply *reply; - int found = FALSE; - - redis_ping(redis); - - reply = redisCommand(*redis, "HGETALL ghash:%s", ghash); - if (reply == NULL) { - fprintf(stderr, "REDIS: %d %s\n", (*redis)->err, (*redis)->errstr); - return (FALSE); - - } - if ( reply->type == REDIS_REPLY_ERROR ) { - fprintf(stderr, "Error: %s\n", reply->str ); - return (FALSE); - } - else if ( reply->type != REDIS_REPLY_ARRAY ) - printf( "Unexpected type: %d\n", reply->type ); - if (reply->type == REDIS_REPLY_ARRAY) { - int i; - char *key, *val; - - if (reply->elements >= 1) { - for (i = 0; i < (reply->elements - 1); i += 2) { - key = reply->element[i]->str; - val = reply->element[i+1]->str; - - if (!strcmp(key, "addr")) - utstring_printf(addr, "%s", val); - else if (!strcmp(key, "cc")) - utstring_printf(cc, "%s", val); - } - found = TRUE; - } - } - - freeReplyObject(reply); - - return (found); -} - -void monitor_update(struct udata *ud, time_t now, char *topic) -{ - redisReply *r; - - redis_ping(&ud->redis); - - r = redisCommand(ud->redis, "HMSET ot-recorder-monitor time %ld topic %s", now, topic); - if (r) /* FIXME */ - return; - -} - -#endif /* !HAVE_REDIS */ - -int ghash_readcache(struct udata *ud, char *ghash, UT_string *addr, UT_string *cc) -{ - int cached = FALSE; - char gfile[BUFSIZ]; - FILE *fp; - - /* FIXME */ - -#ifdef HAVE_REDIS - if (ud->useredis) { - if (ghash_get_redis_cache( &ud->redis, ghash, addr, cc) == TRUE) { - return (TRUE); - } - } -#endif - - if (ud->usefiles) { - /* if ghash file is available, read cc:addr into that */ - snprintf(gfile, BUFSIZ, "%s/ghash/%-3.3s/%s.json", STORAGEDIR, ghash, ghash); - - // fprintf(stderr, "Reading GhashCache from %s\n", gfile); - if ((fp = fopen(gfile, "r")) != NULL) { - char buf[BUFSIZ]; - - /* FIXME: read JSON */ - if (fgets(buf, sizeof(buf), fp) != NULL) { - JsonNode *json, *j; - - if ((json = json_decode(buf)) == NULL) { - puts(" FIXME: can't decode JSON"); - } else { - - if ((j = json_find_member(json, "cc")) != NULL) { - if (j->tag == JSON_STRING) { - utstring_printf(cc, "%s", j->string_); - } - } - if ((j = json_find_member(json, "addr")) != NULL) { - if (j->tag == JSON_STRING) { - utstring_printf(addr, "%s", j->string_); - } - } - - json_delete(json); - cached = TRUE; - } - - fclose(fp); - } - } else { - // fprintf(stderr, "Not cached: ghash: %s\n", gfile); - } - } - - return (cached); -} - - -void ghash_storecache(struct udata *ud, JsonNode *geo, char *ghash, char *addr, char *cc) -{ - char gfile[BUFSIZ]; - FILE *fp; - -#ifdef HAVE_REDIS - if (ud->useredis) { - ghash_store_redis(&ud->redis, ghash, addr, cc); - } -#endif - - if (ud->usefiles) { - snprintf(gfile, BUFSIZ, "%s/ghash", STORAGEDIR); - if (mkpath(gfile) < 0) { - perror(gfile); - } else { - char *js; - - if ((js = json_stringify(geo, NULL)) != NULL) { - snprintf(gfile, BUFSIZ, "%s/ghash/%-3.3s", STORAGEDIR, ghash); - if (mkpath(gfile) != 0) { - perror(gfile); - return; - } - snprintf(gfile, BUFSIZ, "%s/ghash/%-3.3s/%s.json", STORAGEDIR, ghash, ghash); - if ((fp = fopen(gfile, "w")) != NULL) { - fprintf(fp, "%s\n", js); - fclose(fp); - } - - free(js); - } - } - } -} diff --git a/ghash.h b/ghash.h deleted file mode 100644 index 29f5f21..0000000 --- a/ghash.h +++ /dev/null @@ -1,20 +0,0 @@ -#include "config.h" -#include "udata.h" -#ifdef HAVE_REDIS -# include -#endif -#include "geohash.h" -#include "utstring.h" -#include "json.h" - -#ifndef TRUE -# define TRUE 1 -# define FALSE 0 -#endif - -#ifdef HAVE_REDIS -void redis_ping(redisContext **redis); -void last_storeredis(redisContext **redis, char *username, char *device, char *jsonstring); -#endif -int ghash_readcache(struct udata *ud, char *ghash, UT_string *addr, UT_string *cc); -void ghash_storecache(struct udata *ud, JsonNode *geo, char *ghash, char *addr, char *cc); diff --git a/misc.c b/misc.c index e766ba3..c1484c8 100644 --- a/misc.c +++ b/misc.c @@ -32,23 +32,16 @@ char *bindump(char *buf, long buflen) void monitorhook(struct udata *userdata, time_t now, char *topic) { - struct udata *ud = (struct udata *)userdata; + // struct udata *ud = (struct udata *)userdata; -#ifdef HAVE_REDIS - if (ud->useredis) { - monitor_update(ud, now, topic); - return; - } -#else - if (ud->usefiles) { - char mpath[BUFSIZ]; - static UT_string *us = NULL; + /* TODO: add monitor hook to a "monitor" key in LMDB ? */ - utstring_renew(us); - utstring_printf(us, "%ld %s\n", now, topic); + char mpath[BUFSIZ]; + static UT_string *us = NULL; - sprintf(mpath, "%s/monitor", STORAGEDIR); - safewrite(mpath, utstring_body(us)); - } -#endif + utstring_renew(us); + utstring_printf(us, "%ld %s\n", now, topic); + + sprintf(mpath, "%s/monitor", STORAGEDIR); + safewrite(mpath, utstring_body(us)); } diff --git a/ocat.c b/ocat.c index d3f8d8c..6d8e78a 100644 --- a/ocat.c +++ b/ocat.c @@ -235,6 +235,8 @@ int main(int argc, char **argv) argc -= (optind); argv += (optind); + storage_init(); + if (killdata) { JsonNode *obj; //, *killed, *f; diff --git a/ot-recorder.c b/ot-recorder.c index bfec6a9..a9674f0 100644 --- a/ot-recorder.c +++ b/ot-recorder.c @@ -14,14 +14,17 @@ #include "utstring.h" #include "utarray.h" #include "geo.h" -#include "ghash.h" #include "config.h" +#include "geohash.h" #include "file.h" #include "safewrite.h" #include "base64.h" #include "misc.h" #include "util.h" #include "storage.h" +#ifdef HAVE_LMDB +# include "gcache.h" +#endif #ifdef HAVE_HTTP # include "http.h" #endif @@ -136,9 +139,6 @@ int do_info(void *userdata, UT_string *username, UT_string *device, char *payloa struct udata *ud = (struct udata *)userdata; JsonNode *json, *j; static UT_string *name = NULL, *face = NULL; -#ifdef HAVE_REDIS - redisReply *r; -#endif FILE *fp; char *img; int rc = FALSE; @@ -162,15 +162,13 @@ int do_info(void *userdata, UT_string *username, UT_string *device, char *payloa goto cleanup; } - if (ud->usefiles) { - /* I know the payload is valid JSON: write card */ + /* I know the payload is valid JSON: write card */ - if ((fp = pathn("wb", "cards", username, NULL, "json")) != NULL) { - fprintf(fp, "%s\n", payload); - fclose(fp); - } - rc = TRUE; + if ((fp = pathn("wb", "cards", username, NULL, "json")) != NULL) { + fprintf(fp, "%s\n", payload); + fclose(fp); } + rc = TRUE; if ((j = json_find_member(json, "name")) != NULL) { if (j->tag == JSON_STRING) { @@ -190,7 +188,7 @@ int do_info(void *userdata, UT_string *username, UT_string *device, char *payloa fprintf(stderr, "* CARD: %s-%s %s\n", utstring_body(username), utstring_body(device), utstring_body(name)); -#ifdef HAVE_REDIS +#ifdef HAVE_REDIS /* TODO: LMDB? */ if (ud->useredis) { redis_ping(&ud->redis); r = redisCommand(ud->redis, "HMSET card:%s name %s face %s", utstring_body(username), utstring_body(name), utstring_body(face)); @@ -203,14 +201,12 @@ int do_info(void *userdata, UT_string *username, UT_string *device, char *payloa int imglen; if ((imglen = base64_decode(utstring_body(face), img)) > 0) { - if (ud->usefiles) { - if ((fp = pathn("wb", "photos", username, NULL, "png")) != NULL) { - fwrite(img, sizeof(char), imglen, fp); - fclose(fp); - } + if ((fp = pathn("wb", "photos", username, NULL, "png")) != NULL) { + fwrite(img, sizeof(char), imglen, fp); + fclose(fp); } -#ifdef HAVE_REDIS +#ifdef HAVE_REDIS /* TODO: LMDB ? */ if (ud->useredis) { /* Add photo (binary) to Redis as photo:username */ redis_ping(&ud->redis); @@ -250,13 +246,11 @@ void do_msg(void *userdata, UT_string *username, UT_string *device, char *payloa goto cleanup; } - if (ud->usefiles) { - /* I know the payload is valid JSON: write message */ + /* I know the payload is valid JSON: write message */ - if ((fp = pathn("ab", "msg", username, NULL, "json")) != NULL) { - fprintf(fp, "%s\n", payload); - fclose(fp); - } + if ((fp = pathn("ab", "msg", username, NULL, "json")) != NULL) { + fprintf(fp, "%s\n", payload); + fclose(fp); } fprintf(stderr, "* MSG: %s-%s\n", utstring_body(username), utstring_body(device)); @@ -433,12 +427,10 @@ void on_message(struct mosquitto *mosq, void *userdata, const struct mosquitto_m } - if (ud->usefiles) { - if ((fp = pathn("a", "rec", username, device, "rec")) != NULL) { + if ((fp = pathn("a", "rec", username, device, "rec")) != NULL) { - fprintf(fp, RECFORMAT, isotime(now), utstring_body(reltopic), bindump(m->payload, m->payloadlen)); - fclose(fp); - } + fprintf(fp, RECFORMAT, isotime(now), utstring_body(reltopic), bindump(m->payload, m->payloadlen)); + fclose(fp); } mosquitto_sub_topic_tokens_free(&topics, count); @@ -498,17 +490,25 @@ void on_message(struct mosquitto *mosq, void *userdata, const struct mosquitto_m cached = FALSE; if (ud->revgeo == TRUE) { + JsonNode *geo, *j; - /* FIXME: */ + if ((geo = gcache_json_get(ud->gc, utstring_body(ghash))) != NULL) { + /* Habemus cached data */ + + puts("I HAVE THIS IN LMDB"); - cached = ghash_readcache(ud, utstring_body(ghash), addr, cc); - if (!cached) { - JsonNode *geo; + cached = TRUE; + if ((j = json_find_member(geo, "cc")) != NULL) { + utstring_printf(cc, "%s", j->string_); + } + if ((j = json_find_member(geo, "addr")) != NULL) { + utstring_printf(addr, "%s", j->string_); + } + + } else { if ((geo = revgeo(lat, lon, addr, cc)) != NULL) { - // fprintf(stderr, "REVGEO: %s\n", utstring_body(addr)); - ghash_storecache(ud, geo, utstring_body(ghash), utstring_body(addr), utstring_body(cc)); - json_delete(geo); + gcache_json_put(ud->gc, utstring_body(ghash), geo); } else { /* We didn't obtain reverse Geo, maybe because of over * quota; make a note of the missing geohash */ @@ -532,13 +532,10 @@ void on_message(struct mosquitto *mosq, void *userdata, const struct mosquitto_m * We have exactly three topic parts (owntracks/user/device), and valid JSON. */ - /* - * Add a few bits to the JSON, and record it on a per-user/device basis. - json_append_member(json, "ghash", json_mkstring(utstring_body(ghash))); - */ - + /* TODO: shall we store last positions in LMDB? */ if ((jsonstring = json_stringify(json, NULL)) != NULL) { + char *js; #ifdef HAVE_REDIS if (ud->useredis) { @@ -547,35 +544,31 @@ void on_message(struct mosquitto *mosq, void *userdata, const struct mosquitto_m } #endif - if (ud->usefiles) { - char *js; + if ((fp = pathn("a", "rec", username, device, "rec")) != NULL) { - if ((fp = pathn("a", "rec", username, device, "rec")) != NULL) { + fprintf(fp, RECFORMAT, isotime(now), "*", jsonstring); + fclose(fp); + } - fprintf(fp, RECFORMAT, isotime(now), "*", jsonstring); - fclose(fp); + + /* Keep track of original username & device name in LAST. */ + json_append_member(json, "username", json_mkstring(utstring_body(username))); + json_append_member(json, "device", json_mkstring(utstring_body(device))); + json_append_member(json, "topic", json_mkstring(m->topic)); + json_append_member(json, "ghash", json_mkstring(utstring_body(ghash))); + + if ((js = json_stringify(json, NULL)) != NULL) { + /* Now safewrite the last location */ + utstring_printf(ts, "%s/last/%s/%s", + STORAGEDIR, utstring_body(username), utstring_body(device)); + if (mkpath(utstring_body(ts)) < 0) { + perror(utstring_body(ts)); } + utstring_printf(ts, "/%s-%s.json", + utstring_body(username), utstring_body(device)); - - /* Keep track of original username & device name in LAST. */ - json_append_member(json, "username", json_mkstring(utstring_body(username))); - json_append_member(json, "device", json_mkstring(utstring_body(device))); - json_append_member(json, "topic", json_mkstring(m->topic)); - json_append_member(json, "ghash", json_mkstring(utstring_body(ghash))); - - if ((js = json_stringify(json, NULL)) != NULL) { - /* Now safewrite the last location */ - utstring_printf(ts, "%s/last/%s/%s", - STORAGEDIR, utstring_body(username), utstring_body(device)); - if (mkpath(utstring_body(ts)) < 0) { - perror(utstring_body(ts)); - } - utstring_printf(ts, "/%s-%s.json", - utstring_body(username), utstring_body(device)); - - safewrite(utstring_body(ts), js); - free(js); - } + safewrite(utstring_body(ts), js); + free(js); } free(jsonstring); } @@ -617,16 +610,16 @@ void on_connect(struct mosquitto *mosq, void *userdata, int rc) void on_disconnect(struct mosquitto *mosq, void *userdata, int reason) { -#ifdef HAVE_REDIS +#ifdef HAVE_LMDB struct udata *ud = (struct udata *)userdata; #endif syslog(LOG_INFO, "Disconnected. Reason: %d [%s]", reason, mosquitto_strerror(reason)); if (reason == 0) { // client wish - #ifdef HAVE_REDIS - redisFree(ud->redis); - #endif +#ifdef HAVE_LMDB + gcache_close(ud->gc); +#endif } } @@ -642,10 +635,8 @@ void usage(char *prog) printf("Usage: %s [options..] topic [topic ...]\n", prog); printf(" --help -h this message\n"); printf(" --storage -S storage dir (./store)\n"); - printf(" --nofiles -F do not use file storage\n"); printf(" --norevgeo -G disable ghash to reverge-geo lookups\n"); printf(" --skipdemo -D do not handle objects with _demo\n"); - printf(" --noredis -N disable Redis even if compiled in\n"); printf(" --useretained -R process retained messages (default: no)\n"); printf(" --clientid -i MQTT client-ID\n"); printf(" --qos -q MQTT QoS (dflt: 2)\n"); @@ -677,19 +668,17 @@ int main(int argc, char **argv) int http_port = 8083; char *doc_root = "./wdocs"; char *http_host = "localhost"; -#endif -#ifdef HAVE_REDIS - struct timeval timeout = { 1, 500000 }; // 1.5 seconds #endif char *progname = *argv; udata.qos = DEFAULT_QOS; - udata.usefiles = TRUE; udata.ignoreretained = TRUE; udata.pubprefix = NULL; udata.skipdemo = TRUE; - udata.useredis = TRUE; udata.revgeo = TRUE; +#ifdef HAVE_LMDB + udata.gc = NULL; +#endif #ifdef HAVE_HTTP mgserver = udata.server = mg_create_server(NULL, ev_handler); #endif @@ -717,9 +706,7 @@ int main(int argc, char **argv) static struct option long_options[] = { { "help", no_argument, 0, 'h'}, { "skipdemo", no_argument, 0, 'D'}, - { "nofiles", no_argument, 0, 'F'}, { "norevgeo", no_argument, 0, 'G'}, - { "noredis", no_argument, 0, 'N'}, { "useretained", no_argument, 0, 'R'}, { "clientid", required_argument, 0, 'i'}, { "pubprefix", required_argument, 0, 'P'}, @@ -737,7 +724,7 @@ int main(int argc, char **argv) }; int optindex = 0; - ch = getopt_long(argc, argv, "hDFGNRi:P:q:S:H:p:A:", long_options, &optindex); + ch = getopt_long(argc, argv, "hDGRi:P:q:S:H:p:A:", long_options, &optindex); if (ch == -1) break; @@ -760,9 +747,6 @@ int main(int argc, char **argv) case 'D': ud->skipdemo = FALSE; break; - case 'F': - ud->usefiles = FALSE; - break; case 'G': ud->revgeo = FALSE; break; @@ -770,9 +754,6 @@ int main(int argc, char **argv) utstring_clear(clientid); utstring_printf(clientid, "%s", optarg); break; - case 'N': - ud->useredis = FALSE; - break; case 'P': udata.pubprefix = strdup(optarg); /* TODO: do we want this? */ break; @@ -816,6 +797,19 @@ int main(int argc, char **argv) syslog(LOG_DEBUG, "starting"); if (ud->revgeo == TRUE) { +#ifdef HAVE_LMDB + char db_filename[BUFSIZ], *pa; + + snprintf(db_filename, BUFSIZ, "%s/ghash", STORAGEDIR); + pa = strdup(db_filename); + mkpath(pa); + free(pa); + udata.gc = gcache_open(db_filename, FALSE); + if (udata.gc == NULL) { + syslog(LOG_WARNING, "Can't initialize gcache in %s", db_filename); + exit(1); + } +#endif revgeo_init(); } @@ -825,10 +819,6 @@ int main(int argc, char **argv) signal(SIGINT, catcher); -#ifdef HAVE_REDIS - ud->redis = redisConnectWithTimeout("localhost", 6379, timeout); -#endif - mosq = mosquitto_new(utstring_body(clientid), CLEAN_SESSION, (void *)&udata); if (!mosq) { fprintf(stderr, "Error: Out of memory.\n"); diff --git a/storage.c b/storage.c index 50d36db..2f4d94d 100644 --- a/storage.c +++ b/storage.c @@ -20,22 +20,23 @@ char STORAGEDIR[BUFSIZ] = "./store"; #define LINESIZE 8192 -int ghash_readcache(struct udata *ud, char *ghash, UT_string *addr, UT_string *cc); +static struct gcache *gc = NULL; + +void storage_init() +{ + char path[BUFSIZ]; + + snprintf(path, BUFSIZ, "%s/ghash", STORAGEDIR); + gc = gcache_open(path, TRUE); +} void get_geo(JsonNode *o, char *ghash) { - static UT_string *addr = NULL, *cc = NULL; - static struct udata udata; + JsonNode *geo; - /* FIXME!!!! */ - udata.usefiles = 1; - - utstring_renew(addr); - utstring_renew(cc); - - if (ghash_readcache(&udata, ghash, addr, cc) == 1) { - json_append_member(o, "addr", json_mkstring(utstring_body(addr))); - json_append_member(o, "cc", json_mkstring(utstring_body(cc))); + if ((geo = gcache_json_get(gc, ghash)) != NULL) { + json_copy_to_object(o, geo, FALSE); + json_delete(geo); } } diff --git a/storage.h b/storage.h index 93c4c2a..8377187 100644 --- a/storage.h +++ b/storage.h @@ -2,6 +2,7 @@ # define _STORAGE_H_INCL_ #include +#include "gcache.h" #include "json.h" typedef enum { @@ -19,5 +20,6 @@ JsonNode *geo_json(JsonNode *json); JsonNode *kill_datastore(char *username, char *device); JsonNode *last_users(); char *gpx_string(JsonNode *json); +void storage_init(); #endif diff --git a/udata.h b/udata.h index 27c1c3a..4681cae 100644 --- a/udata.h +++ b/udata.h @@ -3,27 +3,25 @@ #include "config.h" #include "utarray.h" -#ifdef HAVE_REDIS -# include -#endif #ifdef HAVE_HTTP # include "mongoose.h" #endif +#ifdef HAVE_LMDB +# include "gcache.h" +#endif struct udata { UT_array *topics; /* Array of topics to subscribe to */ -#ifdef HAVE_REDIS - redisContext *redis; -#endif - int usefiles; /* True if files to be written */ int ignoreretained; /* True if retained messages should be ignored */ char *pubprefix; /* If not NULL (default), republish modified payload to /topic */ int skipdemo; /* True if _demo users are to be skipped */ - int useredis; /* True if we should do Redis (if we have it) */ int revgeo; /* True (default) if we should do reverse Geo lookups */ int qos; /* Subscribe QoS */ +#ifdef HAVE_LMDB + struct gcache *gc; +#endif #ifdef HAVE_HTTP struct mg_server *server; #endif