Selection by Event Type (or Event parameter type) possible with Reactor and Spring?

Question

I want to use Reactor something like that:

reactor.notify(new CreatedEvent<A>(a));
reactor.notify(new CreatedEvent<B>(b));

Where my event is simply:

public class CreatedEvent<E> extends Event<E> {
    public CreatedEvent(E data) {
        super(data);
    }
}

And then consume those events like that:

reactor.on(new Consumer<CreatedEvent<A>>() {
    @Override
    public void accept(CreatedEvent<A> t) {
        /* do something with a */
    }
});

reactor.on(new Consumer<CreatedEvent<B>>() {
    @Override
    public void accept(CreatedEvent<B> t) {
        /* do something with b */
    }
});

But I receive both events with both consumers.

If I use selectors it works, i.e.:

reactor.notify("created.a", Event.wrap(a));

reactor.on(Selectors.$("created.a"), new Consumer<Event<A>>() {
    @Override
    public void accept(Event<A> t) {
        /* do something with a */
    }
});

But using selectors I would have to write and maintain very much String.

Is it all about using selectors? Or can I somehow "select by parameterized class type"?

Answer 1

But I receive both events with both consumers.

Right. Because you send both events to the same reactor just to defaultKey. And subscribe both consumers to the defaultSelector. In this case it doesn't have value which types of Event you use: they all will be dispatched by default Selector strategy.

I think you need to use ClassSelector:

reactor.notify(a);
reactor.notify(b);

reactor.on(Selectors.T(A.class), new Consumer<Event<A>>() {});
reactor.on(Selectors.T(B.class), new Consumer<Event<B>>() {});

Of course, you have to type Selectors.T for each class-consumer, but it is more simpler then ObjectSelector and custom Event.

UPDATE

How to inject a custom consumer filter:

Reactors.reactor()
            .eventFilter(new GenericResolverFilter())
            .get()

And it's up to you how to resolve a generic argument from consumerKey Object and each Consumer from List<T> items. Anyway you should think here about some caching to allow to Reactor continue to be reactive.