how to add a coroutine to running event loop?

Question

I have read how to add a coroutine to a running asyncio loop? but it's not I want

basically, I need a daemon thread to subscribe a redis channel, and I can add a callback method dynamic, my solution is subclass a Thread class, and create an event loop and run forever, but after loop running, I can not call any method of the object,

redis.py

#! /usr/bin/env python
# -*- coding: utf-8 -*-
import redis
import os
import asyncio
import aioredis
from threading import Thread
from collections import defaultdict

assert os.getenv('REDIS_HOST') is not None
assert os.getenv('REDIS_PORT') is not None

class RedisClient(Thread):
    def __init__(self, loop):
        super(RedisClient, self).__init__()
        self.callbacks = defaultdict(list)
        self.channels = {}
        self.loop = loop

    async def pubsub(self):
        address = 'redis://{}:{}'.format(os.getenv('REDIS_HOST'), os.getenv('REDIS_PORT'))
        self.sub = await aioredis.create_redis(address)

    def sync_add_callback(self, channel, callback):
        self.loop.create_task(self.add_callback(channel, callback))

    async def add_callback(self, channel, callback):
        self.callbacks[channel].append(callback)

        if channel not in self.channels or self.channels[channel] is None:
            channels = await self.sub.subscribe(channel)
            ch1 = channels[0]
            assert isinstance(ch1, aioredis.Channel)
            self.channels[channel] = ch1

            async def async_reader(channel):
                while await channel.wait_message():
                    msg = await channel.get(encoding='utf-8')
                    # ... process message ...
                    print(msg)
                    print(channel.name)
                    for c in self.callbacks[channel.name.decode('utf-8')]:
                        c(channel.name, msg)

            tsk1 = asyncio.ensure_future(async_reader(ch1))

    def remove_callback(self, channel, callback):
        self.callbacks[channel].remove(callback)

    def run(self):
        asyncio.set_event_loop(self.loop)
        loop.run_until_complete(self.pubsub())


# Create the new loop and worker thread
loop = asyncio.new_event_loop()
redis_client = RedisClient(loop)
redis_client.start()

usage:

def test(channel, msg):
    print('{}{}'.format(channel, msg))

from redis import redis_client
redis_client.sync_add_callback('test', test)

maybe my solution is not a good practice of Python?

Update 1:

I have tried a solution and it works well, but in the beginning, I want to reuse the sub instance, this method can work as a module to subscribe to different channel, but every subscribe should have it's own sub, or that is to say every subscribe have to create it own redis connection

the solution:

#! /usr/bin/env python
# -*- coding: utf-8 -*-
import os
import asyncio
import aioredis
from threading import Thread

assert os.getenv('REDIS_HOST') is not None
assert os.getenv('REDIS_PORT') is not None

class RedisClient(Thread):
    def __init__(self, channel, callback, *args, **kwargs):
        super(RedisClient, self).__init__(*args, **kwargs)
        self.daemon = True
        self.channel = channel
        self.callback = callback

    async def pubsub(self):
        address = 'redis://{}:{}'.format(os.getenv('REDIS_HOST'), os.getenv('REDIS_PORT'))
        sub = await aioredis.create_redis(address)

        channels = await sub.subscribe(self.channel)
        ch1 = channels[0]
        assert isinstance(ch1, aioredis.Channel)

        async def async_reader(channel):
            while await channel.wait_message():
                msg = await channel.get(encoding='utf-8')
                self.callback(channel.name.decode('utf-8'), msg)

        await async_reader(ch1)

    def run(self):
        loop = asyncio.new_event_loop()
        asyncio.set_event_loop(loop)
        loop.run_until_complete(self.pubsub())

update 2:

finally, it works well

#! /usr/bin/env python
# -*- coding: utf-8 -*-
import redis
import os
import asyncio
import aioredis
from threading import Thread
from collections import defaultdict

assert os.getenv('REDIS_HOST') is not None
assert os.getenv('REDIS_PORT') is not None

class RedisClient(Thread):
    def __init__(self, loop):
        super(RedisClient, self).__init__()
        self.callbacks = defaultdict(list)
        self.channels = {}
        self.loop = loop
        self.sub = None

    async def pubsub(self):
        print('test3')
        address = 'redis://{}:{}'.format(os.getenv('REDIS_HOST'), os.getenv('REDIS_PORT'))
        self.sub = await aioredis.create_redis(address)

    def sync_add_callback(self, channel, callback):
        print('ahhhhhhhhh')
        asyncio.run_coroutine_threadsafe(self.add_callback(channel, callback), self.loop)

    async def add_callback(self, channel, callback):
        print('test2')
        if not self.sub:
            await self.pubsub()
        self.callbacks[channel].append(callback)

        if channel not in self.channels or self.channels[channel] is None:
            channels = await self.sub.subscribe(channel)
            ch1 = channels[0]
            assert isinstance(ch1, aioredis.Channel)
            self.channels[channel] = ch1

            async def async_reader(channel):
                while await channel.wait_message():
                    msg = await channel.get(encoding='utf-8')
                    # ... process message ...
                    print(msg)
                    print(channel.name)
                    print(self.callbacks[channel.name])
                    for c in self.callbacks[channel.name.decode('utf-8')]:
                        c(channel.name, msg)

            tsk1 = asyncio.ensure_future(async_reader(ch1))

    def remove_callback(self, channel, callback):
        self.callbacks[channel].remove(callback)

    def run(self):
        asyncio.set_event_loop(self.loop)
        loop.run_forever()


# Create the new loop and worker thread
loop = asyncio.new_event_loop()
redis_client = RedisClient(loop)
redis_client.start()

How would you handle SIGINT SIGTERM Ctrl + C and keyboard interrupts in this — PirateApp
– PirateApp, Commented Feb 21, 2019 at 8:13

Mark White · Accepted Answer · 2018-11-12 02:58:06Z

1

let me show you a similarly case using aiohttp here.

async def listen_to_redis(app):
    try:
        sub = await aioredis.create_redis(('localhost', 6379), loop=app.loop)
        ch, *_ = await sub.subscribe('news')
        async for msg in ch.iter(encoding='utf-8'):
            # Forward message to all connected websockets:
            for ws in app['websockets']:
                ws.send_str('{}: {}'.format(ch.name, msg))
    except asyncio.CancelledError:
        pass
    finally:
        await sub.unsubscribe(ch.name)
        await sub.quit()


async def start_background_tasks(app):
    app['redis_listener'] = app.loop.create_task(listen_to_redis(app))


async def cleanup_background_tasks(app):
    app['redis_listener'].cancel()
    await app['redis_listener']


app = web.Application()
app.on_startup.append(start_background_tasks)
app.on_cleanup.append(cleanup_background_tasks)
web.run_app(app)

answered Nov 12, 2018 at 2:58

Mark White

6601 gold badge5 silver badges13 bronze badges

Sign up to request clarification or add additional context in comments.

2 Comments

EkkoG Over a year ago

can you add listener after run_app? for my example, I use flask for web app, the app is sync so have no event loop, so I create a loop, and after run the loop, I want to add a listener to the sub to reuse the sub, for now, I must add listener before run, I want to add listener dynamic, that is add listener after loop run

Mark White Over a year ago

@KevinGuo I do think you should give aiohttp a try , because it perfect for your case, and it has better performance in production. It is very easy transform flask code to aiohttp medium.com/@SkyscannerEng/from-flask-to-aiohttp-22f1ddc5dd5e

user4815162342 · Accepted Answer · 2018-11-12 17:32:37Z

0

If the idea is for sync_add_callback to be invoked from other threads, then its implementation should look like this:

def sync_add_callback(self, channel, callback):
    asyncio.run_coroutine_threadsafe(self.add_callback(channel, callback), self.loop)

Please note that the callbacks will be invoked in the event loop thread, so they should not use blocking calls themselves.

answered Nov 12, 2018 at 17:32

user4815162342

159k22 gold badges350 silver badges418 bronze badges

9 Comments

EkkoG Over a year ago

sync_add_callback seems cannot be called from main thread :(

user4815162342 Over a year ago

@KevinGuo What happens when you call it from the main thread? If you get a traceback, please post it to pastebin.com or equivalent.

EkkoG Over a year ago

my fault, yesterday I run the loop on the main thread, so when I call sync_add_callback from the main thread, nothing happened.

EkkoG Over a year ago

now I run the loop in thread obj's run method, sync_add_callback can call from method, but await in add_callback method never excute, code here gist.github.com/cielpy/3164074a34b8d017becd0ef60df29c91,it seems event loop's run* method will block the thread?

user4815162342 Over a year ago

@KevinGuo The exception occurs, but in the other thread. You need to collect it by requesting the result() from the future object returned by asyncio.run_coroutine_threadsafe(...).

|

Collectives™ on Stack Overflow

how to add a coroutine to running event loop?

2 Answers 2

2 Comments

9 Comments

Your Answer

Linked

Hot Network Questions

Collectives™ on Stack Overflow

2 Answers 2

2 Comments

9 Comments

Your Answer

Sign up or log in

Post as a guest

Linked

Related