diff --git a/spring-integration-core/src/main/java/org/springframework/integration/util/IntegrationReactiveUtils.java b/spring-integration-core/src/main/java/org/springframework/integration/util/IntegrationReactiveUtils.java index 305578fb4c..42b0f28baa 100644 --- a/spring-integration-core/src/main/java/org/springframework/integration/util/IntegrationReactiveUtils.java +++ b/spring-integration-core/src/main/java/org/springframework/integration/util/IntegrationReactiveUtils.java @@ -1,5 +1,5 @@ /* - * Copyright 2020 the original author or authors. + * Copyright 2020-2021 the original author or authors. * * Licensed under the Apache License, Version 2.0 (the "License"); * you may not use this file except in compliance with the License. @@ -37,6 +37,7 @@ import reactor.core.publisher.Flux; import reactor.core.publisher.Mono; import reactor.core.publisher.Sinks; import reactor.core.scheduler.Schedulers; +import reactor.util.retry.Retry; /** * Utilities for adapting integration components to/from reactive types. @@ -100,7 +101,7 @@ public final class IntegrationReactiveUtils { Mono.delay(ctx.getOrDefault(DELAY_WHEN_EMPTY_KEY, DEFAULT_DELAY_WHEN_EMPTY))))) .repeat() - .retry(); + .retryWhen(Retry.indefinitely().filter(MessagingException.class::isInstance)); } /** diff --git a/spring-integration-core/src/test/java/org/springframework/integration/channel/MessageChannelReactiveUtilsTests.java b/spring-integration-core/src/test/java/org/springframework/integration/channel/IntegrationReactiveUtilsTests.java similarity index 78% rename from spring-integration-core/src/test/java/org/springframework/integration/channel/MessageChannelReactiveUtilsTests.java rename to spring-integration-core/src/test/java/org/springframework/integration/channel/IntegrationReactiveUtilsTests.java index c695d6ffc3..dc13a1378d 100644 --- a/spring-integration-core/src/test/java/org/springframework/integration/channel/MessageChannelReactiveUtilsTests.java +++ b/spring-integration-core/src/test/java/org/springframework/integration/channel/IntegrationReactiveUtilsTests.java @@ -22,11 +22,14 @@ import java.time.Duration; import java.util.concurrent.CountDownLatch; import java.util.concurrent.TimeUnit; import java.util.concurrent.atomic.AtomicInteger; +import java.util.concurrent.atomic.AtomicReference; import org.junit.jupiter.api.AfterAll; import org.junit.jupiter.api.Test; +import org.springframework.integration.core.MessageSource; import org.springframework.integration.util.IntegrationReactiveUtils; +import org.springframework.messaging.MessagingException; import org.springframework.messaging.support.GenericMessage; import reactor.core.Disposable; @@ -43,7 +46,7 @@ import reactor.util.concurrent.Queues; * * @since 5.1.9 */ -class MessageChannelReactiveUtilsTests { +class IntegrationReactiveUtilsTests { private static final Scheduler SCHEDULER = Schedulers.boundedElastic(); @@ -125,4 +128,28 @@ class MessageChannelReactiveUtilsTests { assertThat(publisherSubscribed.await(10, TimeUnit.SECONDS)).isTrue(); } + + @Test + void testRetryOnMessagingExceptionOnly() { + AtomicInteger retryAttempts = new AtomicInteger(3); + AtomicReference finalException = new AtomicReference<>(); + MessageSource messageSource = + () -> { + if (retryAttempts.getAndDecrement() > 0) { + throw new MessagingException("retryable MessagingException"); + } + else { + throw new RuntimeException("non-retryable RuntimeException"); + } + }; + + StepVerifier.create(IntegrationReactiveUtils.messageSourceToFlux(messageSource).doOnError(finalException::set)) + .expectSubscription() + .expectErrorMessage("non-retryable RuntimeException") + .verify(Duration.ofSeconds(1)); + + assertThat(retryAttempts.get()).isEqualTo(-1); + assertThat(finalException.get()).hasMessage("non-retryable RuntimeException"); + } + }