How to use Reactor to dispatch events to multiple consumers and to filter events based on event data?

I'm evaluating Reactor (https://github.com/reactor/reactor) if it would be suitable for creating an event dispatching framework inside my Spring / enterprise application.

First, consider a scenario in which you have an interface A and concrete event classes B, C, and so on. I want to dispatch concrete events to multiple consumers i.e. observers. Those are registered to a global Reactor instance during bean post processing. However, you can register them dynamically. In most cases there is one producer sending events to multiple consumers at high rate.

I have used Selectors, namely, the ClassSelector to dispatch the correct event types to the correct consumers. This seems to work nicely.

Reactor reactor = ...
B event = ...
Consumer<Event<B>> consumer = ...

// Registration is used to cancel the subscription later
Registration<?> registration = reactor.on(T(event.getClass()), consumer);

To notify, use the type of the event as a key

B event = ...
reactor.notify(event.getClass(), Event.wrap(event));

However, I'm wondering if this is the suggested way to dispatch events efficiently?

Secondly, I was wondering that is it possible to filter events based on the event data? If I understand correctly, Selectors are only for inspecting the key. I'm not referring to event headers here but to the domain specific object properties. I was wondering of using Streams and Stream.filter(Predicate<T> p) for this but is it also possible to filter using Reactor and Selectors? Of course I could write a delegating consumer that inspects the data and delegates it to registered consumers if needed.

Solution

There is a helper object called Selectors that helps to create the various kinds of built-in Selector implementations. There you can see references to the PredicateSelector. The PredicateSelector is very useful as it allows you complete control over the matching of the notification key. It can be a Spring @Bean, an anonymous inner class, a lambda, or anything else conforming to the simple Predicate interface.

Optionally, if you have the JsonPath library in your classpath, then you can use the JsonPathSelector to match based on JsonPath queries.

In either of these cases you don't need to have a separate object for a key if the important data is actually the domain object itself. Just notify on the object and pass the Event<Object> as the second parameter.

MyPojo p = service.next();
reactor.notify(p, Event.wrap(p));