EJB - Message Driven Beans

A message driven bean is a type of enterprise bean, which is invoked by EJB container when it receives a message from queue or topic. Message driven bean is a stateless bean and is used to do task asynchronously.

To demonstrate use of message driven bean, we will make use of EJB-persistence chapter and we need to do the following tasks −

Step 1 − Create table in database (Refer to EJB-Persistence chapter).

Step 2 − Create Entity class corresponding to table (Refer to EJB-Persistence chapter).

Step 3 − Create DataSource and Persistence Unit (Refer to EJB-Persistence chapter).

Step 4 − Create a stateless EJB having EntityManager instance (Refer to EJB-Persistence chapter).

Step 5 − Update stateless ejb.Add methods to add records and get records from database via entity manager (Refer to EJB-Persistence chapter).

Step 6 − Create a Queue named BookQueue in JBoss default apppcation directory.

Step 7 − A console based apppcation cpent will send message to this queue.

Step 8 − Create a Message driven bean, which will use the stateless bean to persist the cpent data.

Step 9 − EJB Container of jboss will call the above message driven bean and pass it the message that cpent will be sending to.

Create Queue

Create a file named jbossmq-destinations-service.xml if not exists in <JBoss Installation Folder> > server > default > deploy folder.

Here we are creating a queue named BookQueue −

jbossmq-destinations-service.xml

<mbean code="org.jboss.mq.server.jmx.Queue"  
   name="jboss.mq.destination:service=Queue,name=BookQueue">  
   <depends optional-attribute-name="DestinationManager">
      jboss.mq:service=DestinationManager
   </depends>  
</mbean>  

When you start the JBoss, you will see a similar entry in jboss log.

...
10:37:06,167 INFO  [QueueService] Queue[/queue/BookQueue] started, fullSize=200000, pageSize=2000, downCacheSize=2000
...

Create Message Driven Bean

@MessageDriven(
   name = "BookMessageHandler",
   activationConfig = {
      @ActivationConfigProperty( propertyName = "destinationType", 
                                 propertyValue = "javax.jms.Queue"),
      @ActivationConfigProperty( propertyName = "destination", 
                                 propertyValue ="/queue/BookQueue")
   }
)
pubpc class LibraryMessageBean implements MessageListener {
 
   @Resource
   private MessageDrivenContext mdctx;  
 
   @EJB
   LibraryPersistentBeanRemote pbraryBean;
 
   pubpc LibraryMessageBean() {        
   }
 
   pubpc void onMessage(Message message) {
   }
}

LibraryMessageBean is annotated with @MessageDriven annotation to mark it as message driven bean.

Its properties are defined as destinationType - Queue and destination - /queue/BookQueue.

It implements MessageListener interface, which exposes onMessage method.

It has MessgeDrivenContext as a resource.

LibraryPersistentBeanRemote stateless bean is injected in this bean for persistence purpose.

Build the EjbComponent project and deploy it on JBoss. After building and deploying the EJB module, we need a cpent to send a message to jboss queue.

Example Apppcation

Let us create a test EJB apppcation to test Message Driven Bean.

EJBComponent (EJB Module)

LibraryMessageBean.java

package com.tutorialspoint.messagebean;
 
import com.tutorialspoint.entity.Book;
import com.tutorialspoint.stateless.LibraryPersistentBeanRemote;
import javax.annotation.Resource;
import javax.ejb.ActivationConfigProperty;
import javax.ejb.EJB;
import javax.ejb.MessageDriven;
import javax.ejb.MessageDrivenContext;
import javax.jms.JMSException;
import javax.jms.Message;
import javax.jms.MessageListener;
import javax.jms.ObjectMessage;
 
@MessageDriven(
   name = "BookMessageHandler",
   activationConfig = {
      @ActivationConfigProperty( propertyName = "destinationType", 
                                 propertyValue = "javax.jms.Queue"),
      @ActivationConfigProperty( propertyName = "destination", 
                                 propertyValue ="/queue/BookQueue")
   }
)
pubpc class LibraryMessageBean implements MessageListener {
 
   @Resource
   private MessageDrivenContext mdctx;  
 
   @EJB
   LibraryPersistentBeanRemote pbraryBean;
 
   pubpc LibraryMessageBean() {        
   }
 
   pubpc void onMessage(Message message) {
      ObjectMessage objectMessage = null;
      try {
         objectMessage = (ObjectMessage) message;
         Book book = (Book) objectMessage.getObject(); 
         pbraryBean.addBook(book);
 
      } catch (JMSException ex) {
         mdctx.setRollbackOnly();
      }       
   }   
}

EJBTester (EJB Cpent)

EJBTester.java

package com.tutorialspoint.test;
   
import com.tutorialspoint.entity.Book;
import com.tutorialspoint.stateless.LibraryPersistentBeanRemote;
import java.io.BufferedReader;
import java.io.FileInputStream;
import java.io.IOException;
import java.io.InputStreamReader;
import java.util.List;
import java.util.Properties;
import javax.jms.ObjectMessage;
import javax.jms.Queue;
import javax.jms.QueueConnection;
import javax.jms.QueueConnectionFactory;
import javax.jms.QueueSender;
import javax.jms.QueueSession;
import javax.naming.InitialContext;
import javax.naming.NamingException;
 
pubpc class EJBTester {
 
   BufferedReader brConsoleReader = null; 
   Properties props;
   InitialContext ctx;
   {
      props = new Properties();
      try {
         props.load(new FileInputStream("jndi.properties"));
      } catch (IOException ex) {
         ex.printStackTrace();
      }
      try {
         ctx = new InitialContext(props);            
      } catch (NamingException ex) {
         ex.printStackTrace();
      }
      brConsoleReader = 
      new BufferedReader(new InputStreamReader(System.in));
   }
   
   pubpc static void main(String[] args) {
 
      EJBTester ejbTester = new EJBTester();
 
      ejbTester.testMessageBeanEjb();
   }
   
   private void showGUI() {
      System.out.println("**********************");
      System.out.println("Welcome to Book Store");
      System.out.println("**********************");
      System.out.print("Options 
1. Add Book
2. Exit 
Enter Choice: ");
   }
   
   private void testMessageBeanEjb() {
 
      try {
         int choice = 1; 
         Queue queue = (Queue) ctx.lookup("/queue/BookQueue");
         QueueConnectionFactory factory =
         (QueueConnectionFactory) ctx.lookup("ConnectionFactory");
         QueueConnection connection =  factory.createQueueConnection();
         QueueSession session = 
         connection.createQueueSession(false, QueueSession.AUTO_ACKNOWLEDGE);
         QueueSender sender = session.createSender(queue);
 
         while (choice != 2) {
            String bookName;
            showGUI();
            String strChoice = brConsoleReader.readLine();
            choice = Integer.parseInt(strChoice);
            if (choice == 1) {
               System.out.print("Enter book name: ");
               bookName = brConsoleReader.readLine();
               Book book = new Book();
               book.setName(bookName);
               ObjectMessage objectMessage = 
                  session.createObjectMessage(book);
               sender.send(objectMessage); 
            } else if (choice == 2) {
               break;
            }
         }
 
         LibraryPersistentBeanRemote pbraryBean = 
         (LibraryPersistentBeanRemote)
         ctx.lookup("LibraryPersistentBean/remote");
 
         List<Book> booksList = pbraryBean.getBooks();
 
         System.out.println("Book(s) entered so far: " + booksList.size());
         int i = 0;
         for (Book book:booksList) {
            System.out.println((i+1)+". " + book.getName());
            i++;
         }           
      } catch (Exception e) {
         System.out.println(e.getMessage());
         e.printStackTrace();
      }finally {
         try {
            if(brConsoleReader !=null) {
               brConsoleReader.close();
            }
         } catch (IOException ex) {
            System.out.println(ex.getMessage());
         }
      }
   }   
}

EJBTester performs the following tasks −

Load properties from jndi.properties and initiapze the InitialContext object.

In testStatefulEjb() method, jndi lookup is done with the name - "/queue/BookQueue" to obtain treference of queue available in Jboss. Then sender is created using queue session.

Then user is shown a pbrary store User Interface and he/she is asked to enter choice.

If user enters 1, the system asks for book name and sender sends the book name to queue. When JBoss container receives this message in queue, it calls our message driven bean s onMessage method. Our message driven bean then saves book using stateful session bean addBook() method. Session Bean is persisting the book in database via EntityManager call.

If user enters 2, then another jndi lookup is done with name - "LibraryStatefulSessionBean/remote" to obtain the remote business object (stateful EJB) again and psting of books is done.

Run Cpent to Access EJB

Locate EJBTester.java in project explorer. Right cpck on EJBTester class and select run file.

Verify the following output in Netbeans console −

run:
**********************
Welcome to Book Store
**********************
Options 
1. Add Book
2. Exit 
Enter Choice: 1
Enter book name: Learn EJB
**********************
Welcome to Book Store
**********************
Options 
1. Add Book
2. Exit 
Enter Choice: 2
Book(s) entered so far: 2
1. learn java
1. learn EJB
BUILD SUCCESSFUL (total time: 15 seconds)

The output shown above states that our Message driven bean is receiving the message and storing the book in persistent storage and books are retrieved from the database.