org.qbroker.node

Class EventDispatcher

java.lang.Object

org.qbroker.node.Node

org.qbroker.node.EventDispatcher

All Implemented Interfaces:

MessageNode

public class EventDispatcher
extends Node

EventDispatcher dispatches the incoming events to multiple destinations according to their content and the registered subscriptions. It supports generic JMS messages.

EventDispatcher contains a number of subscriptions which define a filter and a set of actions. Internally, EventDispatcher also creates a topic pattern based on the filter for each subscription. The topic pattern will be used as a pre-filter on the topic of incoming events. For each incoming event, EventDispatcher extracts the topic first. Then it looks up the topic cache for a list of candidate subscriptions on the topic. If the list is empty, there is no subscibers for the event. The incoming event will be routed to the outlink of nohit. Otherwise, EventDispatcher just loops through the list and applies the filter of each candidates on the event. If it is a match and the subscription is active, EventDispatcher will create an ObjectEvent with the event and actions attached. The new message will be sent to the outlink of pool as a request to invoke all the actions of the subscription on the incoming event. At least one instance of EventPersister is supposed to be linked to the outlink of pool. This operation continues until all candidates in the list are checked. If there is no request event generated, the original incoming event will be routed to the outlink of nohit. Otherwise, it will be routed to the outlink of bypass. Due to the MT-Safty, any messages routed to the outlink of bypass should not be modified downstream. Otherwise, EventPersister may throw exceptions in delivery process. If those messages need to be modified, please work on their copy via DuplicateNode and leave the original messages alone. In case of any failure, the original incoming event will be sent to the outlink of failure.

A subscription is defined in a ruleset. Each ruleset contains a unique name of the subscription, an EventSelector to match events, a TimeWindows to define active time of the subscription and a group of EventActions for the subscription.

EventDispatcher always creates two extra rulesets. The first is the nohit ruleset for those events not hitting any matches. The second is the candidate ruleset for all events matching to at least one of subscriptions. Since a candidate event may hit number of subscriptions and generates a task event for every hit subscriptions, the stats count for the candidate ruleset will be the number of the incoming events rather than the number of the task events. The number of task events will be counted by their own rulesets. The DisplayMask and StringProperty of the ruleset are used to display the details of dispatched event 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.

You are free to choose any names for these four fixed outlinks. But EventDispatcher always assumes the first outlink as the pool for requests, the second for the bypass messages, the third for the failures and the last is for the nohit messages. It is OK for the last three outlinks to share the same name. But the name of the first outlink has to be unique. In case some of them sharing the same name, it means these outlinks are sharing the same output channel.

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
EventDispatcher(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()
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.
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, 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

EventDispatcher

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

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