Interface IConsumer

Namespace

Silverback.Messaging.Broker

Assembly

Silverback.Integration.dll

Consumes from one or more endpoints and pushes the received messages via the message bus.

public interface IConsumer

Properties

Client

Gets the related IBrokerClient.

IBrokerClient Client { get; }

Property Value

IBrokerClient

DisplayName

Gets the name to be displayed in the human-targeted output (e.g. logs, health checks result, etc.).

string DisplayName { get; }

Property Value

string

EndpointsConfiguration

Gets the endpoints configuration.

IReadOnlyCollection<ConsumerEndpointConfiguration> EndpointsConfiguration { get; }

Property Value

IReadOnlyCollection<ConsumerEndpointConfiguration>

Name

Gets the consumer name.

string Name { get; }

Property Value

string

StatusInfo

Gets the IConsumerStatusInfo containing the status details and basic statistics of this consumer.

IConsumerStatusInfo StatusInfo { get; }

Property Value

IConsumerStatusInfo

Methods

CommitAsync(IBrokerMessageIdentifier)

Confirms that the specified message has been successfully processed. The acknowledgement will be sent to the message broker and the message will never be consumed again (by the same logical consumer / consumer group).

ValueTask CommitAsync(IBrokerMessageIdentifier brokerMessageIdentifier)

Parameters

brokerMessageIdentifier IBrokerMessageIdentifier

The identifier of the message to be committed.

Returns

ValueTask

A ValueTask representing the asynchronous operation.

CommitAsync(IReadOnlyCollection<IBrokerMessageIdentifier>)

Confirms that the specified messages have been successfully processed. The acknowledgement will be sent to the message broker, and the message will never be consumed again (by the same logical consumer / consumer group).

ValueTask CommitAsync(IReadOnlyCollection<IBrokerMessageIdentifier> brokerMessageIdentifiers)

Parameters

brokerMessageIdentifiers IReadOnlyCollection<IBrokerMessageIdentifier>

The identifiers of to message be committed.

Returns

ValueTask

A ValueTask representing the asynchronous operation.

IncrementFailedAttempts(IRawInboundEnvelope)

Increments the stored failed attempts count for the specified envelope.

int IncrementFailedAttempts(IRawInboundEnvelope envelope)

Parameters

envelope IRawInboundEnvelope

The envelope.

Returns

int

The current failed attempts count after the increment.

RollbackAsync(IBrokerMessageIdentifier)

Notifies that an error occurred while processing the specified message. If necessary, the information will be sent to the message broker to ensure that the message will be consumed again.

ValueTask RollbackAsync(IBrokerMessageIdentifier brokerMessageIdentifier)

Parameters

brokerMessageIdentifier IBrokerMessageIdentifier

The identifier of the message to be rolled back.

Returns

ValueTask

A ValueTask representing the asynchronous operation.

RollbackAsync(IReadOnlyCollection<IBrokerMessageIdentifier>)

Notifies that an error occured while processing the specified messages. If necessary the information will be sent to the message broker to ensure that the message will be re-processed.

ValueTask RollbackAsync(IReadOnlyCollection<IBrokerMessageIdentifier> brokerMessageIdentifiers)

Parameters

brokerMessageIdentifiers IReadOnlyCollection<IBrokerMessageIdentifier>

The identifiers of to messages be rolled back.

Returns

ValueTask

A ValueTask representing the asynchronous operation.

StartAsync()

Starts consuming. Used after StopAsync(bool) has been called to resume consuming.

ValueTask StartAsync()

Returns

ValueTask

A ValueTask representing the asynchronous operation.

StopAsync(IBrokerMessageIdentifier, bool)

Requests that the consumer stop because processing the specified message failed.

ValueTask StopAsync(IBrokerMessageIdentifier brokerMessageIdentifier, bool waitUntilStopped = true)

Parameters

brokerMessageIdentifier IBrokerMessageIdentifier

The original identifier from the received message that caused the stop request.

waitUntilStopped bool

Whether to wait for message processing to finish. Pass false when calling from message processing to avoid waiting for the calling operation itself.

Returns

ValueTask

A ValueTask representing the stop request. If waitUntilStopped is true, accepted stop requests also wait for message processing to finish.

Remarks

The original message identifier allows the broker implementation to ignore obsolete stop requests. For example, delayed error handling for a Kafka message from an earlier partition assignment must not stop consumption after reassignment. Use StopAsync(bool) to stop unconditionally. The request is ignored once the message processing is canceled, since the consumer is then already being stopped.

StopAsync(bool)

Stops the consumer without disconnecting. Can be used to pause and resume consuming.

ValueTask StopAsync(bool waitUntilStopped = true)

Parameters

waitUntilStopped bool

A value indicating whether the method should wait until the consumer has been effectively stopped.

Returns

ValueTask

A ValueTask representing the stop request. If waitUntilStopped is true, accepted stop requests also wait for message processing to finish.

TriggerReconnectAsync()

Stops the consumer and starts an asynchronous Task to disconnect and reconnect it.

ValueTask TriggerReconnectAsync()

Returns

ValueTask

A ValueTask representing the asynchronous operation. This ValueTask will complete as soon as the stopping signal has been sent, while the process will be completed in another asynchronous Task.

Remarks

This is used to recover when the consumer is stuck in state where it's not able to rollback or commit anymore. The request is ignored if the consumer is stopped or being stopped, since reconnecting would restart it.