org.hobbit.core.components.stream

Class AbstractStreamingTaskGenerator

java.lang.Object

org.hobbit.core.components.AbstractComponent

org.hobbit.core.components.AbstractCommandReceivingComponent

org.hobbit.core.components.AbstractPlatformConnectorComponent

org.hobbit.core.components.stream.AbstractStreamingTaskGenerator

All Implemented Interfaces:

Closeable, AutoCloseable, CommandReceivingComponent, Component, PlatformConnector, StreamingGeneratedDataReceivingComponent

Direct Known Subclasses:

AbstractTaskGenerator

public abstract class AbstractStreamingTaskGenerator
extends AbstractPlatformConnectorComponent
implements StreamingGeneratedDataReceivingComponent

This abstract class implements basic functions that can be used to implement a task generator. The following environment variables are expected:

• Constants.GENERATOR_ID_KEY
• Constants.GENERATOR_COUNT_KEY

Author:

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

Nested Class Summary

Nested Classes

Modifier and Type	Class and Description
protected class	AbstractStreamingTaskGenerator.GeneratedDataHandler A simple internal handler class that calls generateTask(InputStream) with the given InputStream.

Field Summary

Fields

Modifier and Type	Field and Description
protected DataReceiver	dataReceiver Receiver for receiving data from the data generators.
private static int	DEFAULT_MAX_PARALLEL_PROCESSED_MESSAGES Default value of the maxParallelProcessedMsgs attribute.
protected int	generatorId The id of this generator.
private static org.slf4j.Logger	LOGGER
private int	maxParallelProcessedMsgs The maximum number of incoming messages that are processed in parallel.
private long	nextTaskId The task id that will be assigned to the next task generated by this generator.
protected int	numberOfGenerators The number of task generators created by the benchmark controller.
protected DataSender	sender2EvalStore Sender for transferring data to the evaluation storage.
protected DataSender	sender2System Sender for sending tasks to the benchmarked system.
private Semaphore	startTaskGenMutex Mutex used to wait for the start signal after the component has been started and initialized.
private Semaphore	terminateMutex Mutex used to wait for the terminate 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
AbstractStreamingTaskGenerator() Default constructor creating an AbstractStreamingTaskGenerator processing up to DEFAULT_MAX_PARALLEL_PROCESSED_MESSAGES= 10 messages in parallel.
AbstractStreamingTaskGenerator(int maxParallelProcessedMsgs) Constructor setting the maximum number of parallel processed messages.

Method Summary

Modifier and Type	Method and Description
void	close()
protected abstract void	generateTask(InputStream dataStream) Generates a task from the given data, sends it to the system, takes the timestamp of the moment at which the message has been sent to the system and sends it together with the expected response to the evaluation storage.
int	getGeneratorId()
protected String	getNextTaskId() Generates the next unique ID for a task.
int	getNumberOfGenerators()
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	receiveGeneratedData(InputStream dataStream) This method is called if a new data stream is received from a data generator.
void	run() This method executes the component.
protected void	sendTaskToEvalStorage(String taskIdString, long timestamp, InputStream stream) This method sends the data read from the given InputStream and the given timestamp of the task with the given task id to the evaluation storage.
protected void	sendTaskToSystemAdapter(String taskIdString, InputStream stream) Sends the given task with the given task id and data to the system.

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

Field Detail

LOGGER

private static final org.slf4j.Logger LOGGER

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

startTaskGenMutex

private Semaphore startTaskGenMutex

Mutex used to wait for the start signal after the component has been started and initialized.

terminateMutex

private Semaphore terminateMutex

Mutex used to wait for the terminate signal.

generatorId

protected int generatorId

The id of this generator.

numberOfGenerators

protected int numberOfGenerators

The number of task generators created by the benchmark controller.

nextTaskId

private long nextTaskId

The task id that will be assigned to the next task generated by this generator.

maxParallelProcessedMsgs

private final int maxParallelProcessedMsgs

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

sender2System

protected DataSender sender2System

Sender for sending tasks to the benchmarked system.

sender2EvalStore

protected DataSender sender2EvalStore

Sender for transferring data to the evaluation storage.

dataReceiver

protected DataReceiver dataReceiver

Receiver for receiving data from the data generators.

Constructor Detail

AbstractStreamingTaskGenerator

public AbstractStreamingTaskGenerator()

Default constructor creating an AbstractStreamingTaskGenerator processing up to DEFAULT_MAX_PARALLEL_PROCESSED_MESSAGES= 10 messages in parallel.

AbstractStreamingTaskGenerator

public AbstractStreamingTaskGenerator(int maxParallelProcessedMsgs)

Constructor setting the maximum number of parallel processed messages. Note that this parameter has to be larger or equal to 1 or the init() method will throw an exception. Setting maxParallelProcessedMsgs=1 leads to the usage of a QueueingConsumer.

Parameters:

maxParallelProcessedMsgs - the number of messaegs that are processed in parallel

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

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

receiveGeneratedData

public void receiveGeneratedData(InputStream dataStream)

Description copied from interface: StreamingGeneratedDataReceivingComponent

This method is called if a new data stream is received from a data generator.

Specified by:

receiveGeneratedData in interface StreamingGeneratedDataReceivingComponent

Parameters:

dataStream - the stream from which the data received from a data generator is read

generateTask

protected abstract void generateTask(InputStream dataStream)
                              throws Exception

Generates a task from the given data, sends it to the system, takes the timestamp of the moment at which the message has been sent to the system and sends it together with the expected response to the evaluation storage.

Parameters:

dataStream - stream containing incoming data generated by a data generator

Throws:

Exception - if a sever error occurred

getNextTaskId

protected String getNextTaskId()

Generates the next unique ID for a task.

Returns:

the next unique task ID

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

sendTaskToEvalStorage

protected void sendTaskToEvalStorage(String taskIdString,
                                     long timestamp,
                                     InputStream stream)
                              throws IOException

This method sends the data read from the given InputStream and the given timestamp of the task with the given task id to the evaluation storage.

Parameters:

taskIdString - the id of the task

timestamp - the timestamp of the moment in which the task has been sent to the system

stream - stream from which the expected response for the task with the given id will be read

Throws:

IOException - if there is an error during the sending

sendTaskToSystemAdapter

protected void sendTaskToSystemAdapter(String taskIdString,
                                       InputStream stream)
                                throws IOException

Sends the given task with the given task id and data to the system.

Parameters:

taskIdString - the id of the task

data - the data of the task

Throws:

IOException - if there is an error during the sending

getGeneratorId

public int getGeneratorId()

getNumberOfGenerators

public int getNumberOfGenerators()

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