diff --git a/Makefile b/Makefile index a209e24..e5a0be4 100644 --- a/Makefile +++ b/Makefile @@ -1,7 +1,7 @@ include config.mk -CFLAGS =-Wall -Werror $(MOSQUITTO_INC) -LIBS = $(MORELIBS) $(MOSQUITTO_LIB) -lmosquitto -lm +CFLAGS =-Wall -Werror +LIBS = $(MORELIBS) -lm LIBS += -lcurl -lconfig TARGETS= @@ -18,6 +18,12 @@ OTR_EXTRA_OBJS = CFLAGS += -DGHASHPREC=$(GHASHPREC) +ifeq ($(WITH_MQTT),yes) + CFLAGS += -DWITH_MQTT=1 + CFLAGS += $(MOSQUITTO_INC) + LIBS += $(MOSQUITTO_LIB) -lmosquitto -lm +endif + ifeq ($(WITH_PING),yes) CFLAGS += -DWITH_PING=1 endif diff --git a/README.md b/README.md index 05c13db..e00f8f6 100644 --- a/README.md +++ b/README.md @@ -2,7 +2,7 @@ ![Recorder logo](assets/recorder-logo-192.png) -The _OwnTracks Recorder_ is a lightweight program for storing and accessing location data published via MQTT by the [OwnTracks](http://owntracks.org) apps. It is a compiled program which is easily to install and operate even on low-end hardware, and it doesn't require external an external database. It is also suited for you to record and store the data you publish via our [Hosted mode](http://owntracks.org/booklet/features/hosted/). +The _OwnTracks Recorder_ is a lightweight program for storing and accessing location data published via MQTT (or HTTP) by the [OwnTracks](http://owntracks.org) apps. It is a compiled program which is easily to install and operate even on low-end hardware, and it doesn't require external an external database. It is also suited for you to record and store the data you publish via our [Hosted mode](http://owntracks.org/booklet/features/hosted/). ![Architecture of the Recorder](assets/ot-recorder.png) @@ -14,7 +14,7 @@ We developed the _recorder_ as a one-stop solution to storing location data publ The _recorder_ serves two purposes: -1. It subscribes to an MQTT broker and reads messages published from the OwnTracks apps, storing these in a particular fashion into what we call the _store_ which is basically a bunch of plain files on the file system. +1. It subscribes to an MQTT broker and reads messages published from the OwnTracks apps, storing these in a particular fashion into what we call the _store_ which is basically a bunch of plain files on the file system. Alternatively the Recorder can listen on HTTP for OwnTracks-type JSON messages POSTed to its HTTP server. 2. It provides a Web server which serves static pages, a REST API you use to request data from the _store_, and a Websocket server. The distribution comes with a few examples of how to access the data through its HTTP interface (REST API). In particular a _table_ of last locations has been made available as well as a _live map_ which updates via the _recorder_'s Websocket interface when location publishes are received. In addition we provide maps with last points or tracks using the GeoJSON produced by the _recorder_. @@ -870,6 +870,19 @@ location /owntracks/static/ { You would then visit `http://example.com/owntracks/view/loire` to see the `loire` view, assuming `example.com` is your proxy. +## HTTP mode + +If enabled at compile time (`WITH_HTTP`), the Recorder will accept OwnTracks-type JSON payloads via HTTP at the URL endpoint `/pub&u=username&d=device`. You specify the username with the `u` parameter and the device name with the `d` parameter. (Alternatively you can provide `X-Limit-U` and `X-Limit-D` as headers with the username and device name respectively.) If unspecified, the username defaults to `owntracks` and the device to `phone`. For example: + +``` +curl --data "${payload}" 'http://127.0.0.1:8085/pub?u=jane&d=3s' +curl -H 'X-Limit-U: jane' -H 'X-Limit-D: 3s' --data "${payload}" 'http://127.0.0.1:8085/pub' +``` + +The content of the request is used by the Recorder as though it had arrived as an MQTT message. + +If the Recorder is compiled without specifying `WITH_MQTT` at build time, support for MQTT is disabled completely. + ## Advanced topics ### The LMDB database diff --git a/config.mk.in b/config.mk.in index bfeb64f..a3c015f 100644 --- a/config.mk.in +++ b/config.mk.in @@ -2,6 +2,9 @@ INSTALLDIR = /usr/local +# Do you want support for MQTT? +WITH_MQTT ?= yes + # Do you want recorder's built-in HTTP REST API? WITH_HTTP ?= yes diff --git a/http.c b/http.c index 5734cf1..84ecaee 100644 --- a/http.c +++ b/http.c @@ -23,6 +23,7 @@ #include #include #include +#include "recorder.h" #include "json.h" #include "util.h" #include "misc.h" @@ -372,6 +373,47 @@ static int send_status(struct mg_connection *conn, int status, char *text) return (MG_TRUE); } +/* + * Invoked from an HTTP POST to /pub?u=username&d=devicename + * We need u and d in order to contruct a topic name. Obtain + * the content of the POST request and give it to the recorder + * to do the needful. :) + */ + +static int dopublish(struct mg_connection *conn, const char *uri) +{ + struct udata *ud = (struct udata *)conn->server_param; + char *payload, *u, *d; + static UT_string *topic = NULL; + + + if ((u = field(conn, "u")) == NULL) { + u = strdup("owntracks"); + } + + if ((d = field(conn, "d")) == NULL) { + d = strdup("phone"); + } + + utstring_renew(topic); + utstring_printf(topic, "owntracks/%s/%s", u, d); + free(u); + free(d); + + + payload = malloc(conn->content_len + 1); + memcpy(payload, conn->content, conn->content_len); + payload[conn->content_len] = 0; + + debug(ud, "HTTPPUB clen=%zu, topic=%s", conn->content_len, UB(topic)); + + handle_message(ud, UB(topic), payload, conn->content_len, 0); + + free(payload); + + return json_response(conn, NULL); +} + /* * Procure back-end data for a VIEW. `view' is the JSON view from which * we obtain who/what to get. This returns a JSON array of location @@ -1031,6 +1073,12 @@ int ev_handler(struct mg_connection *conn, enum mg_event ev) + if (!strcmp(conn->request_method, "POST")) { + if (!strcmp(conn->uri, "/pub")) { + return dopublish(conn, conn->uri + strlen("/pub")); + } + } + if (!strcmp(conn->request_method, "POST")) { #ifdef WITH_LMDB diff --git a/ocat.c b/ocat.c index 21b519d..6321d59 100644 --- a/ocat.c +++ b/ocat.c @@ -23,7 +23,9 @@ #include #include #include -#include +#if WITH_MQTT +# include +#endif #include "json.h" #include "storage.h" #include "util.h" @@ -116,10 +118,12 @@ void print_versioninfo() printf("\tGHASHPREC = %d\n", GHASHPREC); printf("\tDEFAULT_HISTORY_HOURS = %d\n", DEFAULT_HISTORY_HOURS); printf("\tJSON_INDENT = \"%s\"\n", (JSON_INDENT) ? JSON_INDENT : "NULL"); +#if WITH_MQTT printf("\tLIBMOSQUITTO_VERSION = %d.%d.%d\n", LIBMOSQUITTO_MAJOR, LIBMOSQUITTO_MINOR, LIBMOSQUITTO_REVISION); +#endif #ifdef WITH_LMDB printf("\tMDB VERSION = %s\n", MDB_VERSION_STRING); #endif diff --git a/recorder.c b/recorder.c index 9df763b..84edee8 100644 --- a/recorder.c +++ b/recorder.c @@ -24,7 +24,9 @@ #include #include #include -#include +#if WITH_MQTT +# include +#endif #include #include #include @@ -152,6 +154,7 @@ int do_info(void *userdata, UT_string *username, UT_string *device, JsonNode *js return (rc); } +#ifdef WITH_MQTT void republish(struct mosquitto *mosq, struct udata *userdata, char *username, char *topic, double lat, double lon, char *cc, char *addr, long tst, char *t) { struct udata *ud = (struct udata *)userdata; @@ -190,6 +193,7 @@ void republish(struct mosquitto *mosq, struct udata *userdata, char *username, c json_delete(json); } +#endif /* WITH_MQTT */ /* * Quickly check wheterh the payload looks like @@ -594,7 +598,7 @@ void handle_message(void *userdata, char *topic, char *payload, size_t payloadle double lat, lon, acc; long tst; struct udata *ud = (struct udata *)userdata; - char **topics; + char *topics[42]; int count = 0, cached; static UT_string *basetopic = NULL, *username = NULL, *device = NULL, *addr = NULL, *cc = NULL, *ghash = NULL, *ts = NULL; static UT_string *reltopic = NULL; @@ -617,6 +621,7 @@ void handle_message(void *userdata, char *topic, char *payload, size_t payloadle time(&now); monitorhook(ud, now, topic); + debug(ud, "%s (plen=%d, r=%d)", topic, payloadlen, retain); if (payloadlen == 0) { return; } @@ -633,7 +638,7 @@ void handle_message(void *userdata, char *topic, char *payload, size_t payloadle utstring_renew(username); utstring_renew(device); - if (mosquitto_sub_topic_tokenise(topic, &topics, &count) != MOSQ_ERR_SUCCESS) { + if ((count = splitter(topic, "/", topics)) == -1) { return; } @@ -649,7 +654,7 @@ void handle_message(void *userdata, char *topic, char *payload, size_t payloadle } if (count - skipslash < 3) { fprintf(stderr, "Ignoring short topic %s\n", topic); - mosquitto_sub_topic_tokens_free(&topics, count); + splitterfree(topics); return; } @@ -702,7 +707,7 @@ void handle_message(void *userdata, char *topic, char *payload, size_t payloadle #endif - mosquitto_sub_topic_tokens_free(&topics, count); + splitterfree(topics); /* * Now let's see if this contains some sort of valid JSON @@ -1129,6 +1134,8 @@ void handle_message(void *userdata, char *topic, char *payload, size_t payloadle if (_typestr) free(_typestr); } +#ifdef WITH_MQTT + void on_message(struct mosquitto *mosq, void *userdata, const struct mosquitto_message *m) { struct udata *ud = (struct udata *)userdata; @@ -1183,6 +1190,8 @@ void on_disconnect(struct mosquitto *mosq, void *userdata, int reason) } } +#endif /* WITH_MQTT */ + static void catcher(int sig) { fprintf(stderr, "Going down on signal %d\n", sig); @@ -1191,17 +1200,25 @@ static void catcher(int sig) void usage(char *prog) { - printf("Usage: %s [options..] topic [topic ...]\n", prog); + printf("Usage: %s [options..] ", prog); +#ifdef WITH_MQTT + printf("topic [topic ...]\n"); +#else + printf("\n"); +#endif printf(" --help -h this message\n"); printf(" --storage -S storage dir (%s)\n", STORAGEDEFAULT); printf(" --norevgeo -G disable ghash to reverge-geo lookups\n"); printf(" --skipdemo -D do handle objects with _demo (default: don't)\n"); +#if WITH_MQTT printf(" --useretained -R process retained messages (default: no)\n"); printf(" --clientid -i MQTT client-ID\n"); printf(" --qos -q MQTT QoS (dflt: 2)\n"); printf(" --pubprefix -P republish prefix (dflt: no republish)\n"); printf(" --host -H MQTT host (localhost)\n"); printf(" --port -p MQTT port (1883)\n"); + printf(" --hosted use OwnTracks Hosted\n"); +#endif printf(" --logfacility syslog facility (local0)\n"); printf(" --quiet disable printing of messages to stdout\n"); printf(" --initialize initialize storage\n"); @@ -1215,15 +1232,15 @@ void usage(char *prog) printf(" --lua-script path to Lua script. If unset, no Lua hooks\n"); #endif printf(" --precision ghash precision (dflt: %d)\n", GHASHPREC); - printf(" --hosted use OwnTracks Hosted\n"); printf(" --norec don't maintain REC files\n"); printf(" --geokey optional Google reverse-geo API key\n"); printf(" --debug additional debugging\n"); printf("\n"); printf("Options override these environment variables:\n"); + printf(" $OTR_STORAGEDIR\n"); +#ifdef WITH_MQTT printf(" $OTR_HOST MQTT hostname\n"); printf(" $OTR_PORT MQTT port\n"); - printf(" $OTR_STORAGEDIR\n"); printf(" $OTR_USER\n"); printf(" $OTR_PASS\n"); printf(" $OTR_CAFILE PEM CA certificate chain\n"); @@ -1231,6 +1248,7 @@ void usage(char *prog) printf(" $OTR_USER username as registered on Hosted\n"); printf(" $OTR_DEVICE connect as device\n"); printf(" $OTR_TOKEN device token\n"); +#endif exit(1); } @@ -1238,18 +1256,24 @@ void usage(char *prog) int main(int argc, char **argv) { +#if WITH_MQTT struct mosquitto *mosq = NULL; - char err[1024], *p, *username, *password, *cafile, *device; - char *hostname = "localhost", *logfacility = "local0"; + char *username, *password, *cafile, *device; + char *hostname = "localhost"; + int port = 1883; + int hosted = FALSE; + UT_string *clientid; + int rc, i; + struct utsname uts; +#endif /* WITH_MQTT */ + char err[1024], *p; + char *logfacility = "local0"; #ifdef WITH_LUA char *luascript = NULL; #endif - int port = 1883; int loop_timeout = 0; - int rc, i, ch, hosted = FALSE, initialize = FALSE; + int ch, initialize = FALSE; static struct udata udata, *ud = &udata; - struct utsname uts; - UT_string *clientid; #ifdef WITH_HTTP int http_port = 8083; char *doc_root = DOCROOT; @@ -1257,9 +1281,11 @@ int main(int argc, char **argv) #endif char *progname = *argv; +#if WITH_MQTT udata.qos = DEFAULT_QOS; - udata.ignoreretained = TRUE; udata.pubprefix = NULL; +#endif + udata.ignoreretained = TRUE; udata.skipdemo = TRUE; udata.revgeo = TRUE; udata.verbose = TRUE; @@ -1288,6 +1314,7 @@ int main(int argc, char **argv) get_defaults(CONFIGFILE, &udata); +#if WITH_MQTT if ((p = getenv("OTR_HOST")) != NULL) { hostname = strdup(p); } @@ -1295,33 +1322,38 @@ int main(int argc, char **argv) if ((p = getenv("OTR_PORT")) != NULL) { port = atoi(p); } +#endif if ((p = getenv("OTR_STORAGEDIR")) != NULL) { strcpy(STORAGEDIR, p); } +#if WITH_MQTT utstring_new(clientid); utstring_printf(clientid, "ot-recorder"); if (uname(&uts) == 0) { utstring_printf(clientid, "-%s", uts.nodename); } utstring_printf(clientid, "-%d", getpid()); +#endif while (1) { static struct option long_options[] = { { "help", no_argument, 0, 'h'}, { "skipdemo", no_argument, 0, 'D'}, { "norevgeo", no_argument, 0, 'G'}, +#if WITH_MQTT { "useretained", no_argument, 0, 'R'}, { "clientid", required_argument, 0, 'i'}, { "pubprefix", required_argument, 0, 'P'}, { "qos", required_argument, 0, 'q'}, { "host", required_argument, 0, 'H'}, { "port", required_argument, 0, 'p'}, + { "hosted", no_argument, 0, 6}, +#endif /* !MQTT */ { "storage", required_argument, 0, 'S'}, { "logfacility", required_argument, 0, 4}, { "precision", required_argument, 0, 5}, - { "hosted", no_argument, 0, 6}, { "quiet", no_argument, 0, 8}, { "initialize", no_argument, 0, 9}, { "label", required_argument, 0, 10}, @@ -1369,32 +1401,10 @@ int main(int argc, char **argv) luascript = strdup(optarg); break; #endif +#ifdef WITH_MQTT case 6: hosted = TRUE; break; - case 5: - geohash_setprec(atoi(optarg)); - break; - case 4: - logfacility = strdup(optarg); - break; -#ifdef WITH_HTTP - case 'A': /* API */ - http_port = atoi(optarg); - break; - case 2: /* no short char */ - doc_root = strdup(optarg); - break; - case 3: /* no short char */ - http_host = strdup(optarg); - break; -#endif - case 'D': - ud->skipdemo = FALSE; - break; - case 'G': - ud->revgeo = FALSE; - break; case 'i': utstring_clear(clientid); utstring_printf(clientid, "%s", optarg); @@ -1418,6 +1428,30 @@ int main(int argc, char **argv) case 'p': port = atoi(optarg); break; +#endif /* WITH_MQTT */ + case 5: + geohash_setprec(atoi(optarg)); + break; + case 4: + logfacility = strdup(optarg); + break; +#ifdef WITH_HTTP + case 'A': /* API */ + http_port = atoi(optarg); + break; + case 2: /* no short char */ + doc_root = strdup(optarg); + break; + case 3: /* no short char */ + http_host = strdup(optarg); + break; +#endif + case 'D': + ud->skipdemo = FALSE; + break; + case 'G': + ud->revgeo = FALSE; + break; case 'S': strcpy(STORAGEDIR, optarg); break; @@ -1494,11 +1528,14 @@ int main(int argc, char **argv) argc -= (optind); argv += (optind); +#ifdef WITH_MQTT if (argc < 1) { usage(progname); return (-1); } +#endif +#ifdef WITH_MQTT if (hosted) { char tmp[BUFSIZ]; @@ -1534,6 +1571,7 @@ int main(int argc, char **argv) username = getenv("OTR_USER"); password = getenv("OTR_PASS"); } +#endif /* WITH_MQTT */ #ifdef WITH_HTTP if (http_port) { @@ -1598,13 +1636,13 @@ int main(int argc, char **argv) } #endif - mosquitto_lib_init(); - - signal(SIGINT, catcher); signal(SIGTERM, catcher); signal(SIGPIPE, SIG_IGN); +#ifdef WITH_MQTT + mosquitto_lib_init(); + mosq = mosquitto_new(UB(clientid), CLEAN_SESSION, (void *)&udata); if (!mosq) { fprintf(stderr, "Error: Out of memory.\n"); @@ -1680,6 +1718,7 @@ int main(int argc, char **argv) mosquitto_lib_cleanup(); return rc; } +#endif /* WITH_MQTT */ #ifdef WITH_HTTP if (http_port) { @@ -1720,15 +1759,18 @@ int main(int argc, char **argv) #endif while (run) { +#ifdef WITH_MQTT rc = mosquitto_loop(mosq, loop_timeout, /* max-packets */ 1); if (run && rc) { olog(LOG_INFO, "MQTT connection: rc=%d [%s] (errno=%d; %s). Sleeping...", rc, mosquitto_strerror(rc), errno, strerror(errno)); sleep(10); mosquitto_reconnect(mosq); } +#endif #ifdef WITH_HTTP if (udata.mgserver) { - mg_poll_server(udata.mgserver, 100); + // mg_poll_server(udata.mgserver, 100); + mg_poll_server(udata.mgserver, 50); } #endif } @@ -1753,10 +1795,13 @@ int main(int argc, char **argv) #if WITH_LUA && WITH_LMDB hooks_exit(ud->luadata, "recorder stops"); #endif + +#ifdef WITH_MQTT mosquitto_disconnect(mosq); mosquitto_destroy(mosq); mosquitto_lib_cleanup(); +#endif return (0); } diff --git a/udata.h b/udata.h index c1a7349..dd4ce91 100644 --- a/udata.h +++ b/udata.h @@ -14,10 +14,12 @@ struct udata { JsonNode *topics; /* Array of topics to subscribe to */ int ignoreretained; /* True if retained messages should be ignored */ +#if WITH_MQTT char *pubprefix; /* If not NULL (default), republish modified payload to /topic */ + int qos; /* Subscribe QoS */ +#endif int skipdemo; /* True if _demo users are to be skipped */ int revgeo; /* True (default) if we should do reverse Geo lookups */ - int qos; /* Subscribe QoS */ int verbose; /* TRUE if print verbose messages to stdout */ int norec; /* If TRUE, no .REC files are written to */ #ifdef WITH_LMDB diff --git a/util.c b/util.c index 788d614..7239d4a 100644 --- a/util.c +++ b/util.c @@ -212,6 +212,14 @@ int splitter(char *s, char *sep, char **parts) return (nt); } +void splitterfree(char **parts) +{ + int n; + + for (n = 0; parts[n] != NULL; n++) + free(parts[n]); +} + /* * Split a string separated by characters in `sep' into a JSON * array and return that or NULL on error. diff --git a/util.h b/util.h index 7fdcfda..93dd85e 100644 --- a/util.h +++ b/util.h @@ -23,6 +23,7 @@ int json_copy_to_object(JsonNode *obj, JsonNode * object_or_array, int clobber); int json_copy_element_to_object(JsonNode *obj, char *key, JsonNode *node); int json_copy_from_file(JsonNode * obj, char *filename); int splitter(char *s, char *sep, char **parts); +void splitterfree(char **parts); JsonNode *json_splitter(char *s, char *sep); int syslog_facility_code(char *facility); const char *yyyymm(time_t t);