SimpleEventBus.java

package cleanbank.commands;

import com.google.common.reflect.TypeToken;

import java.util.Collection;
import java.util.concurrent.CopyOnWriteArrayList;
import java.util.concurrent.Executor;

import static com.google.common.util.concurrent.MoreExecutors.directExecutor;
import static java.lang.Thread.currentThread;
import static java.util.concurrent.Executors.newWorkStealingPool;

class EventBus {

  private final Event.Handlers syncHandlers = new Event.Handlers(directExecutor());
  private final Event.Handlers asyncHandlers = new Event.Handlers(newWorkStealingPool());

  public <E extends Event> void publish(E event) {
    syncHandlers.invokeMatching(event);
    asyncHandlers.invokeMatching(event);
  }

  public <T extends Event> void registerSync(Event.Handler<T> handler) {
    syncHandlers.add(handler);
  }

  public <T extends Event> void registerAsync(Event.Handler<T> handler) {
    asyncHandlers.add(handler);
  }
}

interface Event {

  @SuppressWarnings({"unchecked", "rawtypes"})
  class Handlers {

    private final Collection<Event.Handler> handlers = new CopyOnWriteArrayList<>();

    private final Executor pool;

    Handlers(Executor pool) {
      this.pool = pool;
    }

    void add(Handler handler) {
      handlers.add(handler);
    }

    void invokeMatching(Event event) {
      handlers
        .stream()
        .filter(handler -> handler.matches(event))
        .forEach(handler -> pool.execute(() -> handler.handle(event)));
    }
  }

  interface Handler<E extends Event> {
    void handle(E event);

    default boolean matches(E event) {
      var eventType = new TypeToken<E>(getClass()) {};
      return eventType.getRawType().equals(event.getClass()) || eventType.isSupertypeOf(event.getClass());
    }
  }
}


class Demo {

  record Ping(String sender) implements Event { }

  static class Pong1 implements Event.Handler<Ping> {
    @Override
    public void handle(Ping event) {
      System.out.printf("%s: ping received from %s in %s\n", getClass().getName(), event.sender(), currentThread().getName());
    }
  }

  static class Pong2 implements Event.Handler<Ping> {
    @Override
    public void handle(Ping event) {
      System.out.printf("%s: ping received from %s in %s\n", getClass().getName(), event.sender(), currentThread().getName());
    }
  }

  public static void main(String[] args) {
    var bus = new EventBus();

    bus.registerSync(new Pong1());
    bus.registerSync(new Pong2());
    bus.registerAsync(new Pong1());

    bus.publish(new Ping("Hello World"));

  }
}