reactor.rabbitmq

Class Sender

java.lang.Object

reactor.rabbitmq.Sender

All Implemented Interfaces:

AutoCloseable

public class Sender
extends Object
implements AutoCloseable

Reactive abstraction to create resources and send messages.

Nested Class Summary

Nested Classes

Modifier and Type	Class and Description
static class	Sender.ConfirmSendContext<OMSG extends OutboundMessage>
static class	Sender.SendContext<OMSG extends OutboundMessage>

Constructor Summary

Constructors

Constructor and Description
Sender()
Sender(SenderOptions options)

Method Summary

Modifier and Type	Method and Description
Mono<AMQP.Queue.BindOk>	bind(BindingSpecification specification) Bind a queue to an exchange.
Mono<AMQP.Queue.BindOk>	bind(BindingSpecification specification, ResourceManagementOptions options) Bind a queue to an exchange.
Mono<AMQP.Exchange.BindOk>	bindExchange(BindingSpecification specification) Bind an exchange to another exchange.
Mono<AMQP.Exchange.BindOk>	bindExchange(BindingSpecification specification, ResourceManagementOptions options) Bind an exchange to another exchange.
Mono<AMQP.Queue.BindOk>	bindQueue(BindingSpecification specification) Bind a queue to an exchange.
Mono<AMQP.Queue.BindOk>	bindQueue(BindingSpecification specification, ResourceManagementOptions options) Bind a queue to an exchange.
protected <T> Mono<T>	cache(Mono<T> mono)
void	close()
protected Scheduler	createScheduler(String name)
Mono<AMQP.Exchange.DeclareOk>	declare(ExchangeSpecification specification) Declare an exchange.
Mono<AMQP.Exchange.DeclareOk>	declare(ExchangeSpecification specification, ResourceManagementOptions options) Declare an exchange.
Mono<AMQP.Queue.DeclareOk>	declare(QueueSpecification specification) Declare a queue following the specification.
Mono<AMQP.Queue.DeclareOk>	declare(QueueSpecification specification, ResourceManagementOptions options) Declare a queue following the specification and the resource management options.
Mono<AMQP.Exchange.DeclareOk>	declareExchange(ExchangeSpecification specification) Declare an exchange.
Mono<AMQP.Exchange.DeclareOk>	declareExchange(ExchangeSpecification specification, ResourceManagementOptions options) Declare an exchange.
Mono<AMQP.Queue.DeclareOk>	declareQueue(QueueSpecification specification) Declare a queue following the specification.
Mono<AMQP.Queue.DeclareOk>	declareQueue(QueueSpecification specification, ResourceManagementOptions options) Declare a queue following the specification and the resource management options.
Mono<AMQP.Exchange.DeleteOk>	delete(ExchangeSpecification specification) Delete an exchange.
Mono<AMQP.Exchange.DeleteOk>	delete(ExchangeSpecification specification, boolean ifUnused) Delete an exchange.
Mono<AMQP.Exchange.DeleteOk>	delete(ExchangeSpecification specification, boolean ifUnused, ResourceManagementOptions options) Delete an exchange.
Mono<AMQP.Exchange.DeleteOk>	delete(ExchangeSpecification specification, ResourceManagementOptions options) Delete an exchange.
Mono<AMQP.Queue.DeleteOk>	delete(QueueSpecification specification) Delete a queue.
Mono<AMQP.Queue.DeleteOk>	delete(QueueSpecification specification, boolean ifUnused, boolean ifEmpty) Delete a queue.
Mono<AMQP.Queue.DeleteOk>	delete(QueueSpecification specification, boolean ifUnused, boolean ifEmpty, ResourceManagementOptions options) Delete a queue.
Mono<AMQP.Queue.DeleteOk>	delete(QueueSpecification specification, ResourceManagementOptions options) Delete a queue.
Mono<AMQP.Exchange.DeleteOk>	deleteExchange(ExchangeSpecification specification, boolean ifUnused) Delete an exchange.
Mono<AMQP.Exchange.DeleteOk>	deleteExchange(ExchangeSpecification specification, boolean ifUnused, ResourceManagementOptions options) Delete an exchange.
Mono<AMQP.Queue.DeleteOk>	deleteQueue(QueueSpecification specification, boolean ifUnused, boolean ifEmpty) Delete a queue.
Mono<AMQP.Queue.DeleteOk>	deleteQueue(QueueSpecification specification, boolean ifUnused, boolean ifEmpty, ResourceManagementOptions options) Delete a queue.
RpcClient	rpcClient(String exchange, String routingKey)
RpcClient	rpcClient(String exchange, String routingKey, Supplier<String> correlationIdProvider)
Mono<Void>	send(Publisher<OutboundMessage> messages)
Mono<Void>	send(Publisher<OutboundMessage> messages, SendOptions options)
Flux<OutboundMessageResult>	sendWithPublishConfirms(Publisher<OutboundMessage> messages) Publish a flux of messages and expect confirmations.
Flux<OutboundMessageResult>	sendWithPublishConfirms(Publisher<OutboundMessage> messages, SendOptions options) Publish a flux of messages and expect confirmations.
<OMSG extends OutboundMessage> Flux<OutboundMessageResult<OMSG>>	sendWithTypedPublishConfirms(Publisher<OMSG> messages) Publish a flux of subclass of OutboundMessage and expect confirmations.
<OMSG extends OutboundMessage> Flux<OutboundMessageResult<OMSG>>	sendWithTypedPublishConfirms(Publisher<OMSG> messages, SendOptions options) Publish a flux of subclass of OutboundMessage and expect confirmations.
Mono<AMQP.Queue.UnbindOk>	unbind(BindingSpecification specification) Unbind a queue from an exchange.
Mono<AMQP.Queue.UnbindOk>	unbind(BindingSpecification specification, ResourceManagementOptions options) Unbind a queue from an exchange.
Mono<AMQP.Exchange.UnbindOk>	unbindExchange(BindingSpecification specification) Unbind an exchange from another exchange.
Mono<AMQP.Exchange.UnbindOk>	unbindExchange(BindingSpecification specification, ResourceManagementOptions options) Unbind an exchange from another exchange.
Mono<AMQP.Queue.UnbindOk>	unbindQueue(BindingSpecification specification) Unbind a queue from an exchange.
Mono<AMQP.Queue.UnbindOk>	unbindQueue(BindingSpecification specification, ResourceManagementOptions options) Unbind a queue from an exchange.

Methods inherited from class java.lang.Object

clone, equals, finalize, getClass, hashCode, notify, notifyAll, toString, wait, wait, wait

Constructor Detail

Sender

public Sender()

Sender

public Sender(SenderOptions options)

Method Detail

createScheduler

protected Scheduler createScheduler(String name)

cache

protected <T> Mono<T> cache(Mono<T> mono)

send

public Mono<Void> send(Publisher<OutboundMessage> messages)

send

public Mono<Void> send(Publisher<OutboundMessage> messages,
                       @Nullable
                       SendOptions options)

sendWithPublishConfirms

public Flux<OutboundMessageResult> sendWithPublishConfirms(Publisher<OutboundMessage> messages)

Publish a flux of messages and expect confirmations.

This method uses RabbitMQ Publisher Confirms extension to make sure outbound messages made it or not to the broker.

See sendWithPublishConfirms(Publisher, SendOptions) to have more control over the publishing and the confirmations with SendOptions.

Use sendWithTypedPublishConfirms(Publisher) to publish instances of a custom OutboundMessage subclass and get them back in the flux of OutboundMessageResult.

Parameters:

messages - flux of outbound messages

Returns:

flux of confirmations

See Also:

Publisher Confirms

sendWithPublishConfirms

public Flux<OutboundMessageResult> sendWithPublishConfirms(Publisher<OutboundMessage> messages,
                                                           SendOptions options)

Publish a flux of messages and expect confirmations.

This method uses RabbitMQ Publisher Confirms extension to make sure outbound messages made it or not to the broker.

It is also possible to know if a message has been routed to a least one queue by enabling the mandatory flag. The default is to not use this flag.

Use sendWithTypedPublishConfirms(Publisher, SendOptions) to publish instances of a custom OutboundMessage subclass and get them back in the flux of OutboundMessageResult.

Parameters:

messages - flux of outbound messages

options - options to configure publishing

Returns:

flux of confirmations

See Also:

Publisher Confirms, Mandatory flag, SendOptions.trackReturned(boolean)

sendWithTypedPublishConfirms

public <OMSG extends OutboundMessage> Flux<OutboundMessageResult<OMSG>> sendWithTypedPublishConfirms(Publisher<OMSG> messages)

Publish a flux of subclass of OutboundMessage and expect confirmations.

This method can be interesting compared to sendWithPublishConfirms(Publisher) when OutboundMessage instances of a custom class need to be published and then retrieved in the OutboundMessageResult flux. Such a custom class can contain correlation data useful when dealing with the confirmation. Note the CorrelableOutboundMessage class is provided if just an extra property is needed compared to OutboundMessage.

This method uses RabbitMQ Publisher Confirms extension to make sure outbound messages made it or not to the broker.

See sendWithTypedPublishConfirms(Publisher, SendOptions) to have more control over the publishing and the confirmations with SendOptions.

Parameters:

messages - flux of outbound messages with typed correlated metadata

Returns:

flux of confirmations with typed correlated metadata

Since:

1.4.0

See Also:

Publisher Confirms, CorrelableOutboundMessage

sendWithTypedPublishConfirms

public <OMSG extends OutboundMessage> Flux<OutboundMessageResult<OMSG>> sendWithTypedPublishConfirms(Publisher<OMSG> messages,
                                                                                                     @Nullable
                                                                                                     SendOptions options)

Publish a flux of subclass of OutboundMessage and expect confirmations.

This method can be interesting compared to sendWithPublishConfirms(Publisher) when OutboundMessage instances of a custom class need to be published and then retrieved in the OutboundMessageResult flux. Such a custom class can contain correlation data useful when dealing with the confirmation. Note the CorrelableOutboundMessage class is provided if just an extra property is needed compared to OutboundMessage.

This method uses RabbitMQ Publisher Confirms extension to make sure outbound messages made it or not to the broker.

It is also possible to know if a message has been routed to a least one queue by enabling the mandatory flag. The default is to not use this flag.

Parameters:

messages - flux of outbound messages with typed correlated metadata

options - options to configure publishing

Returns:

flux of confirmations

Since:

1.4.0

See Also:

Publisher Confirms, Mandatory flag, SendOptions.trackReturned(boolean)

rpcClient

public RpcClient rpcClient(String exchange,
                           String routingKey)

rpcClient

public RpcClient rpcClient(String exchange,
                           String routingKey,
                           Supplier<String> correlationIdProvider)

declare

public Mono<AMQP.Queue.DeclareOk> declare(QueueSpecification specification)

Declare a queue following the specification. Warning: This method relies on RPC. As AMQP 0.9.1 protocol does not use a correlation ID for requests, a lock is being used to prevent concurrent RPCs, making this publisher potentially blocking.

Parameters:

specification - the specification of the queue

Returns:

a mono wrapping the result of the declaration

See Also:

QueueSpecification

declare

public Mono<AMQP.Queue.DeclareOk> declare(QueueSpecification specification,
                                          @Nullable
                                          ResourceManagementOptions options)

Declare a queue following the specification and the resource management options. Warning: This method relies on RPC. As AMQP 0.9.1 protocol does not use a correlation ID for requests, a lock is being used to prevent concurrent RPCs, making this publisher potentially blocking.

Parameters:

specification - the specification of the queue

options - options for resource management

Returns:

a mono wrapping the result of the declaration

See Also:

QueueSpecification, ResourceManagementOptions

declareQueue

public Mono<AMQP.Queue.DeclareOk> declareQueue(QueueSpecification specification)

Declare a queue following the specification. Warning: This method relies on RPC. As AMQP 0.9.1 protocol does not use a correlation ID for requests, a lock is being used to prevent concurrent RPCs, making this publisher potentially blocking.

Parameters:

specification - the specification of the queue

Returns:

a mono wrapping the result of the declaration

See Also:

QueueSpecification

declareQueue

public Mono<AMQP.Queue.DeclareOk> declareQueue(QueueSpecification specification,
                                               @Nullable
                                               ResourceManagementOptions options)

Declare a queue following the specification and the resource management options. Warning: This method relies on RPC. As AMQP 0.9.1 protocol does not use a correlation ID for requests, a lock is being used to prevent concurrent RPCs, making this publisher potentially blocking.

Parameters:

specification - the specification of the queue

options - options for resource management

Returns:

a mono wrapping the result of the declaration

See Also:

QueueSpecification, ResourceManagementOptions

delete

public Mono<AMQP.Queue.DeleteOk> delete(QueueSpecification specification)

Delete a queue. Warning: This method relies on RPC. As AMQP 0.9.1 protocol does not use a correlation ID for requests, a lock is being used to prevent concurrent RPCs, making this publisher potentially blocking.

delete

public Mono<AMQP.Queue.DeleteOk> delete(QueueSpecification specification,
                                        @Nullable
                                        ResourceManagementOptions options)

Delete a queue. Warning: This method relies on RPC. As AMQP 0.9.1 protocol does not use a correlation ID for requests, a lock is being used to prevent concurrent RPCs, making this publisher potentially blocking.

delete

public Mono<AMQP.Queue.DeleteOk> delete(QueueSpecification specification,
                                        boolean ifUnused,
                                        boolean ifEmpty)

Delete a queue. Warning: This method relies on RPC. As AMQP 0.9.1 protocol does not use a correlation ID for requests, a lock is being used to prevent concurrent RPCs, making this publisher potentially blocking.

delete

public Mono<AMQP.Queue.DeleteOk> delete(QueueSpecification specification,
                                        boolean ifUnused,
                                        boolean ifEmpty,
                                        @Nullable
                                        ResourceManagementOptions options)

Delete a queue. Warning: This method relies on RPC. As AMQP 0.9.1 protocol does not use a correlation ID for requests, a lock is being used to prevent concurrent RPCs, making this publisher potentially blocking.

deleteQueue

public Mono<AMQP.Queue.DeleteOk> deleteQueue(QueueSpecification specification,
                                             boolean ifUnused,
                                             boolean ifEmpty)

Delete a queue. Warning: This method relies on RPC. As AMQP 0.9.1 protocol does not use a correlation ID for requests, a lock is being used to prevent concurrent RPCs, making this publisher potentially blocking.

deleteQueue

public Mono<AMQP.Queue.DeleteOk> deleteQueue(QueueSpecification specification,
                                             boolean ifUnused,
                                             boolean ifEmpty,
                                             @Nullable
                                             ResourceManagementOptions options)

Delete a queue. Warning: This method relies on RPC. As AMQP 0.9.1 protocol does not use a correlation ID for requests, a lock is being used to prevent concurrent RPCs, making this publisher potentially blocking.

declare

public Mono<AMQP.Exchange.DeclareOk> declare(ExchangeSpecification specification)

Declare an exchange. Warning: This method relies on RPC. As AMQP 0.9.1 protocol does not use a correlation ID for requests, a lock is being used to prevent concurrent RPCs, making this publisher potentially blocking.

declare

public Mono<AMQP.Exchange.DeclareOk> declare(ExchangeSpecification specification,
                                             @Nullable
                                             ResourceManagementOptions options)

Declare an exchange. Warning: This method relies on RPC. As AMQP 0.9.1 protocol does not use a correlation ID for requests, a lock is being used to prevent concurrent RPCs, making this publisher potentially blocking.

declareExchange

public Mono<AMQP.Exchange.DeclareOk> declareExchange(ExchangeSpecification specification)

Declare an exchange. Warning: This method relies on RPC. As AMQP 0.9.1 protocol does not use a correlation ID for requests, a lock is being used to prevent concurrent RPCs, making this publisher potentially blocking.

declareExchange

public Mono<AMQP.Exchange.DeclareOk> declareExchange(ExchangeSpecification specification,
                                                     @Nullable
                                                     ResourceManagementOptions options)

Declare an exchange. Warning: This method relies on RPC. As AMQP 0.9.1 protocol does not use a correlation ID for requests, a lock is being used to prevent concurrent RPCs, making this publisher potentially blocking.

delete

public Mono<AMQP.Exchange.DeleteOk> delete(ExchangeSpecification specification)

Delete an exchange. Warning: This method relies on RPC. As AMQP 0.9.1 protocol does not use a correlation ID for requests, a lock is being used to prevent concurrent RPCs, making this publisher potentially blocking.

delete

public Mono<AMQP.Exchange.DeleteOk> delete(ExchangeSpecification specification,
                                           @Nullable
                                           ResourceManagementOptions options)

Delete an exchange. Warning: This method relies on RPC. As AMQP 0.9.1 protocol does not use a correlation ID for requests, a lock is being used to prevent concurrent RPCs, making this publisher potentially blocking.

delete

public Mono<AMQP.Exchange.DeleteOk> delete(ExchangeSpecification specification,
                                           boolean ifUnused)

Delete an exchange. Warning: This method relies on RPC. As AMQP 0.9.1 protocol does not use a correlation ID for requests, a lock is being used to prevent concurrent RPCs, making this publisher potentially blocking.

delete

public Mono<AMQP.Exchange.DeleteOk> delete(ExchangeSpecification specification,
                                           boolean ifUnused,
                                           @Nullable
                                           ResourceManagementOptions options)

Delete an exchange. Warning: This method relies on RPC. As AMQP 0.9.1 protocol does not use a correlation ID for requests, a lock is being used to prevent concurrent RPCs, making this publisher potentially blocking.

deleteExchange

public Mono<AMQP.Exchange.DeleteOk> deleteExchange(ExchangeSpecification specification,
                                                   boolean ifUnused)

Delete an exchange. Warning: This method relies on RPC. As AMQP 0.9.1 protocol does not use a correlation ID for requests, a lock is being used to prevent concurrent RPCs, making this publisher potentially blocking.

deleteExchange

public Mono<AMQP.Exchange.DeleteOk> deleteExchange(ExchangeSpecification specification,
                                                   boolean ifUnused,
                                                   @Nullable
                                                   ResourceManagementOptions options)

Delete an exchange. Warning: This method relies on RPC. As AMQP 0.9.1 protocol does not use a correlation ID for requests, a lock is being used to prevent concurrent RPCs, making this publisher potentially blocking.

unbindQueue

public Mono<AMQP.Queue.UnbindOk> unbindQueue(BindingSpecification specification)

Unbind a queue from an exchange.

Alias of unbind(BindingSpecification). Warning: This method relies on RPC. As AMQP 0.9.1 protocol does not use a correlation ID for requests, a lock is being used to prevent concurrent RPCs, making this publisher potentially blocking.

Parameters:

specification - the unbinding specification

Returns:

the result of the operation

Since:

1.4.1

unbindQueue

public Mono<AMQP.Queue.UnbindOk> unbindQueue(BindingSpecification specification,
                                             @Nullable
                                             ResourceManagementOptions options)

Unbind a queue from an exchange.

Alias of unbind(BindingSpecification, ResourceManagementOptions). Warning: This method relies on RPC. As AMQP 0.9.1 protocol does not use a correlation ID for requests, a lock is being used to prevent concurrent RPCs, making this publisher potentially blocking.

Parameters:

specification - the unbinding specification

options - options to control the operation, e.g. channel to use

Returns:

the result of the operation

Since:

1.4.1

unbind

public Mono<AMQP.Queue.UnbindOk> unbind(BindingSpecification specification)

Unbind a queue from an exchange.

Alias of unbindQueue(BindingSpecification). Warning: This method relies on RPC. As AMQP 0.9.1 protocol does not use a correlation ID for requests, a lock is being used to prevent concurrent RPCs, making this publisher potentially blocking.

Parameters:

specification - the unbinding specification

Returns:

the result of the operation

unbind

public Mono<AMQP.Queue.UnbindOk> unbind(BindingSpecification specification,
                                        @Nullable
                                        ResourceManagementOptions options)

Unbind a queue from an exchange.

Alias of unbindQueue(BindingSpecification, ResourceManagementOptions). Warning: This method relies on RPC. As AMQP 0.9.1 protocol does not use a correlation ID for requests, a lock is being used to prevent concurrent RPCs, making this publisher potentially blocking.

Parameters:

specification - the unbinding specification

options - options to control the operation, e.g. channel to use

Returns:

the result of the operation

unbindExchange

public Mono<AMQP.Exchange.UnbindOk> unbindExchange(BindingSpecification specification)

Unbind an exchange from another exchange. Warning: This method relies on RPC. As AMQP 0.9.1 protocol does not use a correlation ID for requests, a lock is being used to prevent concurrent RPCs, making this publisher potentially blocking.

Parameters:

specification - the unbinding specification

Returns:

the result of the operation

Since:

1.4.1

unbindExchange

public Mono<AMQP.Exchange.UnbindOk> unbindExchange(BindingSpecification specification,
                                                   @Nullable
                                                   ResourceManagementOptions options)

Unbind an exchange from another exchange. Warning: This method relies on RPC. As AMQP 0.9.1 protocol does not use a correlation ID for requests, a lock is being used to prevent concurrent RPCs, making this publisher potentially blocking.

Parameters:

specification - the unbinding specification

options - options to control the operation, e.g. channel to use

Returns:

the result of the operation

Since:

1.4.1

bindQueue

public Mono<AMQP.Queue.BindOk> bindQueue(BindingSpecification specification)

Bind a queue to an exchange.

Alias of bind(BindingSpecification) Warning: This method relies on RPC. As AMQP 0.9.1 protocol does not use a correlation ID for requests, a lock is being used to prevent concurrent RPCs, making this publisher potentially blocking.

Parameters:

specification - the binding specification

Returns:

the result of the operation

Since:

1.4.1

bindQueue

public Mono<AMQP.Queue.BindOk> bindQueue(BindingSpecification specification,
                                         @Nullable
                                         ResourceManagementOptions options)

Bind a queue to an exchange.

Alias of bind(BindingSpecification, ResourceManagementOptions) Warning: This method relies on RPC. As AMQP 0.9.1 protocol does not use a correlation ID for requests, a lock is being used to prevent concurrent RPCs, making this publisher potentially blocking.

Parameters:

specification - the binding specification

options - options to control the operation, e.g. channel to use

Returns:

the result of the operation

Since:

1.4.1

bind

public Mono<AMQP.Queue.BindOk> bind(BindingSpecification specification)

Bind a queue to an exchange.

Alias of bindQueue(BindingSpecification) Warning: This method relies on RPC. As AMQP 0.9.1 protocol does not use a correlation ID for requests, a lock is being used to prevent concurrent RPCs, making this publisher potentially blocking.

Parameters:

specification - the binding specification

Returns:

the result of the operation

Since:

1.4.1

bind

public Mono<AMQP.Queue.BindOk> bind(BindingSpecification specification,
                                    @Nullable
                                    ResourceManagementOptions options)

Bind a queue to an exchange.

Alias of bindQueue(BindingSpecification, ResourceManagementOptions) Warning: This method relies on RPC. As AMQP 0.9.1 protocol does not use a correlation ID for requests, a lock is being used to prevent concurrent RPCs, making this publisher potentially blocking.

Parameters:

specification - the binding specification

options - options to control the operation, e.g. channel to use

Returns:

the result of the operation

Since:

1.4.1

bindExchange

public Mono<AMQP.Exchange.BindOk> bindExchange(BindingSpecification specification)

Bind an exchange to another exchange. Warning: This method relies on RPC. As AMQP 0.9.1 protocol does not use a correlation ID for requests, a lock is being used to prevent concurrent RPCs, making this publisher potentially blocking.

Parameters:

specification - the binding specification

Returns:

the result of the operation

Since:

1.4.1

bindExchange

public Mono<AMQP.Exchange.BindOk> bindExchange(BindingSpecification specification,
                                               @Nullable
                                               ResourceManagementOptions options)

Bind an exchange to another exchange. Warning: This method relies on RPC. As AMQP 0.9.1 protocol does not use a correlation ID for requests, a lock is being used to prevent concurrent RPCs, making this publisher potentially blocking.

Parameters:

specification - the binding specification

options - options to control the operation, e.g. channel to use

Returns:

the result of the operation

Since:

1.4.1

close

public void close()

Specified by:

close in interface AutoCloseable