Package com.azure.messaging.servicebus
Class ServiceBusClientBuilder.ServiceBusSessionProcessorClientBuilder
java.lang.Object
com.azure.messaging.servicebus.ServiceBusClientBuilder.ServiceBusSessionProcessorClientBuilder
- Enclosing class:
- ServiceBusClientBuilder
Builder for creating
ServiceBusProcessorClient
to consume messages from a session-based Service Bus
entity. ServiceBusProcessorClient
processes messages and errors via processMessage(Consumer)
and processError(Consumer)
. When the processor finishes processing a session, it tries to fetch the
next session to process.
By default, the processor:
- Automatically settles messages. Disabled via
disableAutoComplete()
- Processes 1 session concurrently. Configured via
maxConcurrentSessions(int)
- Invokes 1 instance of
processMessage consumer
. Configured viamaxConcurrentCalls(int)
Instantiate a session-enabled processor client
Consumer<ServiceBusReceivedMessageContext> onMessage = context -> { ServiceBusReceivedMessage message = context.getMessage(); System.out.printf("Processing message. Session: %s, Sequence #: %s. Contents: %s%n", message.getSessionId(), message.getSequenceNumber(), message.getBody()); }; Consumer<ServiceBusErrorContext> onError = context -> { System.out.printf("Error when receiving messages from namespace: '%s'. Entity: '%s'%n", context.getFullyQualifiedNamespace(), context.getEntityPath()); if (context.getException() instanceof ServiceBusException) { ServiceBusException exception = (ServiceBusException) context.getException(); System.out.printf("Error source: %s, reason %s%n", context.getErrorSource(), exception.getReason()); } else { System.out.printf("Error occurred: %s%n", context.getException()); } }; // Retrieve 'connectionString/queueName' from your configuration. ServiceBusProcessorClient sessionProcessor = new ServiceBusClientBuilder() .connectionString(connectionString) .sessionProcessor() .queueName(queueName) .maxConcurrentSessions(2) .processMessage(onMessage) .processError(onError) .buildProcessorClient(); // Start the processor in the background sessionProcessor.start();
- See Also:
-
Method Summary
Modifier and TypeMethodDescriptionCreates a session-aware Service Bus processor responsible for readingmessages
from a specific queue or subscription.Disables auto-complete and auto-abandon of received messages.maxAutoLockRenewDuration
(Duration maxAutoLockRenewDuration) Sets the amount of time to continue auto-renewing the lock.maxConcurrentCalls
(int maxConcurrentCalls) Max concurrent messages that this processor should process.maxConcurrentSessions
(int maxConcurrentSessions) Enables session processing roll-over by processing at mostmaxConcurrentSessions
.prefetchCount
(int prefetchCount) Sets the prefetch count of the processor.processError
(Consumer<ServiceBusErrorContext> processError) The error handler for the processor which will be invoked in the event of an error while receiving messages.processMessage
(Consumer<ServiceBusReceivedMessageContext> processMessage) The message processing callback for the processor that will be executed when a message is received.Sets the name of the queue to create a processor for.receiveMode
(ServiceBusReceiveMode receiveMode) Sets the receive mode for the processor.Sets the type of theSubQueue
to connect to.subscriptionName
(String subscriptionName) Sets the name of the subscription in the topic to listen to.Sets the name of the topic.
-
Method Details
-
maxAutoLockRenewDuration
public ServiceBusClientBuilder.ServiceBusSessionProcessorClientBuilder maxAutoLockRenewDuration(Duration maxAutoLockRenewDuration) Sets the amount of time to continue auto-renewing the lock. SettingDuration.ZERO
ornull
disables auto-renewal. ForRECEIVE_AND_DELETE
mode, auto-renewal is disabled.- Parameters:
maxAutoLockRenewDuration
- the amount of time to continue auto-renewing the lock.Duration.ZERO
ornull
indicates that auto-renewal is disabled.- Returns:
- The updated
ServiceBusClientBuilder.ServiceBusSessionProcessorClientBuilder
object. - Throws:
IllegalArgumentException
- If {code maxAutoLockRenewDuration} is negative.
-
maxConcurrentSessions
public ServiceBusClientBuilder.ServiceBusSessionProcessorClientBuilder maxConcurrentSessions(int maxConcurrentSessions) Enables session processing roll-over by processing at mostmaxConcurrentSessions
.- Parameters:
maxConcurrentSessions
- Maximum number of concurrent sessions to process at any given time.- Returns:
- The modified
ServiceBusClientBuilder.ServiceBusSessionProcessorClientBuilder
object. - Throws:
IllegalArgumentException
- ifmaxConcurrentSessions
is less than 1.
-
prefetchCount
public ServiceBusClientBuilder.ServiceBusSessionProcessorClientBuilder prefetchCount(int prefetchCount) Sets the prefetch count of the processor. For bothPEEK_LOCK
andRECEIVE_AND_DELETE
modes the default value is 0. Prefetch speeds up the message flow by aiming to have a message readily available for local retrieval when and before the application starts the processor. Setting a non-zero value will prefetch that number of messages. Setting the value to zero turns prefetch off. Using a non-zero prefetch risks of losing messages even though it has better performance.- Parameters:
prefetchCount
- The prefetch count.- Returns:
- The modified
ServiceBusClientBuilder.ServiceBusProcessorClientBuilder
object. - See Also:
-
queueName
Sets the name of the queue to create a processor for.- Parameters:
queueName
- Name of the queue.- Returns:
- The modified
ServiceBusClientBuilder.ServiceBusSessionProcessorClientBuilder
object.
-
receiveMode
public ServiceBusClientBuilder.ServiceBusSessionProcessorClientBuilder receiveMode(ServiceBusReceiveMode receiveMode) Sets the receive mode for the processor.- Parameters:
receiveMode
- Mode for receiving messages.- Returns:
- The modified
ServiceBusClientBuilder.ServiceBusSessionProcessorClientBuilder
object.
-
subQueue
Sets the type of theSubQueue
to connect to. Azure Service Bus queues and subscriptions provide a secondary sub-queue, called a dead-letter queue (DLQ).- Parameters:
subQueue
- The type of the sub queue.- Returns:
- The modified
ServiceBusClientBuilder.ServiceBusSessionProcessorClientBuilder
object. - See Also:
-
subscriptionName
public ServiceBusClientBuilder.ServiceBusSessionProcessorClientBuilder subscriptionName(String subscriptionName) Sets the name of the subscription in the topic to listen to.topicName(String)
must also be set.- Parameters:
subscriptionName
- Name of the subscription.- Returns:
- The modified
ServiceBusClientBuilder.ServiceBusSessionProcessorClientBuilder
object. - See Also:
-
topicName
Sets the name of the topic.subscriptionName(String)
must also be set.- Parameters:
topicName
- Name of the topic.- Returns:
- The modified
ServiceBusClientBuilder.ServiceBusSessionProcessorClientBuilder
object. - See Also:
-
processMessage
public ServiceBusClientBuilder.ServiceBusSessionProcessorClientBuilder processMessage(Consumer<ServiceBusReceivedMessageContext> processMessage) The message processing callback for the processor that will be executed when a message is received.- Parameters:
processMessage
- The message processing consumer that will be executed when a message is received.- Returns:
- The updated
ServiceBusClientBuilder.ServiceBusProcessorClientBuilder
object.
-
processError
public ServiceBusClientBuilder.ServiceBusSessionProcessorClientBuilder processError(Consumer<ServiceBusErrorContext> processError) The error handler for the processor which will be invoked in the event of an error while receiving messages.- Parameters:
processError
- The error handler which will be executed when an error occurs.- Returns:
- The updated
ServiceBusClientBuilder.ServiceBusProcessorClientBuilder
object
-
maxConcurrentCalls
public ServiceBusClientBuilder.ServiceBusSessionProcessorClientBuilder maxConcurrentCalls(int maxConcurrentCalls) Max concurrent messages that this processor should process.- Parameters:
maxConcurrentCalls
- max concurrent messages that this processor should process.- Returns:
- The updated
ServiceBusClientBuilder.ServiceBusSessionProcessorClientBuilder
object. - Throws:
IllegalArgumentException
- ifmaxConcurrentCalls
is less than 1.
-
disableAutoComplete
Disables auto-complete and auto-abandon of received messages. By default, a successfully processed message iscompleted
. If an error happens when the message is processed, it isabandoned
.- Returns:
- The modified
ServiceBusClientBuilder.ServiceBusSessionProcessorClientBuilder
object.
-
buildProcessorClient
Creates a session-aware Service Bus processor responsible for readingmessages
from a specific queue or subscription.- Returns:
- An new
ServiceBusProcessorClient
that receives messages from a queue or subscription. - Throws:
IllegalStateException
- ifqueueName
ortopicName
are not set or, both of these fields are set. It is also thrown if the Service BusconnectionString
contains anEntityPath
that does not match one set inqueueName
ortopicName
. Lastly, if atopicName
is set, butsubscriptionName
is not.IllegalArgumentException
- Queue or topic name are not set viaqueueName()
ortopicName()
, respectively.NullPointerException
- if theprocessMessage(Consumer)
orprocessError(Consumer)
callbacks are not set.
-