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.

Note

The RPC feature is implemented over Redis Pub/Sub independently of the original subscriber type.

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:

Initiate your FastStream application with RedisBroker

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

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

@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

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

Additionally, you can set a timeout to decide how long the publisher should wait for a response before timing out.

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

    response: RedisMessage = await broker.request(
        "Hi!",
        channel="test-channel",
        timeout=3.0,
    )
    assert await response.decode() == msg

    response: RedisMessage = await broker.request(
        "Hi!",
        list="test-list",
        timeout=3.0,
    )
    assert await response.decode() == msg

    response: RedisMessage = await broker.request(
        "Hi!",
        stream="test-stream",
        timeout=3.0,
    )
    assert await response.decode() == msg

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:

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

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!"

    response: RedisMessage = await broker.request(
        "Hi!",
        channel="test-channel",
        timeout=3.0,
    )
    assert await response.decode() == msg

    response: RedisMessage = await broker.request(
        "Hi!",
        list="test-list",
        timeout=3.0,
    )
    assert await response.decode() == msg

    response: RedisMessage = await broker.request(
        "Hi!",
        stream="test-stream",
        timeout=3.0,
    )
    assert await response.decode() == msg

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.

Creating an RPC subscriber#

To handle an RPC request, you need to create a subscriber that processes the incoming message and returns a response. The subscriber should be decorated with @broker.subscriber and return either a raw value or a Response object.

Below is an example of a simple RPC subscriber that processes a message and returns a response.

from faststream.redis import RedisBroker

broker = RedisBroker()

@broker.subscriber(channel="test-channel")
async def handle(msg):
    return f"Received: {msg}"

When the client sends a request like this:

from faststream.redis import RedisMessage

msg: RedisMessage = await broker.request(
    "Hello, Redis!",
    channel="test-channel",
)
assert await msg.decode() == "Received: Hello, Redis!"

The subscriber processes the request and sends back the response, which is received by the client.

Tip

You can use the no_reply=True flag in the @broker.subscriber decorator to suppress automatic RPC and reply_to responses. This is useful when you want the subscriber to process the message without sending a response back to the client.

Using the Response class#

The Response class allows you to attach metadata, such as headers, to the response message. This is useful for adding context or tracking information to your responses.

Below is an example of how to use the Response class in an RPC subscriber.

from faststream import Response
from faststream.redis import RedisBroker

broker = RedisBroker()

@broker.subscriber(channel="test-channel")
async def handle(msg):
    return Response(
        body=f"Processed: {msg}",
        headers={"x-token": "some-token"},
        correlation_id="some-correlation-id",
    )

When the client sends a request:

from faststream.redis import RedisMessage

msg: RedisMessage = await broker.request(
    "Hello, Redis!",
    channel="test-channel",
)
assert msg.body == b"Processed: Hello, Redis!"
assert msg.headers == {"x-token": "some-token"}
assert msg.correlation_id == "some-correlation-id"

Using the RedisResponse class#

For Redis-specific use cases, you can use the RedisResponse class instead of the generic Response class.

The RedisResponse class extends Response and adds support for specifying a maxlen parameter, which is useful when publishing responses to a Redis stream to limit the stream's length. This option could be helpful with Reply-To feature, when reply-to destination is a Redis stream.

Below is an example of how to use the RedisResponse class in an RPC subscriber.

from faststream.redis import RedisBroker, RedisResponse

broker = RedisBroker()

@broker.subscriber(stream="test-stream")
async def handle(msg):
    return RedisResponse(
        body=f"Processed: {msg}",
        headers={"x-token": "some-token"},
        correlation_id="some-correlation-id",
        maxlen=1000,
    )