org.qbroker.node

Class ServiceNode

java.lang.Object

org.qbroker.node.Node

org.qbroker.node.ServiceNode

All Implemented Interfaces:

MessageNode

public class ServiceNode
extends Node

ServiceNode listens to an input XQ for JMS ObjectMessages with a connected socket as a request to establish a connection. The incoming message is also called a handshake message. It contains the information about the request, the client and the connection. ServiceNode matches the properties against the predefined rulesets to determine which ruleset to handle the connection request. ServiceNode extracts the URI from the handshake message and looks it up in the cache for an XQueue as the transmit queue established for the connection. If there is an XQueue in the cache, it means that the service is ready. Hence ServiceNode will just check out a thread to handle the connection on the XQueue. Otherwise, ServiceNode will generate a new request with a newly created XQueue for either a new consumer or a new producer to serve on the connection. The request will be sent to the right outlink according to the ruleset.

ServiceNode has two types of outlinks, position-fixed and non-fixed. There are three position-fixed outlinks: producer for all requests to a ReceiverPool, consumer for all requests to a PersisterPool, nohit for all requests not covered by any rulesets. The non-fixed outlinks are for the dynamic destinations or sources.

ServiceNode also contains a number of predefined rulesets. These rulesets categorize messages into non-overlapping groups. Therefore, each ruleset defines a unique message group. The ruleset also specifies the way to construct the URI and properties for the new Persisters or receivers. For those messages falling off all defined rulesets, ServiceNode always creates an extra ruleset, nohit, to gather stats on them. Currently, all the nohit messages will not be routed to the nohit outlink. Instead, they will be removed at once by ServiceNode. The stats of pending messages will be tracked by RULE_PEND for each ruleset.

If the XQueue does not exist and the ruleset is up to the consumer pool, ServiceNode will create a new ObjectMessage as the request for a new persister thread. It will also create an XQueue and puts the XQueue and the URI into the request before sending it to consumer outlink. Once the persister is instantiated, ServiceNode will collect the response from the PersisterPool. The response contains the persister thread for delivering the messages to the specific destination. ServiceNode caches the thread and the id of the XQueue using the URI as the key. Then it starts the proxy thread to pipe all messages from the socket to the XQueue. This way, ServiceNode is able to deliver JMS messages to arbitrary destinations according to their URIs and the predefined rulesets dynamically.

If the XQueue does not exist and the ruleset is up to the producer pool, ServiceNode will create a new ObjectMessage as the request for a new receiver thread. It will also create an XQueue and puts the XQueue and the URI into the request before sending it to producer outlink. Once the receiver is instantiated, ServiceNode will collect the response from the ReceiverPool. The response contains the receiver thread for picking up the messages from the specific source. ServiceNode caches the thread and the id of the XQueue using the URI as the key. Then it starts the proxy thread to pipe all messages from the XQueue to the socket. This way, ServiceNode is able to pick up JMS messages from arbitrary sources according to their URIs and the predefined rulesets dynamically.

URI is used to identify destinations or sources. In order to construct the URI string for an arbitrary destination or source, each ruleset has to define a template via URITemplate. Optinally, URISubstitution may be defined also for a simple substitution. With them, ServiceNode will be able to retrieve the URI string from the handshake messages. A ruleset may also contain a hash of DefaultProperty as static or dynamic properties for new persisters or new receivers. If the URI string is not in the cache, ServiceNode will resolve all the dynamic variables in the DefaultProperty from the handshake message first. Then it adds the DefaultProperty to the new ObjectMessage for the either pools to create a new consumer or a new producer. If any of the operations fails, the handshake message will be removed and the socket will be closed as a failure.

For each new destination/source, ServiceNode uses the same Object message as the request containing the URI and the XQueue. The request is sent to either PersisterPool/ReceiverPool via the corresponding outlink. Then ServiceNode frequently checks the response for each outstanding requests. The response is supposed to have the status and the persister/receiver thread for the new destination/source. ServiceNode will use the thread to monitor its status. If the response has no such thread, ServiceNode will remove the incoming message and closes the socket. It will also remove the XQueue and the URI from the cache. The samething will happen if the request for a new persister or receiver times out. MaxRetry is used to control when to timeout the request on the pool. It also controls the timeout on a dead persister or receiver thread.

ServiceNode also maintains an active set of XQueues as the transmit queues for all destinations and sources. Behind each XQueue, there is at least one persister or receiver thread processing the messages. The messages may be stuck in the XQueue until they are processed, as long as the persister or receiver is not stopped. However, if there is no message in the queue, ServiceNode will mark it idle. All the transmit queues are monitored frequently in sessions. If one of them has been idle for over MaxIdleTime, its queue will be stopped. Its persister or receiver thread and the transmit queue will be removed from the cache. The associated socket will be closed too.

You are free to choose any names for the three fixed outlinks. But ServiceNode always assumes the first outlink for the producer, the second for consumer and the third is for nohit. The rest of the outlinks are on-demand outlinks for dynamic destinations or sources. The name of the first outlink has to be unique on the node.

Currently, there is no support for NOHIT outlink yet. But it may be added in the future to support existing static XQueues. If a ruleset is nohit and the XQueue does not exist, ServiceNode will send an ObjectMessage as the request to the nohit outlink. The nohit outlink should always point to the escalation queue. So the manager of the apps should reply with an static XQueue with partition info. Then ServiceNode starts the proxy thread to handle the connection.

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
ServiceNode(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.
void	doPublish(XQueue xq, QList subList, Template template) It keeps publishing until its end of life
protected int	feedback(XQueue in, long milliSec) returns number of done messages removed from in
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.
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) passes the message from an input XQ over to an output XQ as the request for a new persister or receiver.
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	proxy(java.util.Map client) It continuously reads byte stream from the socket and package them into JMS Messages.
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.
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, getOutLinkBoundary, getRuleName, getStatus, getXAMode, getXAMode, list, listPendings, refreshRule, renameRule, 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

ServiceNode

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

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.

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

proxy

public void proxy(java.util.Map client)

It continuously reads byte stream from the socket and package them into JMS Messages. It adds the messages to an XQueue as output. The messages flow downstream and they will be consumed eventually. It acts like a proxy for JMS messages.

doPublish

public void doPublish(XQueue xq,
                      QList subList,
                      Template template)

It keeps publishing until its end of life

passthru

protected int passthru(long currentTime,
                       javax.jms.Message msg,
                       XQueue in,
                       int rid,
                       int oid,
                       int cid,
                       int tid)

passes the message from an input XQ over to an output XQ as the request for a new persister or receiver. The last argument, tid, is the outLink id of the new persister or receiver.

Overrides:

passthru in class Node

feedback

protected int feedback(XQueue in,
                       long milliSec)

returns number of done messages removed from in

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

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

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