mjwestcott / runnel Goto Github PK

Hi. Having a lightweight alternative for Kafka/Faust is awesome, a shame this hasn't got more attention up until now, thanks for sharing!
Wondering what's the current status of the project. Experiences in production? Still being maintained?

Thanks!

Hi, I was wondering if you have plans for adding django integration.

On pypi https://pypi.org/project/runnel/ as well as https://runnel.dev/ there is no link to https://github.com/mjwestcott/runnel/.

That makes searching a bit ... harder :) It would be nice to have links to github at docs and pypi. Thanks!

Running simple example for middlewares:

from datetime import datetime

from runnel import App, Record, Events, Middleware

app = App(name="myapp", redis_url="redis://127.0.0.1")


class Order(Record):
    order_id: int
    created_at: datetime
    amount: int


class Printer(Middleware):
    async def handler(self, parent: Events, **kwargs):
        async for x in parent:
            print(x)
            yield x


orders = app.stream("orders", record=Order, partition_by="order_id")


@app.processor(orders, middleware=[Printer()])
async def printer(events: Events):
    async for order in events.records():
        pass

error stack

Traceback (most recent call last):
  File "/Users/colch/Work/py-word-counter/.venv/lib/python3.9/site-packages/runnel/runner.py", line 157, in consumer
    await tg.cancel_scope.cancel()
  File "/Users/colch/Work/py-word-counter/.venv/lib/python3.9/site-packages/anyio/_backends/_asyncio.py", line 438, in __aexit__
    raise exceptions[0]
  File "/Users/colch/Work/py-word-counter/.venv/lib/python3.9/site-packages/anyio/_backends/_asyncio.py", line 466, in _run_wrapped_task
    await func(*args)
  File "/Users/colch/Work/py-word-counter/app.py", line 26, in printer
    async for order in events.records():
  File "/Users/colch/Work/py-word-counter/.venv/lib/python3.9/site-packages/runnel/events.py", line 175, in iter
    async for x in agen:
  File "/Users/colch/Work/py-word-counter/.venv/lib/python3.9/site-packages/runnel/middleware/deserialize.py", line 11, in handler
    async for x in parent:
  File "/Users/colch/Work/py-word-counter/.venv/lib/python3.9/site-packages/runnel/middleware/ack.py", line 11, in handler
    async for x in parent:
  File "/Users/colch/Work/py-word-counter/app.py", line 17, in handler
    print(x)
  File "/Users/colch/Work/py-word-counter/.venv/lib/python3.9/site-packages/runnel/types.py", line 82, in __repr__
    if os.environ["RUNNEL_TESTING"] == "1":
  File "/usr/local/Cellar/[email protected]/3.9.1_3/Frameworks/Python.framework/Versions/3.9/lib/python3.9/os.py", line 679, in __getitem__
    raise KeyError(key) from None
KeyError: 'RUNNEL_TESTING'

Exporting env variable RUNNEL_TESTING fixes this, indeed. But I suppose, os.environ["RUNNEL_TESTING"] should be os.environ.get("RUNNEL_TESTING"), right?

This is the calling code. Worker exits with this error just after reaching 'new-leader'. Windows systems not having signals seems to be the cause: https://stackoverflow.com/q/45987985/4670905

Python version: 3.8.5

Can it be used in production

Hi,

thank you for developing runnel!

I would like to add a simple redis key/value store functionality to be used within processor functions whereas (for performance purposes) the already open runnel redis connection should be used. Also, a lock mechanism for write procedures should be in place. Any hints on how to get this done?

Sometimes random integration tests on Python 3.7 fail when exiting due to an uncaught ConnectionError from our Redis driver aredis.

I believe this is a bug in aredis caused by suppressing asyncio.CancelledError: NoneGG/aredis#169

Sometimes random integration tests fail with aredis.exceptions.ResponseError: Wrong number of response items from pipeline execution.

I believe this is a bug in aredis caused by failing to disconnect and clean up connections on cancellation: NoneGG/aredis#171