Package org.opennms.core.ipc.sink.aggregation

Class MappingAggregationPolicy<S,T extends Message>

java.lang.Object

org.opennms.core.ipc.sink.aggregation.MappingAggregationPolicy<S,T>

All Implemented Interfaces:

AggregationPolicy<S,T,T>

public abstract class MappingAggregationPolicy<S,T extends Message>
extends Object
implements AggregationPolicy<S,T,T>

An aggregation policy that performs a simple map operation on the given message.

Author:

jwhite

Constructor Summary

Constructors

Constructor	Description
MappingAggregationPolicy()

Method Summary

Modifier and Type	Method	Description
T	aggregate(T oldBucket, S newMessage)	Aggregate the given message into an existing accumulator, or create a new accumulator if no accumulator exists.
T	build(T accumulator)	Build the resulting message from the accumulator.
int	getCompletionIntervalMs()	Maximum number of milliseconds for which buckets should continue accumulating messages after creation.
int	getCompletionSize()	Maximum number of messages to be added to a bucket before dispatching.
Object	key(S message)	Calculate a key for the given message.
abstract T	map(S message)

Methods inherited from class java.lang.Object

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

Constructor Detail

MappingAggregationPolicy

public MappingAggregationPolicy()

Method Detail

map

public abstract T map(S message)

key

public Object key(S message)

Description copied from interface: AggregationPolicy

Calculate a key for the given message. Objects with the same key will be aggregated together. Returned values should non-null.

Specified by:

key in interface AggregationPolicy<S,T extends Message,T extends Message>

Parameters:

message - the message

Returns:

the message's key

getCompletionSize

public int getCompletionSize()

Description copied from interface: AggregationPolicy

Maximum number of messages to be added to a bucket before dispatching. If this value is <= 1, the buckets should be dispatched immediately after adding a single element.

Specified by:

getCompletionSize in interface AggregationPolicy<S,T extends Message,T extends Message>

Returns:

maximum number of messages per bucket

getCompletionIntervalMs

public int getCompletionIntervalMs()

Description copied from interface: AggregationPolicy

Maximum number of milliseconds for which buckets should continue accumulating messages after creation. If a bucket has been created for longer than this interval, it will be dispatched regardless of it's current size. Values <= 0 will disable periodic flushing and buckets will only be dispatched once they have reached the maximum size.

Specified by:

getCompletionIntervalMs in interface AggregationPolicy<S,T extends Message,T extends Message>

Returns:

number of milliseconds to keep a bucket before dispatching

aggregate

public T aggregate(T oldBucket,
                   S newMessage)

Description copied from interface: AggregationPolicy

Aggregate the given message into an existing accumulator, or create a new accumulator if no accumulator exists.

Specified by:

aggregate in interface AggregationPolicy<S,T extends Message,T extends Message>

Parameters:

oldBucket - the existing accumulator, or null if a new accumulator should be created

newMessage - the message to aggregate

Returns:

the new or updated accumulator

build

public T build(T accumulator)

Description copied from interface: AggregationPolicy

Build the resulting message from the accumulator.

Specified by:

build in interface AggregationPolicy<S,T extends Message,T extends Message>

Parameters:

accumulator - an existing accumulator

Returns:

the aggregated message to dispatch