org.qbroker.node

Class JobNode

java.lang.Object

org.qbroker.node.Node

org.qbroker.node.JobNode

All Implemented Interfaces:

java.util.Comparator<int[]>, MessageNode

public class JobNode
extends Node
implements java.util.Comparator<int[]>

JobNode picks up JMS messages as the job requests and job queries from the uplink and processes them according to their content and the pre-defined rulesets. A job request is to start a time consuming job whereas a job query is just to query the status of a running job. If an incoming message is neither a job request nor a job query, JobNode will treat it as a bypass and routes it to the bypass outlink without any delay. When a job request arrives, JobNode creates a job task out of the message and sets the job id and the status to the message. Then it routes the message to the done outlink at once. Meanwhile, JobNode tries to dispatch the new job task to an available worker outlink based on the priority of the job. If there is no enough worker available, the job task will wait until the required worker resource is available. JobNode also periodically monitors all active jobs and sends query events to their worker outlinks. When the query events are collected, They will be escalated to the outlink of notice. Once a job has completed, JobNode will collect the task message from the worker outlink and escalates the final report event to the notice outlink. If an incoming message is a job query, JobNode will forward it to the worker outlink. Once the response is collected, it will be routed to the bypass outlink.

JobNode contains multiple rulesets that categorize the incoming messages into non-overlapped groups. Each group has its own priority for scheduling and threshold for the maximum number of active jobs. JobNode also supports the dynamic threshold via the array of ActiveTime. By default, all messages in the same group are unique. But if KeyTemplate is defined in the ruleset, messages with the same key are treated as identical. JobNode will make sure only the first one of the identical messages will be scheduled. This way, JobNode is able to manage and schedule jobs for each group of messages. A group can also define its own Heartbeat and DisplayMask, as well as the a bunch of property names. Hearbeat controls how often to query the status on active jobs. DisplayMask determines whether to log the escalation messages. Property names specify what properties to be copied over to the escalation messages.

JobNode always creates an extra ruleset, nohit, for those messages not hitting any pre-defined rulesets. All the nohit messages will be routed to the outlink of bypass.

You are free to choose any names for the outlinks. But JobNode always assumes the first outlink for done, the second for bypass, and third for notice and the rest for workers.

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
JobNode(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.
int	compare(int[] a, int[] b)
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 -(BOUNDARY+1) to have the container enable ack regardlessly on the outlinks starting from BOUNDARY and beyond
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 pending jobs
protected int	passthru(long currentTime, javax.jms.Message msg, XQueue in, int rid, int oid, int cid, int tid) It passes the message from an input XQueue over to an output XQ.
void	propagate(XQueue in, XQueue[] out) picks up a message from input queue and evaluate its content to decide which output queue to propagate
void	resetMetaData(XQueue in, XQueue[] out) cleans up MetaData for all XQs and messages

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, internalXQSupported, list, refreshRule, removeRule, renameRule, replaceRule, restore, rotateRule, setDebugMode, setDisplayMask, setDisplayProperties, setStatus, show, takeback, takebackEnabled, updateParameters

Methods inherited from class java.lang.Object

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

Methods inherited from interface java.util.Comparator

comparing, comparing, comparingDouble, comparingInt, comparingLong, equals, naturalOrder, nullsFirst, nullsLast, reversed, reverseOrder, thenComparing, thenComparing, thenComparing, thenComparingDouble, thenComparingInt, thenComparingLong

Constructor Detail

JobNode

public JobNode(java.util.Map props)

Method Detail

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 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

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 an input XQueue over to an output XQ.

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. There is no wait and it tries to collect all cells regardless what milliSec is. If a collected msg is from one of the worker outlinks, the escalation event will be flushed to NOTICE_OUT accordingly.

Overrides:

feedback in class Node

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

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

listPendings

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

lists all propagating messages and pending jobs

Specified by:

listPendings in interface MessageNode

Overrides:

listPendings in class Node

getOutLinkBoundary

public int getOutLinkBoundary()

returns -(BOUNDARY+1) to have the container enable ack regardlessly on the outlinks starting from BOUNDARY and beyond

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

compare

public int compare(int[] a,
                   int[] b)

Specified by:

compare in interface java.util.Comparator<int[]>

finalize

protected void finalize()

Overrides:

finalize in class Node