org.qbroker.node

Class PickupNode

java.lang.Object

org.qbroker.node.Node

org.qbroker.node.PickupNode

All Implemented Interfaces:

MessageNode

public class PickupNode
extends Node

PickupNode listens to an input XQ for JMS messages with an arbitrary URI as a request to retrieve messages. For each incoming messages, the URI identifies the data source. PickupNode looks it up in the local cache for any XQueue as the transmit queue established for the data source. If there is an XQueue in the cache for the data source, PickupNode just packs up all the messages in the XQueue into the incoming message body and sends it back as the response. Otherwise, PickupNode will generate a new request with the URI and a newly created XQueue for the data source. The request will be sent to the outlink connected to a ReceiverPool for a new receiver thread. Once the receiver is instantiated, PickupNode will collect the response from the ReceiverPool. The response is supposed to contain the receiver thread for receiving the messages from the specific data source. PickupNode caches the thread and the id of the XQueue with the URI as the key. Therefore, PickupNode is able to receive JMS messages via the receivers from arbitrary data sources based on their URIs and the predefined rulesets on-demand. The received messages will be cached in their XQueues for picking up until they expires.

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

PickupNode 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 receivers. For those messages falling off all defined rulesets, PickupNode 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 data sources. In order to construct the URI for an arbitrary data source, 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 receivers. If the URI string is not in the cache, PickupNode 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 receiver. So the extra properties are able to passed over to the receiver pool.

For each new data source, PickupNode 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 ReceiverPool via the pool outlink. Then PickupNode frequently checks the response for each outstanding requests. The response is supposed to have the status and the receiver thread for the new data source. PickupNode will use the thread to monitor its status. If the response does not have the thread, PickupNode 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 receiver times out. MaxRetry is used to control when to timeout the request to the pool. It also controls the timeout on a dead receiver thread.

PickupNode also maintains an active set of XQueues as the receiving queues for all data sources. In front of each XQueue, there is at least one receiver thread receiving the messages from the data sources. The messages are available for applications to pick up. If PickupNode has not got any pickup requests for certain queues, it will mark them idle, no matter how many messages available. All dynamic queues are monitored frequently in every heartbeat. If one of them has been idle for over MaxIdleTime, the queue will be stopped. Its receiver thread and the queue as well as the leftover messages will also be removed from cache.

You are free to choose any names for the three fixed outlinks. But PickupNode 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 pool. The rest of the outlinks are for the on-demand data sources.

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
PickupNode(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()
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
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, callback, checkpoint, collectAll, containsRule, containsXQ, feedback, forcethru, getCapacity, getCapacity, getDebugMode, getDisplayMask, getDisplayProperties, getLinkName, getLinkName, getMetaData, getName, getNumberOfOutLinks, getNumberOfRules, getOperation, getRuleName, getStatus, getXAMode, getXAMode, list, listPendings, passthru, 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

PickupNode

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

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