io.micronaut.rabbitmq.reactive

Class ReactorReactivePublisher

java.lang.Object

io.micronaut.rabbitmq.reactive.ReactorReactivePublisher

All Implemented Interfaces:

ReactivePublisher

@Internal
 @EachBean(value=ChannelPool.class)
public class ReactorReactivePublisher
extends java.lang.Object
implements ReactivePublisher

A reactive publisher implementation that uses a single channel per publish operation and returns a Reactor Mono.

This is an internal API and may change at any time

Since:

3.0.0

Author:

James Kleeh, Iván López

Constructor Summary

Constructors

Constructor and Description
ReactorReactivePublisher(ChannelPool channelPool, RabbitConnectionFactoryConfig config) Default constructor.

Method Summary

Modifier and Type	Method and Description
protected reactor.core.Disposable	createConsumer(com.rabbitmq.client.Channel channel, RabbitPublishState publishState, java.lang.String correlationId, reactor.core.publisher.MonoSink<RabbitConsumerState> emitter) Listens for ack/nack from the broker.
protected reactor.core.Disposable	createListener(com.rabbitmq.client.Channel channel, reactor.core.publisher.MonoSink<java.lang.Object> emitter, RabbitPublishState publishState) Listens for ack/nack from the broker.
protected reactor.core.publisher.Mono<com.rabbitmq.client.Channel>	getChannel() Creates a Mono from a channel, emitting an error if the channel could not be retrieved.
protected reactor.core.publisher.Mono<com.rabbitmq.client.Channel>	initializePublish(com.rabbitmq.client.Channel channel) Initializes the channel to allow publisher acknowledgements.
reactor.core.publisher.Mono<java.lang.Void>	publish(RabbitPublishState publishState) Publish the message with the provided arguments and return a reactive type that completes successfully when the message is published.
reactor.core.publisher.Mono<java.lang.Void>	publishAndConfirm(RabbitPublishState publishState) Publish the message with the provided arguments and return a reactive type that completes successfully when the broker acknowledged the message.
reactor.core.publisher.Flux<RabbitConsumerState>	publishAndReply(RabbitPublishState publishState) Publish the message with the provided arguments and return a reactive type that completes successfully when the reply is received from the reply to queue.
protected reactor.core.publisher.Mono<java.lang.Object>	publishInternal(com.rabbitmq.client.Channel channel, RabbitPublishState publishState) Publishes the message to the channel.
protected reactor.core.publisher.Mono<java.lang.Object>	publishInternalNoConfirm(com.rabbitmq.client.Channel channel, RabbitPublishState publishState) Publishes the message to the channel.
protected reactor.core.publisher.Mono<RabbitConsumerState>	publishRpcInternal(com.rabbitmq.client.Channel channel, RabbitPublishState publishState) Publishes the message to the channel.
protected void	returnChannel(com.rabbitmq.client.Channel channel) Removes confirm listeners from the channel and returns the channel to the pool.

Methods inherited from class java.lang.Object

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

Constructor Detail

ReactorReactivePublisher

public ReactorReactivePublisher(@Parameter
                                ChannelPool channelPool,
                                @Parameter
                                RabbitConnectionFactoryConfig config)

Default constructor.

Parameters:

channelPool - The channel pool to retrieve channels

config - Any configuration used in building the publishers

Method Detail

publishAndConfirm

public reactor.core.publisher.Mono<java.lang.Void> publishAndConfirm(RabbitPublishState publishState)

Description copied from interface: ReactivePublisher

Publish the message with the provided arguments and return a reactive type that completes successfully when the broker acknowledged the message.

Specified by:

publishAndConfirm in interface ReactivePublisher

Parameters:

publishState - The RabbitMQ publishing data

Returns:

The publisher

publish

public reactor.core.publisher.Mono<java.lang.Void> publish(RabbitPublishState publishState)

Description copied from interface: ReactivePublisher

Publish the message with the provided arguments and return a reactive type that completes successfully when the message is published.

Specified by:

publish in interface ReactivePublisher

Parameters:

publishState - The RabbitMQ publishing data

Returns:

The publisher

publishAndReply

public reactor.core.publisher.Flux<RabbitConsumerState> publishAndReply(RabbitPublishState publishState)

Description copied from interface: ReactivePublisher

Publish the message with the provided arguments and return a reactive type that completes successfully when the reply is received from the reply to queue.

Specified by:

publishAndReply in interface ReactivePublisher

Parameters:

publishState - The RabbitMQ publishing data

Returns:

The publisher of the received reply

getChannel

protected reactor.core.publisher.Mono<com.rabbitmq.client.Channel> getChannel()

Creates a Mono from a channel, emitting an error if the channel could not be retrieved.

Returns:

A Mono that emits the channel on success

publishInternal

protected reactor.core.publisher.Mono<java.lang.Object> publishInternal(com.rabbitmq.client.Channel channel,
                                                                        RabbitPublishState publishState)

Publishes the message to the channel. The Mono returned from this method should ensure the ConfirmListener is added to the channel prior to publishing and the ConfirmListener is removed from the channel after the publish has been acknowledged.

Parameters:

channel - The channel to publish the message to

publishState - The publishing state

Returns:

A completable that terminates when the publish has been acknowledged

See Also:

Channel.basicPublish(String, String, AMQP.BasicProperties, byte[])

publishRpcInternal

protected reactor.core.publisher.Mono<RabbitConsumerState> publishRpcInternal(com.rabbitmq.client.Channel channel,
                                                                              RabbitPublishState publishState)

Publishes the message to the channel. The Mono returned from this method should ensure the ConfirmListener is added to the channel prior to publishing and the ConfirmListener is removed from the channel after the publish has been acknowledged.

Parameters:

channel - The channel to publish the message to

publishState - The publishing state

Returns:

A completable that terminates when the publish has been acknowledged

See Also:

Channel.basicPublish(String, String, AMQP.BasicProperties, byte[])

publishInternalNoConfirm

protected reactor.core.publisher.Mono<java.lang.Object> publishInternalNoConfirm(com.rabbitmq.client.Channel channel,
                                                                                 RabbitPublishState publishState)

Publishes the message to the channel. The Mono returned from this method should ensure the ConfirmListener is added to the channel prior to publishing and the ConfirmListener is removed from the channel after the publish has been acknowledged.

Parameters:

channel - The channel to publish the message to

publishState - The publishing state

Returns:

A completable that terminates when the publish has been acknowledged

See Also:

Channel.basicPublish(String, String, AMQP.BasicProperties, byte[])

initializePublish

protected reactor.core.publisher.Mono<com.rabbitmq.client.Channel> initializePublish(com.rabbitmq.client.Channel channel)

Initializes the channel to allow publisher acknowledgements.

Parameters:

channel - The channel to enable acknowledgements.

Returns:

A Mono that will complete according to the success of the operation.

returnChannel

protected void returnChannel(com.rabbitmq.client.Channel channel)

Removes confirm listeners from the channel and returns the channel to the pool.

Parameters:

channel - The channel to clean and return

createListener

protected reactor.core.Disposable createListener(com.rabbitmq.client.Channel channel,
                                                 reactor.core.publisher.MonoSink<java.lang.Object> emitter,
                                                 RabbitPublishState publishState)

Listens for ack/nack from the broker. The listener auto disposes itself after receiving a response from the broker. If no response is received, the caller is responsible for disposing the listener.

Parameters:

channel - The channel to listen for confirms

emitter - The emitter to send the event

publishState - The publishing state

Returns:

A disposable to allow cleanup of the listener

createConsumer

protected reactor.core.Disposable createConsumer(com.rabbitmq.client.Channel channel,
                                                 RabbitPublishState publishState,
                                                 java.lang.String correlationId,
                                                 reactor.core.publisher.MonoSink<RabbitConsumerState> emitter)
                                          throws java.io.IOException

Listens for ack/nack from the broker. The listener auto disposes itself after receiving a response from the broker. If no response is received, the caller is responsible for disposing the listener.

Parameters:

channel - The channel to listen for confirms

publishState - The publish state

correlationId - The correlation id

emitter - The emitter to send the response

Returns:

A disposable to allow cleanup of the listener

Throws:

java.io.IOException - If an error occurred subscribing