org.qbroker.wmq

Class TConnector

java.lang.Object

org.qbroker.wmq.TConnector

All Implemented Interfaces:

javax.jms.ExceptionListener, TopicConnector

public class TConnector
extends java.lang.Object
implements TopicConnector

TConnector connects to a topic and initializes one of the operations, such as sub, pub, RegSub or DeregSub.

It supports both MQ Broker (V1) and MQSI Broker (V2) for Publication and Subscription. You can choose the version by BrokerVersion. Due to IBM's implementation on publish, the API for publish has some kind of transaction controls. Within every 25 messages it publishes, the first message has the reply field enabled and the API expects the reply from the broker. This means the publish is no longer asynchronous. In a distributed environment, if the broker is down the messages will stay in the queue. But the API will wait for the reply and hold the flow. If the broker is not available at the publish client startup, the client will fail. There is a parameter to set so that the transaction size is 0.

There are two ways to get around this problem. The first one is to set up a generic publish message flow in MQSI Broker. The flow will pick up the topic string from certain JMS property and add the publish header with the topic. All the messages will be sent to the Publish Node. This way, the client just needs to use point-to-point to publish the messages. The other way is to implement the message flow into publish method. We have implemented a simple Publication and Subscription via point-to-point. In the implementation, the pub/sub header is added to each messages. In case of sub, it only supports QueueStore. You can choose this implementation by setting BrokerVersion to 0. If you want to use this implementation, please remember to specify the full path of each JMS properties in the message selector. For example, if you reference a property called "FileName", you have to use "Root.MQRFH2.usr.FileName".

Here is the detail about the Pub/Sub support with point-to-point:

(1) always assume the subscription queue is set up already.
(2) browse SYSTEM.JMS.ADMIN.QUEUE for the clientID and subscriptionID.
(3) if found the entry, compare the topic and the subscription queue.
(4) if they are same, use get() rather than sub() to receive one message.
(5) if either topic or queue is not same, unsub old topic and sub new topic via broker and continue to get the message.
(6) if there is no match entry, sub the topic via broker and continue to get the messages.

This is NOT MT-Safe due to IBM's implementation on sessions and conn.

Author:

yannanlu@yahoo.com

Field Summary

Fields

Modifier and Type	Field and Description
protected javax.jms.TopicSession	tSession

Fields inherited from interface org.qbroker.jms.TopicConnector

TC_QUEUE, TC_STREAM, TC_TOPIC, TC_XQ, XA_CLIENT, XA_COMMIT

Constructor Summary

Constructors

Constructor and Description
TConnector(java.util.Map props) Creates new TConnector

Method Summary

Modifier and Type	Method and Description
int	check_subscription() check the subscription existence by browsing SYSTEM.JMS.ADMIN.QUEUE It returns 1 for subscribed, 0 for not subscribed and -1 for subscibed but not the same.
void	close() closes all opened resources
protected void	connect()
javax.jms.Message	createMessage(int type) returns a JMS Message with the initialized JMSSession
javax.jms.Queue	createQueue(java.lang.String queueName) returns the queue created for the given name
java.lang.String	getOperation() returns the string of the operation
javax.jms.TopicPublisher	getTopicPublisher() returns an on-demand publisher in a separate session
javax.jms.TopicSubscriber	getTopicSubscriber() returns an on-demand subscriber in a separate session
void	onException(javax.jms.JMSException e) catches JMSException of Connection asynchronously
void	pub(java.io.InputStream in) It reads bytes from the InputStream and publish them into a JMS topic as JMS messages.
void	pub(XQueue xq) It gets JMS messages from the XQueue and publishes them on a JMS topic.
void	publish_v0(java.lang.String topicName, javax.jms.Message msg) To support MQSI V2 broker, this method uses qSender to put the message with the bare V2 header that contains MQSI V2 PUB command.
java.lang.String	reconnect() reconnects to the JMS server for the operation or returns error text
void	sub(java.lang.Object out) It subscribes a JMS topic and sends the JMS messages into a output channel that may be a JMS destination, an OutputStream or an XQueue.
void	tClose() closes the topic
void	tOpen() opens the topic for the operation
int	tReopen() reopens the topic for the operation

Methods inherited from class java.lang.Object

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

Field Detail

tSession

protected javax.jms.TopicSession tSession

Constructor Detail

TConnector

public TConnector(java.util.Map props)
           throws javax.jms.JMSException

Creates new TConnector

Throws:

javax.jms.JMSException

Method Detail

connect

protected void connect()
                throws javax.jms.JMSException

Throws:

javax.jms.JMSException

tOpen

public void tOpen()
           throws javax.jms.JMSException

Description copied from interface: TopicConnector

opens the topic for the operation

Specified by:

tOpen in interface TopicConnector

Throws:

javax.jms.JMSException

tReopen

public int tReopen()

Description copied from interface: TopicConnector

reopens the topic for the operation

Specified by:

tReopen in interface TopicConnector

tClose

public void tClose()
            throws javax.jms.JMSException

Description copied from interface: TopicConnector

closes the topic

Specified by:

tClose in interface TopicConnector

Throws:

javax.jms.JMSException

createMessage

public javax.jms.Message createMessage(int type)
                                throws javax.jms.JMSException

Description copied from interface: TopicConnector

returns a JMS Message with the initialized JMSSession

Specified by:

createMessage in interface TopicConnector

Throws:

javax.jms.JMSException

createQueue

public javax.jms.Queue createQueue(java.lang.String queueName)
                            throws javax.jms.JMSException

Description copied from interface: TopicConnector

returns the queue created for the given name

Specified by:

createQueue in interface TopicConnector

Throws:

javax.jms.JMSException

reconnect

public java.lang.String reconnect()

Description copied from interface: TopicConnector

reconnects to the JMS server for the operation or returns error text

Specified by:

reconnect in interface TopicConnector

onException

public void onException(javax.jms.JMSException e)

Description copied from interface: TopicConnector

catches JMSException of Connection asynchronously

Specified by:

onException in interface javax.jms.ExceptionListener

Specified by:

onException in interface TopicConnector

close

public void close()

Description copied from interface: TopicConnector

closes all opened resources

Specified by:

close in interface TopicConnector

sub

public void sub(java.lang.Object out)
         throws java.io.IOException,
                javax.jms.JMSException

It subscribes a JMS topic and sends the JMS messages into a output channel that may be a JMS destination, an OutputStream or an XQueue. It supports the message acknowledgement at the source.

It catches JMSException in the operations regarding to the message itself, like get/set JMS properties. In this case, it still tries to send the message to the output channel and logs the errors. It does not catch the JMSException or IOException from the message transport operations, like subscribe and send. Such Exception will be thrown so that the caller can handle it either to reset the connection or to do something else.

Specified by:

sub in interface TopicConnector

Throws:

java.io.IOException

javax.jms.JMSException

pub

public void pub(java.io.InputStream in)
         throws java.io.IOException,
                javax.jms.JMSException

It reads bytes from the InputStream and publish them into a JMS topic as JMS messages. It supports flexable but static delimiters at either or both ends. It also has the blind trim support at both ends.

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

Specified by:

pub in interface TopicConnector

Throws:

java.io.IOException

javax.jms.JMSException

pub

public void pub(XQueue xq)
         throws TimeoutException,
                javax.jms.JMSException

It gets JMS messages from the XQueue and publishes them on a JMS topic. It supports both transaction in the delivery session and the message acknowledgement at the source.

It catches JMSException in the operations regarding to the message itself, like get/set JMS properties or acknowlegement. In this case, it removes the message from the XQueue and logs the errors. It does not catch the JMSException from the message transport operations, like publish or commit. Such JMSException will be thrown so that the caller can handle it either to reset the connection or to do something else.

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.

Specified by:

pub in interface TopicConnector

Throws:

TimeoutException

javax.jms.JMSException

getOperation

public java.lang.String getOperation()

Description copied from interface: TopicConnector

returns the string of the operation

Specified by:

getOperation in interface TopicConnector

getTopicPublisher

public javax.jms.TopicPublisher getTopicPublisher()

Description copied from interface: TopicConnector

returns an on-demand publisher in a separate session

Specified by:

getTopicPublisher in interface TopicConnector

getTopicSubscriber

public javax.jms.TopicSubscriber getTopicSubscriber()

Description copied from interface: TopicConnector

returns an on-demand subscriber in a separate session

Specified by:

getTopicSubscriber in interface TopicConnector

check_subscription

public int check_subscription()
                       throws javax.jms.JMSException

check the subscription existence by browsing SYSTEM.JMS.ADMIN.QUEUE It returns 1 for subscribed, 0 for not subscribed and -1 for subscibed but not the same. In case of existence, it also set msgID.

Throws:

javax.jms.JMSException

publish_v0

public void publish_v0(java.lang.String topicName,
                       javax.jms.Message msg)
                throws javax.jms.JMSException

To support MQSI V2 broker, this method uses qSender to put the message with the bare V2 header that contains MQSI V2 PUB command. Therefore, there is no need to set up a convert flow for publish since the header is already in RFH2.

Throws:

javax.jms.JMSException