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.
JMS configuration for Open Liberty
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:
| 1 2 3 4 5 6 7 8 9 10 11 12 13 14 | <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 data source like the following:
| 1 2 3 4 5 6 7 8 9 10 11 | <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> |
Sending and receiving JMS messages
The central message for the JMS communication and for JPA looks like the following:
| 1 2 3 4 5 6 7 8 9 10 11 12 13 | @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:
| 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 | @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 that 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:
| 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 | @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:
| 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 | @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 Message to 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:
| 1 2 3 4 5 6 7 8 9 | <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:
| 1 2 3 4 5 6 7 | 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.
| 1 2 3 4 | 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.
