@Test public void notify_and_wrap() throws Exception { final AtomicConsumer<String> c = new AtomicConsumer<>(); eventBus.on(Selectors.$("foo"), c); eventBus.notifyAndWrap("foo", "bar"); assertThat(c.isReady()).isTrue(); assertThat(c.getData().get()).isEqualTo("bar"); }
@Override public <T extends Event<?>> Registration<Object, Consumer<? extends Event<?>>> on(final Selector selector, final Consumer<T> consumer) { requireNonNull(selector, "Selector cannot be null."); requireNonNull(consumer, "Consumer cannot be null."); final Class<?> tClass = extractGeneric(consumer); final Consumer<T> proxyConsumer = e -> { if (null != selector.getHeaderResolver()) { e.getHeaders().setAll(selector.getHeaderResolver().resolve(e.getKey())); } if (tClass == null || e.getData() == null || tClass.isAssignableFrom(e.getData().getClass())) { consumer.accept(e); } }; return consumerRegistry.register(selector, proxyConsumer); }
EventBus implements Bus<Event<?>>, Consumer<Event<?>> { @Override public <T extends Event<?>> Registration<Object, Consumer<? extends Event<?>>> on(final Selector selector, final Consumer<T> consumer) { requireNonNull(selector, "Selector cannot be null."); requireNonNull(consumer, "Consumer cannot be null."); final Class<?> tClass = extractGeneric(consumer); final Consumer<T> proxyConsumer = e -> { if (null != selector.getHeaderResolver()) { e.getHeaders().setAll(selector.getHeaderResolver().resolve(e.getKey())); } if (tClass == null || e.getData() == null || tClass.isAssignableFrom(e.getData().getClass())) { consumer.accept(e); } }; return consumerRegistry.register(selector, proxyConsumer); } }
EventBus implements Bus<Event<?>>, Consumer<Event<?>> { @Override public <T extends Event<?>> Registration<Object, Consumer<? extends Event<?>>> on(final Selector selector, final Consumer<T> consumer) { requireNonNull(selector, "Selector cannot be null."); requireNonNull(consumer, "Consumer cannot be null."); final Class<?> tClass = extractGeneric(consumer); final Consumer<T> proxyConsumer = e -> { if (null != selector.getHeaderResolver()) { e.getHeaders().setAll(selector.getHeaderResolver().resolve(e.getKey())); } if (tClass == null || e.getData() == null || tClass.isAssignableFrom(e.getData().getClass())) { consumer.accept(e); } }; return consumerRegistry.register(selector, proxyConsumer); } EventBus(@Nullable Dispatcher dispatcher);  EventBus(@Nullable Dispatcher dispatcher,
                  @Nullable Router router);  EventBus(@Nullable Dispatcher dispatcher,
                  @Nullable Router router,
                  @Nullable Consumer<Throwable> dispatchErrorHandler,
                  @Nullable final Consumer<Throwable> uncaughtErrorHandler);  EventBus(@Nonnull Registry<Object, Consumer<? extends Event<?>>> consumerRegistry,
                  @Nullable Dispatcher dispatcher,
                  @Nullable Router router,
                  @Nullable Consumer<Throwable> dispatchErrorHandler,
                  @Nullable final Consumer<Throwable> uncaughtErrorHandler); }
EventBus implements Bus<Event<?>>, Consumer<Event<?>> { @Override public <T extends Event<?>> Registration<Object, Consumer<? extends Event<?>>> on(final Selector selector, final Consumer<T> consumer) { requireNonNull(selector, "Selector cannot be null."); requireNonNull(consumer, "Consumer cannot be null."); final Class<?> tClass = extractGeneric(consumer); final Consumer<T> proxyConsumer = e -> { if (null != selector.getHeaderResolver()) { e.getHeaders().setAll(selector.getHeaderResolver().resolve(e.getKey())); } if (tClass == null || e.getData() == null || tClass.isAssignableFrom(e.getData().getClass())) { consumer.accept(e); } }; return consumerRegistry.register(selector, proxyConsumer); } EventBus(@Nullable Dispatcher dispatcher);  EventBus(@Nullable Dispatcher dispatcher,
                  @Nullable Router router);  EventBus(@Nullable Dispatcher dispatcher,
                  @Nullable Router router,
                  @Nullable Consumer<Throwable> dispatchErrorHandler,
                  @Nullable final Consumer<Throwable> uncaughtErrorHandler);  EventBus(@Nonnull Registry<Object, Consumer<? extends Event<?>>> consumerRegistry,
                  @Nullable Dispatcher dispatcher,
                  @Nullable Router router,
                  @Nullable Consumer<Throwable> dispatchErrorHandler,
                  @Nullable final Consumer<Throwable> uncaughtErrorHandler); static EventBusSpec config(); static EventBus create(); static EventBus create(Environment env); static EventBus create(Dispatcher dispatcher); static EventBus create(Environment env, String dispatcher); static EventBus create(Environment env, Dispatcher dispatcher); synchronized UUID getId(); Registry<Object, Consumer<? extends Event<?>>> getConsumerRegistry(); Dispatcher getDispatcher(); Router getRouter(); Consumer<Throwable> getDispatchErrorHandler(); Consumer<Throwable> getUncaughtErrorHandler(); @Override boolean respondsToKey(Object key); @Override Registration<Object, Consumer<? extends Event<?>>> on(final Selector selector,
                                                                                    final Consumer<T> consumer); Publisher<? extends Event<?>> on(Selector broadcastSelector); @Override EventBus notify(Object key, Event<?> ev); final EventBus notify(@Nonnull final Publisher<?> source, @Nonnull final Object key); final EventBus notify(@Nonnull final Publisher<? extends T> source, @Nonnull final Function<? super T, ?> keyMapper); Registration<?, Consumer<? extends Event<?>>> receive(Selector sel, Function<T, V> fn); EventBus notify(Object key, Supplier<? extends Event<?>> supplier); EventBus notify(Object key); EventBus send(Object key, Event<?> ev); EventBus send(Object key, Supplier<? extends Event<?>> supplier); EventBus send(Object key, Event<?> ev, Bus replyTo); EventBus send(Object key, Supplier<? extends Event<?>> supplier, Bus replyTo); EventBus sendAndReceive(Object key, Event<?> event, Consumer<T> reply); EventBus sendAndReceive(Object key, Supplier<? extends Event<?>> supplier, Consumer<T> reply); Consumer<Event<T>> prepare(final Object key); void schedule(final Consumer<T> consumer, final T data); @Override void accept(Event<?> event); }
EventBus implements Bus<Event<?>>, Consumer<Event<?>> { @Override public <T extends Event<?>> Registration<Object, Consumer<? extends Event<?>>> on(final Selector selector, final Consumer<T> consumer) { requireNonNull(selector, "Selector cannot be null."); requireNonNull(consumer, "Consumer cannot be null."); final Class<?> tClass = extractGeneric(consumer); final Consumer<T> proxyConsumer = e -> { if (null != selector.getHeaderResolver()) { e.getHeaders().setAll(selector.getHeaderResolver().resolve(e.getKey())); } if (tClass == null || e.getData() == null || tClass.isAssignableFrom(e.getData().getClass())) { consumer.accept(e); } }; return consumerRegistry.register(selector, proxyConsumer); } EventBus(@Nullable Dispatcher dispatcher);  EventBus(@Nullable Dispatcher dispatcher,
                  @Nullable Router router);  EventBus(@Nullable Dispatcher dispatcher,
                  @Nullable Router router,
                  @Nullable Consumer<Throwable> dispatchErrorHandler,
                  @Nullable final Consumer<Throwable> uncaughtErrorHandler);  EventBus(@Nonnull Registry<Object, Consumer<? extends Event<?>>> consumerRegistry,
                  @Nullable Dispatcher dispatcher,
                  @Nullable Router router,
                  @Nullable Consumer<Throwable> dispatchErrorHandler,
                  @Nullable final Consumer<Throwable> uncaughtErrorHandler); static EventBusSpec config(); static EventBus create(); static EventBus create(Environment env); static EventBus create(Dispatcher dispatcher); static EventBus create(Environment env, String dispatcher); static EventBus create(Environment env, Dispatcher dispatcher); synchronized UUID getId(); Registry<Object, Consumer<? extends Event<?>>> getConsumerRegistry(); Dispatcher getDispatcher(); Router getRouter(); Consumer<Throwable> getDispatchErrorHandler(); Consumer<Throwable> getUncaughtErrorHandler(); @Override boolean respondsToKey(Object key); @Override Registration<Object, Consumer<? extends Event<?>>> on(final Selector selector,
                                                                                    final Consumer<T> consumer); Publisher<? extends Event<?>> on(Selector broadcastSelector); @Override EventBus notify(Object key, Event<?> ev); final EventBus notify(@Nonnull final Publisher<?> source, @Nonnull final Object key); final EventBus notify(@Nonnull final Publisher<? extends T> source, @Nonnull final Function<? super T, ?> keyMapper); Registration<?, Consumer<? extends Event<?>>> receive(Selector sel, Function<T, V> fn); EventBus notify(Object key, Supplier<? extends Event<?>> supplier); EventBus notify(Object key); EventBus send(Object key, Event<?> ev); EventBus send(Object key, Supplier<? extends Event<?>> supplier); EventBus send(Object key, Event<?> ev, Bus replyTo); EventBus send(Object key, Supplier<? extends Event<?>> supplier, Bus replyTo); EventBus sendAndReceive(Object key, Event<?> event, Consumer<T> reply); EventBus sendAndReceive(Object key, Supplier<? extends Event<?>> supplier, Consumer<T> reply); Consumer<Event<T>> prepare(final Object key); void schedule(final Consumer<T> consumer, final T data); @Override void accept(Event<?> event);  }