package com.oreilly.jent.jms;
/**
* In general, you may use the code in this book in your programs and
* documentation. You do not need to contact us for permission unless
* you're reproducing a significant portion of the code. For example,
* writing a program that uses several chunks of code from this book does
* not require permission. Selling or distributing a CD-ROM of examples
* from O'Reilly books does require permission. Answering a question by
* citing this book and quoting example code does not require permission.
* Incorporating a significant amount of example code from this book into
* your product's documentation does require permission.
*
* We appreciate, but do not require, attribution. An attribution usually
* includes the title, author, publisher, and ISBN. For example:
*
* "Java Enterprise in a Nutshell, Third Edition,
* by Jim Farley and William Crawford
* with Prakash Malani, John G. Norman, and Justin Gehtland.
* Copyright 2006 O'Reilly Media, Inc., 0-596-10142-2."
*
* If you feel your use of code examples falls outside fair use or the
* permission given above, feel free to contact us at
* permissions@oreilly.com.
*/
/**
* PubSubMessagingClient: A JMS publish-subscribe messaging client.
* This client can either send or receive messages from a given Topic. The
* main method allows for running the client from the command-line.
*/
import javax.jms.JMSException;
import javax.jms.Message;
import javax.jms.MessageListener;
import javax.jms.TextMessage;
import javax.jms.Topic;
import javax.jms.TopicConnection;
import javax.jms.TopicConnectionFactory;
import javax.jms.TopicPublisher;
import javax.jms.TopicSession;
import javax.jms.TopicSubscriber;
import javax.naming.Context;
import javax.naming.InitialContext;
import javax.naming.NamingException;
public class PubSubMessagingClient implements Runnable {
// Our connection to the JMS provider. Only one is needed for this client.
private TopicConnection mTopicConn = null;
// The topic used for message-passing
private Topic mTopic = null;
// Our message subscriber - only need one.
private TopicSubscriber mSubscriber = null;
// A single session for sending and receiving from all remote peers.
private TopicSession mSession = null;
// The message type we tag all our messages with
private static String MSG_TYPE = "JavaEntMessage";
// Constructor, with client name, and the JNDI location of the JMS
// connection factory and topic that we want to use.
public PubSubMessagingClient(String cFactJNDIName, String topicJNDIName) {
init(cFactJNDIName, topicJNDIName);
}
// Do all the JMS-setup for this client. Assumes that the JVM is
// configured (perhaps using jndi.properties) so that the default JNDI
// InitialContext points to the JMS provider's JNDI service.
protected boolean init(String cFactoryJNDIName, String topicJNDIName) {
boolean success = true;
Context ctx = null;
// Attempt to make connection to JNDI service
try {
ctx = new InitialContext();
}
catch (NamingException ne) {
System.out.println("Failed to connect to JNDI provider:");
ne.printStackTrace();
success = false;
}
// If no JNDI context, bail out here
if (ctx == null) {
return success;
}
// Attempt to lookup JMS connection factory from JNDI service
TopicConnectionFactory connFactory = null;
try {
connFactory = (TopicConnectionFactory)ctx.lookup(cFactoryJNDIName);
System.out.println("Got JMS connection factory.");
}
catch (NamingException ne2) {
System.out.println("Failed to get JMS connection factory: ");
ne2.printStackTrace();
success = false;
}
try {
// Make a connection to the JMS provider and keep it
// At this point, the connection is not started, so we aren't
// receiving any messages.
mTopicConn = connFactory.createTopicConnection();
// Try to find our designated topic
mTopic = (Topic)ctx.lookup(topicJNDIName);
// Make a session for topicing messages
// no transactions, auto-acknowledge
mSession =
mTopicConn.createTopicSession(false,
javax.jms.Session.AUTO_ACKNOWLEDGE);
}
catch (JMSException e) {
System.out.println("Failed to establish connection/topic:");
e.printStackTrace();
success = false;
}
catch (NamingException ne) {
System.out.println("JNDI Error looking up factory or topic:");
ne.printStackTrace();
success = false;
}
try {
// Make our subscriber, for incoming messages
// Set the message selector to only receive our type of messages,
// in case the same topic is being used for other purposes
// Also indicate that we don't want any message sent from this connection
mSubscriber =
mSession.createSubscriber(mTopic, "JMSType = '" + MSG_TYPE + "'",
true);
}
catch (JMSException je) {
System.out.println("Error establishing message subscriber:");
je.printStackTrace();
}
return success;
}
// Send a message to the topic
public void publishMessage(String msg) {
try {
// Create a JMS msg publisher connected to the destination topic
TopicPublisher publisher = mSession.createPublisher(mTopic);
// Use the session to create a text message
TextMessage tMsg = mSession.createTextMessage();
tMsg.setJMSType(MSG_TYPE);
// Set the body of the message
tMsg.setText(msg);
// Send the message using the publisher
publisher.publish(tMsg);
System.out.println("Published the message");
}
catch (JMSException je) {
System.out.println("Error sending message " + msg + " to topic");
je.printStackTrace();
}
}
// Register a MessageListener with the topic to receive
// messages asynchronously
public void registerListener(MessageListener listener) {
try {
// Set the listener on the subscriber
mSubscriber.setMessageListener(listener);
// Start the connection, in case it's still stopped
mTopicConn.start();
}
catch (JMSException je) {
System.out.println("Error registering listener: ");
je.printStackTrace();
}
}
// Perform an synchronous receive of a message from the topic. If it's a
// TextMessage, print the contents.
public String receiveMessage() {
String msg = "-- No message --";
try {
// (Re)start the connection, in case it's not already started
mTopicConn.start();
// Check for a message
Message m = mSubscriber.receive();
if (m instanceof TextMessage) {
msg = ((TextMessage)m).getText();
}
else {
msg = "-- Unsupported message type received --";
}
}
catch (JMSException je) {
}
return msg;
}
// When run within a thread, just wait for messages to be delivered to us
public void run() {
while (true) {
try { this.wait(); } catch (Exception we) {}
}
}
// Take command-line arguments and send or receive messages from the
// named topic
public static void main(String args[]) {
if (args.length < 3) {
System.out.println("Usage: PubSubMessagingClient" +
" connFactoryName topicName" +
" [publish|subscribe|recv_synch] <messageToSend>");
System.exit(1);
}
// Get our client name, and the JNDI name of the connection factory and
// topic, from the command-line
String factoryName = args[0];
String topicName = args[1];
// Get the command to execute (publish, subscribe, recv_synch)
String cmd = args[2];
// Create and initialize the messaging participant
PubSubMessagingClient msger =
new PubSubMessagingClient(factoryName, topicName);
// Run the participant in its own thread, so that it can react to
// incoming messages
Thread listen = new Thread(msger);
listen.start();
// Send a message to the topic
if (cmd.equals("publish")) {
String msg = args[3];
msger.publishMessage(msg);
System.exit(0);
}
// Register a listener
else if (cmd.equals("subscribe")) {
MessageListener listener = new TextLogger();
msger.registerListener(listener);
System.out.println("Client listening to topic " + topicName
+ ". . .");
try { listen.wait(); } catch (Exception we) {}
}
// Synchronously receive a message from the topic
else if (cmd.equals("recv_synch")) {
String msg = msger.receiveMessage();
System.out.println("Received message: " + msg);
System.exit(0);
}
}
}
|
