org.openeai.jms.consumer
Class PubSubConsumer

java.lang.Object
  |
  +--org.openeai.OpenEaiObject
        |
        +--org.openeai.jms.consumer.MessageConsumer
              |
              +--org.openeai.jms.consumer.PubSubConsumer

public class PubSubConsumer

extends MessageConsumer

This consumer consumes messages from a JMS Topic. Then based on the content of the message it executes 'SyncCommands' associated to the consumer and the message consumed as specified in it's configuration document (deployment descriptor).

The actual business logic related to the message consumed is performed by the Sync command implementations.

Version:

3.0 beta2 - 28 January 2003

Author:

Tod Jackson (tod@openeai.org), Steve Wheat (steve@openeai.org)

Nested Class Summary

protected class	PubSubConsumer.ConsumerShutdownHook This Thread will be started when the consumer receives a shutdown signal from the os.
protected class	PubSubConsumer.MonitorConsumer This Thread will sleep for a specified period of time and then wake up and check the status of the consumer by attempting to create/delete a TemporaryTopic.
protected class	PubSubConsumer.MyTopicListener This is the JMS MessageListener implementation for OpenEAI PubSubConsumers.

Nested classes inherited from class org.openeai.jms.consumer.MessageConsumer

MessageConsumer.MessageTransaction

Field Summary

Fields inherited from class org.openeai.jms.consumer.MessageConsumer

CONNECTED, DISCONNECTED, INITIALIZED, m_commandConfigs, m_inProcessMessages, m_messages, m_numThreads, m_password, m_startOnInitialization, m_threadPool, m_userName, NOT_INITIALIZED, STARTED, STOPPED

Fields inherited from class org.openeai.OpenEaiObject

logger

Constructor Summary

PubSubConsumer()

PubSubConsumer(ConsumerConfig cConfig)
As AppConfig reads through a gateway's deployment document, it will build a ConsumerConfig Java object and pass that object to this constructor.

PubSubConsumer(java.lang.String cfName, java.lang.String tName)

Method Summary

PubSubMessageBalancer	getBalancer() Returns the PubSubMessageBalancer associated to this Consumer.
javax.jms.Topic	getTopic() Returns the Consumer's Topic object.
javax.jms.TopicConnection	getTopicConnection() Returns the Consumer's TopicConnection object.
javax.jms.TopicConnectionFactory	getTopicConnectionFactory() Returns the Consumer's TopicConnectionFactory object.
javax.jms.TopicSession	getTopicSession() Returns the Consumer's TopicSession object.
javax.jms.TopicSubscriber	getTopicSubscriber() Returns the Consumer's TopicSubscriber object.
protected void	init(java.util.Properties props) Invokes MessageConsumer.init(Properties) and adds the ConsumerShutDownHook for this consumer.
void	setBalancer(PubSubMessageBalancer balancer) Sets the PubSubMessageBalancer associated to this Consumer.
void	setTopic(javax.jms.Topic topic) Sets the Consumer's Topic object.
void	setTopicConnection(javax.jms.TopicConnection topicConnection) Sets the Consumer's TopicConnection object.
void	setTopicConnectionFactory(javax.jms.TopicConnectionFactory tcf) Sets the Consumer's TopicConnectionFactory object.
void	setTopicSession(javax.jms.TopicSession session) Sets the Consumer's TopicSession object.
void	setTopicSubscriber(javax.jms.TopicSubscriber topicSubscriber) Sets the Consumer's TopicSubscriber object.
void	startConsumer() Starts the consumer making it ready to consume messages from the Topic that it connects to.
void	startMonitor() This method starts the Consumer's "Monitor Thread".
void	stopConsumer() Attempts to cleanly shutdown the Consumer.
void	stopMonitor() This method stops the Consumer's "Monitor Thread" so it won't attempt to restart the consumer.

Methods inherited from class org.openeai.jms.consumer.MessageConsumer

addCommandConfig, addSyncCommand, consumptionStopped, getAbsoluteCommandName, getCommandConfig, getConfig, getConnectionFactoryName, getConsumerName, getConsumerStatus, getDefaultCommandName, getDestinationName, getGenericErrorDoc, getInitialContext, getInitialContextFactory, getInitializationStatus, getInstanceName, getPassword, getProviderUrl, getSecurityCredentials, getSecurityPrincipal, getStartOnInitialization, getSyncCommand, getThreadPool, getTransacted, getUserName, handleSync, initializeConsumer, initSyncCommands, isStarted, setAbsoluteCommandName, setCommandConfigs, setConfig, setConnectionFactoryName, setConsumerName, setConsumerStatus, setDefaultCommandName, setDestinationName, setGenericErrorDoc, setInitialContextFactory, setInitializationStatus, setInstanceName, setPassword, setProviderUrl, setSecurityCredentials, setSecurityPrincipal, setStartOnInitialization, setThreadPool, setTransacted, setUserName, stopConsumption

Methods inherited from class org.openeai.OpenEaiObject

addLog4jProperty, getAppName, getDebug, getFromAddr, getLog4jProperties, getMailHost, getMailService, getProperties, getToAddr, initializeLog4j, initializeLog4j, initializeLog4j, setAppName, setDebug, setFromAddr, setMailHost, setMailService, setProperties, setToAddr

Methods inherited from class java.lang.Object

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

Constructor Detail

PubSubConsumer

public PubSubConsumer()

PubSubConsumer

public PubSubConsumer(java.lang.String cfName,
                      java.lang.String tName)

PubSubConsumer

public PubSubConsumer(ConsumerConfig cConfig)
               throws javax.jms.JMSException,
                      javax.naming.NamingException,
                      java.io.IOException

As AppConfig reads through a gateway's deployment document, it will build a ConsumerConfig Java object and pass that object to this constructor. Then this consumer will have all the information it needs to initialize itself which includes:

• Initializing the consumer itself
• Initializing all Commands that the consumer may execute

  See Also:
  ConsumerConfig

Method Detail

stopMonitor

public void stopMonitor()

This method stops the Consumer's "Monitor Thread" so it won't attempt to restart the consumer.

When the consumer is started it starts a Thread that monitors the Consumer's connection to the broker. If that connection is broken for some reason, that "Monitor Thread" will attempt to restart the consumer. This continues indefinitely until the consumer is able to re-connect to the broker.

This method allows an application to in effect stop that monitor thread so they can shut the consumer down without it restarting itself.

startMonitor

public void startMonitor()

This method starts the Consumer's "Monitor Thread". This is a thread that runs for the life of the consumer and checks the status of the consumer's connection to the broker every thirty seconds. If that connection is broken for some reason, the Monitor Thread will attempt to restart the consumer, re-connecting it to the broker. It will continue to do this until either the consumer is able to re-connect or the consumer is shutdown.

getBalancer

public final PubSubMessageBalancer getBalancer()

Returns the PubSubMessageBalancer associated to this Consumer.

Returns:

PubSubMessageBalancer

See Also:

PubSubMessageBalancer

setBalancer

public final void setBalancer(PubSubMessageBalancer balancer)

Sets the PubSubMessageBalancer associated to this Consumer.

See Also:

PubSubMessageBalancer

getTopicConnectionFactory

public final javax.jms.TopicConnectionFactory getTopicConnectionFactory()

Returns the Consumer's TopicConnectionFactory object.

See the JMS Specification to learn more about JMS objects.

Returns:

javax.jms.TopicConnectionFactory

setTopicConnectionFactory

public final void setTopicConnectionFactory(javax.jms.TopicConnectionFactory tcf)

Sets the Consumer's TopicConnectionFactory object.

See the JMS Specification to learn more about JMS objects.

getTopic

public final javax.jms.Topic getTopic()

Returns the Consumer's Topic object.

See the JMS Specification to learn more about JMS objects.

Returns:

javax.jms.Topic

setTopic

public final void setTopic(javax.jms.Topic topic)

Sets the Consumer's Topic object.

See the JMS Specification to learn more about JMS objects.

getTopicConnection

public final javax.jms.TopicConnection getTopicConnection()

Returns the Consumer's TopicConnection object.

See the JMS Specification to learn more about JMS objects.

Returns:

javax.jms.TopicConnection

setTopicConnection

public final void setTopicConnection(javax.jms.TopicConnection topicConnection)

Sets the Consumer's TopicConnection object.

See the JMS Specification to learn more about JMS objects.

getTopicSession

public final javax.jms.TopicSession getTopicSession()

Returns the Consumer's TopicSession object.

See the JMS Specification to learn more about JMS objects.

Returns:

javax.jms.TopicSession

setTopicSession

public final void setTopicSession(javax.jms.TopicSession session)

Sets the Consumer's TopicSession object.

See the JMS Specification to learn more about JMS objects.

getTopicSubscriber

public final javax.jms.TopicSubscriber getTopicSubscriber()

Returns the Consumer's TopicSubscriber object.

See the JMS Specification to learn more about JMS objects.

Returns:

javax.jms.TopicSubscriber

setTopicSubscriber

public final void setTopicSubscriber(javax.jms.TopicSubscriber topicSubscriber)

Sets the Consumer's TopicSubscriber object.

See the JMS Specification to learn more about JMS objects.

startConsumer

public final void startConsumer()
                         throws javax.naming.NamingException,
                                javax.jms.JMSException

Starts the consumer making it ready to consume messages from the Topic that it connects to. This follows the typical JMS pattern of starting a message consumer. This includes:

• Retrieving the JMS Administered objects (TopicConnectionFactory and Topic) from a directory server or other JNDI source
• Creating a TopicConnection with the TopicConnectionFactory
• Creating a TopicSession with the TopicConnection
• Creating a TopicSubscriber with the TopicSession and Topic. NOTE: All OpenEAI PubSubConsumers establish a durable subscription when they start.
• Establishing the MessageListener that will be used when messages are delivered to the Topic.

Additionally, this method starts the Consumer's Monitor that will monitor and attempt to resolve any broker connection issues encountered for the life of the Consumer.

Throws:

javax.jms.JMSException

javax.naming.NamingException

See Also:

PointToPointConsumer.startConsumer(), PubSubConsumer.MonitorConsumer

stopConsumer

public final void stopConsumer()

Attempts to cleanly shutdown the Consumer. This includes closing all JMS resources (TopicSubscriber, TopicSession and TopicConnection). If errors occur, it will log those errors as warnings. However, regardless of the outcome of the "clean" shutdown attempt, the consumer will be stopped. This method is called anytime the consumer detects connection problems to the broker or when the consumer receives a shutdown hook from the operating system.

NOTE: This method DOES NOT unsubscribe from durable subscriptions. This way, when the gateway is brought back up, its consumers will have any missed messages available for consumption. Durable subscriptions must be cleaned up manually if that is desired.

Specified by:

stopConsumer in class MessageConsumer

See Also:

PubSubConsumer.MonitorConsumer, PubSubConsumer.ConsumerShutdownHook

init

protected void init(java.util.Properties props)
             throws java.io.IOException

Invokes MessageConsumer.init(Properties) and adds the ConsumerShutDownHook for this consumer.

Overrides:

init in class MessageConsumer

Throws:

java.io.IOException

See Also:

MessageConsumer.init(Properties)