package com.codenotfound.jms.ptp;

import javax.jms.Connection;
import javax.jms.ConnectionFactory;
import javax.jms.JMSException;
import javax.jms.MessageProducer;
import javax.jms.Queue;
import javax.jms.Session;
import javax.jms.TextMessage;

import org.apache.activemq.ActiveMQConnection;
import org.apache.activemq.ActiveMQConnectionFactory;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

public class Producer {

  private static final Logger LOGGER =
      LoggerFactory.getLogger(Producer.class);

  private String clientId;
  private Connection connection;
  private Session session;
  private MessageProducer messageProducer;

  public void create(String clientId, String queueName)
      throws JMSException {
    this.clientId = clientId;

    // create a Connection Factory
    ConnectionFactory connectionFactory =
        new ActiveMQConnectionFactory(
            ActiveMQConnection.DEFAULT_BROKER_URL);

    // create a Connection
    connection = connectionFactory.createConnection();
    connection.setClientID(clientId);

    // create a Session
    session =
        connection.createSession(false, Session.AUTO_ACKNOWLEDGE);

    // create the Queue to which messages will be sent
    Queue queue = session.createQueue(queueName);

    // create a MessageProducer for sending messages
    messageProducer = session.createProducer(queue);
  }

  public void closeConnection() throws JMSException {
    connection.close();
  }

  public void sendName(String firstName, String lastName)
      throws JMSException {
    String text = firstName + " " + lastName;

    // create a JMS TextMessage
    TextMessage textMessage = session.createTextMessage(text);

    // send the message to the queue destination
    messageProducer.send(textMessage);

    LOGGER.debug(clientId + ": sent message with text='{}'", text);
  }
}

package com.codenotfound.jms.ptp;
import javax.jms.Connection;
import javax.jms.ConnectionFactory;
import javax.jms.JMSException;
import javax.jms.Message;
import javax.jms.MessageConsumer;
import javax.jms.Queue;
import javax.jms.Session;
import javax.jms.TextMessage;
import org.apache.activemq.ActiveMQConnection;
import org.apache.activemq.ActiveMQConnectionFactory;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
public class Consumer {
private static final Logger LOGGER =
LoggerFactory.getLogger(Consumer.class);
private static String NO_GREETING = "no greeting";
private String clientId;
private Connection connection;
private MessageConsumer messageConsumer;
public void create(String clientId, String queueName)
throws JMSException {
this.clientId = clientId;
// create a Connection Factory
ConnectionFactory connectionFactory =
new ActiveMQConnectionFactory(
ActiveMQConnection.DEFAULT_BROKER_URL);
// create a Connection
connection = connectionFactory.createConnection();
connection.setClientID(clientId);
// create a Session
Session session =
connection.createSession(false, Session.CLIENT_ACKNOWLEDGE);
// create the Queue from which messages will be received
Queue queue = session.createQueue(queueName);
// create a MessageConsumer for receiving messages
messageConsumer = session.createConsumer(queue);
// start the connection in order to receive messages
connection.start();
}
public void closeConnection() throws JMSException {
connection.close();
}
public String getGreeting(int timeout, boolean acknowledge)
throws JMSException {
String greeting = NO_GREETING;
// read a message from the queue destination
Message message = messageConsumer.receive(timeout);
// check if a message was received
if (message != null) {
// cast the message to the correct type
TextMessage textMessage = (TextMessage) message;
// retrieve the message content
String text = textMessage.getText();
LOGGER.debug(clientId + ": received message with text='{}'",
text);
if (acknowledge) {
// acknowledge the successful processing of the message
message.acknowledge();
LOGGER.debug(clientId + ": message acknowledged");
} else {
LOGGER.debug(clientId + ": message not acknowledged");
}
// create greeting
greeting = "Hello " + text + "!";
} else {
LOGGER.debug(clientId + ": no message received");
}
LOGGER.info("greeting={}", greeting);
return greeting;
}
}

package com.codenotfound.jms.ptp;
import static org.junit.Assert.assertEquals;
import static org.junit.Assert.fail;
import javax.jms.JMSException;
import javax.naming.NamingException;
import org.junit.AfterClass;
import org.junit.BeforeClass;
import org.junit.Test;
public class ConsumerTest {
private static Producer producerPointToPoint,
producerOnlyOneConsumer, producerNoTimingDependencies,
producerAcknowledgeProcessing;
private static Consumer consumerPointToPoint,
consumer1OnlyOneConsumer, consumer2OnlyOneConsumer,
consumerNoTimingDependencies, consumer1AcknowledgeProcessing,
consumer2AcknowledgeProcessing;
@BeforeClass
public static void setUpBeforeClass()
throws JMSException, NamingException {
producerPointToPoint = new Producer();
producerPointToPoint.create("producer-pointtopoint",
"pointtopoint.q");
producerOnlyOneConsumer = new Producer();
producerOnlyOneConsumer.create("producer-onlyoneconsumer",
"onlyoneconsumer.q");
producerNoTimingDependencies = new Producer();
producerNoTimingDependencies.create(
"producer-notimingdependencies", "notimingdependencies.q");
producerAcknowledgeProcessing = new Producer();
producerAcknowledgeProcessing.create(
"producer-acknowledgeprocessing", "acknowledgeprocessing.q");
consumerPointToPoint = new Consumer();
consumerPointToPoint.create("consumer-pointtopoint",
"pointtopoint.q");
consumer1OnlyOneConsumer = new Consumer();
consumer1OnlyOneConsumer.create("consumer1-onlyoneconsumer",
"onlyoneconsumer.q");
consumer2OnlyOneConsumer = new Consumer();
consumer2OnlyOneConsumer.create("consumer2-onlyoneconsumer",
"onlyoneconsumer.q");
// consumerNoTimingDependencies
consumer1AcknowledgeProcessing = new Consumer();
consumer1AcknowledgeProcessing.create(
"consumer1-acknowledgeprocessing", "acknowledgeprocessing.q");
consumer2AcknowledgeProcessing = new Consumer();
consumer2AcknowledgeProcessing.create(
"consumer2-acknowledgeprocessing", "acknowledgeprocessing.q");
}
@AfterClass
public static void tearDownAfterClass() throws JMSException {
producerPointToPoint.closeConnection();
producerOnlyOneConsumer.closeConnection();
producerNoTimingDependencies.closeConnection();
producerAcknowledgeProcessing.closeConnection();
consumerPointToPoint.closeConnection();
consumer1OnlyOneConsumer.closeConnection();
consumer2OnlyOneConsumer.closeConnection();
consumerNoTimingDependencies.closeConnection();
// consumer1AcknowledgeProcessing
consumer2AcknowledgeProcessing.closeConnection();
}
@Test
public void testGetGreeting() {
try {
producerPointToPoint.sendName("Frodo", "Baggins");
String greeting = consumerPointToPoint.getGreeting(1000, true);
assertEquals("Hello Frodo Baggins!", greeting);
} catch (JMSException e) {
fail("a JMS Exception occurred");
}
}
@Test
public void testOnlyOneConsumer() throws InterruptedException {
try {
producerOnlyOneConsumer.sendName("Legolas", "Greenleaf");
String greeting1 =
consumer1OnlyOneConsumer.getGreeting(1000, true);
assertEquals("Hello Legolas Greenleaf!", greeting1);
Thread.sleep(1000);
String greeting2 =
consumer2OnlyOneConsumer.getGreeting(1000, true);
// each message has only one consumer
assertEquals("no greeting", greeting2);
} catch (JMSException e) {
fail("a JMS Exception occurred");
}
}
@Test
public void testNoTimingDependencies() {
try {
producerNoTimingDependencies.sendName("Samwise", "Gamgee");
// a receiver can fetch the message whether or not it was running
// when the client sent the message
consumerNoTimingDependencies = new Consumer();
consumerNoTimingDependencies.create(
"consumer-notimingdependencies", "notimingdependencies.q");
String greeting =
consumerNoTimingDependencies.getGreeting(1000, true);
assertEquals("Hello Samwise Gamgee!", greeting);
} catch (JMSException e) {
fail("a JMS Exception occurred");
}
}
@Test
public void testAcknowledgeProcessing()
throws InterruptedException {
try {
producerAcknowledgeProcessing.sendName("Gandalf", "the Grey");
// consume the message without an acknowledgment
String greeting1 =
consumer1AcknowledgeProcessing.getGreeting(1000, false);
assertEquals("Hello Gandalf the Grey!", greeting1);
// close the MessageConsumer so the broker knows there is no
// acknowledgment
consumer1AcknowledgeProcessing.closeConnection();
String greeting2 =
consumer2AcknowledgeProcessing.getGreeting(1000, true);
assertEquals("Hello Gandalf the Grey!", greeting2);
} catch (JMSException e) {
fail("a JMS Exception occurred");
}
}
}