Guava EventBus Spring Advanced Integration

Guava EventBus Spring Advanced Integration

In previous tutorials you saw how to use Google Guava EventBus and integrate it with Spring framework. Today I will show you advance technique how to integrate EventBus with Spring.

Project set up is same as previous tutorial but for recap I will be showing here as well:

• Maven project structure:

• pom.xml:

<?xml version="1.0" encoding="UTF-8"?>
<project xmlns="http://maven.apache.org/POM/4.0.0" xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
	xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/xsd/maven-4.0.0.xsd">
	<modelVersion>4.0.0</modelVersion>

	<groupId>com.javahonk</groupId>
	<version>0.0.1-SNAPSHOT</version>
	<packaging>jar</packaging>

	<name>EventBusSpringIntegration</name>
	<url>http://maven.apache.org</url>

	<properties>
		<org.springframework.version>4.1.5.RELEASE</org.springframework.version>
		<log4j.version>1.2.16</log4j.version>
		<org.apache.log4j.version>2.1</org.apache.log4j.version>
		<guava.version>18.0</guava.version>
	</properties>

	<dependencies>

		<dependency>
			<groupId>org.springframework</groupId>
			<artifactId>spring-context</artifactId>
			<version>${org.springframework.version}</version>
		</dependency>

	   <!-- Log4j -->
		<dependency>
        	<groupId>org.apache.logging.log4j</groupId>
            <artifactId>log4j-api</artifactId>
            <version>${org.apache.log4j.version}</version>
		</dependency>
        
        <dependency>
        	<groupId>org.apache.logging.log4j</groupId>
            <artifactId>log4j-core</artifactId>
            <version>${org.apache.log4j.version}</version>
		</dependency>
		
		<!-- Guava -->
		<dependency>
    		<groupId>com.google.guava</groupId>
    		<artifactId>guava</artifactId>
    		<version>${guava.version}</version>
		</dependency>

	</dependencies>


	<artifactId>EventBusSpringIntegration</artifactId>
</project>

If you are working on big project and want to integrate EventBus you should follow below approach where you don’t need to explicitly register your subscriber every time and below class will take care subscription part automatically:

• EventBusSubscriberProcessor.java: This class implements BeanPostProcessor where you will have to override two method postProcessAfterInitialization and postProcessBeforeInitialization and during application initialization time we are checking if bean method has @Subscribe annotation then register with EventBus:

package com.javahonk.eventbusprocessor;

import java.lang.annotation.Annotation;
import java.lang.reflect.Method;

import org.apache.logging.log4j.LogManager;
import org.apache.logging.log4j.Logger;
import org.springframework.beans.BeansException;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.beans.factory.config.BeanPostProcessor;

import com.google.common.eventbus.EventBus;
import com.google.common.eventbus.Subscribe;

public class EventBusSubscriberProcessor implements BeanPostProcessor {
	
	private static final Logger LOGGER = LogManager.getLogger(EventBusSubscriberProcessor.class.getName());
	
	@Autowired
    private EventBus eventBus;

	@Override
	public Object postProcessAfterInitialization(Object beanObject, String beanObjectName)
			throws BeansException {
		return beanObject;
	}

	@Override
	public Object postProcessBeforeInitialization(Object beanObject, String beanObjectName)
			throws BeansException {
		
		Method[] methods = beanObject.getClass().getMethods();
		for (Method method : methods) {
			Annotation[] annotations = method.getAnnotations();
			
			for (Annotation methodAnnotation : annotations) {
				if (methodAnnotation.annotationType().equals(Subscribe.class)) {
					eventBus.register(beanObject);
					LOGGER.info("Bean {} method {} has been subscribed to the EventBus.", new Object[] { beanObjectName, method.getName() });
					return beanObject;
				}
			}
		}

		return beanObject;
	}

}

Important: Don’t use EventBusSubscriberProcessorShorten.java or it’s your choice: Google Guava EventBus is smart enough to find if bean has @Subscribe annotation defined or not then register and you could shorten above EventBusSubscriberProcessor.java as below but my preference is use above class where we are checking @Subscribe annotation:

• EventBusSubscriberProcessorShorten.java:

package com.javahonk.eventbusprocessor;

import org.apache.logging.log4j.LogManager;
import org.apache.logging.log4j.Logger;
import org.springframework.beans.BeansException;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.beans.factory.config.BeanPostProcessor;

import com.google.common.eventbus.EventBus;

public class EventBusSubscriberProcessorShorten implements BeanPostProcessor {
	
	private static final Logger LOGGER = LogManager.getLogger(EventBusSubscriberProcessorShorten.class.getName());
	
	@Autowired
    private EventBus eventBus;

	@Override
	public Object postProcessAfterInitialization(Object beanObject, String beanObjectName)
			throws BeansException {
		return beanObject;
	}

	@Override
	public Object postProcessBeforeInitialization(Object beanObject, String beanObjectName)
			throws BeansException {
		eventBus.register(beanObject);
		LOGGER.info("Bean {} has been subscribed to the EventBus.", new Object[] { beanObjectName });
		return beanObject;
	}

}

• DeadEventProcessor.java:

package com.javahonk.subscriber;

import org.apache.logging.log4j.LogManager;
import org.apache.logging.log4j.Logger;

import com.google.common.eventbus.DeadEvent;
import com.google.common.eventbus.Subscribe;

public class DeadEventProcessor {

	private static final Logger logger = LogManager.getLogger(DeadEventProcessor.class.getName());
	
	@Subscribe
	public void processDeadEvent(DeadEvent deadEvent){
		
		logger.error("DEADEVENT DETECTED:{}",deadEvent.getEvent().getClass());
		
	}
}

• SendPostThourghClass.java:

package com.javahonk.subscriber;

public class SendPostThourghClass {
	
	private String name;

	public SendPostThourghClass(String name) {
		this.name = name;
	}

	public String getName() {
		return name;
	}

	public void setName(String name) {
		this.name = name;
	}

}

• Subscriber1.java:

package com.javahonk.subscriber;

import com.google.common.eventbus.Subscribe;

public class Subscriber1 {
	
	@Subscribe
	public void getPublishData(SendPostThourghClass sendPostThourghClass){
		System.out.println("Got post data on Subscriber1: "+sendPostThourghClass.getName());
	}

}

• Subscriber2.java:

package com.javahonk.subscriber;

import com.google.common.eventbus.Subscribe;

public class Subscriber2 {
	
	@Subscribe
	public void getPublishData(SendPostThourghClass sendPostThourghClass){
		System.out.println("Got post data on Subscriber2: "+sendPostThourghClass.getName());
	}

}

• spring-context.xml:

<?xml version="1.0" encoding="UTF-8"?>
<beans xmlns="http://www.springframework.org/schema/beans"
	xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
	xmlns:context="http://www.springframework.org/schema/context"
	xmlns:jdbc="http://www.springframework.org/schema/jdbc"
	xmlns:tx="http://www.springframework.org/schema/tx"
	xmlns:util="http://www.springframework.org/schema/util"
	xsi:schemaLocation="http://www.springframework.org/schema/beans http://www.springframework.org/schema/beans/spring-beans.xsd
		http://www.springframework.org/schema/context http://www.springframework.org/schema/context/spring-context-4.1.xsd
		http://www.springframework.org/schema/jdbc http://www.springframework.org/schema/jdbc/spring-jdbc-4.1.xsd
		http://www.springframework.org/schema/tx http://www.springframework.org/schema/tx/spring-tx-4.1.xsd
		http://www.springframework.org/schema/util http://www.springframework.org/schema/util/spring-util-4.1.xsd">
	
	<context:annotation-config />
	<context:component-scan base-package="com.javahonk, com.javahonk.eventbusprocessor, com.javahonk.subscriber" />
	
	<bean class="org.springframework.beans.factory.config.PropertyPlaceholderConfigurer">
		<property name="locations">
			<list>
				<value>application.properties</value>				
			</list>
		</property>
		<property name="ignoreUnresolvablePlaceholders" value="true"/>
	</bean>	

	<bean id="eventBus" class="com.google.common.eventbus.EventBus" />
	
	<bean id="deadEventProcessor" class="com.javahonk.subscriber.DeadEventProcessor" />
	<bean id="subscriber1" class="com.javahonk.subscriber.Subscriber1" />
	<bean id="subscriber2" class="com.javahonk.subscriber.Subscriber2" />
	
	<bean id="eventBusSubscriberProcessor" class="com.javahonk.eventbusprocessor.EventBusSubscriberProcessor" />
	
	<!-- <bean id="eventBusSubscriberProcessorShorten" class="com.javahonk.eventbusprocessor.EventBusSubscriberProcessorShorten" /> -->
	
	<!-- <bean id="subscriber3NotSingleton" class="com.javahonk.subscriber.Subscriber3NotSingleton" scope="prototype"/> -->
	
	<!-- <bean id="eventBusPostProcessor" class="com.javahonk.eventbusprocessor.EventBusPostProcessorSingletonTest" /> -->
	
		
</beans>

• MainApplication.java: Main class to test application:

package com.javahonk;

import org.apache.logging.log4j.LogManager;
import org.apache.logging.log4j.Logger;
import org.springframework.context.ApplicationContext;
import org.springframework.context.support.AbstractApplicationContext;
import org.springframework.context.support.ClassPathXmlApplicationContext;

import com.google.common.eventbus.EventBus;
import com.javahonk.subscriber.SendPostThourghClass;

public class MainApplication {
	
	private static final Logger logger = LogManager.getLogger(MainApplication.class.getName());
	
	public static void main(String[] args) {
		
		logger.info("EventBus Spring application starting...");
		
		ApplicationContext applicationContext = new ClassPathXmlApplicationContext("spring-context.xml");
		
		EventBus eventBus = (EventBus)applicationContext.getBean(EventBus.class);
		
		eventBus.post("This event will go to DeadEventProcessor because no subscriber");
		
		eventBus.post( new SendPostThourghClass ("This will go to subscribers"));
		
		((AbstractApplicationContext)applicationContext).close();	
		
	}
}

• That’s it. Please run MainApplication.java as java application you will see below output:

Most important: People who are expert in Spring definitely raise question what if Subscriber is not Singleton then it will cause memory leak and in that case use below class:

• EventBusPostProcessorSingletonTest.java:

package com.javahonk.eventbusprocessor;

import java.lang.reflect.Method;

import org.apache.logging.log4j.LogManager;
import org.apache.logging.log4j.Logger;
import org.springframework.beans.BeansException;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.beans.factory.config.BeanPostProcessor;
import org.springframework.context.ApplicationContext;

import com.google.common.eventbus.EventBus;
import com.google.common.eventbus.Subscribe;

public class EventBusPostProcessorSingletonTest implements BeanPostProcessor
{
	@Autowired private ApplicationContext context;
    @Autowired private EventBus eventBus;
	
	private static final Logger LOGGER = LogManager.getLogger(EventBusSubscriberProcessor.class.getName());

    @Override
    public Object postProcessBeforeInitialization(Object beanObject, String beanObjectName) throws BeansException
    {
        return beanObject;
    }

    @Override
    public Object postProcessAfterInitialization(Object beanObject, String beanObjectName)
                  throws BeansException
    {
        //If bean is singleton register with event bus
    	if(context.isSingleton(beanObjectName)) {
            eventBus.register(beanObject);
            LOGGER.info("Bean {} method {} has been subscribed to the EventBus.", new Object[] { beanObjectName});			
        } else {        	
        	LOGGER.info("Bean {} method {} has not been subscribed to the EventBus because it's not Singleton class.", new Object[] { beanObjectName});	
            if(isSubscribe(beanObject)) {
				LOGGER.warn("Bean {} class method @Subscribe annotation not registered with EventBus "
						+ "because it's not Singleton class and will create memory leak. "
						+ "Please change your bean class to singleton by making it's scope singleton.",
						beanObjectName);
			}
        }
        return beanObject;
    }
    
    public static boolean isSubscribe(Object beanObject)
    {
        Method[] methods = beanObject.getClass().getMethods();
        for(Method method : methods)
        {
            Subscribe subscribe = method.getAnnotation(Subscribe.class);
            if(subscribe != null)
            {
                return true;
            }
        }
        return false;
    }
    
    
}

• For test use this class: Subscriber3NotSingleton.java:

package com.javahonk.subscriber;

import com.google.common.eventbus.Subscribe;

public class Subscriber3NotSingleton {
	
	@Subscribe
	public void getPublishData(SendPostThourghClass sendPostThourghClass){
		System.out.println("Got post data on Subscriber3NotSingleton: "+sendPostThourghClass.getName());
	}

}

• For more information please visit Google Guava EventBus official documentation here

 Download Project:  EventBusSpringIntegration