Package dev.aherscu.qa.jgiven.rabbitmq.utils

Class QueueHandler<K,V>

java.lang.Object

dev.aherscu.qa.jgiven.rabbitmq.utils.QueueHandler<K,V>

Type Parameters:

K - type of message-key; it should have a proper hash function in order to get O(1) access time, otherwise it may degrade to O(n)

V - type of message-value

All Implemented Interfaces:

AutoCloseable

public class QueueHandler<K,V>
extends Object
implements AutoCloseable

Retrieves RabbitMQ messages from specified queue via provided channel. Messages are converted by specified value function and indexed by specified key function.

Messages are consumed in background threads and made available for retrieval by their key according to a specified retry policy. If a retry policy is not specified, then an internal default will be used.

Typical workflow:

1. build it
2. start consumption -- this will run in background until closing
3. retrieve messages

Nested Class Summary

Nested Classes

Modifier and Type	Class	Description
static class	QueueHandler.QueueHandlerBuilder<K,V,C extends QueueHandler<K,V>,B extends QueueHandler.QueueHandlerBuilder<K,V,C,B>>
static class	QueueHandler.Singleton<K,V>	Utility for building a singleton QueueHandler.

Field Summary

Fields

Modifier and Type	Field	Description
com.rabbitmq.client.Channel	channel
Function<byte[],V>	consumingBy
Function<Message<V>,K>	indexingBy
Function<V,byte[]>	publishingBy
String	queue

Constructor Summary

Constructors

Modifier	Constructor	Description
protected	QueueHandler(QueueHandler.QueueHandlerBuilder<K,V,?,?> b)

Method Summary

Modifier and Type	Method	Description
static <K,V> QueueHandler.QueueHandlerBuilder<K,V,?,?>	builder()
String	cancel()	Cancels the consumption, previously started by consume().
void	close()
static com.rabbitmq.client.ConnectionFactory	connectionFactoryFrom(String uri)
String	consume()	Starts the message consumption process and returns immediately.
void	publish(Stream<Message<V>> messages)
void	publishValues(Stream<V> values)
Map<K,Message<V>>	recievedMessages()	Unmodifiable view of retrieved messages.

Methods inherited from class java.lang.Object

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

Field Detail

channel

public final com.rabbitmq.client.Channel channel

queue

public final String queue

indexingBy

public final Function<Message<V>,K> indexingBy

consumingBy

public final Function<byte[],V> consumingBy

publishingBy

public final Function<V,byte[]> publishingBy

Constructor Detail

QueueHandler

protected QueueHandler(QueueHandler.QueueHandlerBuilder<K,V,?,?> b)

Method Detail

connectionFactoryFrom

public static com.rabbitmq.client.ConnectionFactory connectionFactoryFrom(String uri)

Parameters:

uri - an amqp or amqps uri; e.g., amqps://username:password@host/virtual-host

Returns:

a connection factory

Throws:

RuntimeException - upon malformed uri

cancel

public String cancel()
              throws IOException

Cancels the consumption, previously started by consume(). Must be called on same thread as last consume() call.

Returns:

the consumer tag

Throws:

IOException - upon connection failures, or if consumer was not started

close

public void close()
           throws IOException

Specified by:

close in interface AutoCloseable

Throws:

IOException

consume

public String consume()

Starts the message consumption process and returns immediately.

Returns:

consumer tag

Throws:

RuntimeException - upon connection failures

See Also:

for canceling the consumption

publish

public void publish(Stream<Message<V>> messages)

publishValues

public void publishValues(Stream<V> values)

recievedMessages

public Map<K,Message<V>> recievedMessages()

Unmodifiable view of retrieved messages.

Returns:

the messages

builder

public static <K,V> QueueHandler.QueueHandlerBuilder<K,V,?,?> builder()