PipeNode

java.lang.Object
- org.qbroker.node.Node
- - org.qbroker.node.PipeNode

All Implemented Interfaces:

java.util.Comparator<int[]>, MessageNode
```
public class PipeNode
extends Node
implements java.util.Comparator<int[]>
```
PipeNode picks up JMS messages from the input XQueue and caches them first. Then it dispatches them based on the priorities and thresholds for each of groups. There is only one outlink, pipe. All the groups share the same pipe with different priorities and thresholds.

PipeNode contains multiple rulesets that categorize the incoming messages into non-overlapped groups. Each group has its own priority for scheduling and threshold as the maximum number of active messages waiting for feedback. PipeNode 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. PipeNode will make sure only the first one of the identical messages will be scheduled. This way, PipeNode is able to control flow thruput on each group of messages.

By default, PipeNode has the cache enabled. So all the overflown messages will be cached. Due to the scheduling process, the message affinity may get broken. However, if the cache is disabled, the message affinity will be guaranteed. In this case, the first overflown message will block until the required resource is available.

PipeNode always creates an extra ruleset, nohit, for those messages not hitting any pre-defined rulesets. It is always the first ruleset with the id of 0. For nohit, there is no flow control by default. It means that there is no caching on nohit messages and no scheduling on them either. However, if nohit is the only ruleset and the cache is disabled, the flow control is activated. The behavior of the flow control is determined by the values of Heartbeat and Throughput. Throughput is the number of the messages allowed to propagate within a session. The duration of a session is defined via Heartbeat in second. If both Throughput and Heartbeat are larger than zero, the throughput of the nohit messages will be controlled not to exceed the pre-defined Throughput in a session. If Throughput is zero but Heartbeat is positive, there is no flow control at all. But the ruleset will still save the number of messages in the previous session into RULE_PEND for monitoring. If Throughput is less than zero, the nohit messages will be delayed for a while according to the value of Heartbeat. If Heartbeat is a positive number, the length of the delay will be a random number generated via the ordered random number generator with N = 1000000. If Heartbeat is zero, however, the delay will be based on the original intervals between the successive messages. In this case, the ruleset will track the internal timestamps of messages and figures out the original intervals between two successive messages. These intervals will be used to delay the message propagations. It can be used to replay messages in their original throughput rate.

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

PipeNode(java.util.Map props)

Constructors
Constructor and Description
`PipeNode(java.util.Map props)`

Method Summary

All Methods Instance Methods Concrete Methods
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)` returns the 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.
`java.util.Map<java.lang.String,java.lang.String>`	`listPendings(XQueue xq, int type)` lists all propagating and pending messages
`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
`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, internalXQSupported, list, passthru, refreshRule, removeRule, renameRule, replaceRule, 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

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
  - PipeNode
```
public PipeNode(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 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
  - feedback
```
protected int feedback(XQueue in,
                       long milliSec)
```
    returns the 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
  - listPendings
```
public java.util.Map<java.lang.String,java.lang.String> listPendings(XQueue xq,
                                                                     int type)
```
    lists all propagating and pending messages
    
    Specified by:
    
    listPendings in interface MessageNode
    
    Overrides:
    
    listPendings 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

Class PipeNode

Field Summary

Fields inherited from class org.qbroker.node.Node

Fields inherited from interface org.qbroker.node.MessageNode

Constructor Summary

Method Summary

Methods inherited from class org.qbroker.node.Node

Methods inherited from class java.lang.Object

Methods inherited from interface java.util.Comparator

Constructor Detail

PipeNode

Method Detail

updateParameters

initRuleset

propagate

feedback

callback

resetMetaData

listPendings

close

compare

finalize