org.qbroker.node

Class ActionNode

java.lang.Object

org.qbroker.node.Node

org.qbroker.node.ActionNode

All Implemented Interfaces:

MessageNode

public class ActionNode
extends Node

ActionNode processes incoming JMS messages and takes the predefined actions according to the rulesets and the content of the incoming messages. As the escalation of actions, ActionNode may generate a new JMS message. All the escalation messages will be routed to the outlink of done. For all the incoming messages, ActionNode routes them into three outlinks: bypass for all the processed incoming messages, nohit for those messages do not belong to any rulesets, failure for the incoming messages failed in the process. However, the outlink of done can be shared with other outlinks.

ActionNode contains a number of predefined rulesets. These rulesets categorize incoming messages into non-overlapping groups. Therefore, each rule defines a unique message group. The ruleset also defines its actions and the rules to generate new messages for the group. Different groups may have different actions and/or different parameters. For an escalation message, ActionOrder determines its delivery order relative to the incoming message. ActionOrder takes one of the three values: "none", "first" and "last". By default, it is set to "none" and there is no order defined on the newly generated messages. If it is set to "first", ActionNode will send out the new message first and wait for its delivery. Then it will have the incoming message going through the bypass outlink. If it is set to "last", the incoming message will go through the bypass outlink first. Upon the completion of its delivery, ActionNode sends the escalation message through the outlink of done. In case of "first" or "last", the ruleset also allows ActionDelay defined in millisec. It specifies how long the escalation gets delayed for the ruleset of "last". For the ruleset of "first", the delay happens after the escalation. EscalationMask controls the display mask on escalation messages.

Apart from the user defined rulesets, ActionNode always creates one extra ruleset, nohit. The ruleset of nohit is for those messages not hitting any patterns. Please remember, there is no one-to-one relationship between the candidate messages and the escalation messages. The total number of escalation messages will be stored into the RULE_PEND field of the rulesets.

Each ruleset contains the name of the rule, the classname of the action and a Map of ActionArgument. The ActionArgument has the configuration parameters for the constructor. If the classname is not defined in a rule, the default action will be used so that it just converts the msg into a different JMS message type, including the same type for copy. In case of the same type for the escalation, the rule can have a formatter to format the escalation messages. On the other hand, developers can plugin their own actions. The requirement is minimum. The class should have a public method of escalate() that takes a JMS Message as the only argument and returns a Java Object. The returned object must be null for bypass or the new JMS message for escalations. If something is wrong, the returned object should be a String as the text of the error.

The plugin must also have a constructor taking a Map, or a List, or just a String as the only argument for configurations. Based on the data type of the constructor argument, developers should define configuration parameters in the base of ActionArgument. ActionNode will pass the data to the plugin's constructor as an opaque object during the instantiation of the plugin. In the normal operation, ActionNode will invoke the method to process the incoming messages. It is OK for the method to modify the incoming messages. But it is not recommented to acknowledge them. If any of the operations fails, the incoming message will be routed to failure outlink.

In case a plugin needs to connect to external resources for dynamic escalations, it should define an extra method of close() to close all the external resources gracefully. Its method of escalate() should also be able to detect the disconnections and cleanly reconnect to the resources automatically. If the container wants to stop the node, it will call the resetMetaData() in which the methods of close() of all actions will be called in order to release all external resources.

You are free to choose any names for the four fixed outlinks. But ActionNode always assumes the first outlink for done, the second for bypass, 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
ActionNode(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 int	feedback(XQueue in, long milliSec) It returns the number of done messages removed from the input XQueue.
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.
protected int	passthru(long currentTime, javax.jms.Message msg, XQueue in, int rid, int oid, int cid, int tid) It passes the message from the input XQueue over to an output XQueue and returns 1 upon success or 0 otherwise.
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.

Methods inherited from class org.qbroker.node.Node

addRule, callback, checkpoint, collectAll, containsRule, containsXQ, forcethru, getCapacity, getCapacity, getDebugMode, getDisplayMask, getDisplayProperties, getLinkName, getLinkName, getMetaData, getName, getNumberOfOutLinks, getNumberOfRules, getOperation, getOutLinkBoundary, getRuleName, getStatus, getXAMode, getXAMode, internalXQSupported, list, listPendings, refreshRule, renameRule, resetMetaData, 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

ActionNode

public ActionNode(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

passthru

protected int passthru(long currentTime,
                       javax.jms.Message msg,
                       XQueue in,
                       int rid,
                       int oid,
                       int cid,
                       int tid)

It passes the message from the input XQueue over to an output XQueue and returns 1 upon success or 0 otherwise. In case of cid less than 0, the message has nothing to do with the input XQueue. Nothing will be added to msgList for tracking. No callbacks and no update on rule stats either. If tid is less than 0, message will not be put back to uplink in case of failure. There is no callbacks and no update on rule stats either. In case of that cid is larger than or equal to 0 and tid is less than 0, it will add msg stats to msgList for external tracking. But the end user has to clean it up after the usage since it is not collectible by feedback() without the callback.

Overrides:

passthru in class Node

feedback

protected int feedback(XQueue in,
                       long milliSec)

It returns the number of done messages removed from the input XQueue. If milliSec is less than 0, there is no wait and it tries to collect all cells. Otherwise, it just tries to collect the first collectible cell.

Overrides:

feedback in class Node

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