io.micronaut.configuration.rabbitmq.reactive

Class RxJavaReactivePublisher

java.lang.Object

io.micronaut.configuration.rabbitmq.reactive.RxJavaReactivePublisher

All Implemented Interfaces:

ReactivePublisher

@Internal
 @EachBean(value=ChannelPool.class)
public class RxJavaReactivePublisher
extends Object
implements ReactivePublisher

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, RabbitConnectionFactoryConfig config) Default constructor.

Method Summary

Modifier and Type	Method and Description
protected io.reactivex.disposables.Disposable	createConsumer(com.rabbitmq.client.Channel channel, RabbitPublishState publishState, String correlationId, io.reactivex.SingleEmitter<RabbitConsumerState> emitter) Listens for ack/nack from the broker.
protected io.reactivex.disposables.Disposable	createListener(com.rabbitmq.client.Channel channel, io.reactivex.CompletableEmitter emitter, RabbitPublishState publishState) 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.Flowable<Void>	publish(RabbitPublishState publishState) Publish the message with the provided arguments and return a reactive type that completes successfully when the message is published.
io.reactivex.Flowable<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.
io.reactivex.Flowable<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 io.reactivex.Completable	publishInternal(com.rabbitmq.client.Channel channel, RabbitPublishState publishState) Publishes the message to the channel.
protected io.reactivex.Completable	publishInternalNoConfirm(com.rabbitmq.client.Channel channel, RabbitPublishState publishState) Publishes the message to the channel.
protected io.reactivex.Single<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

RxJavaReactivePublisher

public RxJavaReactivePublisher(@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 io.reactivex.Flowable<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 io.reactivex.Flowable<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 io.reactivex.Flowable<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 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[])

publishRpcInternal

protected io.reactivex.Single<RabbitConsumerState> publishRpcInternal(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[])

publishInternalNoConfirm

protected io.reactivex.Completable publishInternalNoConfirm(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,
                                                             io.reactivex.CompletableEmitter 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 io.reactivex.disposables.Disposable createConsumer(com.rabbitmq.client.Channel channel,
                                                             RabbitPublishState publishState,
                                                             String correlationId,
                                                             io.reactivex.SingleEmitter<RabbitConsumerState> emitter)
                                                      throws 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:

IOException - If an error occurred subscribing