Skip to content

Publisher Direct Usage#

The Publisher Object provides a full-featured way to publish messages. It has an AsyncAPI representation and includes testability features.

This method creates a reusable Publisher object that can be used directly to publish a message:

Pros and Cons

AsyncAPI support - AsyncAPI is a specification for describing asynchronous APIs used in messaging applications. This method supports the specification render.

Testing support - This method has full Testing support.

Broker availability from Context - You can leverage FastStream's Context, a built-in Dependency Injection (DI) container, to work with brokers or other external services.

Optional publication - You can create optional publications.

Reusable - This method is reusable.

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
from faststream import FastStream
from faststream.kafka import KafkaBroker

broker = KafkaBroker("localhost:9092")
app = FastStream(broker)

publisher = broker.publisher("another-topic")

@broker.subscriber("test-topic")
async def handle():
    await publisher.publish("Hi!")


@broker.subscriber("another-topic")
async def handle_next(msg: str):
    assert msg == "Hi!"

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
from faststream import FastStream
from faststream.confluent import KafkaBroker

broker = KafkaBroker("localhost:9092")
app = FastStream(broker)

publisher = broker.publisher("another-topic")

@broker.subscriber("test-topic")
async def handle():
    await publisher.publish("Hi!")


@broker.subscriber("another-topic")
async def handle_next(msg: str):
    assert msg == "Hi!"

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
from faststream import FastStream
from faststream.rabbit import RabbitBroker

broker = RabbitBroker("amqp://guest:guest@localhost:5672/")
app = FastStream(broker)

publisher = broker.publisher("another-queue")

@broker.subscriber("test-queue")
async def handle():
    await publisher.publish("Hi!")


@broker.subscriber("another-queue")
async def handle_next(msg: str):
    assert msg == "Hi!"

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
from faststream import FastStream
from faststream.nats import NatsBroker

broker = NatsBroker("nats://localhost:4222")
app = FastStream(broker)

publisher = broker.publisher("another-subject")

@broker.subscriber("test-subject")
async def handle():
    await publisher.publish("Hi!")


@broker.subscriber("another-subject")
async def handle_next(msg: str):
    assert msg == "Hi!"

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
from faststream import FastStream
from faststream.redis import RedisBroker

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

publisher = broker.publisher("another-channel")

@broker.subscriber("test-channel")
async def handle():
    await publisher.publish("Hi!")


@broker.subscriber("another-channel")
async def handle_next(msg: str):
    assert msg == "Hi!"

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
from faststream import FastStream
from faststream.mqtt import MQTTBroker

broker = MQTTBroker("localhost", port=1883)
app = FastStream(broker)

publisher = broker.publisher("another-topic")

@broker.subscriber("test-topic")
async def handle():
    await publisher.publish("Hi!")


@broker.subscriber("another-topic")
async def handle_next(msg: str):
    assert msg == "Hi!"

It is something in the middle between broker publish and object decorator. It has an AsyncAPI representation and testability features (like the object decorator), but allows you to send different messages to different outputs (like the broker publish).

@broker.subscriber("in")
async def handle(msg) -> str:
    await publisher1.publish("Response-1")
    await publisher2.publish("Response-2")

Overriding the Destination#

The destination passed to broker.publisher(...) is only a default. Every publisher.publish(...) call accepts the same destination arguments as broker.publish(...), and a value passed there overrides the one from the constructor for that single call:

publisher = broker.publisher("another-topic")

@broker.subscriber("in")
async def handle(msg: str) -> None:
    await publisher.publish(msg, topic=f"another-topic.{msg}")

publisher = broker.publisher("another-topic")

@broker.subscriber("in")
async def handle(msg: str) -> None:
    await publisher.publish(msg, topic=f"another-topic.{msg}")

publisher = broker.publisher("another-queue")

@broker.subscriber("in")
async def handle(msg: str) -> None:
    await publisher.publish(msg, routing_key=f"another-queue.{msg}")

publisher = broker.publisher("another-subject")

@broker.subscriber("in")
async def handle(msg: str) -> None:
    await publisher.publish(msg, subject=f"another-subject.{msg}")

publisher = broker.publisher("another-channel")

@broker.subscriber("in")
async def handle(msg: str) -> None:
    await publisher.publish(msg, channel=f"another-channel.{msg}")

publisher = broker.publisher("another-topic")

@broker.subscriber("in")
async def handle(msg: str) -> None:
    await publisher.publish(msg, topic=f"another-topic/{msg}")

The constructor argument still matters: the AsyncAPI schema describes the publisher by it, and the test mock is attached to it. If a publisher never publishes to its default destination, it is probably broker publish you want.

Note

When using this method, FastStream doesn't reuse the incoming correlation_id to mark outgoing messages with it. You should set it manually if it is required:

3
4
5
@broker.subscriber("test-queue")
async def handle(message: KafkaMessage):
    await publisher.publish("Hi!", correlation_id=message.correlation_id)

3
4
5
@broker.subscriber("test-queue")
async def handle(message: KafkaMessage):
    await publisher.publish("Hi!", correlation_id=message.correlation_id)

3
4
5
@broker.subscriber("test-queue")
async def handle(message: RabbitMessage):
    await publisher.publish("Hi!", correlation_id=message.correlation_id)

3
4
5
@broker.subscriber("test-queue")
async def handle(message: NatsMessage):
    await publisher.publish("Hi!", correlation_id=message.correlation_id)

3
4
5
@broker.subscriber("test-queue")
async def handle(message: RedisMessage):
    await publisher.publish("Hi!", correlation_id=message.correlation_id)

3
4
5
@broker.subscriber("test-topic")
async def handle(message: MQTTMessage):
    await publisher.publish("Hi!", correlation_id=message.correlation_id)