EventBus support in Micronaut

[luistrigueiros]

I would love to see support in Micronaut for a type of EventBus mechanics similar to the one that exists in Guava or Spring.

The would be that a bean method annotated with a given annotation i.e (@Subscribe) would be automatically registered as a listener for events that could be fired via an implementation of this EventBus given by Micronaut container.

Additionally optionally it would be cool if this would be handled async maybe using the equivalent of one elastic I/O executor service.

[RyanHoldren]

This sounds to me a lot like the method adapter advice example in the documentation

[brandonlamb]

+1

Editing for more clarity/details, example fixes.

My team builds microservices (duh right) implementing Domain-Driven Design and Ports and Adapters. We implement application services as command/event handlers (command pattern?).

For those stumbling upon this thread and wondering what examples could be:

• Command: CreateProduct (a command, do something, take some action)
• Event: ProductCreated (an immutable event or thing that happened in the past)

The built-in application events felt a bit funky/clumsy so we added MBassador which is a no-deps (hopefully Im not lying on that) event bus. I didnt want to pull in the entire guava library to use its event bus.

Event Listener Class

@Singleton
@Listener
class ThingCreatedHandler @Inject constructor(private val thingService: ThingService) {
  @Handler
  fun handle(event: ThingCreated) {
    try {
      thingService.create(event.thing)
    } catch (e: Exception) {
      Logger.error("log_type=thing_create_failed, message={}, data={}", e.message, event)
    }
  }
}

There is an extra step where I could see the awesome MN compile-time stuff fix of instantiating (via MN DI) and registering classes.

Event Subscribe Class

@Context
class EventSubscriber @Inject constructor(
  eventBus: MBassador<Any>,
  thingCreatedHandler: ThingCreatedHandler
) {
  init {
    eventBus.subscribe(thingCreatedHandler)
    //... and the next one
  }
}

Publish messages (command or event) supports both sync/async

eventBus.publish(ThingCreated(thing)
eventBus.publishAsync(ThingCreated(thing)

[luistrigueiros]

Thank you both very useful inputs I will have a look at MBassador and the Adapter. I my case what I am doing is just when persisting objects to Redis using Lettuce reactive API for some reason the the doOnNext does not gets called, but I need to create additional aux keys and data structures for example to support multi-facet search. So I am using an event listener to create the additional support indexes for a given domain object been persisted.

[graemerocher]

Like @RyanHoldren said I don't believe this is necessary. You can achieve 90% of what an event bus offers by publishing an event:

class MyEvent {}

@Inject ApplicationEventPublisher eventPublisher

eventPublisher.publishEvent(new MyEvent())

And adding an async event listener:

@Listener
@Async
void myEventListener(MyEvent event) {
    ..
}

Other features like filtering etc. can be added as necessary at the method level.

[luistrigueiros]

Thank you @graemerocher this was I was looking for.

[luistrigueiros]

Hi @graemerocher:

How can one obtain an instance of ApplicationEventListener ? I my case I don't have of these service discovery services yet.

https://docs.micronaut.io/snapshot/api/io/micronaut/context/event/ApplicationEventListener.html All Known Implementing Classes: AutoRegistration, ConsulAutoRegistration, DiscoveryServiceAutoRegistration, EnvironmentConfiguration, EnvironmentDeploymentContext, EurekaAutoRegistration, HeartbeatTask, RefreshScope, Route53AutoNamingRegistrationClient, SessionWebSocketEventListener, SimpleMeterRegistryEventListener

[rodolfodpk]

90% of the features? Well, Quarkus certainly got some inspiration from Micronaut and I wish Micronaut could do it too: https://quarkus.io/guides/reactive-event-bus An event bus with request/reply (not only pub/sub), blocking code annotation, and even Mutiny as an alternative reactive glue (compatible with RxJava but not Reactor, AFAIK). I have some use cases I usually implemented in Vertx/Quarkus with its event bus but now I have to use Micronaut and stucked. At least I can use Vertx reactive drivers with Micronaut, I probably will make use of its event bus too but will miss some nice building blocks.