From 09dec2eab5216586f13cdcf45308238e5072a931 Mon Sep 17 00:00:00 2001 From: Artem Bilan Date: Fri, 11 Sep 2020 14:12:09 -0400 Subject: [PATCH] Handle new Reactor Emission FAIL_NON_SERIALIZED * Rework WebFlux test to JUnit 5 --- .../integration/channel/FluxMessageChannel.java | 5 ++++- .../integration/util/IntegrationReactiveUtils.java | 6 +++--- .../WebFluxInboundChannelAdapterParserTests.java | 11 +++++------ .../config/WebFluxInboundGatewayParserTests.java | 11 +++++------ .../config/WebFluxOutboundGatewayParserTests.java | 7 +++---- .../webflux/inbound/WebFluxInboundEndpointTests.java | 9 ++++----- .../management/IntegrationGraphControllerTests.java | 11 ++++------- .../WebFluxRequestExecutingMessageHandlerTests.java | 4 ++-- 8 files changed, 30 insertions(+), 34 deletions(-) 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 index 9a8118fc6a..41a4c9f318 100644 --- 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,6 +83,9 @@ public class FluxMessageChannel extends AbstractMessageChannel 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: @@ -90,7 +93,7 @@ public class FluxMessageChannel extends AbstractMessageChannel 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 index 9d5ee1bc0c..ec20cd24bc 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 @@ -124,14 +124,14 @@ public final class IntegrationReactiveUtils { } } + @SuppressWarnings("unchecked") private static Flux> adaptSubscribableChannelToPublisher(SubscribableChannel inputChannel) { return Flux.defer(() -> { Sinks.Many> sink = Sinks.many().unicast().onBackpressureError(); MessageHandler messageHandler = (message) -> { while (true) { - @SuppressWarnings("unchecked") - Sinks.Emission emission = sink.tryEmitNext((Message) message); - switch (emission) { + switch (sink.tryEmitNext((Message) 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 index e2c632eb0f..a796afc7bd 100644 --- 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 static org.assertj.core.api.Assertions.assertThat; 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.mapping.HeaderMapper; 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 @@ import org.springframework.web.reactive.accept.RequestedContentTypeResolver; * * @since 5.0.1 */ -@RunWith(SpringRunner.class) +@SpringJUnitConfig @DirtiesContext public class WebFluxInboundChannelAdapterParserTests { @@ -120,7 +119,7 @@ public class WebFluxInboundChannelAdapterParserTests { 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 index 77ed4af71d..433e82ed89 100644 --- 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 static org.assertj.core.api.Assertions.assertThat; 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.mapping.HeaderMapper; 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; /** @@ -44,7 +43,7 @@ import org.springframework.web.reactive.accept.RequestedContentTypeResolver; * * @since 5.0.1 */ -@RunWith(SpringRunner.class) +@SpringJUnitConfig @DirtiesContext public class WebFluxInboundGatewayParserTests { @@ -115,7 +114,7 @@ public class WebFluxInboundGatewayParserTests { 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 index 326fd42a80..be823dc776 100644 --- 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.Charset; 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.endpoint.AbstractEndpoint; 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 @@ import org.springframework.web.reactive.function.client.WebClient; * * @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 index 93da402b40..36b3a2e8d1 100644 --- 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 @@ package org.springframework.integration.webflux.inbound; 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.config.EnableIntegration; 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 @@ import reactor.core.publisher.Flux; * * @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 index e3e0ad20ff..106dfd3b8d 100644 --- 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.config.EnableIntegration; 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 @@ import org.springframework.web.reactive.config.EnableWebFlux; * * @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 index 129ccacfb9..6d5ccab8f7 100644 --- 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 @@ class WebFluxRequestExecutingMessageHandlerTests { .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