Adding implementation

Author: JBris

README.md

@@ -1 +1,3 @@
-# redis-streams-python
+# redis-streams-python
+
+Testing out Redis Streams using Python client

broadcast_redis.py

@@ -0,0 +1,16 @@
+import redis
+import random
+
+r = redis.Redis(host="localhost", port=6379)
+
+stream = "controller"
+group = f"group_uuid"
+consumer = "subscriber_uuid"
+
+try:
+    r.xgroup_create(stream, group, id="0", mkstream=True)
+except redis.exceptions.ResponseError as e:
+    if "BUSYGROUP" not in str(e):
+        raise
+
+r.xadd(stream, {"value": "broadcast"})

docker-compose.yaml

@@ -0,0 +1,15 @@
+version: '3.8'
+
+services:
+  redis:
+    image: redis:7
+    ports:
+      - 6379:6379
+    stop_grace_period: 60s
+    restart: always
+
+
+networks:
+  default:
+    name: redis-streams
+    driver: bridge

exit_redis.py

@@ -0,0 +1,16 @@
+import redis
+import random
+
+r = redis.Redis(host="localhost", port=6379)
+
+stream = "controller"
+group = f"group_uuid"
+consumer = "subscriber_uuid"
+
+try:
+    r.xgroup_create(stream, group, id="0", mkstream=True)
+except redis.exceptions.ResponseError as e:
+    if "BUSYGROUP" not in str(e):
+        raise
+
+r.xadd(stream, {"value": "exit"})

poetry.lock

@@ -0,0 +1,48 @@
+import redis
+import time
+import random
+
+r = redis.Redis(host="localhost", port=6379)
+
+stream = "controller"
+group = f"group_uuid"
+consumer = "subscriber_uuid"
+
+exit_condition = False
+
+try:
+    r.xgroup_create(stream, group, id="0", mkstream=True)
+except redis.exceptions.ResponseError as e:
+    if "BUSYGROUP" not in str(e):
+        raise
+
+while not exit_condition:
+    entries = r.xreadgroup(
+        groupname=group,
+        consumername=consumer,
+        streams={stream: '>'},
+        block=1,   
+        count=1
+    )
+
+    for _, msgs in entries:
+        for msg_id, data in msgs:
+            value = data[b'value'].decode()
+            print(f"Received controller value: {value}")
+            r.xack(stream, group, msg_id)
+
+            if value == "exit":
+                exit_condition = True
+                r.xadd("temperature", {"value": "exit"})
+                r.xtrim(stream, maxlen=0)
+                r.xtrim("temperature", maxlen=0)
+                break
+            elif value == "broadcast":
+                r.xadd("temperature", {"value": "broadcast"})
+    
+    if not exit_condition:
+        value = random.randint(0, 100)
+        r.xadd("temperature", {"value": value})
+        print(f"Published: {value}")
+
+        time.sleep(5)

publish_redis.py

@@ -0,0 +1,15 @@
+[tool.poetry]
+name = "redis-streams-python"
+version = "0.1.0"
+description = ""
+authors = ["James Bristow <[email protected]>"]
+readme = "README.md"
+
+[tool.poetry.dependencies]
+python = "^3.10"
+redis = "^6.2.0"
+
+
+[build-system]
+requires = ["poetry-core"]
+build-backend = "poetry.core.masonry.api"

pyproject.toml

@@ -0,0 +1,35 @@
+import redis
+import uuid
+
+r = redis.Redis(host="localhost", port=6379)
+exit_condition = False
+
+stream = "temperature"
+group = f"group_{str(uuid.uuid4())}"
+consumer = "subscriber_uuid"
+
+try:
+    r.xgroup_create(stream, group, id="0", mkstream=True)
+except redis.exceptions.ResponseError as e:
+    if "BUSYGROUP" not in str(e):
+        raise
+
+print("Waiting...")
+
+while not exit_condition:
+    entries = r.xreadgroup(
+        groupname=group,
+        consumername=consumer,
+        streams={stream: '>'},
+        block=0
+    )
+
+    for _, msgs in entries:
+        for msg_id, data in msgs:
+            value = data[b'value'].decode()
+            print(f"Received value: {value}")
+            r.xack(stream, group, msg_id)
+
+            if value == "exit":
+                exit_condition = True
+                break