org.qbroker.jms

Class MessageStream

java.lang.Object

org.qbroker.jms.MessageStream

public class MessageStream
extends java.lang.Object

MessageStream handles Stream IO on JMS Messages. The Stream has to be established before using this object.

There are four methods, read(), write(), request() and respond() on Stream. The method of read() is MT-Safe. If read() gets the end of stream, it will stop and return. The method of write() is for asynchronous output whereas request() is for synchronous requests. For testing purpose, MessageStream supports two methods, echo() and collect().

MessageStream reacts on EOF differently according to the Mode. If its value is set to command mode, read() just exits when an EOF is read. For Linux comm port, the timeout implementation returns an EOF. Therefore, setting the value to ignore mode will have most of read operation to ignore the EOF and continue to read. If the value is set to daemon, read() will throw an IOException whenever it reads an EOF to indicate end of stream unexpectedly.

For charSet and encoding, you can overwrite the default encoding by starting JVM with "-Dfile.encoding=UTF8 your_class".

Author:

yannanlu@yahoo.com

Constructor Summary

Constructors

Constructor and Description
MessageStream(java.util.Map props)

Method Summary

Modifier and Type	Method and Description
void	collect(java.io.InputStream in, XQueue xq, int timeout) collects the response for a request for testing purpose
void	echo(XQueue xq) echoes the message back for testing purpose
byte[]	getEotBytes()
int	getOffhead()
int	getOfftail()
java.lang.String	getOperation()
byte[]	getSotBytes()
long	read(java.io.InputStream in, XQueue xq) It reads the byte stream from an InputStream, packages them into JMS Messages and adds the messages to an XQueue.
long	read(java.io.InputStream in, XQueue xq, java.lang.String[] pNames, java.lang.String[] pValues) It reads the byte stream from an InputStream, packages them into JMS Messages and sets the properties before adding the messages to an XQueue.
long	request(java.io.InputStream in, XQueue xq, java.io.OutputStream out) It listens on an XQueue for an incoming JMS messages and writes all bytes to an OutputStream as the request.
long	request(java.io.InputStream in, XQueue xq, java.io.OutputStream out, QList msgList, DelimitedBuffer sBuf) It listens on an XQueue for incoming JMS messages and writes all bytes to an OutputStream as the requests.
long	respond(java.io.InputStream in, XQueue xq, java.io.OutputStream out, QList msgList, DelimitedBuffer sBuf, CollectibleCells cells) It reads the byte stream from an InputStream, packages them into JMS Messages and adds the messages to an XQueue as the request.
long	write(XQueue xq, java.io.OutputStream out) It gets JMS Messages from an XQueue and write all bytes to the OutputStream continuously.

Methods inherited from class java.lang.Object

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

Constructor Detail

MessageStream

public MessageStream(java.util.Map props)

Method Detail

read

public long read(java.io.InputStream in,
                 XQueue xq)
          throws java.io.IOException,
                 javax.jms.JMSException

It reads the byte stream from an InputStream, packages them into JMS Messages and adds the messages to an XQueue. It supports flexible but static delimiters at either or both ends. It also has the blind trim support at both ends.

Most of the IOExceptions from the stream IO operations, like read, will be thrown to notify the caller. The caller is supposed to catch the exception, either to reset the stream or to do other stuff. It also throws JMSException from the operations regarding to the message itself, like get/set JMS properties. This does not require to reset the stream at all.

This is MT-Safe as long as the threads share the same offhead, offtail, sotBytes and eotBytes.

Throws:

java.io.IOException

javax.jms.JMSException

read

public long read(java.io.InputStream in,
                 XQueue xq,
                 java.lang.String[] pNames,
                 java.lang.String[] pValues)
          throws java.io.IOException,
                 javax.jms.JMSException

It reads the byte stream from an InputStream, packages them into JMS Messages and sets the properties before adding the messages to an XQueue. It supports flexible but static delimiters at either or both ends. It also has the blind trim support at both ends. At the end, it returns number of messages read from input stream.

Most of the IOExceptions from the stream IO operations, like read, will be thrown to notify the caller. The caller is supposed to catch the exception, either to reset the stream or to do other stuff. It also throws JMSException from the operations regarding to the message itself, like get/set JMS properties. This does not require to reset the stream at all.

The class variable controls the reaction on EOF. If the value is set to command mode, it just exits when an EOF is read. For Linux comm port, the timeout implementation returns EOF. Therefore, setting the value to ignore mode will have the method to ignore EOF and continue to read. If the value is set to daemon, it will throw IOException whenever it reads an EOF to indicate connection reset by peer.

This is MT-Safe as long as the threads share the same offhead, offtail, sotBytes and eotBytes.

Throws:

java.io.IOException

javax.jms.JMSException

respond

public long respond(java.io.InputStream in,
                    XQueue xq,
                    java.io.OutputStream out,
                    QList msgList,
                    DelimitedBuffer sBuf,
                    CollectibleCells cells)
             throws java.io.IOException,
                    javax.jms.JMSException

It reads the byte stream from an InputStream, packages them into JMS Messages and adds the messages to an XQueue as the request. Then it continually monitors the XQueue for the reply messages. If it finds one, the message will be sent back to the OutputStream associated with the InputStream as the response. Since there is no blocking between the request and the response, it is possible for the thread to process two requests before any response sent back. In order for the requester to tell a response is corresponding to which request, each request message should contains an ID even though it is unknown to this method. So the downstream message flow is not supposed to modify or remove that ID. It supports flexible but static delimiters at either or both ends. It also has the blind trim support at both ends.

The msgList is a dedicated instance of QList for tracking the collectible responses. The sBuf is also a dedicated buffer with delimiter supports. The last argument is a dedicate instance of CollectibleCells for reverse propagations. Its capacity should be same as the xq.

Most of the IOExceptions from the stream IO operations, like read, will be thrown to notify the caller. The caller is supposed to catch the exception, either to reset the stream or to do other stuff. It also throws JMSException from the operations regarding to the message itself, like get/set JMS properties. This does not require to reset the stream at all.

To support re-entrance, it depends on msgList and sBuf for state info. Please make sure sBuf has the same properties as the instance. This is MT-Safe as long as the threads share the same offhead, offtail, sotBytes and eotBytes.

Throws:

java.io.IOException

javax.jms.JMSException

request

public long request(java.io.InputStream in,
                    XQueue xq,
                    java.io.OutputStream out,
                    QList msgList,
                    DelimitedBuffer sBuf)
             throws java.io.IOException,
                    javax.jms.JMSException,
                    TimeoutException

It listens on an XQueue for incoming JMS messages and writes all bytes to an OutputStream as the requests. Meanwhile, it reads the byte stream from the InputStream associated with the OutputStream, loads them into the request messages and removes them from the XQueue as the responses. In order to track the incoming responses, it will add the SID to the rcField of each outgoing message. Whoever receives the request should ensure the response message also contains the same SID. Hence this method will be able to retrieve the SID and to associate the response with the request. It also supports flexible but static delimiters at either or both ends. It also has the blind trim support at both ends.

Most of the IOExceptions from the stream IO operations, like read, will be thrown to notify the caller. The caller is supposed to catch the exception, either to reset the stream or to do other stuff. It also throws JMSException from the operations regarding to the message itself, like get/set JMS properties. This does not require to reset the stream at all.

Since the request operation relies on a request, this method will intercept the request and process it. The incoming message is required to be writeable. The method will set a String property of the message specified by the RCField with the return code, indicating the status of the process to fulfill the request. The requester is required to check the value of the property once it gets the message back. If the return code is 0, the request is successful. This method will not acknowledge messages. It is up to the requester to do that.

To support multi-plexing, it depends on msgList and sBuf to track the pending requests and the state info. It does not support re-entrance.

Throws:

java.io.IOException

javax.jms.JMSException

TimeoutException

request

public long request(java.io.InputStream in,
                    XQueue xq,
                    java.io.OutputStream out)
             throws java.io.IOException,
                    TimeoutException,
                    javax.jms.JMSException

It listens on an XQueue for an incoming JMS messages and writes all bytes to an OutputStream as the request. After the write operation, it reads the byte stream from the InputStream associated with the OutputStream until the pattern is hit. Then it loads all the bytes into the request message and removes it from the XQueue as the response. Both write and read operations block the message flow. Therefore, there is no need to track SID for each messages. It supports flexible terminate patterns with timeout.

Most of the IOExceptions from the stream IO operations, like read, will be thrown to notify the caller. The caller is supposed to catch the exception, either to reset the stream or to do other stuff. It also throws JMSException from the operations regarding to the message itself, like get/set JMS properties. This does not require to reset the stream at all.

Since the request operation relies on a request, this method will intercept the request and process it. The incoming message is required to be writeable. The method will set a String property of the message specified by the RCField with the return code, indicating the status of the process to fulfill the request. The requester is required to check the value of the property once it gets the message back. If the return code is 0, the request is successful. This method will not acknowledge messages. It is up to the requester to do that.

It is NOT MT-safe.

Throws:

java.io.IOException

TimeoutException

javax.jms.JMSException

write

public long write(XQueue xq,
                  java.io.OutputStream out)
           throws java.io.IOException,
                  TimeoutException,
                  javax.jms.JMSException

It gets JMS Messages from an XQueue and write all bytes to the OutputStream continuously.

Most of the IOExceptions from the stream IO operations, like write, will be thrown to notify the caller. The caller is supposed to catch the exception, either to reset the stream or to do other stuff. It also throws JMSException from the operations regarding to the message itself, like get/set JMS properties. This does not require to reset the stream at all.

If the MaxIdleTime is set to none-zero, it will monitor the idle time and will throw TimeoutException once the idle time exceeds MaxIdleTime. it is up to the caller to handle this exception.

It will not modify the original messages. If the XQueue has enabled the EXTERNAL_XA, it will also acknowledge every messages. This method is MT-Safe.

Throws:

java.io.IOException

TimeoutException

javax.jms.JMSException

collect

public void collect(java.io.InputStream in,
                    XQueue xq,
                    int timeout)
             throws javax.jms.JMSException

collects the response for a request for testing purpose

Throws:

javax.jms.JMSException

echo

public void echo(XQueue xq)
          throws javax.jms.JMSException

echoes the message back for testing purpose

Throws:

javax.jms.JMSException

getOperation

public java.lang.String getOperation()

getSotBytes

public byte[] getSotBytes()

getEotBytes

public byte[] getEotBytes()

getOfftail

public int getOfftail()

getOffhead

public int getOffhead()