From 17fc4eae998a44e3d46d6a8357ffdc3e5d67c38a Mon Sep 17 00:00:00 2001 From: Artem Bilan Date: Mon, 10 Aug 2020 12:32:47 -0400 Subject: [PATCH] Fix IntReactiveUtils for the proper emission * Enable `MessageChannelReactiveUtilsTests.testOverproducingWithSubscribableChannel()` back * Disable `StompServerIntegrationTests` again since build on CI is stalled again --- .../integration/util/IntegrationReactiveUtils.java | 7 ++++++- .../channel/MessageChannelReactiveUtilsTests.java | 2 -- .../stomp/client/StompServerIntegrationTests.java | 3 ++- 3 files changed, 8 insertions(+), 4 deletions(-) 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 045ff2716b..dfb2bac5a4 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 @@ -17,6 +17,7 @@ package org.springframework.integration.util; import java.time.Duration; +import java.util.concurrent.locks.LockSupport; import org.reactivestreams.Publisher; @@ -127,7 +128,11 @@ public final class IntegrationReactiveUtils { return Flux.defer(() -> { Sinks.Many> sink = Sinks.many().multicast().onBackpressureBuffer(1); @SuppressWarnings("unchecked") - MessageHandler messageHandler = (message) -> sink.emitNext((Message) message); + MessageHandler messageHandler = (message) -> { + while (!sink.emitNext((Message) message).hasEmitted()) { + LockSupport.parkNanos(10); + } + }; inputChannel.subscribe(messageHandler); return sink.asFlux() .doOnCancel(() -> inputChannel.unsubscribe(messageHandler)); 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/MessageChannelReactiveUtilsTests.java index 697ecdc766..bc0216db05 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/MessageChannelReactiveUtilsTests.java @@ -21,7 +21,6 @@ import static org.assertj.core.api.Assertions.assertThat; import java.time.Duration; import java.util.concurrent.atomic.AtomicInteger; -import org.junit.jupiter.api.Disabled; import org.junit.jupiter.api.Test; import org.springframework.integration.util.IntegrationReactiveUtils; @@ -71,7 +70,6 @@ class MessageChannelReactiveUtilsTests { } @Test - @Disabled("Backpressure is not honored") void testOverproducingWithSubscribableChannel() { DirectChannel channel = new DirectChannel(); diff --git a/spring-integration-stomp/src/test/java/org/springframework/integration/stomp/client/StompServerIntegrationTests.java b/spring-integration-stomp/src/test/java/org/springframework/integration/stomp/client/StompServerIntegrationTests.java index 84077a2203..d6905b9197 100644 --- a/spring-integration-stomp/src/test/java/org/springframework/integration/stomp/client/StompServerIntegrationTests.java +++ b/spring-integration-stomp/src/test/java/org/springframework/integration/stomp/client/StompServerIntegrationTests.java @@ -22,6 +22,7 @@ import static org.assertj.core.api.Assertions.assertThatExceptionOfType; import org.apache.activemq.broker.BrokerService; import org.junit.jupiter.api.AfterAll; import org.junit.jupiter.api.BeforeAll; +import org.junit.jupiter.api.Disabled; import org.junit.jupiter.api.Test; import org.springframework.context.ApplicationEvent; @@ -62,7 +63,7 @@ import org.springframework.util.SocketUtils; * * @since 4.2 */ -//@Disabled("Until the fix in reactor-netty-core") +@Disabled("Until the fix in reactor-netty-core") public class StompServerIntegrationTests { private static BrokerService activeMQBroker;