MessageFlow

java.lang.Object
- org.qbroker.flow.MessageFlow

All Implemented Interfaces:

java.lang.Runnable
```
public class MessageFlow
extends java.lang.Object
implements java.lang.Runnable
```
MessageFlow is a message-driven application with a group of inter-linked message nodes. Each message node is well designed for certain purpose. It is running in a separate thread and communicates with the linked nodes via the internal queues. There are three types of message nodes, MessageReceiver, MessageNode and MessagePersister. The nodes of MessageReceiver and MessagePersister are for JMSMessage I/O. The nodes of MessageNode are for JMSMessage processing. With these three basic types of nodes, you can build a message flow for various purposes, just like you can build objects with Lego components. For the I/O nodes, MessageFlow uses them to pick up or deliver messages via various transports, like JMS, TCP Socket, FTP, HTTP, JDBC, Serial Port, UDP Packet, SYSLOG, etc. It is also expandable and extensible by implementing new message nodes.

Between two linked message nodes, there is a communication channel for them to transport JMSMessages internally. MessageFlow uses XQueue for the inter-node communications. As an instance of XQueue, also called a link, it supports both FIFO and transactions. A MessageFlow can have at least one link for the minimum of two nodes. The properties of a link, like Capacity and XAMode, are always determined by a linked node of MessageNode type. If there is no MessageNode on the link, the properties of the link will be determined by the default values of MessageFlow. If the nodes at both ends of the link are type of MessageNode, the node at the sending side wins. Among all the links, root is a special link. For example, the root link controls the XAMode of all other links. XAMode and Capacity of MessageFlow are used only as the default values. For a node of either MessageReceiver or MessagePersister, its XAMode only determines its JMS Transaction type.

MessageFlow supports the escalation channel for message nodes to escalate messages to the container or the caller. Its name of "escalation" is a reserved name and it can only be used in MessageNodes as an outLink with its partition specified explicitly.

MessageFlow supports the static ConfigTemplate for both Receiver and Persister. It means a list of persisters or receivers can share the same configuration template. But it does not allow duplicated names for either generated persisters or generated recievers.

FLOW_READY initialized and ready to work
FLOW_RUNNING running normally
FLOW_RETRYING trying to get back to running state
FLOW_PAUSE pausing for a bit while still running
FLOW_STANDBY standby for a bit longer while still running
FLOW_DISABLED flow is disabled
FLOW_STOPPED all threads stopped, can be restart later
FLOW_CLOSED end of life time, is not runable any more

Author:

yannanlu@yahoo.com

Field Summary

Fields
Modifier and Type	Field and Description
`static int`	`FLOW_CLOSED`
`static int`	`FLOW_DISABLED`
`static int`	`FLOW_PAUSE`
`static int`	`FLOW_READY`
`static int`	`FLOW_RETRYING`
`static int`	`FLOW_RUNNING`
`static int`	`FLOW_STANDBY`
`static int`	`FLOW_STOPPED`
`static int`	`GROUP_COUNT`
`static int`	`GROUP_HBEAT`
`static int`	`GROUP_ID`
`static int`	`GROUP_RETRY`
`static int`	`GROUP_SIZE`
`static int`	`GROUP_SPAN`
`static int`	`GROUP_STATUS`
`static int`	`GROUP_TID`
`static int`	`GROUP_TIME`
`static int`	`GROUP_TIMEOUT`
`static int`	`GROUP_TYPE`
`static int`	`MF_NODE`
`static int`	`MF_PSTR`
`static int`	`MF_RCVR`

Constructor Summary

Constructors
Constructor and Description

MessageFlow(java.util.Map props)

MessageFlow(java.util.Map props, XQueue escalation)
Creates new MessageFlow

Constructors
Constructor and Description
`MessageFlow(java.util.Map props)`
`MessageFlow(java.util.Map props, XQueue escalation)` Creates new MessageFlow

Method Summary

All Methods Static Methods Instance Methods Concrete Methods
Modifier and Type	Method and Description
`void`	`close()`
`int`	`demote()` demotes the flow from the current status to its default status
`java.util.Map<java.lang.String,java.lang.Object>`	`diff(java.util.Map props)` It returns a Map with only modified or new props.
`int`	`disable()` disables the flow from the current status to the disabled status
`int`	`enable()` enables the disabled flow to its previous status
`int`	`getDebugMode()`
`int`	`getDefaultStatus()`
`java.lang.String`	`getName()`
`int`	`getNumberOfNodes(int type)` returns number of nodes of the specific type
`int`	`getNumberOfXQueues()` returns number of XQueues of the flow
`java.lang.String`	`getOperation()`
`int[]`	`getPartition()` returns partition of root queue for external requests
`XQueue`	`getRootQueue()`
`int`	`getStatus()`
`boolean`	`hasMajorChange(java.util.Map<java.lang.String,java.lang.Object> change)` It takes the Map returned from diff() and analizes what have been changed.
`int`	`ingest(java.lang.String key, javax.jms.Message msg)` It ingests the message into the XQueue with the name of key and returns the id upon success.
`static java.lang.Object`	`initNode(java.util.Map props, java.lang.String prefix)` It instantiates a node object from the given props and returns the object upon success.
`java.lang.String`	`list(int type)` returns a text of list with all nodes in either JSON or XML
`int`	`promote()` promotes the flow from the current status to running status
`java.util.Map<java.lang.String,java.lang.String>`	`queryInfo(long currentTime, long sessionTime, java.lang.String target, java.lang.String key, int type)` returns a Map containing queried information according to the result type
`int`	`refreshRules(java.lang.String key)` It refreshes the external rules on all the nodes contains the rule with the given name and returns the number of external rulesets updated.
`int`	`reload(java.util.Map<java.lang.String,java.lang.Object> change)` It reloads the changes for the entire flow and returns the number of failures.
`void`	`run()`
`void`	`setDebugMode(int debug)`
`java.lang.String`	`show(java.lang.String key, int id, int type)` returns the display text on the message at the cell id of the queue
`void`	`showChange(java.lang.String prefix, java.lang.String base, java.util.Map<java.lang.String,java.lang.Object> change, boolean detail)` It loggs the details of the change returned by diff()
`void`	`start()` starts all threads and/or nodes
`void`	`stop()` stops all nodes and the threads
`javax.jms.Message`	`takeback(java.lang.String key, int id)` takes back the message at the cell of id from the queue

Methods inherited from class java.lang.Object
clone, equals, finalize, getClass, hashCode, notify, notifyAll, toString, wait, wait, wait

- Field Detail
  - MF_RCVR
```
public static final int MF_RCVR
```
    See Also:
    
    Constant Field Values
  - MF_NODE
```
public static final int MF_NODE
```
    See Also:
    
    Constant Field Values
  - MF_PSTR
```
public static final int MF_PSTR
```
    See Also:
    
    Constant Field Values
  - GROUP_ID
```
public static final int GROUP_ID
```
    See Also:
    
    Constant Field Values
  - GROUP_TID
```
public static final int GROUP_TID
```
    See Also:
    
    Constant Field Values
  - GROUP_TYPE
```
public static final int GROUP_TYPE
```
    See Also:
    
    Constant Field Values
  - GROUP_SIZE
```
public static final int GROUP_SIZE
```
    See Also:
    
    Constant Field Values
  - GROUP_HBEAT
```
public static final int GROUP_HBEAT
```
    See Also:
    
    Constant Field Values
  - GROUP_TIMEOUT
```
public static final int GROUP_TIMEOUT
```
    See Also:
    
    Constant Field Values
  - GROUP_STATUS
```
public static final int GROUP_STATUS
```
    See Also:
    
    Constant Field Values
  - GROUP_COUNT
```
public static final int GROUP_COUNT
```
    See Also:
    
    Constant Field Values
  - GROUP_SPAN
```
public static final int GROUP_SPAN
```
    See Also:
    
    Constant Field Values
  - GROUP_RETRY
```
public static final int GROUP_RETRY
```
    See Also:
    
    Constant Field Values
  - GROUP_TIME
```
public static final int GROUP_TIME
```
    See Also:
    
    Constant Field Values
  - FLOW_READY
```
public static final int FLOW_READY
```
    See Also:
    
    Constant Field Values
  - FLOW_RUNNING
```
public static final int FLOW_RUNNING
```
    See Also:
    
    Constant Field Values
  - FLOW_RETRYING
```
public static final int FLOW_RETRYING
```
    See Also:
    
    Constant Field Values
  - FLOW_PAUSE
```
public static final int FLOW_PAUSE
```
    See Also:
    
    Constant Field Values
  - FLOW_STANDBY
```
public static final int FLOW_STANDBY
```
    See Also:
    
    Constant Field Values
  - FLOW_DISABLED
```
public static final int FLOW_DISABLED
```
    See Also:
    
    Constant Field Values
  - FLOW_STOPPED
```
public static final int FLOW_STOPPED
```
    See Also:
    
    Constant Field Values
  - FLOW_CLOSED
```
public static final int FLOW_CLOSED
```
    See Also:
    
    Constant Field Values
- Constructor Detail
  - MessageFlow
```
public MessageFlow(java.util.Map props,
                   XQueue escalation)
```
    Creates new MessageFlow
  - MessageFlow
```
public MessageFlow(java.util.Map props)
```
- Method Detail
  - run
```
public void run()
```
    Specified by:
    
    run in interface java.lang.Runnable
  - start
```
public void start()
```
    starts all threads and/or nodes
  - stop
```
public void stop()
```
    stops all nodes and the threads
  - enable
```
public int enable()
```
    enables the disabled flow to its previous status
  - disable
```
public int disable()
```
    disables the flow from the current status to the disabled status
  - promote
```
public int promote()
```
    promotes the flow from the current status to running status
  - demote
```
public int demote()
```
    demotes the flow from the current status to its default status
  - ingest
```
public int ingest(java.lang.String key,
                  javax.jms.Message msg)
```
    It ingests the message into the XQueue with the name of key and returns the id upon success. Otherwise, it returns -1. It assumes that the 2nd half of cells for the XQueue are not used.
  - show
```
public java.lang.String show(java.lang.String key,
                             int id,
                             int type)
```
    returns the display text on the message at the cell id of the queue
  - takeback
```
public javax.jms.Message takeback(java.lang.String key,
                                  int id)
```
    takes back the message at the cell of id from the queue
  - queryInfo
```
public java.util.Map<java.lang.String,java.lang.String> queryInfo(long currentTime,
                                                                  long sessionTime,
                                                                  java.lang.String target,
                                                                  java.lang.String key,
                                                                  int type)
```
    returns a Map containing queried information according to the result type
  - list
```
public java.lang.String list(int type)
```
    returns a text of list with all nodes in either JSON or XML
  - getName
```
public java.lang.String getName()
```
  - getOperation
```
public java.lang.String getOperation()
```
  - getStatus
```
public int getStatus()
```
  - getDefaultStatus
```
public int getDefaultStatus()
```
  - getDebugMode
```
public int getDebugMode()
```
  - setDebugMode
```
public void setDebugMode(int debug)
```
  - getRootQueue
```
public XQueue getRootQueue()
```
  - getPartition
```
public int[] getPartition()
```
    returns partition of root queue for external requests
  - getNumberOfNodes
```
public int getNumberOfNodes(int type)
```
    returns number of nodes of the specific type
  - getNumberOfXQueues
```
public int getNumberOfXQueues()
```
    returns number of XQueues of the flow
  - refreshRules
```
public int refreshRules(java.lang.String key)
```
    It refreshes the external rules on all the nodes contains the rule with the given name and returns the number of external rulesets updated.
  - reload
```
public int reload(java.util.Map<java.lang.String,java.lang.Object> change)
```
    It reloads the changes for the entire flow and returns the number of failures.
  - diff
```
public java.util.Map<java.lang.String,java.lang.Object> diff(java.util.Map props)
```
    It returns a Map with only modified or new props. The deleted items will be set to null in the returned Map. If the container has been changed, it will be reflected by the object keyed by base.
  - hasMajorChange
```
public boolean hasMajorChange(java.util.Map<java.lang.String,java.lang.Object> change)
```
    It takes the Map returned from diff() and analizes what have been changed. If the Map contains major changes that require to stop or disable the flow, it returns true. Otherwise, it returns false to indicate the reload of the flow is good enough.
  - showChange
```
public void showChange(java.lang.String prefix,
                       java.lang.String base,
                       java.util.Map<java.lang.String,java.lang.Object> change,
                       boolean detail)
```
    It loggs the details of the change returned by diff()
  - initNode
```
public static java.lang.Object initNode(java.util.Map props,
                                        java.lang.String prefix)
```
    It instantiates a node object from the given props and returns the object upon success. In case of failure, it returns the exception.
  - close
```
public void close()
```

Class MessageFlow

Field Summary

Constructor Summary

Method Summary

Methods inherited from class java.lang.Object

Field Detail

MF_RCVR

MF_NODE

MF_PSTR

GROUP_ID

GROUP_TID

GROUP_TYPE

GROUP_SIZE

GROUP_HBEAT

GROUP_TIMEOUT

GROUP_STATUS

GROUP_COUNT

GROUP_SPAN

GROUP_RETRY

GROUP_TIME

FLOW_READY

FLOW_RUNNING

FLOW_RETRYING

FLOW_PAUSE

FLOW_STANDBY

FLOW_DISABLED

FLOW_STOPPED

FLOW_CLOSED

Constructor Detail

MessageFlow

MessageFlow

Method Detail

run

start

stop

enable

disable

promote

demote

ingest

show

takeback

queryInfo

list

getName

getOperation

getStatus

getDefaultStatus

getDebugMode

setDebugMode

getRootQueue

getPartition

getNumberOfNodes

getNumberOfXQueues

refreshRules

reload

diff

hasMajorChange

showChange

initNode

close