#HOWTO: Simple JMS application with the embedded messaging engine in Open Liberty

For messaging, I have always thought of Apache Kafka as the central solution for messaging. While learning more about Java EE I came along the Java Message Service (JMS) specification and gave it a try. In addition, I wanted to learn more about the JMS capabilities of the Open Liberty application server.

In this blog post, I’ll show you how to create a simple messaging application with JMS, JPA and JSON-B which will be deployed to Open Liberty 18.0.0.2 within a Docker container. For simplification, I’ll use JMS to communicate within one .war  using a JMS queue. The application will periodically publish a simple entity as JSON and the message will be received by a Message Driven Bean and stored to the database.

With JMS you can choose between queues for point-to-point communication or topics for publish-subscribe use cases. The Open Liberty server has the built-in feature wasJmsServer-1.0 which gets activated while using the Java EE 8 full profile feature javaee-8.0. With this feature enabled you can configure the embedded messaging engine like the following:

<server>
  <!-- ... other configuration -->
 <messagingEngine>
        <queue id="QUEUE1" />
    </messagingEngine>

    <jmsQueueConnectionFactory jndiName="jms/JmsFactory">
        <properties.wasJms remoteServerAddress="localhost:7276:BootStrapBasicMessaging" />
    </jmsQueueConnectionFactory>

    <jmsQueue id="simpleJmsQueue" jndiName="jms/JmsQueue">
        <properties.wasJms queueName="QUEUE1" />
    </jmsQueue>
</server>

As Open Liberty doesn’t provide an embedded relational database out of the box, I have to configure it in the server.xml file. Therefore I have to add the embedded database (Apache Derby.jar file to the application server and define a datasource like the following:

<server>
  <!-- ... other configuration -->
 <dataSource id="DefaultDataSource">
        <jdbcDriver libraryRef="DERBY_JDBC_LIB" />
        <properties.derby.embedded databaseName="test" createDatabase="create" />
    </dataSource>

    <library id="DERBY_JDBC_LIB">
        <file name="${server.config.dir}/derby.jar" />
    </library>
</server>

The central message for the JMS communication and for JPA looks like the following:

@Entity
public class CustomMessage {

    @Id
    @GeneratedValue(strategy = GenerationType.AUTO)
    private Long id;

    private String content;
    private String author;
    private long createdAt;

    // getters & setters
}

To publish a new message every two seconds, I’ll use a simple EJB to schedule this task. In addition, I’ll check the number of messages in the database every ten seconds:

@Singleton
public class SimpleTimer {

    @Inject
    private JmsMessageSender sender;

    @PersistenceContext
    private EntityManager entityManager;

    @Schedule(second = "*/2", minute = "*", hour = "*", persistent = false)
    public void sendJmsMessage() {
        sender.send();
    }

    @Schedule(second = "*/10", minute = "*", hour = "*", persistent = false)
    public void checkAmountOfMessage() {
        System.out.println(entityManager.createQuery("SELECT m FROM CustomMessage m").getResultList().size() + " messages are stored in the database");
    }
}

The JmsMessgeSender class is an EJB which connects to the embedded JmsFactory and creates a MessageProducer for the JMS queue. As the JMS classes Connection, Session and MessageProducer all extend the AutoClosable interface, I’ll use Java’s try-with-resources statement to automatically close the opened connections:

@Stateless
public class JmsMessageSender {

    @Resource(lookup = "jms/JmsFactory")
    private ConnectionFactory jmsFactory;

    @Resource(lookup = "jms/JmsQueue")
    private Queue jmsQueue;

    public void send() {

        TextMessage message;

        try (Connection connection = jmsFactory.createConnection();
             Session session = connection.createSession(false, Session.AUTO_ACKNOWLEDGE);
             MessageProducer producer = session.createProducer(jmsQueue)) {

            System.out.println("Sending a new message");
            message = session.createTextMessage();
            message.setText(createCustomMessage());
            producer.send(message);

        } catch (JMSException e) {
            e.printStackTrace();
        }
    }

    private String createCustomMessage() {
        CustomMessage msg = new CustomMessage("Hello World!", "Duke", Instant.now().getEpochSecond());
        Jsonb jsonb = JsonbBuilder.create();
        return jsonb.toJson(msg);
    }
}

For message serialization, I use JSON-B to create a JSON representation of the object.

To receive the message within the JMS queue, I’ll use a Message Driven Bean where I just have to implement the onMessage(Message message) method:

@MessageDriven
public class JmsMessageReader implements MessageListener {

    @PersistenceContext
    private EntityManager entityManager;

    @Override
    public void onMessage(Message message) {

        TextMessage textMessage = (TextMessage) message;

        try {
            String incomingText = textMessage.getText();
            System.out.println("-- a new message arrived: " + incomingText);

            Jsonb jsonb = JsonbBuilder.create();

            CustomMessage parsedMessage = jsonb.fromJson(incomingText, CustomMessage.class);
            entityManager.persist(parsedMessage);

        } catch (JMSException e) {
            System.err.println(e.getMessage());
        }
    }
}

This bean will cast the incoming generic Messageto a TextMessage and use JSON-B again to parse the JSON.

For developing Message Driven Beans, Open Liberty provides a feature called mdb-3.2 which is activated with the javaee-8.0 feature. To receive the incoming message within the Message Driven Bean you need this feature and you have to configure a so-called jmsActivationSpec within the server.xml:

<server>
 <!-- ... other configuration --> 
 <jmsActivationSpec id="embedded-messaging-engine-open-liberty/JmsMessageReader">
        <properties.wasJms
                destinationRef="simpleJmsQueue"
                destinationType="javax.jms.Queue"
                remoteServerAddress="localhost:7276:BootstrapBasicMessaging"/>
 </jmsActivationSpec>
</server>

The id attribute has the following structure: {nameOfTheWar}/{classNameOfTheMessageDrivenBean}. The destinationRef has to point to the already configured JmsQueue within the server.xml.

Running the application should result in the following log:

Sending a new message
-- a new message arrived: {"author":"Duke","content":"Hello World!","createdAt":1534686348}
8 messages are stored in the database
Sending a new message
-- a new message arrived: {"author":"Duke","content":"Hello World!","createdAt":1534686350}
Sending a new message
-- a new message arrived: {"author":"Duke","content":"Hello World!","createdAt":1534686352}

For a simple deployment on your machine, I created a Dockerfile which will copy the required .jar file and the server configuration to the Open Liberty server.

FROM open-liberty:javaee8
COPY server.xml /config/
COPY derby.jar /config/
ADD target/embedded-messaging-engine-open-liberty.war /config/dropins/

You can find the whole codebase as always on GitHub.

Happy messaging,

Phil.

Leave a comment

Your email address will not be published. Required fields are marked *