Package com.azure.messaging.servicebus

Class ServiceBusSenderAsyncClient

java.lang.Object
com.azure.messaging.servicebus.ServiceBusSenderAsyncClient
All Implemented Interfaces:
AutoCloseable
public final class ServiceBusSenderAsyncClient extends Object implements AutoCloseable

An asynchronous client to send messages to a Service Bus resource.

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 an instance of sender

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

 // 'fullyQualifiedNamespace' will look similar to "{your-namespace}.servicebus.windows.net"
 ServiceBusSenderAsyncClient asyncSender = new ServiceBusClientBuilder()
     .credential(fullyQualifiedNamespace, credential)
     .sender()
     .queueName(queueName)
     .buildAsyncClient();

 // When users are done with the sender, they should dispose of it.
 // Clients should be long-lived objects as they require resources
 // and time to establish a connection to the service.
 asyncSender.close();

Sample: Send messages to a Service Bus resource

 // `subscribe` is a non-blocking call. The program will move onto the next line of code when it starts the
 // operation.  Users should use the callbacks on `subscribe` to understand the status of the send operation.
 asyncSender.createMessageBatch().flatMap(batch -> {
     batch.tryAddMessage(new ServiceBusMessage("test-1"));
     batch.tryAddMessage(new ServiceBusMessage("test-2"));
     return asyncSender.sendMessages(batch);
 }).subscribe(unused -> {
 }, error -> {
     System.err.println("Error occurred while sending batch:" + error);
 }, () -> {
     System.out.println("Send complete.");
 });

Sample: Send messages using a size-limited ServiceBusMessageBatch to a Service Bus resource 

 Flux<ServiceBusMessage> telemetryMessages = Flux.just(firstMessage, secondMessage);

 // Setting `setMaximumSizeInBytes` when creating a batch, limits the size of that batch.
 // In this case, all the batches created with these options are limited to 256 bytes.
 CreateMessageBatchOptions options = new CreateMessageBatchOptions()
     .setMaximumSizeInBytes(256);
 AtomicReference<ServiceBusMessageBatch> currentBatch = new AtomicReference<>();

 // Sends the current batch if it is not null and not empty.  If the current batch is null, sets it.
 // Returns the batch to work with.
 Mono<ServiceBusMessageBatch> sendBatchAndGetCurrentBatchOperation = Mono.defer(() -> {
     ServiceBusMessageBatch batch = currentBatch.get();

     if (batch == null) {
         return asyncSender.createMessageBatch(options);
     }

     if (batch.getCount() > 0) {
         return asyncSender.sendMessages(batch).then(
             asyncSender.createMessageBatch(options)
                 .handle((ServiceBusMessageBatch newBatch, SynchronousSink<ServiceBusMessageBatch> sink) -> {
                     // Expect that the batch we just sent is the current one. If it is not, there's a race
                     // condition accessing currentBatch reference.
                     if (!currentBatch.compareAndSet(batch, newBatch)) {
                         sink.error(new IllegalStateException(
                             "Expected that the object in currentBatch was batch. But it is not."));
                     } else {
                         sink.next(newBatch);
                     }
                 }));
     } else {
         return Mono.just(batch);
     }
 });

 // The sample Flux contains two messages, but it could be an infinite stream of telemetry messages.
 Flux<Void> sendMessagesOperation = telemetryMessages.flatMap(message -> {
     return sendBatchAndGetCurrentBatchOperation.flatMap(batch -> {
         if (batch.tryAddMessage(message)) {
             return Mono.empty();
         } else {
             return sendBatchAndGetCurrentBatchOperation
                 .handle((ServiceBusMessageBatch newBatch, SynchronousSink<Void> sink) -> {
                     if (!newBatch.tryAddMessage(message)) {
                         sink.error(new IllegalArgumentException(
                             "Message is too large to fit in an empty batch."));
                     } else {
                         sink.complete();
                     }
                 });
         }
     });
 });

 // `subscribe` is a non-blocking call. The program will move onto the next line of code when it starts the
 // operation.  Users should use the callbacks on `subscribe` to understand the status of the send operation.
 Disposable disposable = sendMessagesOperation.then(sendBatchAndGetCurrentBatchOperation)
     .subscribe(batch -> {
         System.out.println("Last batch should be empty: " + batch.getCount());
     }, error -> {
         System.err.println("Error sending telemetry messages: " + error);
     }, () -> {
         System.out.println("Completed.");

         // Continue using the sender and finally, dispose of the sender.
         // Clients should be long-lived objects as they require resources
         // and time to establish a connection to the service.
         asyncSender.close();
     });

Sample: Sending a message to a session-enabled queue

The snippet below demonstrates sending a message to a Service Bus sessions enabled queue. Setting ServiceBusMessage.setMessageId(String) property to "greetings" will send the message to a Service Bus session with an id of "greetings".

 // 'fullyQualifiedNamespace' will look similar to "{your-namespace}.servicebus.windows.net"
 ServiceBusSenderAsyncClient sender = new ServiceBusClientBuilder()
     .credential(fullyQualifiedNamespace, new DefaultAzureCredentialBuilder().build())
     .sender()
     .queueName(sessionEnabledQueueName)
     .buildAsyncClient();

 // Setting sessionId publishes that message to a specific session, in this case, "greeting".
 ServiceBusMessage message = new ServiceBusMessage("Hello world")
     .setSessionId("greetings");

 // `subscribe` is a non-blocking call. The program will move onto the next line of code when it starts the
 // operation.  Users should use the callbacks on `subscribe` to understand the status of the send operation.
 sender.sendMessage(message).subscribe(unused -> {
 }, error -> {
     System.err.println("Error occurred publishing batch: " + error);
 }, () -> {
     System.out.println("Send complete.");
 });

 // Continue using the sender and finally, dispose of the sender.
 // Clients should be long-lived objects as they require resources
 // and time to establish a connection to the service.
 sender.close();
See Also: