RabbitMQEventBus implements EventBus, Startable { @Override public Mono<Registration> register(MailboxListener.ReactiveMailboxListener listener, RegistrationKey key) { Preconditions.checkState(isRunning, NOT_RUNNING_ERROR_MESSAGE); return keyRegistrationHandler.register(listener, key); } @Inject  RabbitMQEventBus(Sender sender, ReceiverProvider receiverProvider, EventSerializer eventSerializer,
                            RetryBackoffConfiguration retryBackoff,
                            RoutingKeyConverter routingKeyConverter,
                            EventDeadLetters eventDeadLetters, MetricFactory metricFactory, ReactorRabbitMQChannelPool channelPool); void start(); @PreDestroy void stop(); @Override Mono<Registration> register(MailboxListener.ReactiveMailboxListener listener, RegistrationKey key); @Override Registration register(MailboxListener.ReactiveMailboxListener listener, Group group); @Override Mono<Void> dispatch(Event event, Set<RegistrationKey> key); @Override Mono<Void> reDeliver(Group group, Event event);  }
@Test void eventProcessingShouldStoreInvalidMessagesInDeadLetterQueue() { EventCollector listener = new EventCollector(); EventBusTestFixture.GroupA registeredGroup = new EventBusTestFixture.GroupA(); eventBus.register(listener, registeredGroup); String emptyRoutingKey = ""; rabbitMQExtension.getSender() .send(Mono.just(new OutboundMessage(MAILBOX_EVENT_EXCHANGE_NAME, emptyRoutingKey, "BAD_PAYLOAD!".getBytes(StandardCharsets.UTF_8)))) .block(); AtomicInteger deadLetteredCount = new AtomicInteger(); rabbitMQExtension.getRabbitChannelPool() .createReceiver() .consumeAutoAck(MAILBOX_EVENT_DEAD_LETTER_QUEUE) .doOnNext(next -> deadLetteredCount.incrementAndGet()) .subscribeOn(Schedulers.elastic()) .subscribe(); Awaitility.await().atMost(org.awaitility.Duration.TEN_SECONDS) .untilAsserted(() -> assertThat(deadLetteredCount.get()).isEqualTo(1)); }
@Test void registerGroupShouldCreateRetryExchange() throws Exception { MailboxListener listener = newListener(); EventBusTestFixture.GroupA registeredGroup = new EventBusTestFixture.GroupA(); eventBus.register(listener, registeredGroup); GroupConsumerRetry.RetryExchangeName retryExchangeName = GroupConsumerRetry.RetryExchangeName.of(registeredGroup); assertThat(rabbitMQExtension.managementAPI().listExchanges()) .anyMatch(exchange -> exchange.getName().equals(retryExchangeName.asString())); }