public final class ServiceBusClientBuilder.ServiceBusSessionProcessorClientBuilder extends Object
ServiceBusProcessorClient
to consume messages from a session-based Service Bus
entity. ServiceBusProcessorClient
processes messages and errors via ServiceBusClientBuilder.ServiceBusSessionProcessorClientBuilder.processMessage(Consumer)
and ServiceBusClientBuilder.ServiceBusSessionProcessorClientBuilder.processError(Consumer)
. When the processor finishes processing a session, it tries to fetch the
next session to process.
By default, the processor:
ServiceBusClientBuilder.ServiceBusSessionProcessorClientBuilder.disableAutoComplete()
ServiceBusClientBuilder.ServiceBusSessionProcessorClientBuilder.maxConcurrentSessions(int)
processMessage consumer
. Configured via
ServiceBusClientBuilder.ServiceBusSessionProcessorClientBuilder.maxConcurrentCalls(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();
ServiceBusProcessorClient
Modifier and Type | Method and Description |
---|---|
ServiceBusProcessorClient |
buildProcessorClient()
Creates a session-aware Service Bus processor responsible for reading
messages from a specific queue or subscription. |
ServiceBusClientBuilder.ServiceBusSessionProcessorClientBuilder |
disableAutoComplete()
Disables auto-complete and auto-abandon of received messages.
|
ServiceBusClientBuilder.ServiceBusSessionProcessorClientBuilder |
maxAutoLockRenewDuration(Duration maxAutoLockRenewDuration)
Sets the amount of time to continue auto-renewing the lock.
|
ServiceBusClientBuilder.ServiceBusSessionProcessorClientBuilder |
maxConcurrentCalls(int maxConcurrentCalls)
Max concurrent messages that this processor should process.
|
ServiceBusClientBuilder.ServiceBusSessionProcessorClientBuilder |
maxConcurrentSessions(int maxConcurrentSessions)
Enables session processing roll-over by processing at most
maxConcurrentSessions . |
ServiceBusClientBuilder.ServiceBusSessionProcessorClientBuilder |
prefetchCount(int prefetchCount)
Sets the prefetch count of the processor.
|
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.
|
ServiceBusClientBuilder.ServiceBusSessionProcessorClientBuilder |
processMessage(Consumer<ServiceBusReceivedMessageContext> processMessage)
The message processing callback for the processor that will be executed when a message is received.
|
ServiceBusClientBuilder.ServiceBusSessionProcessorClientBuilder |
queueName(String queueName)
Sets the name of the queue to create a processor for.
|
ServiceBusClientBuilder.ServiceBusSessionProcessorClientBuilder |
receiveMode(ServiceBusReceiveMode receiveMode)
Sets the receive mode for the processor.
|
ServiceBusClientBuilder.ServiceBusSessionProcessorClientBuilder |
subscriptionName(String subscriptionName)
Sets the name of the subscription in the topic to listen to.
|
ServiceBusClientBuilder.ServiceBusSessionProcessorClientBuilder |
topicName(String topicName)
Sets the name of the topic.
|
public ServiceBusClientBuilder.ServiceBusSessionProcessorClientBuilder maxAutoLockRenewDuration(Duration maxAutoLockRenewDuration)
Duration.ZERO
or null
disables auto-renewal. For RECEIVE_AND_DELETE
mode,
auto-renewal is disabled.maxAutoLockRenewDuration
- the amount of time to continue auto-renewing the lock. Duration.ZERO
or null
indicates that auto-renewal is disabled.ServiceBusClientBuilder.ServiceBusSessionProcessorClientBuilder
object.IllegalArgumentException
- If {code maxAutoLockRenewDuration} is negative.public ServiceBusClientBuilder.ServiceBusSessionProcessorClientBuilder maxConcurrentSessions(int maxConcurrentSessions)
maxConcurrentSessions
.maxConcurrentSessions
- Maximum number of concurrent sessions to process at any given time.ServiceBusClientBuilder.ServiceBusSessionProcessorClientBuilder
object.IllegalArgumentException
- if maxConcurrentSessions
is less than 1.public ServiceBusClientBuilder.ServiceBusSessionProcessorClientBuilder prefetchCount(int prefetchCount)
PEEK_LOCK
and
RECEIVE_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.prefetchCount
- The prefetch count.ServiceBusClientBuilder.ServiceBusProcessorClientBuilder
object.public ServiceBusClientBuilder.ServiceBusSessionProcessorClientBuilder queueName(String queueName)
queueName
- Name of the queue.ServiceBusClientBuilder.ServiceBusSessionProcessorClientBuilder
object.public ServiceBusClientBuilder.ServiceBusSessionProcessorClientBuilder receiveMode(ServiceBusReceiveMode receiveMode)
receiveMode
- Mode for receiving messages.ServiceBusClientBuilder.ServiceBusSessionProcessorClientBuilder
object.public ServiceBusClientBuilder.ServiceBusSessionProcessorClientBuilder subscriptionName(String subscriptionName)
ServiceBusClientBuilder.ServiceBusSessionProcessorClientBuilder.topicName(String)
must also be set.
subscriptionName
- Name of the subscription.ServiceBusClientBuilder.ServiceBusSessionProcessorClientBuilder
object.A topic name should be set as well.
public ServiceBusClientBuilder.ServiceBusSessionProcessorClientBuilder topicName(String topicName)
ServiceBusClientBuilder.ServiceBusSessionProcessorClientBuilder.subscriptionName(String)
must also be set.topicName
- Name of the topic.ServiceBusClientBuilder.ServiceBusSessionProcessorClientBuilder
object.A subscription name should be set as well.
public ServiceBusClientBuilder.ServiceBusSessionProcessorClientBuilder processMessage(Consumer<ServiceBusReceivedMessageContext> processMessage)
processMessage
- The message processing consumer that will be executed when a message is received.ServiceBusClientBuilder.ServiceBusProcessorClientBuilder
object.public ServiceBusClientBuilder.ServiceBusSessionProcessorClientBuilder processError(Consumer<ServiceBusErrorContext> processError)
processError
- The error handler which will be executed when an error occurs.ServiceBusClientBuilder.ServiceBusProcessorClientBuilder
objectpublic ServiceBusClientBuilder.ServiceBusSessionProcessorClientBuilder maxConcurrentCalls(int maxConcurrentCalls)
maxConcurrentCalls
- max concurrent messages that this processor should process.ServiceBusClientBuilder.ServiceBusSessionProcessorClientBuilder
object.IllegalArgumentException
- if maxConcurrentCalls
is less than 1.public ServiceBusClientBuilder.ServiceBusSessionProcessorClientBuilder disableAutoComplete()
completed
. If an error happens when
the message is processed, it is abandoned
.ServiceBusClientBuilder.ServiceBusSessionProcessorClientBuilder
object.public ServiceBusProcessorClient buildProcessorClient()
messages
from a specific queue or subscription.ServiceBusProcessorClient
that receives messages from a queue or subscription.IllegalStateException
- if queueName
or topicName
are not set or, both of these fields are set. It is also thrown if the Service Bus connectionString
contains an EntityPath
that does not match one set in
queueName
or topicName
. Lastly, if a topicName
is set, but subscriptionName
is not.IllegalArgumentException
- Queue or topic name are not set via queueName()
or topicName()
, respectively.NullPointerException
- if the ServiceBusClientBuilder.ServiceBusSessionProcessorClientBuilder.processMessage(Consumer)
or ServiceBusClientBuilder.ServiceBusSessionProcessorClientBuilder.processError(Consumer)
callbacks are not set.Copyright © 2021 Microsoft Corporation. All rights reserved.