io.micronaut.configuration.rabbitmq.reactive

Class RxJavaReactivePublisher

java.lang.Object

io.micronaut.configuration.rabbitmq.reactive.RxJavaReactivePublisher

All Implemented Interfaces:

ReactivePublisher<io.reactivex.Completable>

@Singleton
 @Internal
public class RxJavaReactivePublisher
extends Object
implements ReactivePublisher<io.reactivex.Completable>

A reactive publisher implementation that uses a single channel per publish operation and returns an RxJava2 Completable. This is an internal API and may change at any time

Since:

1.1.0

Author:

James Kleeh

Constructor Summary

Constructors

Constructor and Description
RxJavaReactivePublisher(ChannelPool channelPool) Default constructor.

Method Summary

Modifier and Type	Method and Description
protected io.reactivex.disposables.Disposable	createListener(com.rabbitmq.client.Channel channel, AtomicReference<BrokerResponse> acknowledgement) Listens for ack/nack from the broker.
protected io.reactivex.Single<com.rabbitmq.client.Channel>	getChannel() Creates a Single from a channel, emitting an error if the channel could not be retrieved.
protected io.reactivex.Single<com.rabbitmq.client.Channel>	initializePublish(com.rabbitmq.client.Channel channel) Initializes the channel to allow publisher acknowledgements.
io.reactivex.Completable	publish(String exchange, String routingKey, com.rabbitmq.client.AMQP.BasicProperties properties, byte[] body) Publish the message with the provided arguments and return a reactive type that completes successfully when the broker acknowledged the message.
protected io.reactivex.Completable	publishInternal(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

RxJavaReactivePublisher

public RxJavaReactivePublisher(ChannelPool channelPool)

Default constructor.

Parameters:

channelPool - The channel pool to retrieve channels

Method Detail

publish

public io.reactivex.Completable publish(String exchange,
                                        String routingKey,
                                        com.rabbitmq.client.AMQP.BasicProperties properties,
                                        byte[] body)

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:

publish in interface ReactivePublisher<io.reactivex.Completable>

Parameters:

exchange - The exchange

routingKey - The routing key

properties - The properties

body - The body

Returns:

The reactive type to subscribe to

getChannel

protected io.reactivex.Single<com.rabbitmq.client.Channel> getChannel()

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

Returns:

A Single that emits the channel on success

publishInternal

protected io.reactivex.Completable publishInternal(com.rabbitmq.client.Channel channel,
                                                   RabbitPublishState publishState)

Publishes the message to the channel. The Completable 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 io.reactivex.Single<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 Single 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 io.reactivex.disposables.Disposable createListener(com.rabbitmq.client.Channel channel,
                                                             AtomicReference<BrokerResponse> acknowledgement)

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

acknowledgement - The acknowledgement object to update on response

Returns:

A disposable to allow cleanup of the listener