@Test void givenTwoTerminationSubscribersWhenAnEventIsSentItShouldBeReceivedByBoth() { TerminationSubscriber subscriber1 = subscriber(); TerminationSubscriber subscriber2 = subscriber(); Flux<Event> firstListener = Flux.from(subscriber1.listenEvents()); Flux<Event> secondListener = Flux.from(subscriber2.listenEvents()); sendEvents(subscriber1, COMPLETED_EVENT); List<Event> receivedEventsFirst = new ArrayList<>(); firstListener.subscribe(receivedEventsFirst::add); List<Event> receivedEventsSecond = new ArrayList<>(); secondListener.subscribe(receivedEventsSecond::add); Awaitility.await().atMost(ONE_MINUTE).until(() -> receivedEventsFirst.size() == 1 && receivedEventsSecond.size() == 1); assertThat(receivedEventsFirst).containsExactly(COMPLETED_EVENT); assertThat(receivedEventsSecond).containsExactly(COMPLETED_EVENT); }
@Override public Publisher<Event> listenEvents() { return listener .share(); }
RabbitMQTerminationSubscriber implements TerminationSubscriber, Startable, Closeable { @Override public Publisher<Event> listenEvents() { return listener .share(); } }
RabbitMQTerminationSubscriber implements TerminationSubscriber, Startable, Closeable { @Override public Publisher<Event> listenEvents() { return listener .share(); } @Inject  RabbitMQTerminationSubscriber(Sender sender, ReceiverProvider receiverProvider, JsonEventSerializer serializer); }
RabbitMQTerminationSubscriber implements TerminationSubscriber, Startable, Closeable { @Override public Publisher<Event> listenEvents() { return listener .share(); } @Inject  RabbitMQTerminationSubscriber(Sender sender, ReceiverProvider receiverProvider, JsonEventSerializer serializer); void start(); @Override void addEvent(Event event); @Override Publisher<Event> listenEvents(); @Override @PreDestroy void close(); }
RabbitMQTerminationSubscriber implements TerminationSubscriber, Startable, Closeable { @Override public Publisher<Event> listenEvents() { return listener .share(); } @Inject  RabbitMQTerminationSubscriber(Sender sender, ReceiverProvider receiverProvider, JsonEventSerializer serializer); void start(); @Override void addEvent(Event event); @Override Publisher<Event> listenEvents(); @Override @PreDestroy void close();  }