Detailed Description

Receives zero or more input events, and produces zero or more output events.

The processor is the fundamental class where all computation occurs. All of BeepBeep's processors (including yours) are descendants of this class.

A processor is depicted graphically as a "box", with "pipes" representing its input and output streams.

Processor

This class itself is abstract; nevertheless, it provides important methods for handling input/output event queues, connecting processors together, etc. However, if you write your own processor, you will most likely want to inherit from its child, SynchronousProcessor, which does some more work for you.

The Processor class does not assume anything about the type of events being input or output. All its input and output queues are therefore declared as containing instances of Object, Java's most generic type.

Author: Sylvain Hallé

Since: 0.1

Definition at line 68 of file Processor.java.

Classes
class	InternalProcessorState
	An object capturing the internal state of a processor, including the current contents of its input and output queues. More...

Public Member Functions
	Processor (int in_arity, int out_arity)
	Initializes a processor.
final Object	getContext (String key)
	Retrieves an object from the processor's context.
Context	getContext ()
	Gets the context associated to this object.
void	setContext (String key, Object value)
	Adds an object to the object's context.
void	setContext (Context context)
	Adds a complete context to this object.
final int	hashCode ()
	Implementation of `hashCode()` specific to processors.
final boolean	equals (Object o)
	Implementation of `equals()` specific to processors.
final int	getId ()
	Fetches the processor instance's unique ID.
void	reset ()
	Resets the processor.
abstract Pushable	getPushableInput (int index)
	Returns the `Pushable` corresponding to the processor's i-th input trace.
final Pushable	getPushableInput ()
	Returns the `Pushable` corresponding to the processor's first input trace.
abstract Pullable	getPullableOutput (int index)
	Returns the `Pullable` corresponding to the processor's i-th output trace.
final Pullable	getPullableOutput ()
	Returns the `Pullable` corresponding to the processor's first output trace.
void	setPullableInput (int i, Pullable p)
	Assigns a `Pullable` to the processor's i-th input.
Pullable	getPullableInput (int i)
	Returns the `Pullable` corresponding to the processor's i-th input.
void	setPushableOutput (int i, Pushable p)
	Assigns a `Pushable` to the processor's i-th output.
Pushable	getPushableOutput (int i)
	Retrieves the `Pushable` associated to the processor's i-th output.
final int	getInputArity ()
	Returns the processor's input arity.
final int	getOutputArity ()
	Returns the processor's output arity.
void	duplicateInto (Processor p)
	Copies the contents and state of the current processor into another.
final Set< Class<?> >	getInputType (int index)
	Gets the type of events the processor accepts for its i-th input trace.
void	getInputTypesFor (Set< Class<?> > classes, int index)
	Populates the set of classes accepted by the processor for its i-th input.
Class<?>	getOutputType (int index)
	Returns the type of the events produced by the processor for its i-th output.
void	start ()
	Starts the processor.
void	stop ()
	Stops the processor.
final EventTracker	getEventTracker ()
	Gets the instance of event tracker associated to this processor.
Processor	setEventTracker (EventTracker tracker)
	Associates an event tracker to this processor.
void	associateToInput (int in_stream_index, int in_stream_pos, int out_stream_index, int out_stream_pos)
	Associates an input event to an output event.
void	associateTo (NodeFunction f, int out_stream_index, int out_stream_pos)
	Associates a node function to a particular event of processor's output stream.
void	associateToOutput (int in_stream_index, int in_stream_pos, int out_stream_index, int out_stream_pos)
	Associates an input event to an output event.
final int	getInputCount ()
	Gets the number of event fronts received so far by this processor.
final int	getOutputCount ()
	Gets the number of event fronts produced so far by this processor.
final Object	print (ObjectPrinter<?> printer) throws ProcessorException
	Prints the contents of this processor into an object printer.
final Processor	read (ObjectReader<?> reader, Object o) throws ProcessorException
	Reads the content of a processor from a serialized object.
final Processor	duplicate ()
	Duplicates an object and sets it to its initial state.
void	copyInputQueue (int index, Collection< Object > to)
	Copies the content of one of the processor's input queue to a collection.
void	copyOutputQueue (int index, Collection< Object > to)
	Copies the content of one of the processor's output queue to a collection.
abstract Processor	duplicate (boolean with_state)
	Duplicates an object.
Processor	or (Processor p)
	Connects the first output pipe of this processor to the first input pipe of another processor.
Processor	or (CallAfterConnect c)
	Operates similar to `or(Processor)`, but also calls a method after the connection has been established.
Processor	or (SelectedInputPipe p)
	Connects the output at index 0 of the current processor to the input of another processor.
PipeSelector	getAt (int index)
	Gets the `PipeSelector` object corresponding to the processor's input or output pipe for a given index.
Pushable	rightShift (int index)
Pullable	leftShift (int index)
Queue< Object >	getInputQueue (int index)
	Gets the content of the processor's input queue at a given index.
Queue< Object >	getOutputQueue (int index)
	Gets the content of the processor's output queue at a given index.

Static Public Member Functions
static boolean	allNull (Object[] v)
	Checks if all objects in the array are null.
static Queue< Object[]>	getEmptyQueue ()
	Gets an instance of an empty event queue.
static void	startAll (Processor ... procs)
	Starts all processors given as an argument.
static void	stopAll (Processor ... procs)
	Stops all processors given as an argument.
static List< ProvenanceNode >	getLeaves (ProvenanceNode root)
	Gets the leaves of a provenance tree.

Static Public Attributes
static final transient String	s_versionString = "0.11.2"
	A string used to identify the program's version.
static final transient int	MAX_PULL_RETRIES = 10000000
	Number of times the `Pullable#hasNext()` method tries to produce an output from the input before giving up.

Protected Member Functions
boolean	allNotifiedEndOfTrace ()
	Determines if all the upstream pushables have sent the end of trace notification.
final Context	newContext ()
	Creates a new empty context map.
boolean	onEndOfTrace (Queue< Object[]> outputs) throws ProcessorException
	Allows to describe a specific behavior when the trace of input fronts has reached its end.
Object	printState ()
	Produces an object that represents the state of the current processor.
Processor	readState (Object o)
	Reads the state of a processor and uses it to create a new instance.

Static Protected Member Functions
static void	getLeaves (ProvenanceNode root, List< ProvenanceNode > leaves)
	Accumulates the leaves of a provenance tree in a list.

Protected Attributes
int	m_inputArity
	The processor's input arity, i.e.
int	m_outputArity
	The processor's output arity, i.e.
transient Queue< Object >[]	m_inputQueues
	An array of input event queues.
transient EventTracker	m_eventTracker = null
	An object that keeps track of the relationship between input and output events.
transient Queue< Object >[]	m_outputQueues
	An array of output event queues.
transient Pullable[]	m_inputPullables
	An array of `Pullable`s, one for each input trace this processor receives.
transient Pushable[]	m_outputPushables
	An array of `Pushable`s, one for each output trace this processor produces.
int	m_inputCount = 0
	A counter incremented upon each input front processed.
int	m_outputCount = 0
	A counter incremented upon each output front processed.
Context	m_context = null
	The context in which the processor is instantiated.
boolean[]	m_hasBeenNotifiedOfEndOfTrace
	Indicates whether the processor has been notified of the end of trace or not.
boolean	m_notifiedEndOfTraceDownstream
	Indicates whether the processor has notified the end of the trace to the downstream processors it is connected to.

Constructor & Destructor Documentation

◆ Processor()

ca.uqac.lif.cep.Processor.Processor	(	int	in_arity,
		int	out_arity )

Initializes a processor.

This has for effect of executing the basic operations common to every processor:

Giving a unique ID
Determining its input and output arity
Creating arrays of empty input and output queues, as well as arrays of Pullables and Pushables

If you create your own processor, its constructor must start by calling its ancestor's constructor (which ultimately ends up calling this constructor). Otherwise, expect a plethora of null pointers and other oddities.

Parameters

in_arity	The processor's input arity
out_arity	The processor's output arity

Definition at line 192 of file Processor.java.

Member Function Documentation

◆ allNotifiedEndOfTrace()

boolean ca.uqac.lif.cep.Processor.allNotifiedEndOfTrace ( )

protected

Determines if all the upstream pushables have sent the end of trace notification.

Returns: true if all notifications have been sent, false otherwise

Definition at line 224 of file Processor.java.

◆ allNull()

boolean ca.uqac.lif.cep.Processor.allNull ( Object[] v )

static

Checks if all objects in the array are null.

This is a convenience method used by other processor classes (e.g. SynchronousProcessor to make sure that some output was generated from a given input

Parameters

v The array

Returns: true if all elements in the array are null, false otherwise

Definition at line 508 of file Processor.java.

◆ associateTo()

void ca.uqac.lif.cep.Processor.associateTo	(	NodeFunction	f,
		int	out_stream_index,
		int	out_stream_pos )

Associates a node function to a particular event of processor's output stream.

Parameters

f	The node function
out_stream_index	The index of the processor's output stream
out_stream_pos	The position of the event in the output stream

Definition at line 718 of file Processor.java.

◆ associateToInput()

void ca.uqac.lif.cep.Processor.associateToInput	(	int	in_stream_index,
		int	in_stream_pos,
		int	out_stream_index,
		int	out_stream_pos )

Associates an input event to an output event.

Parameters

in_stream_index	The index of the processor's input stream
in_stream_pos	The position of the event in the input stream
out_stream_index	The index of the processor's output stream
out_stream_pos	The position of the event in the output stream

Definition at line 701 of file Processor.java.

◆ associateToOutput()

void ca.uqac.lif.cep.Processor.associateToOutput	(	int	in_stream_index,
		int	in_stream_pos,
		int	out_stream_index,
		int	out_stream_pos )

Associates an input event to an output event.

Parameters

in_stream_index	The index of the processor's input stream
in_stream_pos	The position of the event in the input stream
out_stream_index	The index of the processor's output stream
out_stream_pos	The position of the event in the output stream

Definition at line 733 of file Processor.java.

◆ copyInputQueue()

void ca.uqac.lif.cep.Processor.copyInputQueue	(	int	index,
		Collection< Object >	to )

Copies the content of one of the processor's input queue to a collection.

Parameters

index	The index of the input queue
to	The collection where queue contents are copied to

Since: 0.11

Definition at line 915 of file Processor.java.

◆ copyOutputQueue()

void ca.uqac.lif.cep.Processor.copyOutputQueue	(	int	index,
		Collection< Object >	to )

Copies the content of one of the processor's output queue to a collection.

Parameters

index	The index of the output queue
to	The collection where queue contents are copied to

Since: 0.11

Definition at line 926 of file Processor.java.

◆ duplicate() [1/2]

final Processor ca.uqac.lif.cep.Processor.duplicate ( )

Duplicates an object and sets it to its initial state.

This should be the same thing as calling duplicate(false).

Returns: Another object

Implements ca.uqac.lif.cep.DuplicableProcessor.

Definition at line 904 of file Processor.java.

◆ duplicate() [2/2]

abstract Processor ca.uqac.lif.cep.Processor.duplicate ( boolean with_state )

abstract

Duplicates an object.

Optionally, set the object into the same state as the source object.

Parameters

with_state Set to true to replicate the object's state, false to create a new copy in the initial state.

Returns: Another object

Implements ca.uqac.lif.cep.Duplicable.

◆ duplicateInto()

void ca.uqac.lif.cep.Processor.duplicateInto ( Processor p )

Copies the contents and state of the current processor into another.

Parameters

p	The processor to copy contents into

Definition at line 526 of file Processor.java.

◆ equals()

final boolean ca.uqac.lif.cep.Processor.equals ( Object o )

Implementation of equals() specific to processors.

Since every processor has a unique ID, equality amounts to equality of the field m_uniqueId. This behavior cannot be overridden by descendants.

Definition at line 321 of file Processor.java.

◆ getAt()

PipeSelector ca.uqac.lif.cep.Processor.getAt ( int index )

Gets the PipeSelector object corresponding to the processor's input or output pipe for a given index.

Java programmers probably won't use this method, but users of the Groovy language can benefit from its operator overloading conventions, which map the construct p[x] to p.getAt(x). Combined with the definition of or(Connector.SelectedInputPipe), this can be used to easily pipe two processors together:



def p = (some processor)

def q = (some other processor)

p | q[1] // Connects p to pipe index 1 of q

Parameters

index The input pipe index

Returns: The pushable object

Since: 0.11

Definition at line 1059 of file Processor.java.

◆ getContext() [1/2]

Context ca.uqac.lif.cep.Processor.getContext ( )

Gets the context associated to this object.

Returns: The context

Implements ca.uqac.lif.cep.Contextualizable.

Definition at line 263 of file Processor.java.

◆ getContext() [2/2]

final Object ca.uqac.lif.cep.Processor.getContext ( String key )

Retrieves an object from the processor's context.

Parameters

key	The key associated to that object

Returns: The object, or null if no object exists with such key

Definition at line 253 of file Processor.java.

◆ getEmptyQueue()

Queue< Object[]> ca.uqac.lif.cep.Processor.getEmptyQueue ( )

static

Gets an instance of an empty event queue.

It is recommended that processors call this method to get a queue instance, rather than instantiate their own manually.

Returns: The queue

Definition at line 612 of file Processor.java.

◆ getEventTracker()

final EventTracker ca.uqac.lif.cep.Processor.getEventTracker ( )

Gets the instance of event tracker associated to this processor.

Returns: The event tracker, or null of no event tracker is associated to this processor

Definition at line 675 of file Processor.java.

◆ getId()

final int ca.uqac.lif.cep.Processor.getId ( )

Fetches the processor instance's unique ID.

Returns: The ID

Definition at line 336 of file Processor.java.

◆ getInputArity()

final int ca.uqac.lif.cep.Processor.getInputArity ( )

Returns the processor's input arity.

Returns: The arity

Definition at line 483 of file Processor.java.

◆ getInputCount()

final int ca.uqac.lif.cep.Processor.getInputCount ( )

Gets the number of event fronts received so far by this processor.

Returns: The number of fronts

Definition at line 768 of file Processor.java.

◆ getInputQueue()

Queue< Object > ca.uqac.lif.cep.Processor.getInputQueue ( int index )

Gets the content of the processor's input queue at a given index.

Note that the method returns a copy of the queue, and not its own internal queue. This means that modifications to the returned queue have no effect on the processor's internal state.

Parameters

index The index of the input

Returns: The contents of the queue

Since: 0.11.2

Definition at line 1084 of file Processor.java.

◆ getInputType()

final Set< Class<?> > ca.uqac.lif.cep.Processor.getInputType ( int index )

Gets the type of events the processor accepts for its i-th input trace.

Note that this method returns a set, in the case where the processor accepts various types of objects (for example, a processor accepting Numbers, but also Strings it converts into numbers internally).

Parameters

index The index of the input to query

Returns: A set of classes. If index is less than 0 or greater than the processor's declared input arity, the set will be empty.

Definition at line 553 of file Processor.java.

◆ getInputTypesFor()

void ca.uqac.lif.cep.Processor.getInputTypesFor	(	Set< Class<?> >	classes,
		int	index )

Populates the set of classes accepted by the processor for its i-th input.

By default, a processor returns the Connector.Variant type for all its inputs and all its outputs, meaning that the checking of types in Connector#connect(Processor...) will be skipped. A descendant of this class may choose to define specific types for its input and output, thereby activating runtime type checking.

Parameters

classes	The set of to fill with classes
index	The index of the input to query

Reimplemented in ca.uqac.lif.cep.functions.ApplyFunction, and ca.uqac.lif.cep.functions.TurnInto.

Definition at line 578 of file Processor.java.

◆ getLeaves() [1/2]

List< ProvenanceNode > ca.uqac.lif.cep.Processor.getLeaves ( ProvenanceNode root )

static

Gets the leaves of a provenance tree.

Parameters

root	The root of the tree

Returns: A list of nodes that correspond to the leaves

Definition at line 939 of file Processor.java.

◆ getLeaves() [2/2]

void ca.uqac.lif.cep.Processor.getLeaves	(	ProvenanceNode	root,
		List< ProvenanceNode >	leaves )

staticprotected

Accumulates the leaves of a provenance tree in a list.

Parameters

root	The current node in the tree
leaves	The list of leaves

Definition at line 951 of file Processor.java.

◆ getOutputArity()

final int ca.uqac.lif.cep.Processor.getOutputArity ( )

Returns the processor's output arity.

Returns: The arity

Definition at line 493 of file Processor.java.

◆ getOutputCount()

final int ca.uqac.lif.cep.Processor.getOutputCount ( )

Gets the number of event fronts produced so far by this processor.

Returns: The number of fronts

Definition at line 777 of file Processor.java.

◆ getOutputQueue()

Queue< Object > ca.uqac.lif.cep.Processor.getOutputQueue ( int index )

Gets the content of the processor's output queue at a given index.

Note that the method returns a copy of the queue, and not its own internal queue. This means that modifications to the returned queue have no effect on the processor's internal state.

Parameters

index The index of the input

Returns: The contents of the queue

Since: 0.11.2

Definition at line 1101 of file Processor.java.

◆ getOutputType()

Class<?> ca.uqac.lif.cep.Processor.getOutputType ( int index )

Returns the type of the events produced by the processor for its i-th output.

By default, a processor returns the Connector.Variant type for all its inputs and all its outputs, meaning that the checking of types in Connector#connect(Processor...) will be skipped. A descendant of this class may choose to define specific types for its input and output, thereby activating runtime type checking.

Parameters

index The index of the output to query

Returns: The type of the output. If index it less than 0 or greater than the processor's declared output arity, this method may throw an IndexOutOfBoundsException.

Reimplemented in ca.uqac.lif.cep.functions.ApplyFunction, ca.uqac.lif.cep.functions.TurnInto, ca.uqac.lif.cep.tmf.QueueSource, ca.uqac.lif.cep.util.Lists.ListUpdateProcessor, ca.uqac.lif.cep.util.Maps.MapPutInto, ca.uqac.lif.cep.util.Maps.PutInto, and ca.uqac.lif.cep.util.Sets.SetUpdateProcessor.

Definition at line 600 of file Processor.java.

◆ getPullableInput()

Pullable ca.uqac.lif.cep.Processor.getPullableInput ( int i )

Returns the Pullable corresponding to the processor's i-th input.

Parameters

i	The index of the input. Should be greater than 0 and less than the processor's output arity. Outside these bounds, an ArrayIndexOutOfBounds will be thrown.

Returns: The pullable

Reimplemented in ca.uqac.lif.cep.GroupProcessor.

Definition at line 443 of file Processor.java.

◆ getPullableOutput() [1/2]

final Pullable ca.uqac.lif.cep.Processor.getPullableOutput ( )

Returns the Pullable corresponding to the processor's first output trace.

Returns: The pullable if the processor has at least one output. An ArrayIndexOutOfBounds will be thrown if the processor has an output arity of 0.

Definition at line 414 of file Processor.java.

◆ getPullableOutput() [2/2]

abstract Pullable ca.uqac.lif.cep.Processor.getPullableOutput ( int index )

abstract

Returns the Pullable corresponding to the processor's i-th output trace.

Parameters

index The index. Should be between 0 and the processor's output arity - 1 (since indices start at 0).

Returns: The pullable if the index is within the appropriate range, null otherwise.

Reimplemented in ca.uqac.lif.cep.AsynchronousProcessor, ca.uqac.lif.cep.functions.ApplyFunctionPartial, ca.uqac.lif.cep.GroupProcessor, ca.uqac.lif.cep.io.ReadInputStream, ca.uqac.lif.cep.SynchronousProcessor, ca.uqac.lif.cep.tmf.Divert, ca.uqac.lif.cep.tmf.Multiplex, ca.uqac.lif.cep.tmf.Pump, ca.uqac.lif.cep.tmf.Tank, ca.uqac.lif.cep.UniformProcessor, and ca.uqac.lif.cep.util.Lists.TimePack.

◆ getPushableInput() [1/2]

final Pushable ca.uqac.lif.cep.Processor.getPushableInput ( )

Returns the Pushable corresponding to the processor's first input trace.

Returns: The pushable if the processor has at least one input. An ArrayIndexOutOfBounds will be thrown if the processor has an input arity of 0.

Definition at line 389 of file Processor.java.

◆ getPushableInput() [2/2]

abstract Pushable ca.uqac.lif.cep.Processor.getPushableInput ( int index )

abstract

Returns the Pushable corresponding to the processor's i-th input trace.

Parameters

index The index. Should be between 0 and the processor's input arity - 1 (since indices start at 0).

Returns: The pushable if the index is within the appropriate range. Outside of the range,

Reimplemented in ca.uqac.lif.cep.AsynchronousProcessor, ca.uqac.lif.cep.functions.ApplyFunctionPartial, ca.uqac.lif.cep.GroupProcessor, ca.uqac.lif.cep.SynchronousProcessor, ca.uqac.lif.cep.tmf.Divert, ca.uqac.lif.cep.tmf.Multiplex, ca.uqac.lif.cep.tmf.Pump, ca.uqac.lif.cep.tmf.Tank, ca.uqac.lif.cep.tmf.TankLast, and ca.uqac.lif.cep.UniformProcessor.

◆ getPushableOutput()

Pushable ca.uqac.lif.cep.Processor.getPushableOutput ( int i )

Retrieves the Pushable associated to the processor's i-th output.

Parameters

i	The index of the output. Should be greater than 0 (not checked) and less than the processor's output arity. Outside these bounds, an ArrayIndexOutOfBounds will be thrown.

Returns: The pushable

Reimplemented in ca.uqac.lif.cep.GroupProcessor.

Definition at line 473 of file Processor.java.

◆ hashCode()

final int ca.uqac.lif.cep.Processor.hashCode ( )

Implementation of hashCode() specific to processors.

Every processor instance in BeepBeep is given a unique ID; even a clone of a processor (created using Processor#duplicate() will be identical to the original, except for this ID. This behavior cannot be overridden by descendants.

Definition at line 309 of file Processor.java.

◆ leftShift()

Pullable ca.uqac.lif.cep.Processor.leftShift ( int index )

Definition at line 1069 of file Processor.java.

◆ newContext()

final Context ca.uqac.lif.cep.Processor.newContext ( )

protected

Creates a new empty context map.

Returns: The context map

Definition at line 241 of file Processor.java.

◆ onEndOfTrace()

boolean ca.uqac.lif.cep.Processor.onEndOfTrace ( Queue< Object[]> outputs ) throws ProcessorException

protected

Allows to describe a specific behavior when the trace of input fronts has reached its end.

Called in "push mode" only. In "pull mode", implementing such a behavior can be done by using Pullable#hasNext() or Pullable#hasNextSoft().

Parameters

outputs A queue of arrays of objects. The processor should push arrays into this queue for every output front it produces. The size of each array should be equal to the processor's output arity, although this is not enforced.

Returns: true if the processor should output one or several output fronts, false otherwise and by default.

Exceptions

ProcessorException An exception thrown when a problem occurs with the operation

Reimplemented in ca.uqac.lif.cep.GroupProcessor, ca.uqac.lif.cep.tmf.AbstractSlice, ca.uqac.lif.cep.tmf.Decimate, ca.uqac.lif.cep.tmf.KeepLast, ca.uqac.lif.cep.tmf.SinkLast, ca.uqac.lif.cep.UniformProcessor, and ca.uqac.lif.cep.util.Lists.AbstractPack.

Definition at line 759 of file Processor.java.

◆ or() [1/3]

Processor ca.uqac.lif.cep.Processor.or ( CallAfterConnect c )

Operates similar to or(Processor), but also calls a method after the connection has been established.

Currently there is a single use for this method, which is when GroupProcessor#or(Processor) is called —which, again, typically only occurs in the context of a Groovy script.

Parameters

c	The object that contains the method to call

Returns: The processor encapsulated in c

Since: 0.11.4

Definition at line 1004 of file Processor.java.

◆ or() [2/3]

Processor ca.uqac.lif.cep.Processor.or ( Processor p )

Connects the first output pipe of this processor to the first input pipe of another processor.

Java programmers probably won't use this method, but users of the Groovy language can benefit from its operator overloading conventions, which map the construct p | q to p.or(q). This can be used to easily pipe two processors together:



def p = (some processor)

def q = (some other processor)

p | q // Connects p to q

Parameters

p	The other processor

Returns: The other processor

Since: 0.10.9

Definition at line 988 of file Processor.java.

◆ or() [3/3]

Processor ca.uqac.lif.cep.Processor.or ( SelectedInputPipe p )

Connects the output at index 0 of the current processor to the input of another processor.

Java programmers probably won't use this method. However, combined with the definition of getAt(int), users of the Groovy language can benefit from its operator overloading conventions, which map the construct p | q to p.or(q). This can be used to easily pipe two processors together:



def p = (some processor)

def q = (some other processor)

p | q[1] // Connects p to pipe index 1 of q

The above example works because q[1] returns q's input pushable for pipe index 1.

Parameters

p	The pushable object representing the input of the other processor to which the current output should be connected.

Returns: The other processor

Since: 0.10.9

Definition at line 1033 of file Processor.java.

◆ print()

final Object ca.uqac.lif.cep.Processor.print ( ObjectPrinter<?> printer ) throws ProcessorException

Prints the contents of this processor into an object printer.

Parameters

printer The printer to print this processor to

Returns: The printed processor

Since: 0.10.2

Definition at line 789 of file Processor.java.

◆ printState()

Object ca.uqac.lif.cep.Processor.printState ( )

protected

Produces an object that represents the state of the current processor.

A concrete processor should override this method to add whatever state information that needs to be preserved in the serialization process.

Returns: Any object representing the processor's state (including null)

Since: 0.10.2

Definition at line 827 of file Processor.java.

◆ read()

final Processor ca.uqac.lif.cep.Processor.read	(	ObjectReader<?>	reader,
		Object	o ) throws ProcessorException

Reads the content of a processor from a serialized object.

Parameters

reader	An object reader
o	The object to read from

Returns: The serialized processor

Exceptions

ProcessorException If the read operation failed for some reason

Definition at line 841 of file Processor.java.

◆ readState()

Processor ca.uqac.lif.cep.Processor.readState ( Object o )

protected

Reads the state of a processor and uses it to create a new instance.

Parameters

o	The object containing the processor's state

Returns: A new processor instance

Since: 0.10.2

Definition at line 898 of file Processor.java.

◆ reset()

void ca.uqac.lif.cep.Processor.reset ( )

Resets the processor.

This has for effect of flushing the contents of all input and output event queues. If the processor has an internal state, this should also reset this state to its "initial" settings (whatever that means in your context).

Definition at line 347 of file Processor.java.

◆ rightShift()

Pushable ca.uqac.lif.cep.Processor.rightShift ( int index )

Definition at line 1064 of file Processor.java.

◆ setContext() [1/2]

void ca.uqac.lif.cep.Processor.setContext ( Context context )

Adds a complete context to this object.

Parameters

context The context to add

Implements ca.uqac.lif.cep.Contextualizable.

Reimplemented in ca.uqac.lif.cep.GroupProcessor, and ca.uqac.lif.cep.tmf.Splice.

Definition at line 287 of file Processor.java.

◆ setContext() [2/2]

void ca.uqac.lif.cep.Processor.setContext	(	String	key,
		Object	value )

Adds an object to the object's context.

Parameters

key	The key associated to that object
value	The object

Implements ca.uqac.lif.cep.Contextualizable.

Reimplemented in ca.uqac.lif.cep.GroupProcessor, and ca.uqac.lif.cep.tmf.Splice.

Definition at line 275 of file Processor.java.

◆ setEventTracker()

Processor ca.uqac.lif.cep.Processor.setEventTracker ( EventTracker tracker )

Associates an event tracker to this processor.

Parameters

tracker The event tracker, or null to remove the association to an existing tracker

Returns: This processor

Reimplemented in ca.uqac.lif.cep.GroupProcessor, and ca.uqac.lif.cep.SynchronousProcessor.

Definition at line 688 of file Processor.java.

◆ setPullableInput()

void ca.uqac.lif.cep.Processor.setPullableInput	(	int	i,
		Pullable	p )

Assigns a Pullable to the processor's i-th input.

Parameters

i	The index of the input. An ArrayIndexOutOfBounds will be thrown if it is out of range.
p	The pullable to assign it to

Reimplemented in ca.uqac.lif.cep.GroupProcessor.

Definition at line 428 of file Processor.java.

◆ setPushableOutput()

void ca.uqac.lif.cep.Processor.setPushableOutput	(	int	i,
		Pushable	p )

Assigns a Pushable to the processor's i-th output.

Parameters

i	The index of the output. Should be greater than 0 and less than the processor's output arity. Outside these bounds, an ArrayIndexOutOfBounds will be thrown.
p	The pushable to assign it to

Reimplemented in ca.uqac.lif.cep.GroupProcessor, and ca.uqac.lif.cep.tmf.Divert.

Definition at line 458 of file Processor.java.

◆ start()

void ca.uqac.lif.cep.Processor.start ( )

Starts the processor.

This has no effect, except for processors that use threads; in such a case, calling this method should start the thread.

Reimplemented in ca.uqac.lif.cep.GroupProcessor, ca.uqac.lif.cep.io.SpliceSource, ca.uqac.lif.cep.tmf.Pump, and ca.uqac.lif.cep.util.Lists.TimePack.

Definition at line 621 of file Processor.java.

◆ startAll()

void ca.uqac.lif.cep.Processor.startAll ( Processor ... procs )

static

Starts all processors given as an argument.

Parameters

procs The processors

Definition at line 641 of file Processor.java.

◆ stop()

void ca.uqac.lif.cep.Processor.stop ( )

Stops the processor.

This has no effect, except for processors that use threads; in such a case, calling this method should stop the thread.

Reimplemented in ca.uqac.lif.cep.GroupProcessor, ca.uqac.lif.cep.io.Print, ca.uqac.lif.cep.io.SpliceSource, ca.uqac.lif.cep.tmf.Pump, ca.uqac.lif.cep.tmf.Splice, and ca.uqac.lif.cep.util.Lists.TimePack.

Definition at line 630 of file Processor.java.

◆ stopAll()

void ca.uqac.lif.cep.Processor.stopAll ( Processor ... procs )

static

Stops all processors given as an argument.

Parameters

procs The processors

Definition at line 658 of file Processor.java.

Member Data Documentation

◆ m_context

Context ca.uqac.lif.cep.Processor.m_context = null

protected

The context in which the processor is instantiated.

Definition at line 144 of file Processor.java.

◆ m_eventTracker

transient EventTracker ca.uqac.lif.cep.Processor.m_eventTracker = null

protected

An object that keeps track of the relationship between input and output events.

Definition at line 98 of file Processor.java.

◆ m_hasBeenNotifiedOfEndOfTrace

boolean [] ca.uqac.lif.cep.Processor.m_hasBeenNotifiedOfEndOfTrace

protected

Indicates whether the processor has been notified of the end of trace or not.

Each input pushable has its own flag, and the end of trace signal is propagated only once all upstream processors have sent the notification.

Definition at line 160 of file Processor.java.

◆ m_inputArity

int ca.uqac.lif.cep.Processor.m_inputArity

protected

The processor's input arity, i.e.

the number of input events it requires to produce an output

Definition at line 75 of file Processor.java.

◆ m_inputCount

int ca.uqac.lif.cep.Processor.m_inputCount = 0

protected

A counter incremented upon each input front processed.

Definition at line 122 of file Processor.java.

◆ m_inputPullables

transient Pullable [] ca.uqac.lif.cep.Processor.m_inputPullables

protected

An array of Pullables, one for each input trace this processor receives.

Definition at line 111 of file Processor.java.

◆ m_inputQueues

transient Queue<Object> [] ca.uqac.lif.cep.Processor.m_inputQueues

protected

An array of input event queues.

This is where the input events will be stored before the processor consumes them. There are as many input queues as the input arity of the processor.

Definition at line 92 of file Processor.java.

◆ m_notifiedEndOfTraceDownstream

boolean ca.uqac.lif.cep.Processor.m_notifiedEndOfTraceDownstream

protected

Indicates whether the processor has notified the end of the trace to the downstream processors it is connected to.

The end of trace signal should be sent at most once.

Definition at line 167 of file Processor.java.

◆ m_outputArity

int ca.uqac.lif.cep.Processor.m_outputArity

protected

The processor's output arity, i.e.

the number of output events it produces

Definition at line 80 of file Processor.java.

◆ m_outputCount

int ca.uqac.lif.cep.Processor.m_outputCount = 0

protected

A counter incremented upon each output front processed.

Definition at line 127 of file Processor.java.

◆ m_outputPushables

transient Pushable [] ca.uqac.lif.cep.Processor.m_outputPushables

protected

An array of Pushables, one for each output trace this processor produces.

Definition at line 117 of file Processor.java.

◆ m_outputQueues

transient Queue<Object> [] ca.uqac.lif.cep.Processor.m_outputQueues

protected

An array of output event queues.

This is where the output events will be stored when the processor does its computation. There are as many output queues as the output arity of the processor.

Definition at line 105 of file Processor.java.

◆ MAX_PULL_RETRIES

final transient int ca.uqac.lif.cep.Processor.MAX_PULL_RETRIES = 10000000

static

Number of times the Pullable#hasNext() method tries to produce an output from the input before giving up.

While in theory, the method tries "as long as necessary", in practice a bound was put on the number of attempts as a safeguard to avoid infinite loops.

Definition at line 152 of file Processor.java.

◆ s_versionString

final transient String ca.uqac.lif.cep.Processor.s_versionString = "0.11.2"

static

A string used to identify the program's version.

Definition at line 85 of file Processor.java.

The documentation for this class was generated from the following file:

/home/sylvain/Workspaces/beepbeep/core/src/ca/uqac/lif/cep/Processor.java

Detailed Description

Classes

Public Member Functions

Static Public Member Functions

Static Public Attributes

Protected Member Functions

Static Protected Member Functions

Protected Attributes

Constructor & Destructor Documentation

◆ Processor()

Member Function Documentation

◆ allNotifiedEndOfTrace()

◆ allNull()

◆ associateTo()

◆ associateToInput()

◆ associateToOutput()

◆ copyInputQueue()

◆ copyOutputQueue()

◆ duplicate() [1/2]

◆ duplicate() [2/2]

◆ duplicateInto()

◆ equals()

◆ getAt()

◆ getContext() [1/2]

◆ getContext() [2/2]

◆ getEmptyQueue()

◆ getEventTracker()

◆ getId()

◆ getInputArity()

◆ getInputCount()

◆ getInputQueue()

◆ getInputType()

◆ getInputTypesFor()

◆ getLeaves() [1/2]

◆ getLeaves() [2/2]

◆ getOutputArity()

◆ getOutputCount()

◆ getOutputQueue()

◆ getOutputType()

◆ getPullableInput()

◆ getPullableOutput() [1/2]

◆ getPullableOutput() [2/2]

◆ getPushableInput() [1/2]

◆ getPushableInput() [2/2]

◆ getPushableOutput()

◆ hashCode()

◆ leftShift()

◆ newContext()

◆ onEndOfTrace()

◆ or() [1/3]

◆ or() [2/3]

◆ or() [3/3]

◆ print()

◆ printState()

◆ read()

◆ readState()

◆ reset()

◆ rightShift()

◆ setContext() [1/2]

◆ setContext() [2/2]

◆ setEventTracker()

◆ setPullableInput()

◆ setPushableOutput()

◆ start()

◆ startAll()

◆ stop()

◆ stopAll()

Member Data Documentation

◆ m_context

◆ m_eventTracker

◆ m_hasBeenNotifiedOfEndOfTrace

◆ m_inputArity

◆ m_inputCount

◆ m_inputPullables

◆ m_inputQueues

◆ m_notifiedEndOfTraceDownstream

◆ m_outputArity

◆ m_outputCount

◆ m_outputPushables

◆ m_outputQueues