Fix IntReactiveUtils for the proper emission

* Enable `MessageChannelReactiveUtilsTests.testOverproducingWithSubscribableChannel()` back
* Disable `StompServerIntegrationTests` again since build on CI is stalled again

Author: artembilan

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 @@ private static <T> Flux<Message<T>> adaptSubscribableChannelToPublisher(Subscrib
 		return Flux.defer(() -> {
 			Sinks.Many<Message<T>> sink = Sinks.many().multicast().onBackpressureBuffer(1);
 			@SuppressWarnings("unchecked")
-			MessageHandler messageHandler = (message) -> sink.emitNext((Message<T>) message);
+			MessageHandler messageHandler = (message) -> {
+				while (!sink.emitNext((Message<T>) message).hasEmitted()) {
+					LockSupport.parkNanos(10);
+				}
+			};
 			inputChannel.subscribe(messageHandler);
 			return sink.asFlux()
 					.doOnCancel(() -> inputChannel.unsubscribe(messageHandler));

spring-integration-core/src/test/java/org/springframework/integration/channel/MessageChannelReactiveUtilsTests.java

@@ -21,7 +21,6 @@
 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 @@ void testBackpressureWithSubscribableChannel() {
 	}
 
 	@Test
-	@Disabled("Backpressure is not honored")
 	void testOverproducingWithSubscribableChannel() {
 		DirectChannel channel = new DirectChannel();
 

spring-integration-stomp/src/test/java/org/springframework/integration/stomp/client/StompServerIntegrationTests.java

@@ -22,6 +22,7 @@
 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 @@
  *
  * @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;