Handle new Reactor Emission FAIL_NON_SERIALIZED

artembilan · artembilan · commit 09dec2eab521 · 2020-09-11T14:12:09.000-04:00
* Rework WebFlux test to JUnit 5
diff --git a/spring-integration-core/src/main/java/org/springframework/integration/channel/FluxMessageChannel.java b/spring-integration-core/src/main/java/org/springframework/integration/channel/FluxMessageChannel.java
@@ -83,14 +83,17 @@ protected boolean doSend(Message<?> message, long timeout) {
 
 	private boolean tryEmitMessage(Message<?> message) {
 		switch (this.sink.tryEmitNext(message)) {
+			case OK:
+				return true;
+			case FAIL_NON_SERIALIZED:
 			case FAIL_OVERFLOW:
 				return false;
 			case FAIL_TERMINATED:
 			case FAIL_CANCELLED:
 				throw new IllegalStateException("Cannot emit messages into the cancelled or terminated sink: "
 						+ this.sink);
 			default:
-				return true;
+				throw new UnsupportedOperationException();
 		}
 	}
 
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
@@ -124,14 +124,14 @@ else if (messageChannel instanceof PollableChannel) {
 		}
 	}
 
+	@SuppressWarnings("unchecked")
 	private static <T> Flux<Message<T>> adaptSubscribableChannelToPublisher(SubscribableChannel inputChannel) {
 		return Flux.defer(() -> {
 			Sinks.Many<Message<T>> sink = Sinks.many().unicast().onBackpressureError();
 			MessageHandler messageHandler = (message) -> {
 				while (true) {
-					@SuppressWarnings("unchecked")
-					Sinks.Emission emission = sink.tryEmitNext((Message<T>) message);
-					switch (emission) {
+					switch (sink.tryEmitNext((Message<T>) message)) {
+						case FAIL_NON_SERIALIZED:
 						case FAIL_OVERFLOW:
 							LockSupport.parkNanos(1000); // NOSONAR
 							break;
diff --git a/spring-integration-webflux/src/test/java/org/springframework/integration/webflux/config/WebFluxInboundChannelAdapterParserTests.java b/spring-integration-webflux/src/test/java/org/springframework/integration/webflux/config/WebFluxInboundChannelAdapterParserTests.java
@@ -1,5 +1,5 @@
 /*
- * Copyright 2018-2019 the original author or authors.
+ * Copyright 2018-2020 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.
@@ -20,8 +20,7 @@
 
 import java.util.Map;
 
-import org.junit.Test;
-import org.junit.runner.RunWith;
+import org.junit.jupiter.api.Test;
 
 import org.springframework.beans.DirectFieldAccessor;
 import org.springframework.beans.factory.annotation.Autowired;
@@ -36,7 +35,7 @@
 import org.springframework.integration.webflux.inbound.WebFluxInboundEndpoint;
 import org.springframework.messaging.MessageChannel;
 import org.springframework.test.annotation.DirtiesContext;
-import org.springframework.test.context.junit4.SpringRunner;
+import org.springframework.test.context.junit.jupiter.SpringJUnitConfig;
 import org.springframework.validation.Validator;
 import org.springframework.web.reactive.accept.RequestedContentTypeResolver;
 
@@ -45,7 +44,7 @@
  *
  * @since 5.0.1
  */
-@RunWith(SpringRunner.class)
+@SpringJUnitConfig
 @DirtiesContext
 public class WebFluxInboundChannelAdapterParserTests {
 
@@ -120,7 +119,7 @@ public void reactiveFullConfig() {
 
 		CrossOrigin crossOrigin = (CrossOrigin) endpointAccessor.getPropertyValue("crossOrigin");
 		assertThat(crossOrigin).isNotNull();
-		assertThat(crossOrigin.getOrigin()).isEqualTo(new String[] { "foo" });
+		assertThat(crossOrigin.getOrigin()).isEqualTo(new String[]{ "foo" });
 
 		assertThat(endpointAccessor.getPropertyValue("requestPayloadType"))
 				.isEqualTo(ResolvableType.forClass(byte[].class));
diff --git a/spring-integration-webflux/src/test/java/org/springframework/integration/webflux/config/WebFluxInboundGatewayParserTests.java b/spring-integration-webflux/src/test/java/org/springframework/integration/webflux/config/WebFluxInboundGatewayParserTests.java
@@ -1,5 +1,5 @@
 /*
- * Copyright 2018-2019 the original author or authors.
+ * Copyright 2018-2020 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.
@@ -20,8 +20,7 @@
 
 import java.util.Map;
 
-import org.junit.Test;
-import org.junit.runner.RunWith;
+import org.junit.jupiter.api.Test;
 
 import org.springframework.beans.DirectFieldAccessor;
 import org.springframework.beans.factory.annotation.Autowired;
@@ -36,15 +35,15 @@
 import org.springframework.integration.webflux.inbound.WebFluxInboundEndpoint;
 import org.springframework.messaging.MessageChannel;
 import org.springframework.test.annotation.DirtiesContext;
-import org.springframework.test.context.junit4.SpringRunner;
+import org.springframework.test.context.junit.jupiter.SpringJUnitConfig;
 import org.springframework.web.reactive.accept.RequestedContentTypeResolver;
 
 /**
  * @author Artem Bilan
  *
  * @since 5.0.1
  */
-@RunWith(SpringRunner.class)
+@SpringJUnitConfig
 @DirtiesContext
 public class WebFluxInboundGatewayParserTests {
 
@@ -115,7 +114,7 @@ public void reactiveFullConfig() {
 
 		CrossOrigin crossOrigin = (CrossOrigin) endpointAccessor.getPropertyValue("crossOrigin");
 		assertThat(crossOrigin).isNotNull();
-		assertThat(crossOrigin.getOrigin()).isEqualTo(new String[] { "foo" });
+		assertThat(crossOrigin.getOrigin()).isEqualTo(new String[]{ "foo" });
 
 		assertThat(endpointAccessor.getPropertyValue("requestPayloadType"))
 				.isEqualTo(ResolvableType.forClass(byte[].class));
diff --git a/spring-integration-webflux/src/test/java/org/springframework/integration/webflux/config/WebFluxOutboundGatewayParserTests.java b/spring-integration-webflux/src/test/java/org/springframework/integration/webflux/config/WebFluxOutboundGatewayParserTests.java
@@ -22,8 +22,7 @@
 import java.nio.charset.StandardCharsets;
 import java.util.Map;
 
-import org.junit.Test;
-import org.junit.runner.RunWith;
+import org.junit.jupiter.api.Test;
 
 import org.springframework.beans.DirectFieldAccessor;
 import org.springframework.beans.factory.annotation.Autowired;
@@ -35,7 +34,7 @@
 import org.springframework.integration.test.util.TestUtils;
 import org.springframework.messaging.MessageChannel;
 import org.springframework.test.annotation.DirtiesContext;
-import org.springframework.test.context.junit4.SpringRunner;
+import org.springframework.test.context.junit.jupiter.SpringJUnitConfig;
 import org.springframework.util.ObjectUtils;
 import org.springframework.web.reactive.function.BodyExtractor;
 import org.springframework.web.reactive.function.client.WebClient;
@@ -45,7 +44,7 @@
  *
  * @since 5.0
  */
-@RunWith(SpringRunner.class)
+@SpringJUnitConfig
 @DirtiesContext
 public class WebFluxOutboundGatewayParserTests {
 
diff --git a/spring-integration-webflux/src/test/java/org/springframework/integration/webflux/inbound/WebFluxInboundEndpointTests.java b/spring-integration-webflux/src/test/java/org/springframework/integration/webflux/inbound/WebFluxInboundEndpointTests.java
@@ -1,5 +1,5 @@
 /*
- * Copyright 2017-2019 the original author or authors.
+ * Copyright 2017-2020 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.
@@ -18,8 +18,7 @@
 
 import java.util.Objects;
 
-import org.junit.Test;
-import org.junit.runner.RunWith;
+import org.junit.jupiter.api.Test;
 
 import org.springframework.beans.factory.annotation.Autowired;
 import org.springframework.context.ApplicationContext;
@@ -35,7 +34,7 @@
 import org.springframework.integration.http.inbound.RequestMapping;
 import org.springframework.messaging.MessageChannel;
 import org.springframework.test.annotation.DirtiesContext;
-import org.springframework.test.context.junit4.SpringRunner;
+import org.springframework.test.context.junit.jupiter.SpringJUnitConfig;
 import org.springframework.test.web.reactive.server.WebTestClient;
 import org.springframework.web.reactive.config.EnableWebFlux;
 
@@ -48,7 +47,7 @@
  *
  * @since 5.0
  */
-@RunWith(SpringRunner.class)
+@SpringJUnitConfig
 @DirtiesContext
 public class WebFluxInboundEndpointTests {
 
diff --git a/spring-integration-webflux/src/test/java/org/springframework/integration/webflux/management/IntegrationGraphControllerTests.java b/spring-integration-webflux/src/test/java/org/springframework/integration/webflux/management/IntegrationGraphControllerTests.java
@@ -1,5 +1,5 @@
 /*
- * Copyright 2018-2019 the original author or authors.
+ * Copyright 2018-2020 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.
@@ -16,8 +16,7 @@
 
 package org.springframework.integration.webflux.management;
 
-import org.junit.Test;
-import org.junit.runner.RunWith;
+import org.junit.jupiter.api.Test;
 
 import org.springframework.beans.factory.annotation.Autowired;
 import org.springframework.context.annotation.Configuration;
@@ -26,8 +25,7 @@
 import org.springframework.integration.http.config.EnableIntegrationGraphController;
 import org.springframework.test.annotation.DirtiesContext;
 import org.springframework.test.context.TestPropertySource;
-import org.springframework.test.context.junit4.SpringRunner;
-import org.springframework.test.context.web.WebAppConfiguration;
+import org.springframework.test.context.junit.jupiter.web.SpringJUnitWebConfig;
 import org.springframework.test.web.reactive.server.WebTestClient;
 import org.springframework.web.context.WebApplicationContext;
 import org.springframework.web.reactive.config.EnableWebFlux;
@@ -37,8 +35,7 @@
  *
  * @since 5.0.2
  */
-@RunWith(SpringRunner.class)
-@WebAppConfiguration
+@SpringJUnitWebConfig
 @TestPropertySource(properties = "spring.application.name:testApplication")
 @DirtiesContext
 public class IntegrationGraphControllerTests {
diff --git a/spring-integration-webflux/src/test/java/org/springframework/integration/webflux/outbound/WebFluxRequestExecutingMessageHandlerTests.java b/spring-integration-webflux/src/test/java/org/springframework/integration/webflux/outbound/WebFluxRequestExecutingMessageHandlerTests.java
@@ -1,5 +1,5 @@
 /*
- * Copyright 2017-2019 the original author or authors.
+ * Copyright 2017-2020 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.
@@ -78,7 +78,7 @@ void testReactiveReturn() {
 				.assertNext(m -> assertThat(m.getHeaders()).containsEntry(HttpHeaders.STATUS_CODE, HttpStatus.OK))
 				.expectNoEvent(Duration.ofMillis(100))
 				.thenCancel()
-				.verify(Duration.ofSeconds(1));
+				.verify(Duration.ofSeconds(10));
 	}
 
 	@Test

Original file line number	Diff line number	Diff line change
`@@ -1,5 +1,5 @@`
`1`	`1`	`/*`
`2`		`- * Copyright 2017-2019 the original author or authors.`
	`2`	`+ * Copyright 2017-2020 the original author or authors.`
`3`	`3`	`*`
`4`	`4`	`* Licensed under the Apache License, Version 2.0 (the "License");`
`5`	`5`	`* you may not use this file except in compliance with the License.`
`@@ -78,7 +78,7 @@ void testReactiveReturn() {`
`78`	`78`	`.assertNext(m -> assertThat(m.getHeaders()).containsEntry(HttpHeaders.STATUS_CODE, HttpStatus.OK))`
`79`	`79`	`.expectNoEvent(Duration.ofMillis(100))`
`80`	`80`	`.thenCancel()`
`81`		`- .verify(Duration.ofSeconds(1));`
	`81`	`+ .verify(Duration.ofSeconds(10));`
`82`	`82`	`}`
`83`	`83`
`84`	`84`	`@Test`