org.hobbit.core.components

Class AbstractEvaluationStorage

java.lang.Object

org.hobbit.core.components.AbstractComponent

org.hobbit.core.components.AbstractCommandReceivingComponent

org.hobbit.core.components.AbstractPlatformConnectorComponent

org.hobbit.core.components.AbstractEvaluationStorage

All Implemented Interfaces:

Closeable, AutoCloseable, CommandReceivingComponent, Component, ExpectedResponseReceivingComponent, PlatformConnector, ResponseReceivingComponent

Direct Known Subclasses:

InMemoryEvaluationStore

public abstract class AbstractEvaluationStorage
extends AbstractPlatformConnectorComponent
implements ResponseReceivingComponent, ExpectedResponseReceivingComponent

This abstract class implements basic functions that can be used to implement an evaluation storage. Incoming messages should have the structure:
timestamp length id data
where

• timestamp is a long value (only expected if the message is received from a task generator)
• length is an int value containing the length of the following id string
• id is a string with the given length
• data is the remaining bytes that are received

Author:

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

Nested Class Summary

Nested Classes

Modifier and Type	Class and Description
protected class	AbstractEvaluationStorage.ExpectedResponseReceiver Receiver handling the expected responses.
protected class	AbstractEvaluationStorage.IterationRequestReceiver
protected class	AbstractEvaluationStorage.SystemResponseReceiver Receiver handling the responses coming from the system.

Field Summary

Fields

Modifier and Type	Field and Description
protected com.rabbitmq.client.Channel	ackChannel Channel on which the acknowledgements are send.
protected String	ackExchangeName
private static int	DEFAULT_MAX_PARALLEL_PROCESSED_MESSAGES Default value of the maxParallelProcessedMsgs attribute.
private static byte[]	EMPTY_RESPONSE The empty response that is sent if an error occurs.
protected RabbitQueue	evalModule2EvalStoreQueue The incoming queue from the evaluation module.
static int	EXPECTED_RESPONSE_STREAM_ID
protected DataReceiverImpl	expResponseReceiver The incoming queue from the task generator.
static int	ITERATOR_ID_STREAM_ID
private static org.slf4j.Logger	LOGGER
private int	maxParallelProcessedMsgs The maximum number of incoming messages of a single queue that are processed in parallel.
static byte	NEW_ITERATOR_ID If a request contains this iterator ID, a new iterator is created and its first result as well as its Id are returned.
static String	RECEIVE_TIMESTAMP_FOR_SYSTEM_RESULTS_KEY
static int	RECEIVED_RESPONSE_STREAM_ID
protected boolean	receiveTimeStamp
protected Map<String,PairedDataSender>	replyingSenders
protected List<Iterator<? extends ResultPair>>	resultPairIterators Iterators that have been started.
protected DataReceiverImpl	systemResponseReceiver The incoming queue from the system.
protected DataReceiver	systemResultReceiver The incoming queue from the system.
protected DataReceiver	taskResultReceiver The incoming queue from the task generator.
private Semaphore	terminationMutex Mutex used to wait for the termination signal.

Fields inherited from class org.hobbit.core.components.AbstractCommandReceivingComponent

cmdChannel, cmdQueueFactory, DEFAULT_CMD_RESPONSE_TIMEOUT, defaultContainerType

Fields inherited from class org.hobbit.core.components.AbstractComponent

connectionFactory, incomingDataQueueFactory, NUMBER_OF_RETRIES_TO_CONNECT_TO_RABBIT_MQ, outgoingDataQueuefactory, rabbitMQHostName, START_WAITING_TIME_BEFORE_RETRY

Constructor Summary

Constructors

Constructor and Description
AbstractEvaluationStorage() Constructor using the DEFAULT_MAX_PARALLEL_PROCESSED_MESSAGES= 50.
AbstractEvaluationStorage(int maxParallelProcessedMsgs) Constructor setting the maximum number of messages processed in parallel.

Method Summary

Modifier and Type	Method and Description
protected void	acknowledgeResponse(String taskId)
void	close()
protected abstract Iterator<? extends ResultPair>	createIterator() Creates a new iterator that iterates over the response pairs.
void	init() This method initializes the component.
void	receiveCommand(byte command, byte[] data) This method is called if a command is received and might be interesting for this particular component.
void	run() This method executes the component.

Methods inherited from class org.hobbit.core.components.AbstractPlatformConnectorComponent

addContainerObserver, createContainer, getFactoryForIncomingCmdQueues, getFactoryForIncomingDataQueues, getFactoryForOutgoingCmdQueues, getFactoryForOutgoingDataQueues, stopContainer

Methods inherited from class org.hobbit.core.components.AbstractCommandReceivingComponent

addCommandHeaderId, createContainer, createContainer, handleCmd, sendToCmdQueue, sendToCmdQueue, sendToCmdQueue

Methods inherited from class org.hobbit.core.components.AbstractComponent

createConnection, generateSessionQueueName, getHobbitSessionId

Methods inherited from class java.lang.Object

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

Methods inherited from interface org.hobbit.core.components.ResponseReceivingComponent

receiveResponseData

Methods inherited from interface org.hobbit.core.components.ExpectedResponseReceivingComponent

receiveExpectedResponseData

Field Detail

LOGGER

private static final org.slf4j.Logger LOGGER

ITERATOR_ID_STREAM_ID

public static final int ITERATOR_ID_STREAM_ID

See Also:

Constant Field Values

EXPECTED_RESPONSE_STREAM_ID

public static final int EXPECTED_RESPONSE_STREAM_ID

See Also:

Constant Field Values

RECEIVED_RESPONSE_STREAM_ID

public static final int RECEIVED_RESPONSE_STREAM_ID

See Also:

Constant Field Values

RECEIVE_TIMESTAMP_FOR_SYSTEM_RESULTS_KEY

public static final String RECEIVE_TIMESTAMP_FOR_SYSTEM_RESULTS_KEY

See Also:

Constant Field Values

NEW_ITERATOR_ID

public static final byte NEW_ITERATOR_ID

If a request contains this iterator ID, a new iterator is created and its first result as well as its Id are returned.

See Also:

Constant Field Values

EMPTY_RESPONSE

private static final byte[] EMPTY_RESPONSE

The empty response that is sent if an error occurs.

DEFAULT_MAX_PARALLEL_PROCESSED_MESSAGES

private static final int DEFAULT_MAX_PARALLEL_PROCESSED_MESSAGES

Default value of the maxParallelProcessedMsgs attribute.

See Also:

Constant Field Values

terminationMutex

private Semaphore terminationMutex

Mutex used to wait for the termination signal.

expResponseReceiver

protected DataReceiverImpl expResponseReceiver

The incoming queue from the task generator.

systemResponseReceiver

protected DataReceiverImpl systemResponseReceiver

The incoming queue from the system.

maxParallelProcessedMsgs

private final int maxParallelProcessedMsgs

The maximum number of incoming messages of a single queue that are processed in parallel. Additional messages have to wait.

resultPairIterators

protected List<Iterator<? extends ResultPair>> resultPairIterators

Iterators that have been started.

taskResultReceiver

protected DataReceiver taskResultReceiver

The incoming queue from the task generator.

systemResultReceiver

protected DataReceiver systemResultReceiver

The incoming queue from the system.

evalModule2EvalStoreQueue

protected RabbitQueue evalModule2EvalStoreQueue

The incoming queue from the evaluation module.

ackChannel

protected com.rabbitmq.client.Channel ackChannel

Channel on which the acknowledgements are send.

ackExchangeName

protected String ackExchangeName

replyingSenders

protected Map<String,PairedDataSender> replyingSenders

receiveTimeStamp

protected boolean receiveTimeStamp

Constructor Detail

AbstractEvaluationStorage

public AbstractEvaluationStorage()

Constructor using the DEFAULT_MAX_PARALLEL_PROCESSED_MESSAGES= 50.

AbstractEvaluationStorage

public AbstractEvaluationStorage(int maxParallelProcessedMsgs)

Constructor setting the maximum number of messages processed in parallel.

Parameters:

maxParallelProcessedMsgs - The maximum number of incoming messages of a single queue that are processed in parallel. Additional messages have to wait.

Method Detail

init

public void init()
          throws Exception

Description copied from interface: Component

This method initializes the component.

Specified by:

init in interface Component

Overrides:

init in class AbstractCommandReceivingComponent

Throws:

Exception - if an error occurs during the initialization

createIterator

protected abstract Iterator<? extends ResultPair> createIterator()

Creates a new iterator that iterates over the response pairs.

Returns:

a new iterator or null if an error occurred

run

public void run()
         throws Exception

Description copied from interface: Component

This method executes the component.

Specified by:

run in interface Component

Throws:

Exception - if an error occurs during the execution

receiveCommand

public void receiveCommand(byte command,
                           byte[] data)

Description copied from interface: CommandReceivingComponent

This method is called if a command is received and might be interesting for this particular component.

Specified by:

receiveCommand in interface CommandReceivingComponent

Overrides:

receiveCommand in class AbstractPlatformConnectorComponent

Parameters:

command - the byte encoding the command

data - additional data that was sent together with the command

acknowledgeResponse

protected void acknowledgeResponse(String taskId)

close

public void close()
           throws IOException

Specified by:

close in interface Closeable

Specified by:

close in interface AutoCloseable

Overrides:

close in class AbstractCommandReceivingComponent

Throws:

IOException