public final class ServiceBusSenderAsyncClient extends Object implements AutoCloseable
Create an instance of sender
// The required parameters is connectionString, a way to authenticate with Service Bus using credentials. // The connectionString/queueName must be set by the application. The 'connectionString' format is shown below. // "Endpoint={fully-qualified-namespace};SharedAccessKeyName={policy-name};SharedAccessKey={key}" ServiceBusSenderAsyncClient sender = new ServiceBusClientBuilder() .connectionString(connectionString) .sender() .queueName(queueName) .buildAsyncClient();
Create an instance of sender using default credential
// The required parameter is a way to authenticate with Service Bus using credentials. // The connectionString provides a way to authenticate with Service Bus. ServiceBusSenderAsyncClient sender = new ServiceBusClientBuilder() .credential("<<fully-qualified-namespace>>", new DefaultAzureCredentialBuilder().build()) .sender() .queueName("<< QUEUE NAME >>") .buildAsyncClient();
Send messages to a Service Bus resource
// The required parameters is connectionString, a way to authenticate with Service Bus using credentials. // The connectionString/queueName must be set by the application. The 'connectionString' format is shown below. // "Endpoint={fully-qualified-namespace};SharedAccessKeyName={policy-name};SharedAccessKey={key}" ServiceBusSenderAsyncClient sender = new ServiceBusClientBuilder() .connectionString(connectionString) .sender() .queueName(queueName) .buildAsyncClient(); // Creating a batch without options set, will allow for automatic routing of events to any partition. sender.createMessageBatch().flatMap(batch -> { batch.tryAddMessage(new ServiceBusMessage(BinaryData.fromBytes("test-1".getBytes(UTF_8)))); batch.tryAddMessage(new ServiceBusMessage(BinaryData.fromBytes("test-2".getBytes(UTF_8)))); return sender.sendMessages(batch); }).subscribe(unused -> { }, error -> System.err.println("Error occurred while sending batch:" + error), () -> System.out.println("Send complete."));
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<>( sender.createMessageBatch(options).block()); // The sample Flux contains two messages, but it could be an infinite stream of telemetry messages. telemetryMessages.flatMap(message -> { ServiceBusMessageBatch batch = currentBatch.get(); if (batch.tryAddMessage(message)) { return Mono.empty(); } return Mono.when( sender.sendMessages(batch), sender.createMessageBatch(options).map(newBatch -> { currentBatch.set(newBatch); // Add the message that did not fit in the previous batch. if (!newBatch.tryAddMessage(message)) { throw Exceptions.propagate(new IllegalArgumentException( "Message was too large to fit in an empty batch. Max size: " + newBatch.getMaxSizeInBytes())); } return newBatch; })); }).then() .doFinally(signal -> { ServiceBusMessageBatch batch = currentBatch.getAndSet(null); if (batch != null && batch.getCount() > 0) { sender.sendMessages(batch).block(); } });
Modifier and Type | Method and Description |
---|---|
Mono<Void> |
cancelScheduledMessage(long sequenceNumber)
Cancels the enqueuing of a scheduled message, if it was not already enqueued.
|
Mono<Void> |
cancelScheduledMessages(Iterable<Long> sequenceNumbers)
Cancels the enqueuing of an already scheduled message, if it was not already enqueued.
|
void |
close()
Disposes of the
ServiceBusSenderAsyncClient . |
Mono<Void> |
commitTransaction(ServiceBusTransactionContext transactionContext)
Commits the transaction given
ServiceBusTransactionContext . |
Mono<ServiceBusMessageBatch> |
createMessageBatch()
Creates a
ServiceBusMessageBatch that can fit as many messages as the transport allows. |
Mono<ServiceBusMessageBatch> |
createMessageBatch(CreateMessageBatchOptions options)
Creates an
ServiceBusMessageBatch configured with the options specified. |
Mono<ServiceBusTransactionContext> |
createTransaction()
Starts a new transaction on Service Bus.
|
String |
getEntityPath()
Gets the name of the Service Bus resource.
|
String |
getFullyQualifiedNamespace()
Gets the fully qualified namespace.
|
Mono<Void> |
rollbackTransaction(ServiceBusTransactionContext transactionContext)
Rollbacks the transaction given
ServiceBusTransactionContext . |
Mono<Long> |
scheduleMessage(ServiceBusMessage message,
OffsetDateTime scheduledEnqueueTime)
Sends a scheduled message to the Azure Service Bus entity this sender is connected to.
|
Mono<Long> |
scheduleMessage(ServiceBusMessage message,
OffsetDateTime scheduledEnqueueTime,
ServiceBusTransactionContext transactionContext)
Sends a scheduled message to the Azure Service Bus entity this sender is connected to.
|
Flux<Long> |
scheduleMessages(Iterable<ServiceBusMessage> messages,
OffsetDateTime scheduledEnqueueTime)
Sends a batch of scheduled messages to the Azure Service Bus entity this sender is connected to.
|
Flux<Long> |
scheduleMessages(Iterable<ServiceBusMessage> messages,
OffsetDateTime scheduledEnqueueTime,
ServiceBusTransactionContext transactionContext)
Sends a scheduled messages to the Azure Service Bus entity this sender is connected to.
|
Mono<Void> |
sendMessage(ServiceBusMessage message)
Sends a message to a Service Bus queue or topic.
|
Mono<Void> |
sendMessage(ServiceBusMessage message,
ServiceBusTransactionContext transactionContext)
Sends a message to a Service Bus queue or topic.
|
Mono<Void> |
sendMessages(Iterable<ServiceBusMessage> messages)
Sends a set of messages to a Service Bus queue or topic using a batched approach.
|
Mono<Void> |
sendMessages(Iterable<ServiceBusMessage> messages,
ServiceBusTransactionContext transactionContext)
Sends a set of messages to a Service Bus queue or topic using a batched approach.
|
Mono<Void> |
sendMessages(ServiceBusMessageBatch batch)
Sends a message batch to the Azure Service Bus entity this sender is connected to.
|
Mono<Void> |
sendMessages(ServiceBusMessageBatch batch,
ServiceBusTransactionContext transactionContext)
Sends a message batch to the Azure Service Bus entity this sender is connected to.
|
public String getFullyQualifiedNamespace()
public String getEntityPath()
public Mono<Void> sendMessage(ServiceBusMessage message)
message
- Message to be sent to Service Bus queue or topic.Mono
the finishes this operation on service bus resource.NullPointerException
- if message
is null
.IllegalStateException
- if sender is already disposed.ServiceBusException
- if message
is larger than the maximum allowed size of a single message or
the message could not be sent.public Mono<Void> sendMessage(ServiceBusMessage message, ServiceBusTransactionContext transactionContext)
message
- Message to be sent to Service Bus queue or topic.transactionContext
- to be set on batch message before sending to Service Bus.Mono
the finishes this operation on service bus resource.NullPointerException
- if message
, transactionContext
or
transactionContext.transactionId
is null
.IllegalStateException
- if sender is already disposed.ServiceBusException
- if message
is larger than the maximum allowed size of a single message or
the message could not be sent.public Mono<Void> sendMessages(Iterable<ServiceBusMessage> messages, ServiceBusTransactionContext transactionContext)
messages
- Messages to be sent to Service Bus queue or topic.transactionContext
- to be set on batch message before sending to Service Bus.Mono
that completes when all messages have been sent to the Service Bus resource.NullPointerException
- if batch
, transactionContext
or
transactionContext.transactionId
is null
.IllegalStateException
- if sender is already disposed.ServiceBusException
- if messages
are larger than the maximum allowed size of a single message or
the message could not be sent.public Mono<Void> sendMessages(Iterable<ServiceBusMessage> messages)
messages
- Messages to be sent to Service Bus queue or topic.Mono
that completes when all messages have been sent to the Service Bus resource.NullPointerException
- if messages
is null
.ServiceBusException
- if messages
are larger than the maximum allowed size of a single message or
the message could not be sent.IllegalStateException
- if sender is already disposed.public Mono<Void> sendMessages(ServiceBusMessageBatch batch)
batch
- of messages which allows client to send maximum allowed size for a batch of messages.Mono
the finishes this operation on service bus resource.NullPointerException
- if batch
is null
.ServiceBusException
- if the message batch could not be sent.IllegalStateException
- if sender is already disposed.public Mono<Void> sendMessages(ServiceBusMessageBatch batch, ServiceBusTransactionContext transactionContext)
batch
- of messages which allows client to send maximum allowed size for a batch of messages.transactionContext
- to be set on batch message before sending to Service Bus.Mono
the finishes this operation on service bus resource.NullPointerException
- if batch
, transactionContext
or
transactionContext.transactionId
is null
.ServiceBusException
- if the message batch could not be sent.IllegalStateException
- if sender is already disposed.public Mono<ServiceBusMessageBatch> createMessageBatch()
ServiceBusMessageBatch
that can fit as many messages as the transport allows.ServiceBusMessageBatch
that can fit as many messages as the transport allows.ServiceBusException
- if the message batch could not be created.IllegalStateException
- if sender is already disposed.public Mono<ServiceBusMessageBatch> createMessageBatch(CreateMessageBatchOptions options)
ServiceBusMessageBatch
configured with the options specified.options
- A set of options used to configure the ServiceBusMessageBatch
.ServiceBusMessageBatch
configured with the given options.NullPointerException
- if options
is null.ServiceBusException
- if the message batch could not be created.IllegalStateException
- if sender is already disposed.IllegalArgumentException
- if CreateMessageBatchOptions.getMaximumSizeInBytes()
is larger than
maximum allowed size.public Mono<Long> scheduleMessage(ServiceBusMessage message, OffsetDateTime scheduledEnqueueTime, ServiceBusTransactionContext transactionContext)
message
- Message to be sent to the Service Bus Queue.scheduledEnqueueTime
- OffsetDateTime at which the message should appear in the Service Bus queue or topic.transactionContext
- to be set on message before sending to Service Bus.NullPointerException
- if message
, scheduledEnqueueTime
, transactionContext
or
transactionContext.transactionID
is null
.ServiceBusException
- If the message could not be scheduled.IllegalStateException
- if sender is already disposed.public Mono<Long> scheduleMessage(ServiceBusMessage message, OffsetDateTime scheduledEnqueueTime)
message
- Message to be sent to the Service Bus Queue.scheduledEnqueueTime
- OffsetDateTime at which the message should appear in the Service Bus queue or topic.NullPointerException
- if message
or scheduledEnqueueTime
is null
.ServiceBusException
- If the message could not be scheduled.public Flux<Long> scheduleMessages(Iterable<ServiceBusMessage> messages, OffsetDateTime scheduledEnqueueTime)
messages
- Messages to be sent to the Service Bus queue or topic.scheduledEnqueueTime
- OffsetDateTime at which the message should appear in the Service Bus queue or topic.NullPointerException
- If messages
or scheduledEnqueueTime
is null
.ServiceBusException
- If the messages could not be scheduled.IllegalStateException
- if sender is already disposed.public Flux<Long> scheduleMessages(Iterable<ServiceBusMessage> messages, OffsetDateTime scheduledEnqueueTime, ServiceBusTransactionContext transactionContext)
messages
- Messages to be sent to the Service Bus Queue.scheduledEnqueueTime
- OffsetDateTime at which the messages should appear in the Service Bus queue or topic.transactionContext
- Transaction to associate with the operation.NullPointerException
- If messages
, scheduledEnqueueTime
, transactionContext
or
transactionContext.transactionId
is null
.ServiceBusException
- If the messages could not be scheduled.IllegalStateException
- if sender is already disposed.public Mono<Void> cancelScheduledMessage(long sequenceNumber)
sequenceNumber
- of the scheduled message to cancel.Mono
that finishes this operation on service bus resource.IllegalArgumentException
- if sequenceNumber
is negative.ServiceBusException
- If the messages could not be cancelled.IllegalStateException
- if sender is already disposed.public Mono<Void> cancelScheduledMessages(Iterable<Long> sequenceNumbers)
sequenceNumbers
- of the scheduled messages to cancel.Mono
that finishes this operation on service bus resource.NullPointerException
- if sequenceNumbers
is null.IllegalStateException
- if sender is already disposed.ServiceBusException
- if the scheduled messages cannot cancelled.public Mono<ServiceBusTransactionContext> createTransaction()
ServiceBusTransactionContext
should be passed along with
ServiceBusReceivedMessage
all operations that needs to be in this transaction.ServiceBusTransactionContext
.IllegalStateException
- if sender is already disposed.ServiceBusException
- if a transaction cannot be created.ServiceBusReceiverAsyncClient.createTransaction()
public Mono<Void> commitTransaction(ServiceBusTransactionContext transactionContext)
ServiceBusTransactionContext
. This will make a call to Service Bus.transactionContext
- to be committed.Mono
that finishes this operation on Service Bus resource.IllegalStateException
- if sender is already disposed.NullPointerException
- if transactionContext
or transactionContext.transactionId
is null.ServiceBusException
- if the transaction could not be committed.ServiceBusReceiverAsyncClient.commitTransaction(ServiceBusTransactionContext)
public Mono<Void> rollbackTransaction(ServiceBusTransactionContext transactionContext)
ServiceBusTransactionContext
. This will make a call to Service Bus.transactionContext
- Transaction to rollback.Mono
that finishes this operation on the Service Bus resource.IllegalStateException
- if sender is already disposed.NullPointerException
- if transactionContext
or transactionContext.transactionId
is null.ServiceBusException
- if the transaction could not be rolled back.ServiceBusReceiverAsyncClient.rollbackTransaction(ServiceBusTransactionContext)
public void close()
ServiceBusSenderAsyncClient
. If the client has a dedicated connection, the underlying
connection is also closed.close
in interface AutoCloseable
Copyright © 2021 Microsoft Corporation. All rights reserved.