org.qbroker.node

Class DistinguishNode

java.lang.Object

org.qbroker.node.Node

org.qbroker.node.DistinguishNode

All Implemented Interfaces:

MessageNode

public class DistinguishNode
extends Node

DistinguishNode picks up JMS messages from an input XQ and distinguishes them according to their keys and optional versions. It filters them into four outlinks: done for all messages of newer versions, bypass for those messages either duplicate or older versions, failure for those messages failed in the distinguishing processes, and nohit for all the messages that do not hit any explicitly defined ruleset.

DistinguishNode contains a number of predefined rulesets. These rulesets categorize messages into non-overlapping groups. Therefore, each rule defines a unique message group. The ruleset also specifies the parameters to control the message filtering process on the group, such as KeyTemplate. If KeyTemplate is defined for a ruleset, it is the ruleset for distinction. DistinguishNode maintains a global cache for all the keys in order to track them. Furthermore, the content filtering can be enabled or disabled dynamically via the sessions and monitors. If SessionTimeout is set to 0, it will only do static filtering. The number of the keys cached is tracked via the RULE_PEND field for each rule. If VersionTemplate is defined in a ruleset, DistinguishNode will retrieve the version info from the messages to enforce the version control. It means the messages with older versions will be treated as the duplicates. DistinguishNode also creates one extra ruleset by default for all nohit messages. The reserved word, nohit, is the name of this default ruleset.

DistinguishNode also supports the cache ruleset which is similar to a regular distinction ruleset. The only difference on the configurations is that the former has also defined RuleType as "cache". However, the routing policy for a cache ruleset is different from what of the distinction ruleset. It is designed for message flow clusters only. Even though both distinction rulesets and cache rulesets are defined for the node, their activities are mutually exclusive on any given instance of the node. Therefore, a DistinguishNode with active cache rulesets is called on the cache mode.

One of the use cases of cache rulesets is that an active-passive message flow cluster with the only one DistinguishNode. In this case, the master flow processes all the incoming messages. By desing, those messages only activate the distinction rulesets of the DistinguishNode. As the contrast, the worker flow does not process any incoming messages. Therefore, the distinction rulesets of the DistinguishNode are not active at all. Meanwhile, the master flow also escalates each processed message to the worker flow via the cluster communication channels. On the worker side, the container of the worker flow is supposed to forward the escalated messages to the DistinguishNode. Those escalated messages are supposed to activate the cache rulesets of the DistinguishNode. Once an escalated message arrives, the DistinguishNode on the cache mode extracts the key from the message and saves it to the global cache. The cache ruleset on the worker flow has a different routing policy. Since those messages have already been processed on the master, the messages with the unique keys will be routed to nohit outlink. The rest of the messages will be routed to failure outlink, if there is any. This way the cache on the worker flow always keeps in sync with the master. When a failover occurs, the worker flow with the latest cache will be ready to be promoted to the new master. Since the cache on the new master is current, any duplicated message will not be routed to the done outlink. If you are going to implement the active-passive message flow cluster with a DistinguishNode with cache rules, please make sure its uplink has the 2nd half of the cells unused. The unused partition will be reserved for escalations so that the container is able to forward the escalations.

DistinguishNode also supports checkpointing of the internal data cache via the container. In order to checkpoint, the CheckpointDir must be defined on the container level.

You are free to choose any names for the four fixed outlinks. But DistiguishNode always assumes the first outlink for done, the second for bypass, the third for failure and the last for nohit. The outlink of failure or nohit can share the same name with the first two others. It means these outlinks may share 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
DistinguishNode(java.util.Map props)

Method Summary

Modifier and Type	Method and Description
java.util.Map<java.lang.String,java.lang.Object>	checkpoint() returns the checkpoint hashmap for current state
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.
java.util.Map<java.lang.String,java.lang.String>	listPendings(XQueue xq, int type) lists all propagating messages and cached keys
void	propagate(XQueue in, XQueue[] out) picks up a message from input queue and evaluate 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.
void	restore(java.util.Map<java.lang.String,java.lang.Object> chkpt) restores the state from checkpoint hashmap
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, collectAll, containsRule, containsXQ, feedback, forcethru, getCapacity, getCapacity, getDebugMode, getDisplayMask, getDisplayProperties, getLinkName, getLinkName, getMetaData, getName, getNumberOfOutLinks, getNumberOfRules, getOperation, getOutLinkBoundary, getRuleName, getStatus, getXAMode, getXAMode, internalXQSupported, list, passthru, refreshRule, renameRule, replaceRule, resetMetaData, 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

DistinguishNode

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

propagate

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

picks up a message from input queue and evaluate 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

listPendings

public java.util.Map<java.lang.String,java.lang.String> listPendings(XQueue xq,
                                                                     int type)

lists all propagating messages and cached keys

Specified by:

listPendings in interface MessageNode

Overrides:

listPendings in class Node

checkpoint

public java.util.Map<java.lang.String,java.lang.Object> checkpoint()

returns the checkpoint hashmap for current state

Specified by:

checkpoint in interface MessageNode

Overrides:

checkpoint in class Node

restore

public void restore(java.util.Map<java.lang.String,java.lang.Object> chkpt)

restores the state from checkpoint hashmap

Specified by:

restore in interface MessageNode

Overrides:

restore 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