Table of Contents

Consuming Messages

A consumer reads messages from the broker and dispatches them to your subscribers via the Silverback message bus.

This guide covers the basics: configuring consumers, handling messages, parallelism, and error handling.

Configure a Consumer

A consumer (Kafka) or client (MQTT) can consume one or more endpoints. Each endpoint defines the message type and the broker address to consume from (topic for Kafka, topic filter for MQTT).

services.AddSilverback()
    .WithConnectionToMessageBroker(options => options.AddKafka())
    .AddKafkaClients(clients => clients
        .WithBootstrapServers("PLAINTEXT://localhost:9092")
        .AddConsumer("consumer1", consumer => consumer
            .Consume<MyMessage>("endpoint1", endpoint => endpoint
                .ConsumeFrom("my-topic"))));
Tip

Configuration Notes

  • AddConsumer (Kafka) / AddClient (MQTT) creates a broker connection with its own settings.
  • Call Consume<TMessage> multiple times to add multiple endpoints to the same consumer/client. This usually reduces resource usage compared to creating one consumer per endpoint.
  • In Kafka, producers and consumers are separate entities. In MQTT, the same client can both produce and consume.
  • Naming consumers/clients and endpoints is optional but recommended for logging and diagnostics. It can also prevent double-registration if the same configuration block runs more than once.
  • For broker-specific options, see the upstream client docs: confluent-kafka-dotnet and MQTTnet.

Handle Messages

Consumed messages are published to the Using the Message Bus and handled by your subscribers.

You can subscribe to:

  • The deserialized message (MyMessage).
  • The envelope (IInboundEnvelope<MyMessage>) to access metadata (headers, broker identifier, etc.).
public class MySubscriber
{
    public async Task HandleAsync(MyMessage message)
    {
        Console.WriteLine($"Received message: {message.Content}");
    }

    public async Task HandleAsync(IInboundEnvelope<MyMessage> envelope)
    {
        Console.WriteLine($"Received message: {envelope.Message.Content}");

        // Kafka-only metadata example.
        Console.WriteLine($"Kafka key: {envelope.GetKafkaKey()}");
    }
}

Subscriber methods can be synchronous or asynchronous (Task/ValueTask). A message is considered successfully processed when all matching subscribers complete without throwing.

Warning

The same message can be handled by multiple subscribers. If you do that, make sure handlers are idempotent: if any subscriber fails, the message is considered failed and the configured error policy is applied.

For higher throughput, consider Batch Processing Consumed Messages.

Parallelism

Kafka and MQTT have different parallelism models:

  • Kafka: parallelism is enabled by default. Messages from different partitions can be processed concurrently; messages within the same partition are always processed sequentially.
  • MQTT: processing is sequential by default; you can enable parallel processing.

Kafka

Limit the maximum degree of parallelism:

.AddConsumer("consumer1", consumer => consumer
    .LimitParallelism(3)
    .Consume<MyMessage>("endpoint1", endpoint => endpoint
        .ConsumeFrom("my-topic-1", "my-topic-2")))

Process all partitions in a single stream (disables partition-based parallelism):

.AddConsumer("consumer1", consumer => consumer
    .ProcessAllPartitionsTogether()
    .Consume<MyMessage>("endpoint1", endpoint => endpoint
        .ConsumeFrom("my-topic-1", "my-topic-2")))

Both settings apply to the whole consumer, not to a single endpoint. Use LimitBackpressure(n) to set the buffer capacity per partition when processing partitions independently. With ProcessAllPartitionsTogether(), it sets the capacity of the shared buffer for all partitions. The default limit is 50.

MQTT

Enable parallel processing and set the maximum degree of parallelism:

.AddClient("my-client", client => client
    .WithClientId("client1")
    .EnableParallelProcessing(10)
    .Consume<MyMessage>("endpoint1", endpoint => endpoint
        .ConsumeFrom("messages/topic1", "messages/topic2")
        .WithAtLeastOnceQoS()));

Use LimitBackpressure(n) to limit how many incoming MQTT messages Silverback buffers per parallel execution. The number of parallel executions is controlled by MaxDegreeOfParallelism, configured with EnableParallelProcessing(...). The total buffer capacity per client is n × MaxDegreeOfParallelism, excluding messages already being processed. The default backpressure limit is 2.

For example, EnableParallelProcessing(4) with LimitBackpressure(7) allows up to 28 buffered messages, plus four being processed. This setting controls Silverback's processing buffers, not buffers in MQTTnet or the broker.

MQTT Sessions

Use RequestPersistentSession() together with a stable client identifier (WithClientId(...)) to resume the broker session after reconnecting and get the messages published in the meantime (QoS 1 or 2 only). See MQTT Specifics for all the session options and their details.

Error Handling

If a subscriber throws, Silverback applies the endpoint error policy. The default policy stops the consumer.

Built-in policies:

Policies can be chained:

services.AddSilverback()
    .WithConnectionToMessageBroker(options => options.AddKafka())
    .AddKafkaClients(clients => clients
        .WithBootstrapServers("PLAINTEXT://localhost:9092")
        .AddConsumer("consumer1", consumer => consumer
            .Consume<MyMessage>("endpoint1", endpoint => endpoint
                .ConsumeFrom("my-topic")
                .OnError(policy => policy.Retry(2).ThenSkip()))));
Important

If processing still fails after the last policy in the chain, the exception is returned to the consumer and the consumer stops.

Tip

See Error Handling Policies for all the available policies and options.

Important

RetryErrorPolicy waits before retrying and may delay polling. With Kafka, ensure broker/client timeouts (for example max.poll.interval.ms) are configured accordingly.

Apply Rules

Use ApplyTo, Exclude, and ApplyWhen to control which exceptions are handled and when. The ApplyWhen and Publish functions receive the IRawInboundEnvelope (and optionally the exception), not the deserialized message.

The MoveTo policy accepts either the name of a topic or the friendly name of an endpoint already configured in a producer.

.OnError(policy => policy
    .MoveTo(
        "some-other-topic",
        movePolicy => movePolicy
            .ApplyTo<MyException>()
            .ApplyWhen((envelope, exception) => envelope.Headers.Contains("x-move-on-error")))
    .ThenSkip());

Publish Events

You can publish a message when a policy is applied (for logging, alerts, etc.).

.OnError(policy => policy
    .Retry(3, retryPolicy => retryPolicy.WithInterval(TimeSpan.FromSeconds(1)))
    .ThenSkip(skipPolicy => skipPolicy
        .Publish(envelope => new ProcessingFailedEvent(envelope))));

Additional Resources

Edit this page