Class ServiceBusSessionReceiverAsyncClient

java.lang.Object
com.azure.messaging.servicebus.ServiceBusSessionReceiverAsyncClient
All Implemented Interfaces:
AutoCloseable
public final class ServiceBusSessionReceiverAsyncClient extends Object implements AutoCloseable
This asynchronous session receiver client is used to acquire session locks from a queue or topic and create ServiceBusReceiverAsyncClient instances that are tied to the locked sessions. Sessions can be used as a first in, first out (FIFO) processing of messages. Queues and topics/subscriptions support Service Bus sessions, however, it must be enabled at the time of entity creation.

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: Receive messages from a specific session

Use acceptSession(String) to acquire the lock of a session if you know the session id. ServiceBusReceiveMode.PEEK_LOCK and disableAutoComplete() are strongly recommended so users have control over message settlement.

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

// 'fullyQualifiedNamespace' will look similar to "{your-namespace}.servicebus.windows.net"
// 'disableAutoComplete' indicates that users will explicitly settle their message.
ServiceBusSessionReceiverAsyncClient sessionReceiver = new ServiceBusClientBuilder()
    .credential(fullyQualifiedNamespace, credential)
    .sessionReceiver()
    .disableAutoComplete()
    .queueName(sessionEnabledQueueName)
    .buildAsyncClient();

// acceptSession(String) completes successfully with a receiver when "<<my-session-id>>" session is
// successfully locked.
// `Flux.usingWhen` is used, so we dispose of the receiver resource after `receiveMessages()` and the settlement
// operations complete.
// `Mono.usingWhen` can also be used if the resource closure returns a single item.
Flux<Void> sessionMessages = Flux.usingWhen(
    sessionReceiver.acceptSession("<<my-session-id>>"),
    receiver -> {
        // Receive messages from <<my-session-id>> session.
        return receiver.receiveMessages().flatMap(message -> {
            System.out.printf("Received Sequence #: %s. Contents: %s%n", message.getSequenceNumber(),
                message.getBody());

            // Explicitly settle the message using complete, abandon, defer, dead-letter, etc.
            if (isMessageProcessed) {
                return receiver.complete(message);
            } else {
                return receiver.abandon(message);
            }
        });
    },
    receiver -> Mono.fromRunnable(() -> {
        // Dispose of resources.
        receiver.close();
        sessionReceiver.close();
    }));

// When program ends, or you're done receiving all messages, the `subscription` can be disposed of. This code
// is non-blocking and kicks off the operation.
Disposable subscription = sessionMessages.subscribe(
    unused -> {
    }, error -> System.err.print("Error receiving message from session: " + error),
    () -> System.out.println("Completed receiving from session."));

Sample: Receive messages from the first available session

Use acceptNextSession() to acquire the lock of the next available session without specifying the session id.

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

// 'fullyQualifiedNamespace' will look similar to "{your-namespace}.servicebus.windows.net"
// 'disableAutoComplete' indicates that users will explicitly settle their message.
ServiceBusSessionReceiverAsyncClient sessionReceiver = new ServiceBusClientBuilder()
    .credential(fullyQualifiedNamespace, credential)
    .sessionReceiver()
    .disableAutoComplete()
    .queueName(sessionEnabledQueueName)
    .buildAsyncClient();

// Creates a client to receive messages from the first available session. It waits until
// AmqpRetryOptions.getTryTimeout() elapses. If no session is available within that operation timeout, it
// completes with a retriable error. Otherwise, a receiver is returned when a lock on the session is acquired.
Mono<ServiceBusReceiverAsyncClient> receiverMono = sessionReceiver.acceptNextSession();

Flux<Void> receiveMessagesFlux = Flux.usingWhen(receiverMono,
    receiver -> receiver.receiveMessages().flatMap(message -> {
        System.out.println("Received message: " + message.getBody());

        // Explicitly settle the message via complete, abandon, defer, dead-letter, etc.
        if (isMessageProcessed) {
            return receiver.complete(message);
        } else {
            return receiver.abandon(message);
        }
    }),
    receiver -> Mono.fromRunnable(() -> {
        // Dispose of the receiver and sessionReceiver when done receiving messages.
        receiver.close();
        sessionReceiver.close();
    }));

// This is a non-blocking call that moves onto the next line of code after setting up and starting the receive
// operation. Customers can keep a reference to `subscription` and dispose of it when they want to stop
// receiving messages.
Disposable subscription = receiveMessagesFlux.subscribe(unused -> {
}, error -> System.out.println("Error occurred: " + error),
    () -> System.out.println("Receiving complete."));
See Also:

  • Method Details

    • acceptNextSession

      public Mono<ServiceBusReceiverAsyncClient> acceptNextSession()
      Acquires a session lock for the next available session and creates a ServiceBusReceiverAsyncClient to receive messages from the session. It will wait until a session is available if none is immediately available.
      Returns:
      A ServiceBusReceiverAsyncClient that is tied to the available session.
      Throws:
      UnsupportedOperationException - if the queue or topic subscription is not session-enabled.
      com.azure.core.amqp.exception.AmqpException - if the operation times out. The timeout duration is the tryTimeout of when you build this client with the ServiceBusClientBuilder.retryOptions(AmqpRetryOptions).

    • acceptSession

      public Mono<ServiceBusReceiverAsyncClient> acceptSession(String sessionId)
      Acquires a session lock for sessionId and create a ServiceBusReceiverAsyncClient to receive messages from the session. If the session is already locked by another client, an AmqpException is thrown.
      Parameters:
      sessionId - The session id.
      Returns:
      A ServiceBusReceiverAsyncClient that is tied to the specified session.
      Throws:
      NullPointerException - if sessionId is null.
      IllegalArgumentException - if sessionId is empty.
      UnsupportedOperationException - if the queue or topic subscription is not session-enabled.
      com.azure.core.amqp.exception.AmqpException - if the lock cannot be acquired.

    • listSessions

      public com.azure.core.http.rest.PagedFlux<String> listSessions()
      Lists the IDs of sessions that have active messages or stored session state in this entity.

      Sessions with active messages or stored session state in the queue or subscription are returned. Sessions with neither are excluded. Sessions on the dead-letter queue are not returned.

      The returned PagedFlux fetches additional pages from the broker on demand using cursor-based pagination (server-returned skip plus lastSessionId of the previous page) and terminates when the broker returns a page smaller than the requested page size (a short or empty page signals the end). The default page size is 100; callers can request a different size via ContinuablePagedFluxCore.byPage(int).

      Returns:
      A PagedFlux of session ID strings.

    • listSessions

      public com.azure.core.http.rest.PagedFlux<String> listSessions(OffsetDateTime sessionStateUpdatedAfter)
      Lists the IDs of sessions whose state was set or updated after the specified time.

      The returned PagedFlux fetches additional pages from the broker on demand using cursor-based pagination (server-returned skip plus lastSessionId of the previous page) and terminates when the broker returns a page smaller than the requested page size (a short or empty page signals the end). The default page size is 100; callers can request a different size via ContinuablePagedFluxCore.byPage(int).

      Parameters:
      sessionStateUpdatedAfter - Only sessions whose session state was set or updated after this time are returned.
      Returns:
      A PagedFlux of session ID strings.
      Throws:
      NullPointerException - if sessionStateUpdatedAfter is null.

    • close

      public void close()
      Specified by:
      close in interface AutoCloseable