ServiceBusSenderAsyncClient Class
- java.
lang. Object - com.
azure. messaging. servicebus. ServiceBusSenderAsyncClient
- com.
Implements
public final class ServiceBusSenderAsyncClient
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 setMessageId(String messageId) 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();
Method Summary
Methods inherited from java.lang.Object
Method Details
cancelScheduledMessage
public Mono
Cancels the enqueuing of a scheduled message, if it was not already enqueued.
Parameters:
Returns:
cancelScheduledMessages
public Mono
Cancels the enqueuing of an already scheduled message, if it was not already enqueued.
Parameters:
Returns:
close
public void close()
Disposes of the ServiceBusSenderAsyncClient. If the client has a dedicated connection, the underlying connection is also closed.
commitTransaction
public Mono
Commits the transaction given ServiceBusTransactionContext. This will make a call to Service Bus.
Parameters:
Returns:
createMessageBatch
public Mono
Creates a ServiceBusMessageBatch that can fit as many messages as the transport allows.
Returns:
createMessageBatch
public Mono
Creates an ServiceBusMessageBatch configured with the options specified.
Parameters:
Returns:
createTransaction
public Mono
Starts a new transaction on Service Bus. The ServiceBusTransactionContext should be passed along with ServiceBusReceivedMessage all operations that needs to be in this transaction.
Returns:
getEntityPath
public String getEntityPath()
Gets the name of the Service Bus resource.
Returns:
getFullyQualifiedNamespace
public String getFullyQualifiedNamespace()
Gets the fully qualified namespace.
Returns:
getIdentifier
public String getIdentifier()
Gets the identifier of the instance of ServiceBusSenderAsyncClient.
Returns:
rollbackTransaction
public Mono
Rollbacks the transaction given ServiceBusTransactionContext. This will make a call to Service Bus.
Parameters:
Returns:
scheduleMessage
public Mono
Sends a scheduled message to the Azure Service Bus entity this sender is connected to. A scheduled message is enqueued and made available to receivers only at the scheduled enqueue time.
Parameters:
Returns:
scheduleMessage
public Mono
Sends a scheduled message to the Azure Service Bus entity this sender is connected to. A scheduled message is enqueued and made available to receivers only at the scheduled enqueue time.
Parameters:
Returns:
scheduleMessages
public Flux
Sends a batch of scheduled messages to the Azure Service Bus entity this sender is connected to. A scheduled message is enqueued and made available to receivers only at the scheduled enqueue time.
Parameters:
Returns:
scheduleMessages
public Flux
Sends a scheduled messages to the Azure Service Bus entity this sender is connected to. A scheduled message is enqueued and made available to receivers only at the scheduled enqueue time.
Parameters:
Returns:
sendMessage
public Mono
Sends a message to a Service Bus queue or topic.
Parameters:
Returns:
sendMessage
public Mono
Sends a message to a Service Bus queue or topic.
Parameters:
Returns:
sendMessages
public Mono
Sends a message batch to the Azure Service Bus entity this sender is connected to.
Parameters:
Returns:
sendMessages
public Mono
Sends a message batch to the Azure Service Bus entity this sender is connected to.
Parameters:
Returns:
sendMessages
public Mono
Sends a set of messages to a Service Bus queue or topic using a batched approach. If the size of messages exceed the maximum size of a single batch, an exception will be triggered and the send will fail. By default, the message size is the max amount allowed on the link.
Parameters:
Returns:
sendMessages
public Mono
Sends a set of messages to a Service Bus queue or topic using a batched approach. If the size of messages exceed the maximum size of a single batch, an exception will be triggered and the send will fail. By default, the message size is the max amount allowed on the link.
Parameters:
Returns:
Applies to
Azure SDK for Java
Feedback
Coming soon: Throughout 2024 we will be phasing out GitHub Issues as the feedback mechanism for content and replacing it with a new feedback system. For more information see: https://aka.ms/ContentUserFeedback.
Submit and view feedback for