import java.io.PrintStream;
import javax.jms.BytesMessage;
import javax.jms.Connection;
import javax.jms.Destination;
import javax.jms.ExceptionListener;
import javax.jms.JMSException;
import javax.jms.Message;
import javax.jms.MessageConsumer;
import javax.jms.MessageEOFException;
import javax.jms.MessageListener;
import javax.jms.Queue;
import javax.jms.QueueConnection;
import javax.jms.QueueSender;
import javax.jms.QueueSession;
import javax.jms.Session;
import javax.jms.TextMessage;
import org.apache.activemq.ActiveMQConnection;
import org.apache.activemq.ActiveMQConnectionFactory;
public class Server implements MessageListener, ExceptionListener {
Session session;
public Server() throws Exception {
ActiveMQConnectionFactory connectionFactory = new ActiveMQConnectionFactory(ActiveMQConnection.DEFAULT_USER, ActiveMQConnection.DEFAULT_PASSWORD, ActiveMQConnection.DEFAULT_BROKER_URL);
Connection connection = connectionFactory.createQueueConnection();
connection.setClientID(“clientID”);
connection.setExceptionListener(this);
connection.start();
session = connection.createSession(true, Session.AUTO_ACKNOWLEDGE);
Destination destination = session.createQueue(“ServerSideQ”);
MessageConsumer consumer = session.createConsumer(destination);
consumer.setMessageListener(this);
//if (session != null) session.close();
//if (connection != null) connection.close();
}
public static void main(String[] args) throws Exception {
new Server();
Thread.currentThread().join();
}
public void onMessage(Message message) {
try {
if (message instanceof TextMessage) {
TextMessage txtMsg = (TextMessage) message;
String msg = txtMsg.getText();
System.out.println(“Received: ” + msg);
} else if (message instanceof BytesMessage) {
BytesMessage bytMsg = (BytesMessage) message;
StringBuffer msg = new StringBuffer();
int c;
try {
while ((c=bytMsg.readByte()) != -1) {
msg.append((char) c);
}
} catch (javax.jms.MessageEOFException e) {}
System.out.println(“Received: ” + msg);
}
else {
System.out.println(“Unknown Type ” + message.getClass().getName() + ” Got: ” + message);
}
session.commit();
reply(message);
/*
if (++count % dumpCount == 0) {
dumpStats(connection);
}
*/
}
catch (Exception e) {
System.out.println(“Caught: ” + e);
e.printStackTrace();
}
}
synchronized public void onException(JMSException ex) {
System.out.println(“JMS Exception occured. Shutting down client.”);
}
public void reply(Message msg) throws Exception {
ActiveMQConnectionFactory connectionFactory = new ActiveMQConnectionFactory(ActiveMQConnection.DEFAULT_USER, ActiveMQConnection.DEFAULT_PASSWORD, ActiveMQConnection.DEFAULT_BROKER_URL);
QueueConnection qconnection = connectionFactory.createQueueConnection();
qconnection.setExceptionListener(this);
qconnection.start();
QueueSession qsession = (QueueSession) qconnection.createQueueSession(true, Session.AUTO_ACKNOWLEDGE);
String subject = msg.getJMSReplyTo().toString();
subject = subject.substring(subject.lastIndexOf(‘/’)+1);
//Queue queue = (Queue) session.createQueue(subject);
Queue queue = (Queue) msg.getJMSReplyTo();
QueueSender sender = qsession.createSender(queue);
TextMessage responseMessage = qsession.createTextMessage(“hello back”);
//BytesMessage responseMessage = qsession.createBytesMessage();
//responseMessage.writeUTF(“Response to ” + msg);
responseMessage.setStringProperty(“KEY”, “key”);
sender.send(responseMessage);
qsession.commit();
sender.close();
qconnection.close();
System.out.println(“—Sent ” + responseMessage.toString().substring(0, 20) + ” to ” + subject + “—” + queue);
}
}