org.qbroker.node

Class DeliverNode

java.lang.Object

org.qbroker.node.Node

org.qbroker.node.DeliverNode

All Implemented Interfaces:

MessageNode

public class DeliverNode
extends Node

DeliverNode listens to an input XQ for JMS messages with an arbitray URI as their destinations. For each incoming message, DeliverNode looks up its URI in the local cache for any XQueue as the transmit queue established for the destination. If there is an XQueue in the cache for the destination, DeliverNode just puts the incoming message to the XQueue as the delivery. Otherwise, DeliverNode will generate a request with the URI and a newly created XQueue for the destination. The request will be sent to the outlink connected to a PersisterPool for a new persister thread. Once the persister is instantiated, DeliverNode will collect the response from the PersisterPool. The response is supposed to contain the persister thread for delivering the messages to the specific destination. DeliverNode caches the thread and the id of the XQueue using the URI as the key. Therefore, DeliverNode is able to deliver JMS messages to arbitrary destinations based to their URIs and the predefined rulesets on-demand.

It has two types of outlinks, position-fixed and non-fixed. There are three position-fixed outlinks: pool for all requests to the PersisterPool, failure for the messages failed in the delivery process, nohit for those messages not covered by any rulesets. The non-fixed outlinks are transmit queues for the on-demand destinations.

DeliverNode also 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 way to construct URI and properties for the new persisters. For those messages falling off all defined rulesets, DeliverNode always creates an extra ruleset, nohit, to handle them. Therefore all the nohit messages will be routed to nohit outlink. The downstream node at nohit is supposed to handle all nohit messages.

URI is used to identify destinations. In order to construct the URI for an arbitrary destination, each ruleset contains two sets of format operations. The first one is the array of templates with the name of URITemplate. The other is the array of substitutions with the name of URISubstitution. URITemplate appends the previous non-empty text to URI and sets the next initial text for its corresponding URISubstitutions to modify. The associations between the URITemplates and URISubstitutions are based on their positions. Either URITemplate or URISubstitution can be null for no action and a place holder. Therefore, you can insert multiple null URITemplates so that the associated URISubstitutions will be able to modify the same text in turns. If any of the operations fails, the message will be routed to failure outlink. Besides, a ruleset may contain a hash of DefaultProperty as static or dynamic properties for the new persisters. If the URI string is not in the cache, DeliverNode will try to resolve all the dynamic variables in the hash of DefaultProperty from the incoming message first. Then it adds the DefaultProperty to the new request for the new persister. So the extra properties are able to passed over to the persister pool.

For each new destination, DeliverNode creates an Object message as the request containing the URI and the XQueue, as well as the other properties provided they are defined. The request is sent to the PersisterPool via the pool outlink. Then DeliverNode frequently checks the response for each outstanding requests. The response is supposed to have the status and the persister thread for the new destination. DeliverNode will use the thread to monitor its status. If the response does not have the thread, DeliverNode will route the messages to the failure outlink and remove the XQueue and the URI from the cache. The samething will happen if the request for a new persister times out. MaxRetry is used to control when to timeout the request to the pool. It also controls the timeout on a dead persister thread.

DeliverNode also maintains an active set of XQueues as the transmit queues for all destinations. Behind each XQueue, there is at least one persister thread delivering the messages to the destinations. The messages may be stuck in the XQueue until they are delivered, as long as the persister is not stopped. However, if there is no message in the queue, DeliverNode will mark it idle. All the transmit queues are monitored frequently in every heartbeat. If one of them has been idle for over MaxIdleTime, its queue will be stopped. Its persister thread and the transmit queue will be removed from the cache.

You are free to choose any names for the three fixed outlinks. But DeliverNode always assumes the first outlink for pool, the second for failure and the third for nohit. The name for nohit outlink is allowed to be same as that of failure outlink. But neither nohit nor failure is allowed to share their names with the pool. The rest of the outlinks are for the on-demand destinations.

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

Method Summary

Modifier and Type	Method and Description
void	callback(java.lang.String key, java.lang.String id) It marks the cell in cells with key and id.
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()
int	getOutLinkBoundary() returns 0 to have the ack propagation skipped on the first outlink
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.
boolean	internalXQSupported() always returns true since it supports dynamic XQs
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
void	resetMetaData(XQueue in, XQueue[] out) cleans up MetaData for all XQs and messages
protected boolean	takebackEnabled()
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, checkpoint, collectAll, containsRule, containsXQ, forcethru, getCapacity, getCapacity, getDebugMode, getDisplayMask, getDisplayProperties, getLinkName, getLinkName, getMetaData, getName, getNumberOfOutLinks, getNumberOfRules, getOperation, getRuleName, getStatus, getXAMode, getXAMode, list, listPendings, refreshRule, removeRule, renameRule, replaceRule, restore, rotateRule, setDebugMode, setDisplayMask, setDisplayProperties, setStatus, show, takeback

Methods inherited from class java.lang.Object

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

Constructor Detail

DeliverNode

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

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

callback

public void callback(java.lang.String key,
                     java.lang.String id)

It marks the cell in cells with key and id.

Overrides:

callback in class Node

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.

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

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

internalXQSupported

public boolean internalXQSupported()

always returns true since it supports dynamic XQs

Specified by:

internalXQSupported in interface MessageNode

Overrides:

internalXQSupported in class Node

takebackEnabled

protected boolean takebackEnabled()

Overrides:

takebackEnabled in class Node

getOutLinkBoundary

public int getOutLinkBoundary()

returns 0 to have the ack propagation skipped on the first outlink

Specified by:

getOutLinkBoundary in interface MessageNode

Overrides:

getOutLinkBoundary 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