Class ServiceBusReceiverClient

java.lang.Object
com.azure.messaging.servicebus.ServiceBusReceiverClient
All Implemented Interfaces:
AutoCloseable
public final class ServiceBusReceiverClient extends Object implements AutoCloseable
A synchronous receiver responsible for receiving ServiceBusReceivedMessage from a queue or topic/subscription on Azure Service Bus.

The examples shown in this document use a credential object named DefaultAzureCredential for authentication, which is appropriate for most scenarios, including local development and production environments. Additionally, we recommend using managed identity for authentication in production environments. You can find more information on different ways of authenticating and their corresponding credential types in the Azure Identity documentation".

Sample: Create a receiver and receive messages

The following code sample demonstrates the creation and use of the synchronous client ServiceBusReceiverClient to receive messages from a Service Bus subscription. The receive operation returns when either 10 messages are received or 30 seconds has elapsed. By default, messages are received using ServiceBusReceiveMode.PEEK_LOCK and customers must settle their messages using one of the settlement methods on the receiver client. " "Settling receive operations" provides additional information about message settlement.

TokenCredential tokenCredential = new DefaultAzureCredentialBuilder().build();

// 'fullyQualifiedNamespace' will look similar to "{your-namespace}.servicebus.windows.net"
ServiceBusReceiverClient receiver = new ServiceBusClientBuilder()
    .credential(fullyQualifiedNamespace, tokenCredential)
    .receiver()
    .topicName(topicName)
    .subscriptionName(subscriptionName)
    .buildClient();

// Receives a batch of messages when 10 messages are received or until 30 seconds have elapsed, whichever
// happens first.
IterableStream<ServiceBusReceivedMessage> messages = receiver.receiveMessages(10, Duration.ofSeconds(30));
messages.forEach(message -> {
    System.out.printf("Id: %s. Contents: %s%n", message.getMessageId(), message.getBody());

    // If able to process message, complete it. Otherwise, abandon it and allow it to be
    // redelivered.
    if (isMessageProcessed) {
        receiver.complete(message);
    } else {
        receiver.abandon(message);
    }
});

// When program ends, or you're done receiving all messages, dispose of the receiver.
// Clients should be long-lived objects as they
// require resources and time to establish a connection to the service.
receiver.close();
See Also:

  • Method Details

    • getFullyQualifiedNamespace

      public String getFullyQualifiedNamespace()
      Gets the fully qualified Service Bus namespace that the connection is associated with. This is likely similar to {yournamespace}.servicebus.windows.net.
      Returns:
      The fully qualified Service Bus namespace that the connection is associated with.

    • getEntityPath

      public String getEntityPath()
      Gets the Service Bus resource this client interacts with.
      Returns:
      The Service Bus resource this client interacts with.

    • getSessionId

      public String getSessionId()
      Gets the SessionId of the session if this receiver is a session receiver.
      Returns:
      The SessionId or null if this is not a session receiver.

    • getIdentifier

      public String getIdentifier()
      Gets the identifier of the instance of ServiceBusReceiverClient.
      Returns:
      The identifier that can identify the instance of ServiceBusReceiverClient.

    • abandon

      public void abandon(ServiceBusReceivedMessage message)
      Abandons a message. This will make the message available again for processing. Abandoning a message will increase the delivery count on the message.
      Parameters:
      message - The ServiceBusReceivedMessage to perform this operation.
      Throws:
      NullPointerException - if message is null.
      UnsupportedOperationException - if the receiver was opened in ServiceBusReceiveMode.RECEIVE_AND_DELETE mode or if the message was received from peekMessage.
      IllegalStateException - if receiver is already disposed.
      ServiceBusException - if the message could not be abandoned.
      IllegalArgumentException - if the message has either been deleted or already settled.

    • abandon

      public void abandon(ServiceBusReceivedMessage message, AbandonOptions options)
      Abandons a message and updates the message's properties. This will make the message available again for processing. Abandoning a message will increase the delivery count on the message.
      Parameters:
      message - The ServiceBusReceivedMessage to perform this operation.
      options - The options to set while abandoning the message.
      Throws:
      NullPointerException - if message or options is null. Also if transactionContext.transactionId is null when options.transactionContext is specified.
      UnsupportedOperationException - if the receiver was opened in ServiceBusReceiveMode.RECEIVE_AND_DELETE mode or if the message was received from peekMessage.
      IllegalStateException - if receiver is already disposed.
      IllegalArgumentException - if the message has either been deleted or already settled.
      ServiceBusException - if the message could not be abandoned.

    • complete

      public void complete(ServiceBusReceivedMessage message)
      Completes a message. This will delete the message from the service.
      Parameters:
      message - The ServiceBusReceivedMessage to perform this operation.
      Throws:
      NullPointerException - if message is null.
      UnsupportedOperationException - if the receiver was opened in ServiceBusReceiveMode.RECEIVE_AND_DELETE mode or if the message was received from peekMessage.
      IllegalStateException - if receiver is already disposed.
      IllegalArgumentException - if the message has either been deleted or already settled.
      ServiceBusException - if the message could not be completed.

    • complete

      public void complete(ServiceBusReceivedMessage message, CompleteOptions options)
      Completes a message. This will delete the message from the service.
      Parameters:
      message - The ServiceBusReceivedMessage to perform this operation.
      options - Options used to complete the message.
      Throws:
      NullPointerException - if message or options is null. Also if transactionContext.transactionId is null when options.transactionContext is specified.
      UnsupportedOperationException - if the receiver was opened in ServiceBusReceiveMode.RECEIVE_AND_DELETE mode or if the message was received from peekMessage.
      IllegalStateException - if receiver is already disposed.
      IllegalArgumentException - if the message has either been deleted or already settled.
      ServiceBusException - if the message could not be completed.

    • defer

      public void defer(ServiceBusReceivedMessage message)
      Defers a message. This will move message into the deferred subqueue.
      Parameters:
      message - The ServiceBusReceivedMessage to perform this operation.
      Throws:
      NullPointerException - if message is null.
      UnsupportedOperationException - if the receiver was opened in ServiceBusReceiveMode.RECEIVE_AND_DELETE mode or if the message was received from peekMessage.
      IllegalStateException - if receiver is already disposed.
      ServiceBusException - if the message could not be deferred.
      IllegalArgumentException - if the message has either been deleted or already settled.
      See Also:

    • defer

      public void defer(ServiceBusReceivedMessage message, DeferOptions options)
      Defers a message using its lock token with modified message property. This will move message into the deferred sub-queue.
      Parameters:
      message - The ServiceBusReceivedMessage to perform this operation.
      options - Options used to defer the message.
      Throws:
      NullPointerException - if message or options is null. Also if transactionContext.transactionId is null when options.transactionContext is specified.
      UnsupportedOperationException - if the receiver was opened in ServiceBusReceiveMode.RECEIVE_AND_DELETE mode or if the message was received from peekMessage.
      IllegalStateException - if receiver is already disposed.
      ServiceBusException - if the message could not be deferred.
      IllegalArgumentException - if the message has either been deleted or already settled.
      See Also:

    • deadLetter

      public void deadLetter(ServiceBusReceivedMessage message)
      Moves a message to the dead-letter sub-queue.
      Parameters:
      message - The ServiceBusReceivedMessage to perform this operation.
      Throws:
      NullPointerException - if message is null.
      UnsupportedOperationException - if the receiver was opened in ServiceBusReceiveMode.RECEIVE_AND_DELETE mode or if the message was received from peekMessage.
      IllegalStateException - if receiver is already disposed.
      ServiceBusException - if the message could not be dead-lettered.
      IllegalArgumentException - if the message has either been deleted or already settled.
      See Also:

    • deadLetter

      public void deadLetter(ServiceBusReceivedMessage message, DeadLetterOptions options)
      Moves a message to the dead-letter sub-queue with dead-letter reason, error description, and/or modified properties.
      Parameters:
      message - The ServiceBusReceivedMessage to perform this operation.
      options - Options used to dead-letter the message.
      Throws:
      NullPointerException - if message or options is null. Also if transactionContext.transactionId is null when options.transactionContext is specified.
      UnsupportedOperationException - if the receiver was opened in ServiceBusReceiveMode.RECEIVE_AND_DELETE mode or if the message was received from peekMessage.
      IllegalStateException - if the receiver is already disposed of.
      ServiceBusException - if the message could not be dead-lettered.
      IllegalArgumentException - if the message has either been deleted or already settled.
      See Also:

    • getSessionState

      public byte[] getSessionState()
      Gets the state of the session if this receiver is a session receiver.
      Returns:
      The session state or null if there is no state set for the session.
      Throws:
      IllegalStateException - if the receiver is a non-session receiver or receiver is already disposed.
      ServiceBusException - if the session state could not be acquired.

    • peekMessage

      public ServiceBusReceivedMessage peekMessage()
      Reads the next active message without changing the state of the receiver or the message source. The first call to peekMessage() fetches the first active message for this receiver. Each subsequent call fetches the subsequent message in the entity.
      Returns:
      A peeked ServiceBusReceivedMessage.
      Throws:
      IllegalStateException - if the receiver is already disposed.
      ServiceBusException - if an error occurs while peeking at the message.
      See Also:

    • peekMessage

      public ServiceBusReceivedMessage peekMessage(long sequenceNumber)
      Starting from the given sequence number, reads next the active message without changing the state of the receiver or the message source.
      Parameters:
      sequenceNumber - The sequence number from where to read the message.
      Returns:
      A peeked ServiceBusReceivedMessage.
      Throws:
      IllegalStateException - if the receiver is already disposed.
      ServiceBusException - if an error occurs while peeking at the message.
      See Also:

    • peekMessages

      public com.azure.core.util.IterableStream<ServiceBusReceivedMessage> peekMessages(int maxMessages)
      Reads the next batch of active messages without changing the state of the receiver or the message source.
      Parameters:
      maxMessages - The maximum number of messages to peek.
      Returns:
      An IterableStream of messages that are peeked.
      Throws:
      IllegalArgumentException - if maxMessages is not a positive integer.
      IllegalStateException - if the receiver is already disposed.
      ServiceBusException - if an error occurs while peeking at messages.
      See Also:

    • peekMessages

      public com.azure.core.util.IterableStream<ServiceBusReceivedMessage> peekMessages(int maxMessages, long sequenceNumber)
      Starting from the given sequence number, reads the next batch of active messages without changing the state of the receiver or the message source.
      Parameters:
      maxMessages - The number of messages.
      sequenceNumber - The sequence number from where to start reading messages.
      Returns:
      An IterableStream of messages peeked.
      Throws:
      IllegalArgumentException - if maxMessages is not a positive integer.
      IllegalStateException - if the receiver is already disposed.
      ServiceBusException - if an error occurs while peeking at messages.
      See Also:

    • receiveMessages

      public com.azure.core.util.IterableStream<ServiceBusReceivedMessage> receiveMessages(int maxMessages)
      Receives an iterable stream of messages from the Service Bus entity. The receive operation will wait for a default 1 minute for receiving a message before it times out. You can override it by using receiveMessages(int, Duration).

      The 1-minute timeout is a client-side feature. Each time the application calls receiveMessages, a timer is started on the client that when expires will terminate the IterableStream returned from this method. Timeout being a client-side feature means it is impossible to cancel any message requests that already made it to the broker. The messages can still arrive in the background after the IterableStream is transitioned to terminated state due to the client-side timeout. If there is no active IterableStream, the client will attempt to release any buffered messages back to the broker to avoid messages from going to dead letter. While messages are being released, if a new active IterableStream appears (due to a new receiveMessages call) then client will stop further release, so application may receive some messages from the buffer or already in transit followed by previously released messages when broker redeliver them, which can appear as out of order delivery.

      To keep the lock on each message received from a non-session resource (queue, topic subscription), the client will run a background task that will continuously renew the lock before it expires. By default, the lock renew task will run for a duration of 5 minutes, this duration can be adjusted using the ServiceBusClientBuilder.ServiceBusReceiverClientBuilder.maxAutoLockRenewDuration(Duration) API or can be turned off by setting it Duration.ZERO. A higher maxMessages value means an equivalent number of lock renewal tasks running in the client, which may put more stress on low CPU environments. Given each lock renewal is a network call to the broker, a high number of lock renewal tasks making multiple lock renew calls also may have an adverse effect in namespace throttling. Additionally, if certain lock renewal tasks fail to renew the lock on time because of low CPU, service throttling or overloaded network, then client may lose the lock on the messages, which will cause the application's attempts to settle (e.g., complete, abandon) those messages to fail. The broker will redeliver those messages, but if the settling attempts fail repeatedly beyond the max delivery count, then the message will be transferred to dead letter queue. Keep this in mind when choosing maxMessages. You may consider disabling the client-side lock renewal using maxAutoLockRenewDuration(Duration.ZERO) if you can configure a lock duration at the resource (queue,topic subscription) level that at least exceeds the cumulative expected processing time for maxMessages messages.

      The client uses an AMQP link underneath to receive the messages; the client will transparently transition to a new AMQP link if the current one encounters a retriable error. When the client experiences a non-retriable error or exhausts the retries, the iteration (e.g., forEach) on the IterableStream returned by the further invocations of receiveMessages API will throw the error to the application. Once the application receives this error, the application should reset the client, i.e., close the current ServiceBusReceiverClient and create a new client to continue receiving messages.

      Note: A few examples of non-retriable errors are - the application attempting to connect to a queue that does not exist, deleting or disabling the queue in the middle of receiving, the user explicitly initiating Geo-DR. These are certain events where the Service Bus communicates to the client that a non-retriable error occurred.

      Parameters:
      maxMessages - The maximum number of messages to receive.
      Returns:
      An IterableStream of at most maxMessages messages from the Service Bus entity.
      Throws:
      IllegalArgumentException - if maxMessages is zero or a negative value.
      IllegalStateException - if the receiver is already disposed.
      ServiceBusException - if an error occurs while receiving messages.
      See Also:

    • receiveMessages

      public com.azure.core.util.IterableStream<ServiceBusReceivedMessage> receiveMessages(int maxMessages, Duration maxWaitTime)
      Receives an iterable stream of messages from the Service Bus entity with a timout. The default receive mode is ServiceBusReceiveMode.PEEK_LOCK unless it is changed during creation of ServiceBusReceiverClient using ServiceBusClientBuilder.ServiceBusReceiverClientBuilder.receiveMode(ServiceBusReceiveMode).

      The support for timeout maxWaitTime is a client-side feature. Each time the application calls receiveMessages, a timer is started on the client that when expires will terminate the IterableStream returned from this method. Timeout being a client-side feature means it is impossible to cancel any message requests that already made it to the broker. The messages can still arrive in the background after the IterableStream is transitioned to terminated state due to the client-side timeout. If there is no active IterableStream, the client will attempt to release any buffered messages back to the broker to avoid messages from going to dead letter. While messages are being released, if a new active IterableStream appears (due to a new receiveMessages call) then client will stop further release, so application may receive some messages from the buffer or already in transit followed by previously released messages when broker redeliver them, which can appear as out of order delivery. Consider these when choosing the timeout. For example, a small timeout with a higher maxMessages value while there are a lot of messages in the entity can increase the release network calls to the broker that might have adverse effect in namespace throttling and increases the chances of out of order deliveries. Also, frequent receiveMessages with low timeout means frequent scheduling of timer tasks, which may put more stress on low CPU environments.

      To keep the lock on each message received from a non-session resource (queue, topic subscription), the client will run a background task that will continuously renew the lock before it expires. By default, the lock renew task will run for a duration of 5 minutes, this duration can be adjusted using the ServiceBusClientBuilder.ServiceBusReceiverClientBuilder.maxAutoLockRenewDuration(Duration) API or can be turned off by setting it Duration.ZERO. A higher maxMessages value means an equivalent number of lock renewal tasks running in the client, which may put more stress on low CPU environments. Given each lock renewal is a network call to the broker, a high number of lock renewal tasks making multiple lock renew calls also may have an adverse effect in namespace throttling. Additionally, if certain lock renewal tasks fail to renew the lock on time because of low CPU, service throttling or overloaded network, then client may lose the lock on the messages, which will cause the application's attempts to settle (e.g., complete, abandon) those messages to fail. The broker will redeliver those messages, but if the settling attempts fail repeatedly beyond the max delivery count, then the message will be transferred to dead letter queue. Keep this in mind when choosing maxMessages. You may consider disabling the client-side lock renewal using maxAutoLockRenewDuration(Duration.ZERO) if you can configure a lock duration at the resource (queue,topic subscription) level that at least exceeds the cumulative expected processing time for maxMessages messages.

      The client uses an AMQP link underneath to receive the messages; the client will transparently transition to a new AMQP link if the current one encounters a retriable error. When the client experiences a non-retriable error or exhausts the retries, the iteration (e.g., forEach) on the IterableStream returned by the further invocations of receiveMessages API will throw the error to the application. Once the application receives this error, the application should reset the client, i.e., close the current ServiceBusReceiverClient and create a new client to continue receiving messages.

      Note: A few examples of non-retriable errors are - the application attempting to connect to a queue that does not exist, deleting or disabling the queue in the middle of receiving, the user explicitly initiating Geo-DR. These are certain events where the Service Bus communicates to the client that a non-retriable error occurred.

      Parameters:
      maxMessages - The maximum number of messages to receive.
      maxWaitTime - The time the client waits for receiving a message before it times out.
      Returns:
      An IterableStream of at most maxMessages messages from the Service Bus entity.
      Throws:
      IllegalArgumentException - if maxMessages or maxWaitTime is zero or a negative value.
      IllegalStateException - if the receiver is already disposed.
      NullPointerException - if maxWaitTime is null.
      ServiceBusException - if an error occurs while receiving messages.
      See Also:

    • receiveDeferredMessage

      public ServiceBusReceivedMessage receiveDeferredMessage(long sequenceNumber)
      Receives a deferred message. Deferred messages can only be received by using sequence number.
      Parameters:
      sequenceNumber - The sequence number of the message.
      Returns:
      A deferred message with the matching sequenceNumber.
      Throws:
      IllegalStateException - if receiver is already disposed.
      ServiceBusException - if deferred message cannot be received.

    • receiveDeferredMessageBatch

      public com.azure.core.util.IterableStream<ServiceBusReceivedMessage> receiveDeferredMessageBatch(Iterable<Long> sequenceNumbers)
      Receives a batch of deferred messages. Deferred messages can only be received by using sequence number.
      Parameters:
      sequenceNumbers - The sequence numbers of the deferred messages.
      Returns:
      An IterableStream of deferred messages.
      Throws:
      NullPointerException - if sequenceNumbers is null.
      IllegalStateException - if receiver is already disposed.
      ServiceBusException - if deferred messages cannot be received.

    • renewMessageLock

      public OffsetDateTime renewMessageLock(ServiceBusReceivedMessage message)
      Renews the lock on the specified message. The lock will be renewed based on the setting specified on the entity. When a message is received in ServiceBusReceiveMode.PEEK_LOCK mode, the message is locked on the server for this receiver instance for a duration as specified during the Queue creation (LockDuration). If processing of the message requires longer than this duration, the lock needs to be renewed. For each renewal, the lock is reset to the entity's LockDuration value.
      Parameters:
      message - The ServiceBusReceivedMessage to perform lock renewal.
      Returns:
      The new expiration time for the message.
      Throws:
      NullPointerException - if message or message.getLockToken() is null.
      UnsupportedOperationException - if the receiver was opened in ServiceBusReceiveMode.RECEIVE_AND_DELETE mode or if the message was received from peekMessage.
      IllegalStateException - if the receiver is a session receiver or receiver is already disposed.
      IllegalArgumentException - if message.getLockToken() is an empty value.

    • renewMessageLock

      public void renewMessageLock(ServiceBusReceivedMessage message, Duration maxLockRenewalDuration, Consumer<Throwable> onError)
      Starts the auto lock renewal for a message with the given lock.
      Parameters:
      message - The ServiceBusReceivedMessage to perform auto-lock renewal.
      maxLockRenewalDuration - Maximum duration to keep renewing the lock token.
      onError - A function to call when an error occurs during lock renewal.
      Throws:
      NullPointerException - if message, message.getLockToken(), or maxLockRenewalDuration is null.
      IllegalStateException - if the receiver is a session receiver or the receiver is disposed.
      IllegalArgumentException - if message.getLockToken() is an empty value.
      ServiceBusException - If the message cannot be renewed.

    • renewSessionLock

      public OffsetDateTime renewSessionLock()
      Sets the state of the session if this receiver is a session receiver.
      Returns:
      The next expiration time for the session lock.
      Throws:
      IllegalStateException - if the receiver is a non-session receiver or receiver is already disposed.
      ServiceBusException - if the session lock cannot be renewed.

    • renewSessionLock

      public void renewSessionLock(Duration maxLockRenewalDuration, Consumer<Throwable> onError)
      Starts the auto lock renewal for the session that this receiver works for.
      Parameters:
      maxLockRenewalDuration - Maximum duration to keep renewing the session.
      onError - A function to call when an error occurs during lock renewal.
      Throws:
      NullPointerException - if sessionId or maxLockRenewalDuration is null.
      IllegalArgumentException - if sessionId is an empty string or maxLockRenewalDuration is negative.
      IllegalStateException - if the receiver is a non-session receiver or the receiver is disposed.
      ServiceBusException - if the session lock renewal operation cannot be started.

    • setSessionState

      public void setSessionState(byte[] sessionState)
      Sets the state of the session if this receiver is a session receiver.
      Parameters:
      sessionState - State to set on the session.
      Throws:
      IllegalStateException - if the receiver is a non-session receiver or receiver is already disposed.
      ServiceBusException - if the session state cannot be set.

    • createTransaction

      public ServiceBusTransactionContext createTransaction()
      Starts a new transaction on Service Bus. The ServiceBusTransactionContext should be passed along to all operations that need to be in this transaction.

      Sample: Creating and using a transaction

      ServiceBusTransactionContext transaction = receiver.createTransaction();

// Process messages and associate operations with the transaction.
ServiceBusReceivedMessage deferredMessage = receiver.receiveDeferredMessage(sequenceNumber);
receiver.complete(deferredMessage, new CompleteOptions().setTransactionContext(transaction));
receiver.abandon(receivedMessage, new AbandonOptions().setTransactionContext(transaction));
receiver.commitTransaction(transaction);
      Returns:
      A new ServiceBusTransactionContext.
      Throws:
      IllegalStateException - if the receiver is already disposed.
      ServiceBusException - if a transaction cannot be created.

    • commitTransaction

      public void commitTransaction(ServiceBusTransactionContext transactionContext)
      Commits the transaction and all the operations associated with it.

      Creating and using a transaction

      ServiceBusTransactionContext transaction = receiver.createTransaction();

// Process messages and associate operations with the transaction.
ServiceBusReceivedMessage deferredMessage = receiver.receiveDeferredMessage(sequenceNumber);
receiver.complete(deferredMessage, new CompleteOptions().setTransactionContext(transaction));
receiver.abandon(receivedMessage, new AbandonOptions().setTransactionContext(transaction));
receiver.commitTransaction(transaction);
      Parameters:
      transactionContext - The transaction to be commit.
      Throws:
      IllegalStateException - if the receiver is already disposed.
      NullPointerException - if transactionContext or transactionContext.transactionId is null.
      ServiceBusException - if the transaction could not be committed.

    • rollbackTransaction

      public void rollbackTransaction(ServiceBusTransactionContext transactionContext)
      Rollbacks the transaction given and all operations associated with it.

      Creating and using a transaction

      ServiceBusTransactionContext transaction = receiver.createTransaction();

// Process messages and associate operations with the transaction.
ServiceBusReceivedMessage deferredMessage = receiver.receiveDeferredMessage(sequenceNumber);
receiver.complete(deferredMessage, new CompleteOptions().setTransactionContext(transaction));
receiver.abandon(receivedMessage, new AbandonOptions().setTransactionContext(transaction));
receiver.commitTransaction(transaction);
      Parameters:
      transactionContext - The transaction to be rollback.
      Throws:
      IllegalStateException - if the receiver is already disposed.
      NullPointerException - if transactionContext or transactionContext.transactionId is null.
      ServiceBusException - if the transaction could not be rolled back.

    • close

      public void close()
      Disposes of the consumer by closing the underlying links to the service.
      Specified by:
      close in interface AutoCloseable