feat: add support for broker setting

Author: woile

README.md

@@ -25,24 +25,19 @@ topics = [
     Topic("test-topic-worker", hello_world)
 ]
 
-streamapp = App(topics=topics)
+streamapp = App(topics=topics, broker="kafka:0.0.0.0:29092")
 
 while True:
     # This should be any other worker
     loop.run_until_complete(streamapp.run())
 ```
 
+## Example
+
+- [simple-kafka-consumer](example/simple-kafka-consumer)
+
 ## Usage
 
 1. Clone repository
 2. Run `poetry install`
 3. Write apps inside `example` folder (for now)
-
-## TODO
-
-- [ ] emit events interface
-- [ ] timers
-- [ ] Kstream feature compatibility
-- [ ] general interface stability
-- [ ] documentation
-- [ ] tests

dam/application.py

@@ -5,14 +5,24 @@
 class App:
     def __init__(
         self,
+        broker: str,
         topics: Optional[List] = None,
         loop: Optional[asyncio.AbstractEventLoop] = None,
     ):
         if loop is None:
             loop = asyncio.get_event_loop()
         self.loop = loop
         self.topics = topics
+        self.broker, self.hostname = broker.split(":", 1)
 
     async def run(self):
         for topic in self.topics:
+            topic._configure({
+                "bootstrap_servers": self.hostname
+            })
             await topic.run()
+
+    async def __call__(self, loop):
+        if loop is not None:
+            self.loop = loop
+        await self.run()

dam/topic.py

@@ -1,17 +1,21 @@
 import asyncio
-from typing import Callable
+from typing import Callable, Dict
 
-from aiokafka import AIOKafkaConsumer
+from aiokafka import AIOKafkaConsumer, errors
 
 from dam.stream import Stream
 
 
 class Topic:
     """Kafka topic"""
 
-    def __init__(self, name: str, func: Callable):
+    def __init__(self, name: str, func: Callable, **kwargs):
         self.name = name
         self.func = func
+        self.consumer_kwargs: Dict = kwargs
+
+    def _configure(self, conf: Dict):
+        self.conf = conf
 
     def create_consumer(
         self, name: str, loop: asyncio.AbstractEventLoop = None
@@ -22,13 +26,14 @@ def create_consumer(
         return AIOKafkaConsumer(
             name,
             loop=loop,
-            bootstrap_servers="0.0.0.0:29092",
-            group_id="test-consumer-group",
             auto_offset_reset="earliest",
+            **self.conf,
+            **self.consumer_kwargs
         )
 
     async def run(self):
         consumer = self.create_consumer(self.name)
+
         await consumer.start()
 
         try:
@@ -37,29 +42,3 @@ async def run(self):
         finally:
             # Will leave consumer group; perform autocommit if enabled.
             await consumer.stop()
-
-
-# async def consume_from_kafka(loop):
-#     logger.debug("Task Consuming from kafka initiated...")
-#     consumer = AIOKafkaConsumer(
-#         TOPIC,
-#         loop=loop,
-#         bootstrap_servers=BOOTSTRAP_SERVERS,
-#         group_id=GROUP_ID,
-#         auto_offset_reset=AUTO_OFFSET_RESET,
-#     )
-
-#     # Get cluster layout and join group `my-group`await consumer.start()
-#     await consumer.start()
-
-#     try:
-#         # Consume messages
-#         async for msg in consumer:
-#             print(f"Task Consuming from kafka")
-#             print(
-#                 f"Topic: {msg.topic}, Partition: {msg.partition}, Offset: {msg.offset},"
-#                 f" Key: {msg.key}, Value: {msg.value}, Timestamp: {msg.timestamp}\n"
-#             )
-#     finally:
-#         # Will leave consumer group; perform autocommit if enabled.
-#         await consumer.stop()

example/simple-kafka-consumer/README.md

@@ -0,0 +1,27 @@
+# Simple kafka consumer example
+
+1. Run docker compose
+
+```bash
+docker-compose up
+```
+
+2. Install requirmeents
+
+```bash
+pip install -r requirements.txt
+```
+
+3. Run the code from the root folder of dam
+
+```bash
+python example/simple-kafka-consumer.py
+```
+
+4. Send events
+
+Open a producer console
+
+```bash
+docker-compose exec kafka kafka-console-producer --broker-list kafka:9092 --topic test-topic-worker
+```

example/simple-kafka-consumer/docker-compose.yaml

@@ -0,0 +1,26 @@
+version: '3'
+services:
+  zookeeper:
+    image: "confluentinc/cp-zookeeper"
+    hostname: zookeeper
+    ports:
+      - 32181:32181
+    environment:
+      - ZOOKEEPER_CLIENT_PORT=32181
+      - ALLOW_ANONYMOUS_LOGIN=yes
+  kafka:
+    image: confluentinc/cp-kafka
+    hostname: kafka
+    container_name: kafka
+    ports:
+    - 9092:9092
+    - 29092:29092
+    depends_on:
+    - zookeeper
+    environment:
+      - KAFKA_ZOOKEEPER_CONNECT=zookeeper:32181
+      - KAFKA_OFFSETS_TOPIC_REPLICATION_FACTOR=1
+      - KAFKA_LISTENER_SECURITY_PROTOCOL_MAP=PLAINTEXT:PLAINTEXT,PLAINTEXT_HOST:PLAINTEXT
+      - KAFKA_ADVERTISED_LISTENERS=PLAINTEXT_HOST://localhost:29092,PLAINTEXT://kafka:9092
+      - KAFKA_BROKER_ID=1
+      - ALLOW_PLAINTEXT_LISTENER=yes

example/simple-kafka-consumer/requirements.txt

@@ -0,0 +1 @@
+aioworker

example/simple-kafka-consumer/simple_consumer.py

@@ -0,0 +1,18 @@
+import asyncio
+from aioworker import Worker
+from dam import App, Topic
+
+loop = asyncio.get_event_loop()
+
+
+async def hello_world(stream):
+    async for value in stream:
+        print(value.value)
+
+topics = [
+    Topic("test-topic-worker", hello_world),
+]
+
+streamapp = App(topics=topics, broker="kafka:0.0.0.0:29092")
+
+Worker(tasks=[streamapp]).run()