setup the reader

StreamReader/Dockerfile

@@ -0,0 +1,8 @@
+FROM ubuntu:19.04
+RUN apt update
+RUN apt install -y python git python-pip
+ADD . /myapp
+WORKDIR /myapp
+RUN pip install -r requirements.txt
+#EXPOSE 8080
+CMD ./runner.py

StreamReader/requirements.txt

@@ -0,0 +1,9 @@
+git+git://github.com/RedisTimeSeries/redistimeseries-py
+hiredis>=0.2.0
+redis>=2.10
+rmtest>=0.2
+six>=1.10.0
+gevent==1.2.1
+Flask>=0.12
+Flask-Cors==3.0.2
+python-dateutil==2.6.0

StreamReader/runner.py

@@ -0,0 +1,35 @@
+#!/usr/bin/env python
+
+from redistimeseries.client import Client as RedisTimeSeries
+import redis
+import time
+
+redis_host = "redis"
+redis_port = 6379
+
+rts = RedisTimeSeries(host=redis_host, port=redis_port)
+
+
+
+pool = redis.ConnectionPool(host=redis_host, port=redis_port)
+r = redis.Redis(connection_pool=pool)
+
+try:
+    r.xadd("mystream",  {'event_type': 'startup', 'user': 'root'})
+    r.xgroup_create("mystream", "consumerGroup", '$')
+except:
+    print("group already exists")
+
+while True:
+    msgs = r.xreadgroup("consumerGroup", "consumerName", streams={"mystream": '>'}, count=10, block=1000, noack=False)
+    for msg in msgs:
+        for m in msg[1]:
+            evnt = m[1]['event_type']
+            try:
+                rts.info(evnt)
+            except:
+                rts.create(evnt, retentionSecs=60, labels={'event_type': evnt})
+                rts.create(evnt+"_minute", retentionSecs=0, labels={'event_type': evnt})
+                rts.createrule(evnt, evnt+"_minute", 'count', 60)
+
+            rts.incrby(evnt,1)

docker-compose.yml

@@ -0,0 +1,10 @@
+version: '3'
+services:
+  redis:
+    image: "redislabs/redistimeseries"
+    ports:
+      - "6379:6379"
+  runner:
+    image: "maguec/streams2ts"
+    links:
+      - "redis:redis"