Tag Archives: async

Decoupling event producers and event consumers in Java EE 6 using CDI and JMS

In this post I will show you how to decouple event producer from the event consumers (aka async events).

  1. you must configure your application server (here the glassfish settings):
  2. call the EJB method sendEvent() to send the event (from a servlet or from a web page):
    @LocalBean
    @Stateless
    public class MyEJB {
    
    	@Inject
    	@InForeground
    	private Event<BaseEvent> event;
    
    	/**
    	 * fires the @InForeground event
    	 *
    	 * @throws InterruptedException
    	 */
    	public void sendEvent() throws InterruptedException {
    		event.fire(new MyEvent("Hello world event", new Date()));
    		Thread.sleep(5000);
    	}
    }
  3. the EJB fires the event anotaded with @InForeground qualifier.
  4. An observer method handle the event and send it to the JMS Queue with the help of a message producer:
    @Stateless
    public class BackgroundEventSender implements BackgroundEventSenderItf {
    
    	@Resource(mappedName = "myConnectionFactory") // the JNDI name for this connection factory
    	private ConnectionFactory connectionFactory;
    
    	@Resource(mappedName = "myQueue") // JNDI name for this queue
    	private Queue backgroundEventQueue;
    
    	private Connection connection;
    	private Session session;
    	private MessageProducer producer;
    
    	@PostConstruct
    	public void init() {
    		try {
    			connection = connectionFactory.createConnection();
    			session = connection.createSession(true, Session.AUTO_ACKNOWLEDGE);
    			producer = session.createProducer(backgroundEventQueue);
    		} catch (JMSException ex) {
    			throw new RuntimeException(ex.getMessage(), ex);
    		}
    	}
    
    	@PreDestroy
    	public void destroy() {
    		try {
    			if (connection != null) {
    				connection.close();
    			}
    		} catch (Exception ex) {
    			// todo: handle the exception
    		}
    
    	}
    
    	/**
    	 * sends the foreground event to the queue with the help of the producer object
    	 *
    	 * @param event
    	 */
    	@Override
    	public void event(@Observes @InForeground BaseEvent event) {
    		try {
    			ObjectMessage msg = session.createObjectMessage();
    			msg.setObject(event);
    			producer.send(msg);
    		} catch (JMSException ex) {
    			throw new RuntimeException(ex.getMessage(), ex);
    		}
    
    	}
    
    }
  5. Now, in another (background) thread, the onMessage method handle the message and fires a new event (which wraps the received event) now annotated with @InBackground qualifier:
    @MessageDriven(mappedName = "myQueue", activationConfig = {
    	@ActivationConfigProperty(propertyName = "acknowledgeMode", propertyValue = "Auto-acknowledge"),
    	@ActivationConfigProperty(propertyName = "destinationType", propertyValue = "javax.jms.Queue")
    })
    public class BackgroundEventDispatcher implements MessageListener {
    
    	public BackgroundEventDispatcher() {
    	}
    
    	@Inject
    	@InBackground
    	private Event<BaseEvent> event;
    
    	/**
    	 * dispatches the message received from the queue and fires a new event (now with the @InBackground qualifier)
    	 *
    	 * @param message
    	 */
    	public void onMessage(Message message) {
    		if (!(message instanceof ObjectMessage)) {
    			throw new RuntimeException("Invalid message type received");
    		}
    		ObjectMessage msg = (ObjectMessage) message;
    		try {
    			Serializable eventObject = msg.getObject();
    			if (!(eventObject instanceof BaseEvent)) {
    				throw new RuntimeException("Unknown event type received");
    			}
    			BaseEvent evt = (BaseEvent) eventObject;
    			this.event.fire(evt);
    
    		} catch (JMSException ex) {
    			throw new RuntimeException(ex.getMessage(), ex);
    		}
    	}
    }
  6. And finally, if you have a method that consumes events annotated with @InBackground qualifier, then you receive the event and process it accordingly to your business needs.
    @Stateless
    public class EventConsumer implements EventConsumerItf {
    
    	/**
    	 * consumes the event sent by the JMS queue
    	 *
    	 * @param event
    	 */
    	@Override
    	public void afterMyEvent(@Observes @InBackground MyEvent event) {
    		System.out.println("Event message: " + event.getData());
    		System.out.println("Event time: " + new SimpleDateFormat("yyyy.MM.dd @ HH:mm:ss").format(event.getEventTime()));
    	}
    
    }

The event object it’s just a POJO like this one:

public class MyEvent extends BaseEvent {

	private String data;

	private Date eventTime;

	public String getData() {
		return data;
	}

	public MyEvent(String data, Date eventTime) {
		this.data = data;
		this.eventTime = eventTime;
	}

	public void setData(String data) {
		this.data = data;
	}

	public Date getEventTime() {
		return eventTime;
	}

	public void setEventTime(Date eventTime) {
		this.eventTime = eventTime;
	}
}

You can get the full code from github. Thank you!

Feel free to comment.