org.qbroker.node

Class AggregateNode

java.lang.Object

org.qbroker.node.Node

org.qbroker.node.AggregateNode

All Implemented Interfaces:

MessageNode

public class AggregateNode
extends Node

AggregateNode aggregates JMS messages according to their content and the predefined rulesets. It filters them into three outlinks: bypass for all the aggregated messages, nohit for those messages do not belong to any rulesets, failure for the messages failed at the aggregations. AggregateNode will create a new message, a TextMessage, as the result of the aggregation for each unique key in every session. These new messages will be flushed to the outlink of done when their sessions terminate. They will be formatted right before the flush if the post formatter are defined.

AggregateNode contains a number of predefined rulesets. These rulesets categorize messages into non-overlapping groups. Therefore, each ruleset defines a unique message group. A ruleset also defines the unique keys for the group and how to aggregate the messages sharing the same keys. Each group has its own cache for aggregations. The aggregation process is maintained by sessions with certain lifetime. For each unique key, there is a cached message with aggregated data. A ruleset also defines parameters used for the aggregation, such as DynamicSession, TimeToLive, SessionSize, CountField and TerminatorField, etc. The lifetime of a session is determined by TimeToLive in seconds. Therefore, AggregateNode knows when to terminate an aggregation session. However, a session can belong to a key or all the keys in a ruleset. If DynamicSession is set to true on a ruleset, each key of that ruleset will have its own session. When the session is terminated, only the cached message for that key will be flushed to the outlink of done. By default, DynamicSession is set to false. It means all the keys in a ruleset will share the same session by default. In this case, if the session is terminated, all the cached messages for that ruleset will be flushed out at the same time.

The aggregation on message body is defined separately in BodyAggregation. It supports the first, last, append for text, and merge for JSON or XML payload. In case of merge for JSON payload, JSONPath is required in the hash. JSONPath defines a JSONPath expression that is used to select the data from the responses. The selected data will be merged as a child node to the parent node in the aggregated JSON. In case of merge for XML payload, XPath is required in the hash. XPath defines an XPath expression that is used to select the data from the responses. The selected data will be merged as a child node to the parent node in the aggregated XML.

Currently, AggregateNode supports 4 different ways to terminate a session. First, TimeToLive specifies the lifetime of any key in the cache for the ruleset. If a key expires in the cache, its session is terminated. The expired keys will not participate the aggregation process. But they may not be flushed out until the next flush operation. Second, if the total number of cached keys in a ruleset exceeds the SessionSize, AggregateNode will try to find any expired keys to flush to save space. If CountField is defined in a ruleset, AggregateNode will retrieve the total count from the incoming messages. The total count will be used to terminate the aggration session on the key. If TerminatorField is defined, AggregateNode will check that property on each incoming message. If an incoming message has a non-empty value on that property, it will be treated as the terminator. The aggregation session for the key will be terminated by the terminator. In this case, if the DynamicSession is enabled and CopyOver is set to true, AggregateNode will copy the aggregated result over to the terminate message.

AggregateNode also supports plugins for customized aggregations. In order to support plugins, ClassName of the implementation should be defined in the ruleset. The class should have a public method of aggregate() that takes two JMS messages as the arguments. The first message is the incoming one. The second message is the cached message for aggregations. The method is supposed to return null for success or an error text for failures. It should not modify the incoming message by all means.

Apart from user defined rulesets, AggregateNode always creates one extra ruleset, nohit. The ruleset of nohit is for those messages not hitting any matches. The number of aggregation messages is storded into the RULE_PEND field.

AggregateNode also supports single key sessions. In order to enable the single key session, SessionSize has to be set to 1. In this case, AggregateNode is assuming that all keys of input messages are already grouped in packs. It means all messages with the same key stay together. Any new key will terminate the previous aggregation session.

You are free to choose any names for the four fixed outlinks. But AggregateNode always assumes the first outlink for done, the second for bypass group, the third for failure and the last for nohit. Any two or more outlinks can share the same outlink name. It means these outlinks are sharing the same output channel.

Author:

yannanlu@yahoo.com

Field Summary

Fields inherited from class org.qbroker.node.Node

assetList, bufferSize, callbackMethod, capacity, cbw, cells, cfgList, chkptExpiration, debug, DEBUG_COLL, DEBUG_DIFF, DEBUG_FBAK, DEBUG_INIT, DEBUG_PASS, DEBUG_PROP, DEBUG_REPT, DEBUG_UPDT, displayMask, displayPropertyName, linkName, LOAD_HIGH, LOAD_LOW, LOAD_MEDIUM, msgList, name, operation, RESET_ALL, RESET_MAP, RESET_NONE, RESET_SOME, ruleList, status, TYPE_ACTION, TYPE_AGGREGATE, TYPE_BYPASS, TYPE_CACHE, TYPE_COLLECT, TYPE_DELIVER, TYPE_FORMAT, TYPE_JSONPATH, TYPE_JSONT, TYPE_MAPREDUCE, TYPE_NONE, TYPE_PARSER, TYPE_PICKUP, TYPE_PIPE, TYPE_PUBLISH, TYPE_ROUNDROBIN, TYPE_SCREEN, TYPE_SCRIPT, TYPE_SERVICE, TYPE_SORT, TYPE_SPLIT, TYPE_SPREAD, TYPE_WEIGHTED, TYPE_XPATH, TYPE_XSLT, waitTime, xaMode

Fields inherited from interface org.qbroker.node.MessageNode

ASSET_THR, ASSET_URI, ASSET_XQ, META_MAP, META_MSG, META_OUT, META_RULE, META_XQ, MSG_BID, MSG_CID, MSG_OID, MSG_RID, MSG_TID, MSG_TIME, NODE_CLOSED, NODE_DISABLED, NODE_PAUSE, NODE_READY, NODE_RETRYING, NODE_RUNNING, NODE_STANDBY, NODE_STOPPED, OUT_CAPACITY, OUT_COUNT, OUT_DEQ, OUT_EXTRA, OUT_LENGTH, OUT_MODE, OUT_NRULE, OUT_OFFSET, OUT_ORULE, OUT_QDEPTH, OUT_QIPPS, OUT_QSTATUS, OUT_QTIME, OUT_SIZE, OUT_STATUS, OUT_TIME, RULE_COUNT, RULE_DMASK, RULE_EXTRA, RULE_GID, RULE_MODE, RULE_OID, RULE_OPTION, RULE_PEND, RULE_PID, RULE_SIZE, RULE_STATUS, RULE_TIME, RULE_TTL

Constructor Summary

Constructors

Constructor and Description
AggregateNode(java.util.Map props)

Method Summary

Modifier and Type	Method and Description
void	close() It closes all resources and cleans up all list, cache, metadata, etc.
protected void	finalize()
protected java.util.Map<java.lang.String,java.lang.Object>	initRuleset(long tm, java.util.Map ph, long[] ruleInfo) It initializes a new ruleset with the ruleInfo and returns the rule upon success.
void	propagate(XQueue in, XQueue[] out) It picks up a message from input queue and evaluates its content to decide which output queue to propagate
int	removeRule(java.lang.String key, XQueue in) It removes the rule from the ruleList and returns the rule id upon success.
int	replaceRule(java.lang.String key, java.util.Map ph, XQueue in) It replaces the existing rule of the key and returns its id upon success.
int	updateParameters(java.util.Map props) It updates the parameters of the node with the given property map.

Methods inherited from class org.qbroker.node.Node

addRule, callback, checkpoint, collectAll, containsRule, containsXQ, feedback, forcethru, getCapacity, getCapacity, getDebugMode, getDisplayMask, getDisplayProperties, getLinkName, getLinkName, getMetaData, getName, getNumberOfOutLinks, getNumberOfRules, getOperation, getOutLinkBoundary, getRuleName, getStatus, getXAMode, getXAMode, internalXQSupported, list, listPendings, passthru, refreshRule, renameRule, resetMetaData, restore, rotateRule, setDebugMode, setDisplayMask, setDisplayProperties, setStatus, show, takeback, takebackEnabled

Methods inherited from class java.lang.Object

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

Constructor Detail

AggregateNode

public AggregateNode(java.util.Map props)

Method Detail

updateParameters

public int updateParameters(java.util.Map props)

Description copied from class: Node

It updates the parameters of the node with the given property map. Upon success, it returns the number of parameters modified. It is not MT-Safe.

Specified by:

updateParameters in interface MessageNode

Overrides:

updateParameters in class Node

initRuleset

protected java.util.Map<java.lang.String,java.lang.Object> initRuleset(long tm,
                                                                       java.util.Map ph,
                                                                       long[] ruleInfo)

It initializes a new ruleset with the ruleInfo and returns the rule upon success. Otherwise, it throws an exception or returns null.

Specified by:

initRuleset in class Node

removeRule

public int removeRule(java.lang.String key,
                      XQueue in)

It removes the rule from the ruleList and returns the rule id upon success. It is not MT-Safe.

Specified by:

removeRule in interface MessageNode

Overrides:

removeRule in class Node

replaceRule

public int replaceRule(java.lang.String key,
                       java.util.Map ph,
                       XQueue in)

It replaces the existing rule of the key and returns its id upon success. It is not MT-Safe.

Specified by:

replaceRule in interface MessageNode

Overrides:

replaceRule in class Node

propagate

public void propagate(XQueue in,
                      XQueue[] out)
               throws javax.jms.JMSException

It picks up a message from input queue and evaluates its content to decide which output queue to propagate

Specified by:

propagate in interface MessageNode

Specified by:

propagate in class Node

Throws:

javax.jms.JMSException

close

public void close()

Description copied from class: Node

It closes all resources and cleans up all list, cache, metadata, etc. Once it is closed, it can not be used any more.

Specified by:

close in interface MessageNode

Overrides:

close in class Node

finalize

protected void finalize()

Overrides:

finalize in class Node