1. Overview
So, in a number of other tutorials, we’ve talked about BeanPostProcessor. In this tutorial, we’ll put them to use in a real-world example using Guava’s EventBus.
Spring’s BeanPostProcessor gives us hooks into the Spring bean lifecycle to modify its configuration.
BeanPostProcessor allows for direct modification of the beans themselves.
In this tutorial, we’re going to look at a concrete example of these classes integrating Guava’s EventBus.
2. Setup
First, we need to set up our environment. Let’s add the Spring Context, Spring Expression, and Guava dependencies to our pom.xml:
<dependency>
<groupId>org.springframework</groupId>
<artifactId>spring-context</artifactId>
<version>5.2.6.RELEASE</version>
</dependency>
<dependency>
<groupId>org.springframework</groupId>
<artifactId>spring-expression</artifactId>
<version>5.2.6.RELEASE</version>
</dependency>
<dependency>
<groupId>com.google.guava</groupId>
<artifactId>guava</artifactId>
<version>33.0.0-jre</version>
</dependency>
Next, let’s discuss our goals.
3. Goals and Implementation
For our first goal, we want to utilize Guava’s EventBus to pass messages across various aspects of the system asynchronously.
Next, we want to register and unregister objects for events automatically on bean creation/destruction instead of using the manual method provided by EventBus.
So, we’re now ready to start coding!
Our implementation will consist of a wrapper class for Guava’s EventBus, a custom marker annotation, a BeanPostProcessor, a model object, and a bean to receive stock trade events from the EventBus. In addition, we’ll create a test case to verify the desired functionality.
3.1. EventBus Wrapper
To being with, we’ll define an EventBus wrapper to provide some static methods to easily register and unregister beans for events which will be used by the BeanPostProcessor:
public final class GlobalEventBus {
public static final String GLOBAL_EVENT_BUS_EXPRESSION
= "T(com.baeldung.postprocessor.GlobalEventBus).getEventBus()";
private static final String IDENTIFIER = "global-event-bus";
private static final GlobalEventBus GLOBAL_EVENT_BUS = new GlobalEventBus();
private final EventBus eventBus = new AsyncEventBus(IDENTIFIER, Executors.newCachedThreadPool());
private GlobalEventBus() {}
public static GlobalEventBus getInstance() {
return GlobalEventBus.GLOBAL_EVENT_BUS;
}
public static EventBus getEventBus() {
return GlobalEventBus.GLOBAL_EVENT_BUS.eventBus;
}
public static void subscribe(Object obj) {
getEventBus().register(obj);
}
public static void unsubscribe(Object obj) {
getEventBus().unregister(obj);
}
public static void post(Object event) {
getEventBus().post(event);
}
}
This code provides static methods for accessing the GlobalEventBus and underlying EventBus as well as registering and unregistering for events and posting events. It also has a SpEL expression used as the default expression in our custom annotation to define which EventBus we want to utilize.
3.2. Custom Marker Annotation
Next, let’s define a custom marker annotation which will be used by the BeanPostProcessor to identify beans to automatically register/unregister for events:
@Retention(RetentionPolicy.RUNTIME)
@Target(ElementType.TYPE)
@Inherited
public @interface Subscriber {
String value() default GlobalEventBus.GLOBAL_EVENT_BUS_EXPRESSION;
}
3.3. BeanPostProcessor
Now, we’ll define the BeanPostProcessor which will check each bean for the Subscriber annotation. This class is also a DestructionAwareBeanPostProcessor, which is a Spring interface adding a before-destruction callback to BeanPostProcessor. If the annotation is present, we’ll register it with the EventBus identified by the annotation’s SpEL expression on bean initialization and unregister it on bean destruction:
public class GuavaEventBusBeanPostProcessor
implements DestructionAwareBeanPostProcessor {
Logger logger = LoggerFactory.getLogger(this.getClass());
SpelExpressionParser expressionParser = new SpelExpressionParser();
@Override
public void postProcessBeforeDestruction(Object bean, String beanName)
throws BeansException {
this.process(bean, EventBus::unregister, "destruction");
}
@Override
public boolean requiresDestruction(Object bean) {
return true;
}
@Override
public Object postProcessBeforeInitialization(Object bean, String beanName)
throws BeansException {
return bean;
}
@Override
public Object postProcessAfterInitialization(Object bean, String beanName)
throws BeansException {
this.process(bean, EventBus::register, "initialization");
return bean;
}
private void process(Object bean, BiConsumer<EventBus, Object> consumer, String action) {
// See implementation below
}
}
The code above takes every bean and runs it through the process method, defined below. It processes it after the bean has been initialized and before it is destroyed. The requiresDestruction method returns true by default and we keep that behavior here as we check for the existence of the @Subscriber annotation in the postProcessBeforeDestruction callback.
Let’s now look at the process method:
private void process(Object bean, BiConsumer<EventBus, Object> consumer, String action) {
Object proxy = this.getTargetObject(bean);
Subscriber annotation = AnnotationUtils.getAnnotation(proxy.getClass(), Subscriber.class);
if (annotation == null)
return;
this.logger.info("{}: processing bean of type {} during {}",
this.getClass().getSimpleName(), proxy.getClass().getName(), action);
String annotationValue = annotation.value();
try {
Expression expression = this.expressionParser.parseExpression(annotationValue);
Object value = expression.getValue();
if (!(value instanceof EventBus)) {
this.logger.error(
"{}: expression {} did not evaluate to an instance of EventBus for bean of type {}",
this.getClass().getSimpleName(), annotationValue, proxy.getClass().getSimpleName());
return;
}
EventBus eventBus = (EventBus)value;
consumer.accept(eventBus, proxy);
} catch (ExpressionException ex) {
this.logger.error("{}: unable to parse/evaluate expression {} for bean of type {}",
this.getClass().getSimpleName(), annotationValue, proxy.getClass().getName());
}
}
This code checks for the existence of our custom marker annotation named Subscriber and, if present, reads the SpEL expression from its value property. Then, the expression is evaluated into an object. If it’s an instance of EventBus, we apply the BiConsumer function parameter to the bean. The BiConsumer is used to register and unregister the bean from the EventBus.
The implementation of the method getTargetObject is as follows:
private Object getTargetObject(Object proxy) throws BeansException {
if (AopUtils.isJdkDynamicProxy(proxy)) {
try {
return ((Advised)proxy).getTargetSource().getTarget();
} catch (Exception e) {
throw new FatalBeanException("Error getting target of JDK proxy", e);
}
}
return proxy;
}
3.4. StockTrade Model Object
Next, let’s define our StockTrade model object:
public class StockTrade {
private String symbol;
private int quantity;
private double price;
private Date tradeDate;
// constructor
}
3.5. StockTradePublisher Event Receiver
Then, let’s define a listener class to notify us a trade was received so that we can write our test:
@FunctionalInterface
public interface StockTradeListener {
void stockTradePublished(StockTrade trade);
}
Finally, we’ll define a receiver for new StockTrade events:
@Subscriber
public class StockTradePublisher {
Set<StockTradeListener> stockTradeListeners = new HashSet<>();
public void addStockTradeListener(StockTradeListener listener) {
synchronized (this.stockTradeListeners) {
this.stockTradeListeners.add(listener);
}
}
public void removeStockTradeListener(StockTradeListener listener) {
synchronized (this.stockTradeListeners) {
this.stockTradeListeners.remove(listener);
}
}
@Subscribe
@AllowConcurrentEvents
void handleNewStockTradeEvent(StockTrade trade) {
// publish to DB, send to PubNub, ...
Set<StockTradeListener> listeners;
synchronized (this.stockTradeListeners) {
listeners = new HashSet<>(this.stockTradeListeners);
}
listeners.forEach(li -> li.stockTradePublished(trade));
}
}
The code above marks this class as a Subscriber of Guava EventBus events and Guava’s @Subscribe annotation marks the method handleNewStockTradeEvent as a receiver of events. The type of events it’ll receive is based on the class of the single parameter to the method; in this case, we’ll receive events of type StockTrade.
The @AllowConcurrentEvents annotation allows the concurrent invocation of this method. Once we receive a trade we do any processing we wish then notify any listeners.
3.6. Testing
Now let’s wrap up our coding with an integration test to verify the BeanPostProcessor works correctly. Firstly, we’ll need a Spring context:
@Configuration
public class PostProcessorConfiguration {
@Bean
public GlobalEventBus eventBus() {
return GlobalEventBus.getInstance();
}
@Bean
public GuavaEventBusBeanPostProcessor eventBusBeanPostProcessor() {
return new GuavaEventBusBeanPostProcessor();
}
@Bean
public StockTradePublisher stockTradePublisher() {
return new StockTradePublisher();
}
}
Now we can implement our test:
@RunWith(SpringJUnit4ClassRunner.class)
@ContextConfiguration(classes = PostProcessorConfiguration.class)
public class StockTradeIntegrationTest {
@Autowired
StockTradePublisher stockTradePublisher;
@Test
public void givenValidConfig_whenTradePublished_thenTradeReceived() {
Date tradeDate = new Date();
StockTrade stockTrade = new StockTrade("AMZN", 100, 2483.52d, tradeDate);
AtomicBoolean assertionsPassed = new AtomicBoolean(false);
StockTradeListener listener = trade -> assertionsPassed
.set(this.verifyExact(stockTrade, trade));
this.stockTradePublisher.addStockTradeListener(listener);
try {
GlobalEventBus.post(stockTrade);
await().atMost(Duration.ofSeconds(2L))
.untilAsserted(() -> assertThat(assertionsPassed.get()).isTrue());
} finally {
this.stockTradePublisher.removeStockTradeListener(listener);
}
}
boolean verifyExact(StockTrade stockTrade, StockTrade trade) {
return Objects.equals(stockTrade.getSymbol(), trade.getSymbol())
&& Objects.equals(stockTrade.getTradeDate(), trade.getTradeDate())
&& stockTrade.getQuantity() == trade.getQuantity()
&& stockTrade.getPrice() == trade.getPrice();
}
}
The test code above generates a stock trade and posts it to the GlobalEventBus. We wait at most two seconds for the action to complete and to be notified the trade was received by the stockTradePublisher. Furthermore, we validate the received trade was not modified in transit.
4. Conclusion
In conclusion, Spring’s BeanPostProcessor allows us to customize the beans themselves, providing us with a means to automate bean actions we would otherwise have to do manually.
As always, source code is available over on GitHub.