Package org.hobbit.core.rabbit

Class DataReceiverImpl

java.lang.Object

org.hobbit.core.rabbit.DataReceiverImpl

All Implemented Interfaces:

Closeable, AutoCloseable, DataReceiver

public class DataReceiverImpl
extends Object
implements DataReceiver

Implementation of the DataReceiver interface.

Use the internal DataReceiverImpl.Builder class for creating instances of the DataReceiverImpl class. Note that the created DataReceiverImpl will either use a given RabbitQueue or create a new one. In both cases the receiver will become the owner of the queue, i.e., if the DataReceiverImpl instance is closed the queue will be closed as well.

Internally, the receiver uses an own thread to consume incoming messages. These messages are forwarded to the given DataHandler instance. Note that this forwarding is based on an ExecutorService the called method DataHandler.handleData(byte[]) should be thread safe since it might be called in parallel.

The DataReceiverImpl owns recources that need to be freed if its work is done. This can be achieved by closing the receiver. In most cases, this should be done using the closeWhenFinished() method which waits until all incoming messages are processed and all streams are closed. Note that using the close() method leads to a direct shutdown of the queue which could lead to data loss and threads getting stuck.

Author:

Michael Röder (roeder@informatik.uni-leipzig.de)

Nested Class Summary

Nested Classes

Modifier and Type	Class	Description
static class	DataReceiverImpl.Builder
protected class	DataReceiverImpl.MsgProcessingTask
protected class	DataReceiverImpl.MsgReceivingTask

Field Summary

Fields

Modifier and Type	Field	Description
private DataHandler	dataHandler
private static int	DEFAULT_MAX_PARALLEL_PROCESSED_MESSAGES
private int	errorCount
private ExecutorService	executor
private static org.slf4j.Logger	LOGGER
protected RabbitQueue	queue
private TerminatableRunnable	receiverTask
private Thread	receiverThread

Constructor Summary

Constructors

Modifier	Constructor	Description
protected	DataReceiverImpl(RabbitQueue queue, DataHandler handler, int maxParallelProcessedMsgs)

Method Summary

Modifier and Type	Method	Description
static DataReceiverImpl.Builder	builder()	Returns a newly created DataReceiverImpl.Builder.
protected Runnable	buildMsgProcessingTask(com.rabbitmq.client.QueueingConsumer.Delivery delivery)	This factory method creates a runnable task that processes the given message.
protected TerminatableRunnable	buildMsgReceivingTask(com.rabbitmq.client.QueueingConsumer consumer)	This factory method creates a runnable task that uses the given consumer to receive incoming messages.
void	close()	A rude way to close the receiver.
void	closeWhenFinished()	This method waits for the data receiver to finish its work and closes the incoming queue as well as the internal thread pool after that.
DataHandler	getDataHandler()
int	getErrorCount()	Returns the number of errors that have been encountered while receiving data.
protected ExecutorService	getExecutor()
RabbitQueue	getQueue()
void	increaseErrorCount()

Methods inherited from class java.lang.Object

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

Field Detail

LOGGER

private static final org.slf4j.Logger LOGGER

DEFAULT_MAX_PARALLEL_PROCESSED_MESSAGES

private static final int DEFAULT_MAX_PARALLEL_PROCESSED_MESSAGES

See Also:

Constant Field Values

queue

protected RabbitQueue queue

errorCount

private int errorCount

dataHandler

private DataHandler dataHandler

executor

private ExecutorService executor

receiverTask

private TerminatableRunnable receiverTask

receiverThread

private Thread receiverThread

Constructor Detail

DataReceiverImpl

protected DataReceiverImpl(RabbitQueue queue,
                           DataHandler handler,
                           int maxParallelProcessedMsgs)
                    throws IOException

Throws:

IOException

Method Detail

getDataHandler

public DataHandler getDataHandler()

Specified by:

getDataHandler in interface DataReceiver

increaseErrorCount

public void increaseErrorCount()

Specified by:

increaseErrorCount in interface DataReceiver

getErrorCount

public int getErrorCount()

Description copied from interface: DataReceiver

Returns the number of errors that have been encountered while receiving data. If this number is not 0 the receiver can not guarantee that all data has been received correctly.

Specified by:

getErrorCount in interface DataReceiver

Returns:

the number of errors encountered during the receiving of data

getQueue

public RabbitQueue getQueue()

Specified by:

getQueue in interface DataReceiver

getExecutor

protected ExecutorService getExecutor()

closeWhenFinished

public void closeWhenFinished()

This method waits for the data receiver to finish its work and closes the incoming queue as well as the internal thread pool after that.

Specified by:

closeWhenFinished in interface DataReceiver

close

public void close()

A rude way to close the receiver. Note that this method directly closes the incoming queue and only notifies the internal consumer to stop its work but won't wait for the handler threads to finish their work.

Specified by:

close in interface AutoCloseable

Specified by:

close in interface Closeable

builder

public static DataReceiverImpl.Builder builder()

Returns a newly created DataReceiverImpl.Builder.

Returns:

a new DataReceiverImpl.Builder instance

buildMsgReceivingTask

protected TerminatableRunnable buildMsgReceivingTask(com.rabbitmq.client.QueueingConsumer consumer)

This factory method creates a runnable task that uses the given consumer to receive incoming messages.

Parameters:

consumer - the consumer that can be used to receive messages

Returns:

a Runnable instance that will handle incoming messages as soon as it will be executed

buildMsgProcessingTask

protected Runnable buildMsgProcessingTask(com.rabbitmq.client.QueueingConsumer.Delivery delivery)

This factory method creates a runnable task that processes the given message.

Parameters:

delivery - the message that should be processed

Returns:

a Runnable instance that will process the message as soon as it will be executed