org.qbroker.node

Class PublishNode

java.lang.Object

org.qbroker.node.Node

org.qbroker.node.PublishNode

All Implemented Interfaces:

MessageNode

public class PublishNode
extends Node

PublishNode picks up a JMS messages from the input XQueue and extracts the topic string from it. It looks up the topic cache for a cached list of subscriptions on the topic. If the list is not empty, PublishNode just loops through the list and creates a copy of the message for each subscription. A subscription contains a template to build the unique URI for delivery. PublishNode will set the URIs to the subscribed messages before flushing them downstream.

There are four fixed outlinks, pool, bypass, failure and nohit. All subscribed copies of incoming message will be routed to the outlink of pool. Their original incoming message will be routed to bypass. Any failure will cause the incoming message routed to the outlink of failure. If there is no subscriptions for the topic, the incoming message will be routed to the outlink of nohit. Since there may be multiple subscriptions for an incoming message, please make sure the outlink of pool has the capacity no less than the maximum number of subscriptions.

PublishNode contains a number of rulesets. Each ruleset defines a unique subscription with a name, a group of property patterns, and a URI template for delivery. It may have an active time window for blackout. The number of the rulesets and their content may change dynamically on demand.

PublishNode always creates two extra rulesets. The first one is the nohit ruleset for those messages whose topics have no subscribers. The second is the candidate ruleset for all messages whose topic has at least one subscriber. Since a candidate message may hit number of subscriptions and generates a message copy for every hit subscriptions, the stats count for the candidate ruleset will be the number of the incoming messages rather than the number of the copies. The number of message copies will be counted by their own rulesets. The DisplayMask and StringProperty of the ruleset are used to display the details of dispatched messages for the ruleset. The stats of the rules are stored in the fields of RULE_SIZE, RULE_COUNT and RULE_PEND. RULE_PEND is for number of cached topics, RULE_SIZE for number of preliminary hits, whereas RULE_COUNT is for number of real hits.

SessionTimeout determines how often to clean up expired topics from the cache. If SessionTimeout is larger than zero, any cached topics will be expired if their idle time exceeds SessionTimeout. Those expired topics will be removed from the cache in next session to save resources. TopicTemplate defines a template with multiple property names delimited by a space char. It is required for extracting topic string from a message. URIField specifies where to store the delivery URI in the subscribed messages. By default, it is set to pubURI.

You are free to choose any names for all four outlinks. But PublishNode always assumes the first outlink for pool, the second for bypass, the third for failure, the last for nohit. The names of pool has to be different from other outlinks. The names of bypass, failure and nohit can be same.

Author:

yannanlu@yahoo.com

Field Summary

Fields

Modifier and Type	Field and Description
protected int	sessionTimeout

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

Method Summary

Modifier and Type	Method and Description
int	addRule(java.util.Map ph) It adds a new rule to the ruleList and returns its id upon success.
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.
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.
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

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

Field Detail

sessionTimeout

protected int sessionTimeout

Constructor Detail

PublishNode

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

addRule

public int addRule(java.util.Map ph)

It adds a new rule to the ruleList and returns its id upon success. It is not MT-Safe.

Specified by:

addRule in interface MessageNode

Overrides:

addRule 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. There is no RULE_SIZE checking.

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

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