ServiceBusReceiverClient Class

Package:

com.azure.messaging.servicebus

Maven Artifact:

com.azure:azure-messaging-servicebus:7.16.0

• java.lang.Object
• • com.azure.messaging.servicebus.ServiceBusReceiverClient

Implements

AutoCloseable

public final class ServiceBusReceiverClient
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 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();

Method Summary

Modifier and Type	Method and Description
void	abandon(ServiceBusReceivedMessage message) Abandons a ServiceBusReceivedMessage.
void	abandon(ServiceBusReceivedMessage message, AbandonOptions options) Abandons a ServiceBusReceivedMessage and updates the message's properties.
void	close() Disposes of the consumer by closing the underlying links to the service.
void	commitTransaction(ServiceBusTransactionContext transactionContext) Commits the transaction and all the operations associated with it.
void	complete(ServiceBusReceivedMessage message) Completes a ServiceBusReceivedMessage.
void	complete(ServiceBusReceivedMessage message, CompleteOptions options) Completes a ServiceBusReceivedMessage.
ServiceBusTransactionContext	createTransaction() Starts a new transaction on Service Bus.
void	deadLetter(ServiceBusReceivedMessage message) Moves a ServiceBusReceivedMessage to the dead-letter sub-queue.
void	deadLetter(ServiceBusReceivedMessage message, DeadLetterOptions options) Moves a ServiceBusReceivedMessage to the dead-letter sub-queue with dead-letter reason, error description, and/or modified properties.
void	defer(ServiceBusReceivedMessage message) Defers a ServiceBusReceivedMessage.
void	defer(ServiceBusReceivedMessage message, DeferOptions options) Defers a ServiceBusReceivedMessage using its lock token with modified message property.
String	getEntityPath() Gets the Service Bus resource this client interacts with.
String	getFullyQualifiedNamespace() Gets the fully qualified Service Bus namespace that the connection is associated with.
String	getIdentifier() Gets the identifier of the instance of ServiceBusReceiverClient.
String	getSessionId() Gets the SessionId of the session if this receiver is a session receiver.
byte[]	getSessionState() Gets the state of the session if this receiver is a session receiver.
ServiceBusReceivedMessage	peekMessage() Reads the next active message without changing the state of the receiver or the message source.
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.
IterableStream<ServiceBusReceivedMessage>	peekMessages(int maxMessages) Reads the next batch of active messages without changing the state of the receiver or the message source.
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.
ServiceBusReceivedMessage	receiveDeferredMessage(long sequenceNumber) Receives a deferred ServiceBusReceivedMessage.
IterableStream<ServiceBusReceivedMessage>	receiveDeferredMessageBatch(Iterable<Long> sequenceNumbers) Receives a batch of deferred ServiceBusReceivedMessage.
IterableStream<ServiceBusReceivedMessage>	receiveMessages(int maxMessages) Receives an iterable stream of ServiceBusReceivedMessage from the Service Bus entity.
IterableStream<ServiceBusReceivedMessage>	receiveMessages(int maxMessages, Duration maxWaitTime) Receives an iterable stream of ServiceBusReceivedMessage from the Service Bus entity.
OffsetDateTime	renewMessageLock(ServiceBusReceivedMessage message) Renews the lock on the specified message.
void	renewMessageLock(ServiceBusReceivedMessage message, Duration maxLockRenewalDuration, Consumer<Throwable> onError) Starts the auto lock renewal for a message with the given lock.
OffsetDateTime	renewSessionLock() Sets the state of the session if this receiver is a session receiver.
void	renewSessionLock(Duration maxLockRenewalDuration, Consumer<Throwable> onError) Starts the auto lock renewal for the session that this receiver works for.
void	rollbackTransaction(ServiceBusTransactionContext transactionContext) Rollbacks the transaction given and all operations associated with it.
void	setSessionState(byte[] sessionState) Sets the state of the session if this receiver is a session receiver.

Methods inherited from java.lang.Object

clone equals finalize getClass hashCode notify notifyAll toString wait wait wait

Method Details

abandon

public void abandon(ServiceBusReceivedMessage message)

Abandons a ServiceBusReceivedMessage. 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.

abandon

public void abandon(ServiceBusReceivedMessage message, AbandonOptions options)

Abandons a ServiceBusReceivedMessage 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.

close

public void close()

Disposes of the consumer by closing the underlying links to the service.

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.

complete

public void complete(ServiceBusReceivedMessage message)

Completes a ServiceBusReceivedMessage. This will delete the message from the service.

Parameters:

message - The ServiceBusReceivedMessage to perform this operation.

complete

public void complete(ServiceBusReceivedMessage message, CompleteOptions options)

Completes a ServiceBusReceivedMessage. This will delete the message from the service.

Parameters:

message - The ServiceBusReceivedMessage to perform this operation.

options - Options used to complete the message.

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.

deadLetter

public void deadLetter(ServiceBusReceivedMessage message)

Moves a ServiceBusReceivedMessage to the dead-letter sub-queue.

Parameters:

message - The ServiceBusReceivedMessage to perform this operation.

deadLetter

public void deadLetter(ServiceBusReceivedMessage message, DeadLetterOptions options)

Moves a ServiceBusReceivedMessage 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.

defer

public void defer(ServiceBusReceivedMessage message)

Defers a ServiceBusReceivedMessage. This will move message into the deferred subqueue.

Parameters:

message - The ServiceBusReceivedMessage to perform this operation.

defer

public void defer(ServiceBusReceivedMessage message, DeferOptions options)

Defers a ServiceBusReceivedMessage 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.

getEntityPath

public String getEntityPath()

Gets the Service Bus resource this client interacts with.

Returns:

The Service Bus resource this client interacts with.

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.

getIdentifier

public String getIdentifier()

Gets the identifier of the instance of ServiceBusReceiverClient.

Returns:

The identifier that can identify the instance of ServiceBusReceiverClient.

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.

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.

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.

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.

peekMessages

public IterableStream 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<T> of ServiceBusReceivedMessage that are peeked.

peekMessages

public IterableStream 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<T> of ServiceBusReceivedMessage peeked.

receiveDeferredMessage

public ServiceBusReceivedMessage receiveDeferredMessage(long sequenceNumber)

Receives a deferred ServiceBusReceivedMessage. Deferred messages can only be received by using sequence number.

Parameters:

sequenceNumber - The getSequenceNumber() of the message.

Returns:

A deferred message with the matching sequenceNumber.

receiveDeferredMessageBatch

public IterableStream receiveDeferredMessageBatch(Iterable sequenceNumbers)

Receives a batch of deferred ServiceBusReceivedMessage. Deferred messages can only be received by using sequence number.

Parameters:

sequenceNumbers - The sequence numbers of the deferred messages.

Returns:

An IterableStream<T> of deferred ServiceBusReceivedMessage.

receiveMessages

public IterableStream receiveMessages(int maxMessages)

Receives an iterable stream of ServiceBusReceivedMessage 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 maxMessages, Duration maxWaitTime).

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<T> 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<T> of at most maxMessages messages from the Service Bus entity.

receiveMessages

public IterableStream receiveMessages(int maxMessages, Duration maxWaitTime)

Receives an iterable stream of ServiceBusReceivedMessage from the Service Bus entity. The default receive mode is PEEK_LOCK unless it is changed during creation of ServiceBusReceiverClient using ServiceBusReceiverClientBuilder#receiveMode(ServiceBusReceiveMode).

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<T> 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<T> of at most maxMessages messages from the Service Bus entity.

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 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.

renewMessageLock

public void renewMessageLock(ServiceBusReceivedMessage message, Duration maxLockRenewalDuration, Consumer 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.

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.

renewSessionLock

public void renewSessionLock(Duration maxLockRenewalDuration, Consumer 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.

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.

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.