@Test void eventProcessingShouldNotCrashOnInvalidMessages() { TerminationSubscriber subscriber1 = subscriber(); Flux<Event> firstListener = Flux.from(subscriber1.listenEvents()); IntStream.range(0, 10).forEach(i -> rabbitMQExtension.getSender() .send(Mono.just(new OutboundMessage(EXCHANGE_NAME, ROUTING_KEY, "BAD_PAYLOAD!".getBytes(StandardCharsets.UTF_8)))) .block()); sendEvents(subscriber1, COMPLETED_EVENT); Collection<Event> receivedEventsFirst = new ConcurrentLinkedQueue<>(); firstListener.subscribe(receivedEventsFirst::add); await().atMost(ONE_MINUTE).untilAsserted(() -> SoftAssertions.assertSoftly(soft -> assertThat(receivedEventsFirst).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();  }