org.qbroker.node

Class MapReduceNode

java.lang.Object

org.qbroker.node.Node

org.qbroker.node.MapReduceNode

All Implemented Interfaces:

MessageNode

public class MapReduceNode
extends Node

MapReduceNode picks up JMS messages as the requests and creates multiple copies of the message as the requests according to the predefined rulesets. Those newly created messages will be sent to multiple outlinks. The node collects the responses from them and aggregates the results according to the reducing policies. If all the responses are collected and aggregated, the final result will be loaded to the incoming message as the response. The incoming message will be routed to the outlink of done. In case of failure, the incoming messages will be routed to the outlink of failure. If none of the rulesets matches the incoming messages, they will be put to the outlink of nohit. The rest of the outlinks are collectibles for responses.

MapReduceNode contains a number of predefined rulesets. These rulesets categorize messages into non-overlapping groups via the property filter. Therefore, each ruleset defines a unique message group. If the associated outlink is none of the fixed outlinks and Aggregation is defined in a ruleset, it is a ruleset of map-reduce. There may be other parameters for the map and reduce operations. For example, TimeToLive and Quorum are two important parameters for the reduce process.

The aggregation on responses is defined in BodyAggregation. It supports the first, last, append for text, and merge or union for JSON or XML payload, etc. In case of merge for JSON payload, JSONPath is required in the hash. JSONPath specifies 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 of the JSON document as the result. In case of merge for XML payload, XPath is required in the hash. XPath specifies 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 of the XML document as the result.

The default map process is to copy the message body as the request for each selected outlink. The selected outlinks are defined via the list of SelectedOutLink. If it is defined in a ruleset, its member can be either the name of an outlink or a map containing LinkTemplate and FieldName, etc. If it is not defined for a map-reduce rule, the request will be duplicated to all collectibles. If a member of SelectedOutLink is a map, the value of LinkTemplate will be used to determine the name of the outlink dynamically or statically, depending on whether it contains variables referencing to certain properties of the message. The value of FieldName will be the property name of the message for the request. In this case, the node formats the request out of message and sets the result to that property. Therefore, it may contain Template and Substitution. This way, each outlink may have its own request.

If SelectedOutLink is a string, the ruleset expects a JSONPath defined for selecting a list of items from the JSON payload. In this case, each selected content will be treated as a sub-request. All the sub-requests will be mapped to a single outlink specified by SelectedOutLink. All the properties will be copied over to each of the sub-request.

You are free to choose any names for the three fixed outlinks. But MapReduceNode always assumes the first outlink for done, the second for failure and the third for nohit. The rest of the outlinks are for responses. It is OK for those three fixed outlinks to share the same name. Please make sure the first fixed outlink has the actual capacity no less than that of the uplink.

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
MapReduceNode(java.util.Map props)

Method Summary

Modifier and Type	Method and Description
void	acknowledge(long[] state) acknowledges the request msg only
void	close() It closes all resources and cleans up all list, cache, metadata, etc.
protected void	finalize()
int	getOutLinkBoundary() returns -(BOUNDARY+1) to have the container enable ack regardlessly on the outlinks starting from BOUNDARY and beyond
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) 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.
void	resetMetaData(XQueue in, XQueue[] out) cleans up MetaData for all XQs and messages

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, getRuleName, getStatus, getXAMode, getXAMode, internalXQSupported, list, listPendings, passthru, refreshRule, renameRule, restore, rotateRule, setDebugMode, setDisplayMask, setDisplayProperties, setStatus, show, takeback, takebackEnabled, updateParameters

Methods inherited from class java.lang.Object

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

Constructor Detail

MapReduceNode

public MapReduceNode(java.util.Map props)

Method Detail

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

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

getOutLinkBoundary

public int getOutLinkBoundary()

returns -(BOUNDARY+1) to have the container enable ack regardlessly on the outlinks starting from BOUNDARY and beyond

Specified by:

getOutLinkBoundary in interface MessageNode

Overrides:

getOutLinkBoundary in class Node

resetMetaData

public void resetMetaData(XQueue in,
                          XQueue[] out)

cleans up MetaData for all XQs and messages

Specified by:

resetMetaData in interface MessageNode

Overrides:

resetMetaData in class Node

acknowledge

public void acknowledge(long[] state)
                 throws javax.jms.JMSException

acknowledges the request msg only

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