Create a task without creating a new Redis connection pool and connection (RedisStreamBroker)

[Exagone313]

Hello,

I'm using the RedisStreamBroker. When instantiated, it creates a new Redis connection pool. This is the case even when importing the broker from another module for queuing a new task, as the pool is created in the parent class constructor.

In my application, I'm already creating a Redis connection pool (redis.asyncio.BlockingConnectionPool) for other stuff, and I'd like to use my existing Redis client (redis.asyncio.Redis) to queue a new task.

From what I'm seeing in the code, I should be able to call XADD manually (EDIT: just did this, it works). But ideally I would like to reuse some code.

For example, the BrokerMessage instance is created by first using a private method _prepare_message (it returns a TaskiqMessage, later transformed into BrokerMessage by ProxyFormatter), so the code could easily change in later versions.

What do you think would be the best solution? I would be ok to contribute on code changes, but it would be a bit weird to have a new way to create a task specifically for a broker.

Thanks.

[s3rius]

I was thinking about that, but couldn't find a solution. Because brokers are created when imported and they should have something they can work with. But application's connection pools are created during start of the application.

The problem here is not client code, but workers. Becuase workers know nothing about your application during startup. In order for them to get information about your connection pool, they should first run your startup code. Which, as you can see, an infinite dependency cycle.

[Exagone313]

@s3rius As a middle ground, I think I can rely on calling Redis manually like I did, I'm just not 100% about compatibility going forward, but it seems unlikely that this code will change.