@Test void eventProcessingShouldNotCrashOnInvalidMessage() { TerminationSubscriber subscriber1 = subscriber(); Flux<Event> firstListener = Flux.from(subscriber1.listenEvents()); rabbitMQExtension.getSender() .send(Mono.just(new OutboundMessage(EXCHANGE_NAME, ROUTING_KEY, "BAD_PAYLOAD!".getBytes(UTF_8)))) .block(); sendEvents(subscriber1, COMPLETED_EVENT); List<Event> receivedEventsFirst = new ArrayList<>(); firstListener.subscribe(receivedEventsFirst::add); await().timeout(TEN_SECONDS).untilAsserted(() -> assertThat(receivedEventsFirst).hasSize(1)); }
@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();  }