diff --git a/spring-integration-core/src/main/java/org/springframework/integration/gateway/MessagingGatewaySupport.java b/spring-integration-core/src/main/java/org/springframework/integration/gateway/MessagingGatewaySupport.java index a3f2a68101..20b0161984 100644 --- a/spring-integration-core/src/main/java/org/springframework/integration/gateway/MessagingGatewaySupport.java +++ b/spring-integration-core/src/main/java/org/springframework/integration/gateway/MessagingGatewaySupport.java @@ -606,39 +606,40 @@ public abstract class MessagingGatewaySupport extends AbstractEndpoint private Mono> doSendAndReceiveMessageReactive(MessageChannel requestChannel, Object object, boolean error) { + final Message requestMessage; + try { + Message message = + object instanceof Message + ? (Message) object + : this.requestMapper.toMessage(object); + + message = this.historyWritingPostProcessor.postProcessMessage(message); + requestMessage = message; + } + catch (Exception e) { + throw new MessageMappingException("Cannot map to message: " + object, e); + } return Mono.defer(() -> { - Message message; - try { - message = object instanceof Message - ? (Message) object - : this.requestMapper.toMessage(object); - message = this.historyWritingPostProcessor.postProcessMessage(message); - - } - catch (Exception e) { - throw new MessageMappingException("Cannot map to message: " + object, e); - } - - Object originalReplyChannelHeader = message.getHeaders().getReplyChannel(); - Object originalErrorChannelHeader = message.getHeaders().getErrorChannel(); + Object originalReplyChannelHeader = requestMessage.getHeaders().getReplyChannel(); + Object originalErrorChannelHeader = requestMessage.getHeaders().getErrorChannel(); MonoReplyChannel replyChan = new MonoReplyChannel(); - Message requestMessage = MutableMessageBuilder.fromMessage(message) + Message messageToSend = MutableMessageBuilder.fromMessage(requestMessage) .setReplyChannel(replyChan) .setHeader(this.messagingTemplate.getSendTimeoutHeader(), null) .setHeader(this.messagingTemplate.getReceiveTimeoutHeader(), null) .setErrorChannel(replyChan) .build(); - sendMessageForReactiveFlow(requestChannel, requestMessage); + sendMessageForReactiveFlow(requestChannel, messageToSend); return buildReplyMono(requestMessage, replyChan.replyMono, error, originalReplyChannelHeader, - originalErrorChannelHeader) - .onErrorResume(t -> error ? Mono.error(t) : handleSendError(requestMessage, t)); - }); + originalErrorChannelHeader); + }) + .onErrorResume(t -> error ? Mono.error(t) : handleSendError(requestMessage, t)); } private void sendMessageForReactiveFlow(MessageChannel requestChannel, Message requestMessage) { diff --git a/spring-integration-http/src/test/java/org/springframework/integration/http/dsl/HttpDslTests.java b/spring-integration-http/src/test/java/org/springframework/integration/http/dsl/HttpDslTests.java index 545b70fa83..834463915c 100644 --- a/spring-integration-http/src/test/java/org/springframework/integration/http/dsl/HttpDslTests.java +++ b/spring-integration-http/src/test/java/org/springframework/integration/http/dsl/HttpDslTests.java @@ -31,9 +31,8 @@ import java.util.Collections; import java.util.List; import java.util.Map; -import org.junit.Before; -import org.junit.Test; -import org.junit.runner.RunWith; +import org.junit.jupiter.api.BeforeEach; +import org.junit.jupiter.api.Test; import org.springframework.beans.factory.annotation.Autowired; import org.springframework.beans.factory.annotation.Qualifier; @@ -45,10 +44,12 @@ import org.springframework.http.ResponseEntity; import org.springframework.http.client.ClientHttpRequestFactory; import org.springframework.http.client.ClientHttpResponse; import org.springframework.integration.channel.DirectChannel; +import org.springframework.integration.channel.FixedSubscriberChannel; import org.springframework.integration.config.EnableIntegration; import org.springframework.integration.dsl.IntegrationFlow; import org.springframework.integration.dsl.IntegrationFlows; import org.springframework.integration.dsl.context.IntegrationFlowContext; +import org.springframework.integration.handler.AbstractReplyProducingMessageHandler; import org.springframework.integration.http.multipart.UploadedMultipartFile; import org.springframework.integration.http.outbound.HttpRequestExecutingMessageHandler; import org.springframework.integration.security.channel.ChannelSecurityInterceptor; @@ -68,8 +69,7 @@ import org.springframework.security.core.userdetails.UserDetailsService; import org.springframework.security.crypto.factory.PasswordEncoderFactories; import org.springframework.security.provisioning.InMemoryUserDetailsManager; import org.springframework.test.annotation.DirtiesContext; -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.client.MockMvcClientHttpRequestFactory; import org.springframework.test.web.servlet.MockMvc; import org.springframework.test.web.servlet.setup.MockMvcBuilders; @@ -91,8 +91,7 @@ import org.springframework.web.servlet.DispatcherServlet; * * @since 5.0 */ -@RunWith(SpringRunner.class) -@WebAppConfiguration +@SpringJUnitWebConfig @DirtiesContext public class HttpDslTests { @@ -107,7 +106,7 @@ public class HttpDslTests { private MockMvc mockMvc; - @Before + @BeforeEach public void setup() { this.mockMvc = MockMvcBuilders.webAppContextSetup(this.wac) @@ -236,6 +235,37 @@ public class HttpDslTests { } + @Test + public void testErrorChannelFlow() throws Exception { + IntegrationFlow flow = + IntegrationFlows.from( + Http.inboundGateway("/errorFlow") + .errorChannel(new FixedSubscriberChannel( + new AbstractReplyProducingMessageHandler() { + + @Override + protected Object handleRequestMessage(Message requestMessage) { + return "Error Response"; + } + + }))) + .transform((payload) -> { + throw new RuntimeException("Error!"); + }) + .get(); + + IntegrationFlowContext.IntegrationFlowRegistration flowRegistration = + this.integrationFlowContext.registration(flow).register(); + + this.mockMvc.perform( + get("/errorFlow") + .with(httpBasic("user", "user"))) + .andExpect(status().isOk()) + .andExpect(content().string("Error Response")); + + flowRegistration.destroy(); + } + @Configuration @EnableWebSecurity @EnableIntegration diff --git a/spring-integration-webflux/src/test/java/org/springframework/integration/webflux/dsl/WebFluxDslTests.java b/spring-integration-webflux/src/test/java/org/springframework/integration/webflux/dsl/WebFluxDslTests.java index 72ac27815d..ca5e0afada 100644 --- a/spring-integration-webflux/src/test/java/org/springframework/integration/webflux/dsl/WebFluxDslTests.java +++ b/spring-integration-webflux/src/test/java/org/springframework/integration/webflux/dsl/WebFluxDslTests.java @@ -28,9 +28,8 @@ import java.util.Collections; import javax.annotation.Resource; import org.hamcrest.Matchers; -import org.junit.Before; -import org.junit.Test; -import org.junit.runner.RunWith; +import org.junit.jupiter.api.BeforeEach; +import org.junit.jupiter.api.Test; import org.reactivestreams.Publisher; import org.springframework.beans.DirectFieldAccessor; @@ -44,12 +43,14 @@ import org.springframework.http.HttpMethod; import org.springframework.http.HttpStatus; import org.springframework.http.MediaType; import org.springframework.http.client.reactive.ClientHttpConnector; +import org.springframework.integration.channel.FixedSubscriberChannel; import org.springframework.integration.channel.QueueChannel; import org.springframework.integration.config.EnableIntegration; import org.springframework.integration.dsl.IntegrationFlow; import org.springframework.integration.dsl.IntegrationFlows; import org.springframework.integration.dsl.MessageChannels; import org.springframework.integration.dsl.context.IntegrationFlowContext; +import org.springframework.integration.handler.AbstractReplyProducingMessageHandler; import org.springframework.integration.http.HttpHeaders; import org.springframework.integration.http.dsl.Http; import org.springframework.integration.support.MessageBuilder; @@ -76,8 +77,7 @@ import org.springframework.security.test.web.reactive.server.SecurityMockServerC import org.springframework.security.test.web.servlet.setup.SecurityMockMvcConfigurers; import org.springframework.security.web.server.SecurityWebFilterChain; import org.springframework.test.annotation.DirtiesContext; -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.HttpHandlerConnector; import org.springframework.test.web.reactive.server.WebTestClient; import org.springframework.test.web.servlet.MockMvc; @@ -104,8 +104,7 @@ import reactor.test.StepVerifier; * * @since 5.0 */ -@RunWith(SpringRunner.class) -@WebAppConfiguration +@SpringJUnitWebConfig @DirtiesContext public class WebFluxDslTests { @@ -130,7 +129,7 @@ public class WebFluxDslTests { private WebTestClient webTestClient; - @Before + @BeforeEach public void setup() { this.mockMvc = MockMvcBuilders.webAppContextSetup(this.wac) @@ -320,6 +319,39 @@ public class WebFluxDslTests { flowRegistration.destroy(); } + @Test + public void testErrorChannelFlow() { + IntegrationFlow flow = + IntegrationFlows.from( + WebFlux.inboundGateway("/errorFlow") + .errorChannel(new FixedSubscriberChannel( + new AbstractReplyProducingMessageHandler() { + + @Override + protected Object handleRequestMessage(Message requestMessage) { + return "Error Response"; + } + + }))) + .channel(MessageChannels.flux()) + .transform((payload) -> { + throw new RuntimeException("Error!"); + }) + .get(); + + IntegrationFlowContext.IntegrationFlowRegistration flowRegistration = + this.integrationFlowContext.registration(flow).register(); + + this.webTestClient.get().uri("/errorFlow") + .headers(headers -> headers.setBasicAuth("guest", "guest")) + .exchange() + .expectStatus() + .isOk() + .expectBody(String.class) + .isEqualTo("Error Response"); + + flowRegistration.destroy(); + } @Configuration @EnableWebFlux @@ -445,7 +477,7 @@ public class WebFluxDslTests { .from(WebFlux.inboundGateway("/sse") .requestMapping(m -> m.produces(MediaType.TEXT_EVENT_STREAM_VALUE)) .mappedResponseHeaders("*")) - .enrichHeaders(Collections.singletonMap("aHeader", new String[] { "foo", "bar", "baz" })) + .enrichHeaders(Collections.singletonMap("aHeader", new String[]{"foo", "bar", "baz"})) .handle((p, h) -> Flux.fromArray(h.get("aHeader", String[].class))) .get(); }