Redis RPC with FastStream#

FastStream RedisBroker provides the powerful capability to perform Remote Procedure Calls (RPC) using Redis. This feature enables you to send a message and await a response, effectively creating a synchronous request-response pattern over the inherently asynchronous Redis messaging system. Below is the guide to set up and utilize the Redis RPC publishing feature with FastStream.

RPC with Redis Overview#

In a traditional publish/subscribe setup, the publishing party sends messages without expecting any direct response from the subscribers. However, with RPC, the publisher sends a message and waits for a response from the subscriber, which can then be used for subsequent operations or processing.

FastStream allows you to define RPC-style communication channels, lists, or streams by using the RedisBroker's publishing function with the rpc flag set to True.

Implementing Redis RPC in FastStream#

To implement Redis RPC with RedisBroker in FastStream, follow the steps below:

1. Initiate your FastStream application with RedisBroker

   1 2	broker = RedisBroker("redis://localhost:6379") app = FastStream(broker)

2. Define subscriber handlers for various Redis data types (e.g., channel, list, stream) that can process incoming messages and return responses.

   1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16

   @broker.subscriber(channel="test-channel") async def handle_channel(msg: str, logger: Logger): logger.info(msg) return msg @broker.subscriber(list="test-list") async def handle_list(msg: str, logger: Logger): logger.info(msg) return msg @broker.subscriber(stream="test-stream") async def handle_stream(msg: str, logger: Logger): logger.info(msg) return msg

3. Send RPC messages through RedisBroker and await responses on the correct data type.

   After your application has started and the subscribers are ready to receive messages, you can publish messages with the rpc option enabled. Additionally, you can set an rpc_timeout to decide how long the publisher should wait for a response before timing out.

   1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24

   @app.after_startup async def t(): msg = "Hi!" assert msg == await broker.publish( "Hi!", channel="test-channel", rpc=True, rpc_timeout=3.0, ) assert msg == await broker.publish( "Hi!", list="test-list", rpc=True, rpc_timeout=3.0, ) assert msg == await broker.publish( "Hi!", stream="test-stream", rpc=True, rpc_timeout=3.0, )

In this example, we assert that the msg sent is the same as the response received from the subscriber, demonstrating an operational RPC pattern over three different Redis data types.

Full Example of Redis RPC with FastStream#

Combining all the code snippets above, here is the complete example of how to set up Redis RPC with FastStream RedisBroker:

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49

from faststream import FastStream, Logger
from faststream.redis import RedisBroker

broker = RedisBroker("redis://localhost:6379")
app = FastStream(broker)


@broker.subscriber(channel="test-channel")
async def handle_channel(msg: str, logger: Logger):
    logger.info(msg)
    return msg


@broker.subscriber(list="test-list")
async def handle_list(msg: str, logger: Logger):
    logger.info(msg)
    return msg


@broker.subscriber(stream="test-stream")
async def handle_stream(msg: str, logger: Logger):
    logger.info(msg)
    return msg


@app.after_startup
async def t():
    msg = "Hi!"

    assert msg == await broker.publish(
        "Hi!",
        channel="test-channel",
        rpc=True,
        rpc_timeout=3.0,
    )

    assert msg == await broker.publish(
        "Hi!",
        list="test-list",
        rpc=True,
        rpc_timeout=3.0,
    )

    assert msg == await broker.publish(
        "Hi!",
        stream="test-stream",
        rpc=True,
        rpc_timeout=3.0,
    )

By embracing Redis RPC with FastStream, you can build sophisticated message-based architectures that require direct feedback from message processors. This feature is particularly suitable for cases where immediate processing is necessary or calling functions across different services is essential.