Spring/spring-integration
1
0
Fork 0
Code Issues Packages Projects Releases Wiki Activity

INT-4315: Add WebFlux module

Browse Source 
JIRA: https://jira.spring.io/browse/INT-4315

Move Reactive components outside of HTTP module to the new WebFlux one,
including XSD, tests and documentation
Make an appropriate polishing for the `http.adoc` with cross-link
to the `webflux.adoc`

Exclude transitive `spring-webmvc` for the `spring-integration-webflux`
This commit is contained in:
Artem Bilan
2017-08-11 10:37:08 -04:00
committed by Gary Russell
parent 4fd32d2bd4
commit 84d60f4ab4
52 changed files with 1657 additions and 1003 deletions
Download Patch File Download Diff File Expand all files Collapse all files

8
spring-integration-http/src/test/java/org/springframework/integration/http/config/HttpOutboundChannelAdapterParserTests-context.xml
View File

@@ -17,16 +17,8 @@
<outbound-channel-adapter id="restTemplateConfig" url="http://localhost/test1" channel="requests"
rest-template="customRestTemplate"/>
<outbound-reactive-channel-adapter id="reactiveMinimalConfig" url="http://localhost/test1" channel="requests"/>
<outbound-reactive-channel-adapter id="reactiveWebClientConfig" url="http://localhost/test1" channel="requests"
web-client="webClient"/>
<beans:bean id="customRestTemplate" class="org.springframework.web.client.RestTemplate"/>
<beans:bean id="webClient" class="org.springframework.web.reactive.function.client.WebClient"
factory-method="create"/>
<outbound-channel-adapter id="fullConfig"
url="http://localhost/test2/{foo}"
http-method="GET"

36
spring-integration-http/src/test/java/org/springframework/integration/http/config/HttpOutboundChannelAdapterParserTests.java
View File

@@ -58,7 +58,6 @@ import org.springframework.test.context.junit4.SpringJUnit4ClassRunner;
import org.springframework.util.ObjectUtils;
import org.springframework.web.client.ResponseErrorHandler;
import org.springframework.web.client.RestTemplate;
import org.springframework.web.reactive.function.client.WebClient;
/**
* @author Mark Fisher
@@ -85,21 +84,10 @@ public class HttpOutboundChannelAdapterParserTests {
@Qualifier("restTemplateConfig")
private AbstractEndpoint restTemplateConfig;
@Autowired
@Qualifier("reactiveMinimalConfig")
private AbstractEndpoint reactiveMinimalConfig;
@Autowired
@Qualifier("reactiveWebClientConfig")
private AbstractEndpoint reactiveWebClientConfig;
@Autowired
@Qualifier("customRestTemplate")
private RestTemplate customRestTemplate;
@Autowired
private WebClient webClient;
@Autowired
@Qualifier("withUrlAndTemplate")
private AbstractEndpoint withUrlAndTemplate;
@@ -187,25 +175,6 @@ public class HttpOutboundChannelAdapterParserTests {
assertTrue(ObjectUtils.containsElement(mappedRequestHeaders, "requestHeader2"));
}
@Test
public void reactiveMinimalConfig() {
DirectFieldAccessor endpointAccessor = new DirectFieldAccessor(this.reactiveMinimalConfig);
WebClient webClient =
TestUtils.getPropertyValue(this.reactiveMinimalConfig, "handler.webClient", WebClient.class);
assertNotSame(this.webClient, webClient);
Object handler = endpointAccessor.getPropertyValue("handler");
DirectFieldAccessor handlerAccessor = new DirectFieldAccessor(handler);
assertEquals(false, handlerAccessor.getPropertyValue("expectReply"));
assertEquals(this.applicationContext.getBean("requests"), endpointAccessor.getPropertyValue("inputChannel"));
assertNull(handlerAccessor.getPropertyValue("outputChannel"));
Expression uriExpression = (Expression) handlerAccessor.getPropertyValue("uriExpression");
assertEquals("http://localhost/test1", uriExpression.getValue());
assertEquals(HttpMethod.POST.name(),
TestUtils.getPropertyValue(handler, "httpMethodExpression", Expression.class).getExpressionString());
assertEquals(Charset.forName("UTF-8"), handlerAccessor.getPropertyValue("charset"));
assertEquals(true, handlerAccessor.getPropertyValue("extractPayload"));
}
@Test
public void restTemplateConfig() {
RestTemplate restTemplate =
@@ -213,11 +182,6 @@ public class HttpOutboundChannelAdapterParserTests {
assertEquals(customRestTemplate, restTemplate);
}
@Test
public void reactiveWebClientConfig() {
assertSame(this.webClient, TestUtils.getPropertyValue(this.reactiveWebClientConfig, "handler.webClient"));
}
@Test(expected = BeanDefinitionParsingException.class)
public void failWithRestTemplateAndRestAttributes() {
new ClassPathXmlApplicationContext("HttpOutboundChannelAdapterParserTests-fail-context.xml", this.getClass())

23
spring-integration-http/src/test/java/org/springframework/integration/http/config/HttpOutboundGatewayParserTests-context.xml
View File

@@ -14,9 +14,6 @@
<si:channel id="requests"/>
<beans:bean id="webClient" class="org.springframework.web.reactive.function.client.WebClient"
factory-method="create"/>
<outbound-gateway id="minimalConfig" url="http://localhost/test1" request-channel="requests"/>
<si:channel id="replies">
@@ -43,26 +40,6 @@
<uri-variable name="foo" expression="headers.bar"/>
</outbound-gateway>
<outbound-reactive-gateway id="reactiveMinimalConfig" url="http://localhost/test1" request-channel="requests"
web-client="webClient"/>
<outbound-reactive-gateway id="reactiveFullConfig"
url="http://localhost/test2"
http-method="PUT"
request-channel="requests"
reply-timeout="1234"
extract-request-payload="false"
expected-response-type="java.lang.String"
mapped-request-headers="requestHeader1, requestHeader2"
mapped-response-headers="responseHeader"
reply-channel="replies"
charset="UTF-8"
order="77"
auto-startup="false"
transfer-cookies="true">
<uri-variable name="foo" expression="headers.bar"/>
</outbound-reactive-gateway>
<util:map id="uriVariables">
<beans:entry key="foo1" value="bar1"/>
<beans:entry key="foo2" value="bar2"/>

73
spring-integration-http/src/test/java/org/springframework/integration/http/config/HttpOutboundGatewayParserTests.java
View File

@@ -19,7 +19,6 @@ package org.springframework.integration.http.config;
import static org.junit.Assert.assertEquals;
import static org.junit.Assert.assertNotNull;
import static org.junit.Assert.assertNull;
import static org.junit.Assert.assertSame;
import static org.junit.Assert.assertThat;
import static org.junit.Assert.assertTrue;
import static org.junit.Assert.fail;
@@ -58,7 +57,6 @@ import org.springframework.test.context.ContextConfiguration;
import org.springframework.test.context.junit4.SpringJUnit4ClassRunner;
import org.springframework.util.ObjectUtils;
import org.springframework.web.client.ResponseErrorHandler;
import org.springframework.web.reactive.function.client.WebClient;
/**
* @author Mark Fisher
@@ -79,14 +77,6 @@ public class HttpOutboundGatewayParserTests {
@Qualifier("fullConfig")
private AbstractEndpoint fullConfigEndpoint;
@Autowired
@Qualifier("reactiveMinimalConfig")
private AbstractEndpoint reactiveMinimalConfigEndpoint;
@Autowired
@Qualifier("reactiveFullConfig")
private AbstractEndpoint reactiveFullConfigEndpoint;
@Autowired
@Qualifier("withUrlExpression")
private AbstractEndpoint withUrlExpressionEndpoint;
@@ -99,9 +89,6 @@ public class HttpOutboundGatewayParserTests {
@Qualifier("withPoller1")
private AbstractEndpoint withPoller1;
@Autowired
private WebClient webClient;
@Autowired
private ApplicationContext applicationContext;
@@ -178,66 +165,6 @@ public class HttpOutboundGatewayParserTests {
assertEquals(true, handlerAccessor.getPropertyValue("transferCookies"));
}
@Test
public void reactiveMinimalConfig() {
Object handler = new DirectFieldAccessor(this.reactiveMinimalConfigEndpoint).getPropertyValue("handler");
Object requestChannel = new DirectFieldAccessor(this.reactiveMinimalConfigEndpoint)
.getPropertyValue("inputChannel");
assertEquals(this.applicationContext.getBean("requests"), requestChannel);
DirectFieldAccessor handlerAccessor = new DirectFieldAccessor(handler);
Object replyChannel = handlerAccessor.getPropertyValue("outputChannel");
assertNull(replyChannel);
assertSame(this.webClient, handlerAccessor.getPropertyValue("webClient"));
Expression uriExpression = (Expression) handlerAccessor.getPropertyValue("uriExpression");
assertEquals("http://localhost/test1", uriExpression.getValue());
assertEquals(HttpMethod.POST.name(),
TestUtils.getPropertyValue(handler, "httpMethodExpression", Expression.class).getExpressionString());
assertEquals(Charset.forName("UTF-8"), handlerAccessor.getPropertyValue("charset"));
assertEquals(true, handlerAccessor.getPropertyValue("extractPayload"));
assertEquals(false, handlerAccessor.getPropertyValue("transferCookies"));
}
@Test
@SuppressWarnings("unchecked")
public void reactiveFullConfig() {
DirectFieldAccessor endpointAccessor = new DirectFieldAccessor(this.reactiveFullConfigEndpoint);
Object handler = endpointAccessor.getPropertyValue("handler");
MessageChannel requestChannel = (MessageChannel) new DirectFieldAccessor(
this.reactiveFullConfigEndpoint).getPropertyValue("inputChannel");
assertEquals(this.applicationContext.getBean("requests"), requestChannel);
DirectFieldAccessor handlerAccessor = new DirectFieldAccessor(handler);
assertEquals(77, handlerAccessor.getPropertyValue("order"));
assertEquals(Boolean.FALSE, endpointAccessor.getPropertyValue("autoStartup"));
Object replyChannel = handlerAccessor.getPropertyValue("outputChannel");
assertNotNull(replyChannel);
assertEquals(this.applicationContext.getBean("replies"), replyChannel);
assertEquals(String.class.getName(),
TestUtils.getPropertyValue(handler, "expectedResponseTypeExpression", Expression.class).getValue());
Expression uriExpression = (Expression) handlerAccessor.getPropertyValue("uriExpression");
assertEquals("http://localhost/test2", uriExpression.getValue());
assertEquals(HttpMethod.PUT.name(),
TestUtils.getPropertyValue(handler, "httpMethodExpression", Expression.class).getExpressionString());
assertEquals(Charset.forName("UTF-8"), handlerAccessor.getPropertyValue("charset"));
assertEquals(false, handlerAccessor.getPropertyValue("extractPayload"));
Object sendTimeout = new DirectFieldAccessor(
handlerAccessor.getPropertyValue("messagingTemplate")).getPropertyValue("sendTimeout");
assertEquals(new Long("1234"), sendTimeout);
Map<String, Expression> uriVariableExpressions =
(Map<String, Expression>) handlerAccessor.getPropertyValue("uriVariableExpressions");
assertEquals(1, uriVariableExpressions.size());
assertEquals("headers.bar", uriVariableExpressions.get("foo").getExpressionString());
DirectFieldAccessor mapperAccessor = new DirectFieldAccessor(handlerAccessor.getPropertyValue("headerMapper"));
String[] mappedRequestHeaders = (String[]) mapperAccessor.getPropertyValue("outboundHeaderNames");
String[] mappedResponseHeaders = (String[]) mapperAccessor.getPropertyValue("inboundHeaderNames");
assertEquals(2, mappedRequestHeaders.length);
assertEquals(1, mappedResponseHeaders.length);
assertTrue(ObjectUtils.containsElement(mappedRequestHeaders, "requestHeader1"));
assertTrue(ObjectUtils.containsElement(mappedRequestHeaders, "requestHeader2"));
assertEquals("responseHeader", mappedResponseHeaders[0]);
assertEquals(true, handlerAccessor.getPropertyValue("transferCookies"));
}
@Test
public void withUrlExpression() {
HttpRequestExecutingMessageHandler handler = (HttpRequestExecutingMessageHandler) new DirectFieldAccessor(

130
spring-integration-http/src/test/java/org/springframework/integration/http/dsl/HttpDslTests.java
View File

@@ -16,9 +16,6 @@
package org.springframework.integration.http.dsl;
import static org.hamcrest.Matchers.instanceOf;
import static org.junit.Assert.assertNotNull;
import static org.junit.Assert.assertThat;
import static org.springframework.security.test.web.servlet.request.SecurityMockMvcRequestPostProcessors.httpBasic;
import static org.springframework.security.test.web.servlet.setup.SecurityMockMvcConfigurers.springSecurity;
import static org.springframework.test.web.servlet.request.MockMvcRequestBuilders.get;
@@ -30,28 +27,19 @@ import java.util.List;
import org.junit.Before;
import org.junit.Test;
import org.junit.runner.RunWith;
import org.reactivestreams.Publisher;
import org.springframework.beans.DirectFieldAccessor;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import org.springframework.core.ResolvableType;
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.DirectChannel;
import org.springframework.integration.config.EnableIntegration;
import org.springframework.integration.dsl.IntegrationFlow;
import org.springframework.integration.dsl.IntegrationFlows;
import org.springframework.integration.http.outbound.HttpRequestExecutingMessageHandler;
import org.springframework.integration.http.outbound.ReactiveHttpRequestExecutingMessageHandler;
import org.springframework.integration.security.channel.ChannelSecurityInterceptor;
import org.springframework.integration.security.channel.SecuredChannel;
import org.springframework.messaging.Message;
import org.springframework.messaging.MessageChannel;
import org.springframework.messaging.PollableChannel;
import org.springframework.security.access.AccessDecisionManager;
import org.springframework.security.access.vote.AffirmativeBased;
import org.springframework.security.access.vote.RoleVoter;
@@ -63,20 +51,11 @@ import org.springframework.test.annotation.DirtiesContext;
import org.springframework.test.context.junit4.SpringRunner;
import org.springframework.test.context.web.WebAppConfiguration;
import org.springframework.test.web.client.MockMvcClientHttpRequestFactory;
import org.springframework.test.web.reactive.server.HttpHandlerConnector;
import org.springframework.test.web.reactive.server.WebTestClient;
import org.springframework.test.web.servlet.MockMvc;
import org.springframework.test.web.servlet.setup.MockMvcBuilders;
import org.springframework.util.MultiValueMap;
import org.springframework.web.client.RestTemplate;
import org.springframework.web.context.WebApplicationContext;
import org.springframework.web.reactive.config.EnableWebFlux;
import org.springframework.web.reactive.function.client.WebClient;
import org.springframework.web.util.UriComponentsBuilder;
import reactor.core.publisher.Flux;
import reactor.core.publisher.Mono;
import reactor.test.StepVerifier;
/**
* @author Artem Bilan
@@ -95,23 +74,14 @@ public class HttpDslTests {
@Autowired
private HttpRequestExecutingMessageHandler serviceInternalGatewayHandler;
@Autowired
private ReactiveHttpRequestExecutingMessageHandler serviceInternalReactiveGatewayHandler;
private MockMvc mockMvc;
private WebTestClient webTestClient;
@Before
public void setup() {
this.mockMvc =
MockMvcBuilders.webAppContextSetup(this.wac)
.apply(springSecurity())
.build();
this.webTestClient =
WebTestClient.bindToApplicationContext(this.wac)
.build();
}
@@ -130,71 +100,7 @@ public class HttpDslTests {
.string("FOO"));
}
@Test
public void testHttpReactiveProxyFlow() throws Exception {
ClientHttpConnector httpConnector = new HttpHandlerConnector((request, response) -> {
response.setStatusCode(HttpStatus.OK);
response.getHeaders().setContentType(MediaType.TEXT_PLAIN);
return response.writeWith(Mono.just(response.bufferFactory().wrap("FOO".getBytes())))
.then(Mono.defer(response::setComplete));
});
WebClient webClient = WebClient.builder()
.clientConnector(httpConnector)
.build();
new DirectFieldAccessor(this.serviceInternalReactiveGatewayHandler)
.setPropertyValue("webClient", webClient);
this.mockMvc.perform(
get("/service2")
.with(httpBasic("guest", "guest"))
.param("name", "foo"))
.andExpect(
content()
.string("FOO"));
}
@Autowired
private PollableChannel storeChannel;
@Test
@SuppressWarnings("unchecked")
public void testHttpReactivePost() {
this.webTestClient.post().uri("/reactivePost")
.body(Flux.just("foo", "bar", "baz"), String.class)
.exchange()
.expectStatus().isAccepted();
Message<?> store = this.storeChannel.receive(10_000);
assertNotNull(store);
assertThat(store.getPayload(), instanceOf(Flux.class));
StepVerifier
.create((Publisher<String>) store.getPayload())
.expectNext("foo", "bar", "baz")
.verifyComplete();
}
@Test
public void testSse() {
Flux<String> responseBody =
this.webTestClient.get().uri("/sse")
.exchange()
.returnResult(String.class)
.getResponseBody();
StepVerifier
.create(responseBody)
.expectNext("foo", "bar", "baz")
.verifyComplete();
}
@Configuration
@EnableWebFlux
@EnableWebSecurity
@EnableIntegration
public static class ContextConfiguration extends WebSecurityConfigurerAdapter {
@@ -247,42 +153,6 @@ public class HttpDslTests {
.get();
}
@Bean
public IntegrationFlow httpReactiveProxyFlow() {
return IntegrationFlows
.from(Http.inboundGateway("/service2")
.requestMapping(r -> r.params("name")))
.handle(Http.<MultiValueMap<String, String>>outboundReactiveGateway(m ->
UriComponentsBuilder.fromUriString("http://www.springsource.org/spring-integration")
.queryParams(m.getPayload())
.build()
.toUri())
.httpMethod(HttpMethod.GET)
.expectedResponseType(String.class),
e -> e.id("serviceInternalReactiveGateway"))
.get();
}
@Bean
public IntegrationFlow httpReactiveInboundChannelAdapterFlow() {
return IntegrationFlows
.from(Http.inboundReactiveChannelAdapter("/reactivePost")
.requestMapping(m -> m.methods(HttpMethod.POST))
.requestPayloadType(ResolvableType.forClassWithGenerics(Flux.class, String.class))
.statusCodeFunction(m -> HttpStatus.ACCEPTED))
.channel(c -> c.queue("storeChannel"))
.get();
}
@Bean
public IntegrationFlow sseFlow() {
return IntegrationFlows
.from(Http.inboundReactiveGateway("/sse")
.requestMapping(m -> m.produces(MediaType.TEXT_EVENT_STREAM_VALUE)))
.handle((p, h) -> Flux.just("foo", "bar", "baz"))
.get();
}
@Bean
public AccessDecisionManager accessDecisionManager() {
return new AffirmativeBased(Collections.singletonList(new RoleVoter()));

170
spring-integration-http/src/test/java/org/springframework/integration/http/inbound/ReactiveHttpInboundEndpointTests.java
View File

@@ -1,170 +0,0 @@
/*
* Copyright 2017 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.
* You may obtain a copy of the License at
*
* http://www.apache.org/licenses/LICENSE-2.0
*
* Unless required by applicable law or agreed to in writing, software
* distributed under the License is distributed on an "AS IS" BASIS,
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
* See the License for the specific language governing permissions and
* limitations under the License.
*/
package org.springframework.integration.http.inbound;
import java.util.Objects;
import org.junit.Test;
import org.junit.runner.RunWith;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.context.ApplicationContext;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import org.springframework.http.HttpStatus;
import org.springframework.http.MediaType;
import org.springframework.integration.annotation.ServiceActivator;
import org.springframework.integration.channel.FluxMessageChannel;
import org.springframework.integration.config.EnableIntegration;
import org.springframework.messaging.MessageChannel;
import org.springframework.test.annotation.DirtiesContext;
import org.springframework.test.context.junit4.SpringRunner;
import org.springframework.test.web.reactive.server.WebTestClient;
import org.springframework.web.reactive.config.EnableWebFlux;
import com.fasterxml.jackson.annotation.JsonCreator;
import com.fasterxml.jackson.annotation.JsonProperty;
import reactor.core.publisher.Flux;
/**
* @author Artem Bilan
*
* @since 5.0
*/
@RunWith(SpringRunner.class)
@DirtiesContext
public class ReactiveHttpInboundEndpointTests {
@Autowired
private WebTestClient webTestClient;
@Autowired
private ReactiveHttpInboundEndpoint simpleInboundEndpoint;
@Test
public void testSimpleGet() {
this.webTestClient.get().uri("/test")
.exchange()
.expectStatus().isOk()
.expectBody(String.class).isEqualTo("It works!");
this.simpleInboundEndpoint.stop();
this.webTestClient.get().uri("/test")
.exchange()
.expectStatus().isEqualTo(HttpStatus.SERVICE_UNAVAILABLE)
.expectBody(String.class).isEqualTo("Endpoint is stopped");
}
@Test
public void testJsonResult() {
this.webTestClient.get().uri("/persons")
.accept(MediaType.APPLICATION_JSON_UTF8)
.exchange()
.expectStatus().isOk()
.expectBody()
.jsonPath("$[0].name").isEqualTo("Jane")
.jsonPath("$[1].name").isEqualTo("Jason")
.jsonPath("$[2].name").isEqualTo("John");
}
@Configuration
@EnableWebFlux
@EnableIntegration
public static class ContextConfiguration {
@Bean
public WebTestClient webTestClient(ApplicationContext applicationContext) {
return WebTestClient.bindToApplicationContext(applicationContext).build();
}
@Bean
public ReactiveHttpInboundEndpoint simpleInboundEndpoint() {
ReactiveHttpInboundEndpoint endpoint = new ReactiveHttpInboundEndpoint();
RequestMapping requestMapping = new RequestMapping();
requestMapping.setPathPatterns("/test");
endpoint.setRequestMapping(requestMapping);
endpoint.setRequestChannelName("serviceChannel");
return endpoint;
}
@ServiceActivator(inputChannel = "serviceChannel")
String service() {
return "It works!";
}
@Bean
public ReactiveHttpInboundEndpoint jsonInboundEndpoint() {
ReactiveHttpInboundEndpoint endpoint = new ReactiveHttpInboundEndpoint();
RequestMapping requestMapping = new RequestMapping();
requestMapping.setPathPatterns("/persons");
endpoint.setRequestMapping(requestMapping);
endpoint.setRequestChannel(fluxResultChannel());
return endpoint;
}
@Bean
public MessageChannel fluxResultChannel() {
return new FluxMessageChannel();
}
@ServiceActivator(inputChannel = "fluxResultChannel")
Flux<Person> getPersons() {
return Flux.just(new Person("Jane"), new Person("Jason"), new Person("John"));
}
}
static class Person {
private final String name;
@JsonCreator
Person(@JsonProperty("name") String name) {
this.name = name;
}
public String getName() {
return this.name;
}
@Override
public boolean equals(Object o) {
if (this == o) {
return true;
}
if (o == null || getClass() != o.getClass()) {
return false;
}
Person person = (Person) o;
return Objects.equals(this.name, person.name);
}
@Override
public int hashCode() {
return getName().hashCode();
}
@Override
public String toString() {
return "Person[name='" + this.name + "']";
}
}
}

142
spring-integration-http/src/test/java/org/springframework/integration/http/outbound/ReactiveHttpRequestExecutingMessageHandlerTests.java
View File

@@ -1,142 +0,0 @@
/*
* Copyright 2017 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.
* You may obtain a copy of the License at
*
* http://www.apache.org/licenses/LICENSE-2.0
*
* Unless required by applicable law or agreed to in writing, software
* distributed under the License is distributed on an "AS IS" BASIS,
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
* See the License for the specific language governing permissions and
* limitations under the License.
*/
package org.springframework.integration.http.outbound;
import static org.hamcrest.Matchers.containsString;
import static org.hamcrest.Matchers.instanceOf;
import static org.junit.Assert.assertNotNull;
import static org.junit.Assert.assertThat;
import static org.springframework.integration.test.matcher.HeaderMatcher.hasHeader;
import java.util.List;
import org.junit.Test;
import org.reactivestreams.Subscriber;
import org.springframework.http.HttpStatus;
import org.springframework.http.client.reactive.ClientHttpConnector;
import org.springframework.integration.channel.FluxMessageChannel;
import org.springframework.integration.channel.QueueChannel;
import org.springframework.integration.http.HttpHeaders;
import org.springframework.integration.support.MessageBuilder;
import org.springframework.integration.test.util.TestUtils;
import org.springframework.messaging.Message;
import org.springframework.messaging.support.ErrorMessage;
import org.springframework.test.web.reactive.server.HttpHandlerConnector;
import org.springframework.web.reactive.function.client.WebClient;
import reactor.core.publisher.Mono;
import reactor.test.StepVerifier;
/**
* @author Shiliang Li
* @author Artem Bilan
*
* @since 5.0
*/
public class ReactiveHttpRequestExecutingMessageHandlerTests {
@Test
public void testReactiveReturn() throws Throwable {
ClientHttpConnector httpConnector = new HttpHandlerConnector((request, response) -> {
response.setStatusCode(HttpStatus.OK);
return Mono.empty()
.then(Mono.defer(response::setComplete));
});
WebClient webClient = WebClient.builder()
.clientConnector(httpConnector)
.build();
String destinationUri = "http://www.springsource.org/spring-integration";
ReactiveHttpRequestExecutingMessageHandler reactiveHandler =
new ReactiveHttpRequestExecutingMessageHandler(destinationUri, webClient);
FluxMessageChannel ackChannel = new FluxMessageChannel();
reactiveHandler.setOutputChannel(ackChannel);
reactiveHandler.handleMessage(MessageBuilder.withPayload("hello, world").build());
reactiveHandler.handleMessage(MessageBuilder.withPayload("hello, world").build());
StepVerifier.create(ackChannel, 2)
.assertNext(m -> assertThat(m, hasHeader(HttpHeaders.STATUS_CODE, HttpStatus.OK)))
.assertNext(m -> assertThat(m, hasHeader(HttpHeaders.STATUS_CODE, HttpStatus.OK)))
.then(() ->
((Subscriber<?>) TestUtils.getPropertyValue(ackChannel, "subscribers", List.class).get(0))
.onComplete())
.verifyComplete();
}
@Test
public void testReactiveErrorOneWay() throws Throwable {
ClientHttpConnector httpConnector = new HttpHandlerConnector((request, response) -> {
response.setStatusCode(HttpStatus.UNAUTHORIZED);
return Mono.empty()
.then(Mono.defer(response::setComplete));
});
WebClient webClient = WebClient.builder()
.clientConnector(httpConnector)
.build();
String destinationUri = "http://www.springsource.org/spring-integration";
ReactiveHttpRequestExecutingMessageHandler reactiveHandler =
new ReactiveHttpRequestExecutingMessageHandler(destinationUri, webClient);
reactiveHandler.setExpectReply(false);
QueueChannel errorChannel = new QueueChannel();
reactiveHandler.handleMessage(MessageBuilder.withPayload("hello, world")
.setErrorChannel(errorChannel)
.build());
Message<?> errorMessage = errorChannel.receive(10000);
assertNotNull(errorMessage);
assertThat(errorMessage, instanceOf(ErrorMessage.class));
Throwable throwable = (Throwable) errorMessage.getPayload();
assertThat(throwable.getMessage(), containsString("401 Unauthorized"));
}
@Test
public void testReactiveConnectErrorOneWay() throws Throwable {
ClientHttpConnector httpConnector = new HttpHandlerConnector((request, response) -> {
throw new RuntimeException("Intentional connection error");
});
WebClient webClient = WebClient.builder()
.clientConnector(httpConnector)
.build();
String destinationUri = "http://www.springsource.org/spring-integration";
ReactiveHttpRequestExecutingMessageHandler reactiveHandler =
new ReactiveHttpRequestExecutingMessageHandler(destinationUri, webClient);
reactiveHandler.setExpectReply(false);
QueueChannel errorChannel = new QueueChannel();
reactiveHandler.handleMessage(MessageBuilder.withPayload("hello, world")
.setErrorChannel(errorChannel)
.build());
Message<?> errorMessage = errorChannel.receive(10000);
assertNotNull(errorMessage);
assertThat(errorMessage, instanceOf(ErrorMessage.class));
Throwable throwable = (Throwable) errorMessage.getPayload();
assertThat(throwable.getMessage(), containsString("Intentional connection error"));
}
}