MF-328 - Add message persistence support for MQTT (#327)

* Add redis for message persistence Signed-off-by: Aleksandar Novakovic <aleksandar.novakovic@mainflux.com> * Add redis to docker composition Signed-off-by: Aleksandar Novakovic <aleksandar.novakovic@mainflux.com> * Fix mqtt publish command in documentation Signed-off-by: Aleksandar Novakovic <aleksandar.novakovic@mainflux.com> * Update documentation with new environment variables Signed-off-by: Aleksandar Novakovic <aleksandar.novakovic@mainflux.com>
mainflux · Jun 15, 2018 · d8357b5 · d8357b5
1 parent 993398b
commit d8357b5
Show file tree

Hide file tree

Showing 5 changed files with 42 additions and 9 deletions.
diff --git a/docker/docker-compose.yml b/docker/docker-compose.yml
@@ -94,6 +94,8 @@ services:
     image: mainflux/normalizer:latest
     container_name: mainflux-normalizer
     restart: on-failure
+    depends_on:
+      - nats
     expose:
       - 8184
     environment:
@@ -112,6 +114,7 @@ services:
     container_name: mainflux-ws
     depends_on:
       - things
+      - nats
     expose:
       - 8186
     restart: on-failure
@@ -127,6 +130,7 @@ services:
     container_name: mainflux-http
     depends_on:
       - things
+      - nats
     restart: on-failure
     expose:
       - 8185
@@ -137,15 +141,23 @@ services:
     ports:
       - 8185:8185
 
+  redis:
+    image: redis:4.0.9-alpine
+    container_name: mainflux-redis
+    restart: on-failure
+
   mqtt-adapter:
     image: mainflux/mqtt:latest
     container_name: mainflux-mqtt
     depends_on:
       - things
+      - nats
+      - redis
     restart: on-failure
     environment:
       MF_MQTT_ADAPTER_PORT: 1883
       MF_MQTT_WS_PORT: 8880
+      MF_MQTT_REDIS_HOST: redis
       MF_NATS_URL: nats://nats:4222
       MF_THINGS_URL: things:8183
     ports:

diff --git a/docs/getting-started.md b/docs/getting-started.md
@@ -290,7 +290,7 @@ or [Paho](https://www.eclipse.org/paho/) if you want to use MQTT over WebSocket.
 To publish message over channel, thing should call following command:
 
 ```
-mosquitto_pub -u <thing_id> -P <thing_key> -t channels/<channel_id>/messages -h localhost -m [{"bn":"some-base-name:","bt":1.276020076001e+09, "bu":"A","bver":5, "n":"voltage","u":"V","v":120.1}, {"n":"current","t":-5,"v":1.2}, {"n":"current","t":-4,"v":1.3}]
+mosquitto_pub -u <thing_id> -P <thing_key> -t channels/<channel_id>/messages -h localhost -m '[{"bn":"some-base-name:","bt":1.276020076001e+09, "bu":"A","bver":5, "n":"voltage","u":"V","v":120.1}, {"n":"current","t":-5,"v":1.2}, {"n":"current","t":-4,"v":1.3}]'
 ```
 
 To subscribe to channel, thing should call following command:

diff --git a/mqtt/README.md b/mqtt/README.md
@@ -1,6 +1,6 @@
 # MQTT adapter
 
-MQTT adapter provides an MQTT API for sending and receiving messages through the 
+MQTT adapter provides an MQTT API for sending and receiving messages through the
 platform.
 
 ## Configuration
@@ -14,6 +14,10 @@ default values.
 | MF_MQTT_ADAPTER_PORT | Service MQTT port   | 1883                  |
 | MF_MQTT_WS_PORT      | WebSocket port      | 8880                  |
 | MF_NATS_URL          | NATS instance URL   | nats://localhost:4222 |
+| MF_MQTT_REDIS_PORT   | Redis port          | 6379                  |
+| MF_MQTT_REDIS_HOST   | Redis host          | localhost             |
+| MF_MQTT_REDIS_PASS   | Redis pass          | mqtt                  |
+| MF_MQTT_REDIS_DB     | Redis db            | 0                     |
 | MF_THINGS_URL        | Things service URL  | localhost:8181        |
 
 ## Deployment
@@ -34,6 +38,10 @@ services:
       MF_NATS_URL: [NATS instance URL]
       MF_MQTT_ADAPTER_PORT: [Service MQTT port]
       MF_MQTT_WS_PORT: [Service WS port]
+      MF_MQTT_REDIS_PORT: [Redis port]
+      MF_MQTT_REDIS_HOST: [Redis host]
+      MF_MQTT_REDIS_PASS: [Redis pass]
+      MF_MQTT_REDIS_DB: [Redis db]
 ```
 
 To start the service outside of the container, execute the following shell script:
@@ -48,7 +56,7 @@ cd $GOPATH/src/github.com/mainflux/mainflux/mqtt
 npm install
 
 # set the environment variables and run the service
-MF_THINGS_URL=[Things service URL] MF_NATS_URL=[NATS instance URL] MF_MQTT_ADAPTER_PORT=[Service MQTT port] MF_MQTT_WS_PORT=[Service WS port] node mqtt.js ..
+MF_THINGS_URL=[Things service URL] MF_NATS_URL=[NATS instance URL] MF_MQTT_ADAPTER_PORT=[Service MQTT port] MF_MQTT_WS_PORT=[Service WS port] MF_MQTT_REDIS_PORT=[Redis port] MF_MQTT_REDIS_HOST=[Redis host] MF_MQTT_REDIS_PASS=[Redis pass] MF_MQTT_REDIS_DB=[Redis db] node mqtt.js ..
 ```
 
 ## Usage

diff --git a/mqtt/mqtt.js b/mqtt/mqtt.js
@@ -2,26 +2,38 @@
 
 var http = require('http'),
     net = require('net'),
-    aedes = require('aedes')(),
-    logging = require('aedes-logging'),
     protobuf = require('protocol-buffers'),
     websocket = require('websocket-stream'),
     grpc = require('grpc'),
     fs = require('fs'),
-    bunyan = require('bunyan');
+    bunyan = require('bunyan'),
+    logging = require('aedes-logging');
 
 // pass a proto file as a buffer/string or pass a parsed protobuf-schema object
 var logger = bunyan.createLogger({name: "mqtt"}),
     config = {
-        mqtt_port: process.env.MF_MQTT_ADAPTER_PORT || 1883,
-        ws_port: process.env.MF_MQTT_WS_PORT || 8880,
+        mqtt_port: Number(process.env.MF_MQTT_ADAPTER_PORT) || 1883,
+        ws_port: Number(process.env.MF_MQTT_WS_PORT) || 8880,
         nats_url: process.env.MF_NATS_URL || 'nats://localhost:4222',
+        redis_port: Number(process.env.MF_MQTT_REDIS_PORT) || 6379,
+        redis_host: process.env.MF_MQTT_REDIS_HOST || 'localhost',
+        redis_pass: process.env.MF_MQTT_REDIS_PASS || 'mqtt',
+        redis_db: Number(process.env.MF_MQTT_REDIS_DB) || 0,
         auth_url: process.env.MF_THINGS_URL || 'localhost:8181',
-        schema_dir: process.argv[2] || '.'
+        schema_dir: process.argv[2] || '.',
     },
     message = protobuf(fs.readFileSync(config.schema_dir + '/message.proto')),
     thingsSchema = grpc.load(config.schema_dir + "/internal.proto").mainflux,
     nats = require('nats').connect(config.nats_url),
+    aedesRedis = require('aedes-persistence-redis')({
+        port: config.redis_port,
+        host: config.redis_host,
+        password: config.redis_pass,
+        db: config.redis_db
+    }),
+    aedes = require('aedes')({
+        persistence: aedesRedis
+    }),
     things = new thingsSchema.ThingsService(config.auth_url, grpc.credentials.createInsecure()),
     servers = [
         startMqtt(),

diff --git a/mqtt/package.json b/mqtt/package.json
@@ -15,6 +15,7 @@
   "dependencies": {
     "aedes": "^0.34.0",
     "aedes-logging": "^1.0.1",
+    "aedes-persistence-redis": "^5.1.0",
     "atob": "^2.0.3",
     "bunyan": "^1.5.1",
     "grpc": "^1.11.3",