diff --git a/spring-cloud-gateway-core/pom.xml b/spring-cloud-gateway-core/pom.xml index 3687e260..956738f4 100644 --- a/spring-cloud-gateway-core/pom.xml +++ b/spring-cloud-gateway-core/pom.xml @@ -88,6 +88,11 @@ spring-boot-starter-test test + + org.springframework.boot + spring-boot-starter-validation + test + org.springframework.cloud spring-cloud-test-support diff --git a/spring-cloud-gateway-core/src/main/java/org/springframework/cloud/gateway/config/GatewayAutoConfiguration.java b/spring-cloud-gateway-core/src/main/java/org/springframework/cloud/gateway/config/GatewayAutoConfiguration.java index 99c1b534..2e8b9af0 100644 --- a/spring-cloud-gateway-core/src/main/java/org/springframework/cloud/gateway/config/GatewayAutoConfiguration.java +++ b/spring-cloud-gateway-core/src/main/java/org/springframework/cloud/gateway/config/GatewayAutoConfiguration.java @@ -1,5 +1,5 @@ /* - * Copyright 2013-2017 the original author or authors. + * Copyright 2013-2018 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. @@ -140,8 +140,8 @@ public class GatewayAutoConfiguration { } @Bean - public NettyWriteResponseFilter nettyWriteResponseFilter() { - return new NettyWriteResponseFilter(); + public NettyWriteResponseFilter nettyWriteResponseFilter(GatewayProperties properties) { + return new NettyWriteResponseFilter(properties.getStreamingMediaTypes()); } @Bean diff --git a/spring-cloud-gateway-core/src/main/java/org/springframework/cloud/gateway/config/GatewayProperties.java b/spring-cloud-gateway-core/src/main/java/org/springframework/cloud/gateway/config/GatewayProperties.java index 73a9f163..bd45d167 100644 --- a/spring-cloud-gateway-core/src/main/java/org/springframework/cloud/gateway/config/GatewayProperties.java +++ b/spring-cloud-gateway-core/src/main/java/org/springframework/cloud/gateway/config/GatewayProperties.java @@ -1,5 +1,5 @@ /* - * Copyright 2013-2017 the original author or authors. + * Copyright 2013-2018 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,6 +18,7 @@ package org.springframework.cloud.gateway.config; import java.util.ArrayList; +import java.util.Arrays; import java.util.List; import javax.validation.Valid; @@ -27,6 +28,7 @@ import org.springframework.boot.context.properties.ConfigurationProperties; import org.springframework.cloud.gateway.filter.FilterDefinition; import org.springframework.cloud.gateway.filter.factory.RemoveNonProxyHeadersGatewayFilterFactory; import org.springframework.cloud.gateway.route.RouteDefinition; +import org.springframework.http.MediaType; import org.springframework.validation.annotation.Validated; import static org.springframework.cloud.gateway.support.NameUtils.normalizeFilterName; @@ -50,6 +52,9 @@ public class GatewayProperties { */ private List defaultFilters = loadDefaults(); + private List streamingMediaTypes = Arrays.asList(MediaType.TEXT_EVENT_STREAM, + MediaType.APPLICATION_STREAM_JSON); + private ArrayList loadDefaults() { ArrayList defaults = new ArrayList<>(); FilterDefinition definition = new FilterDefinition(); @@ -73,4 +78,21 @@ public class GatewayProperties { public void setDefaultFilters(List defaultFilters) { this.defaultFilters = defaultFilters; } + + public List getStreamingMediaTypes() { + return streamingMediaTypes; + } + + public void setStreamingMediaTypes(List streamingMediaTypes) { + this.streamingMediaTypes = streamingMediaTypes; + } + + @Override + public String toString() { + return "GatewayProperties{" + + "routes=" + routes + + ", defaultFilters=" + defaultFilters + + ", streamingMediaTypes=" + streamingMediaTypes + + '}'; + } } diff --git a/spring-cloud-gateway-core/src/main/java/org/springframework/cloud/gateway/filter/NettyWriteResponseFilter.java b/spring-cloud-gateway-core/src/main/java/org/springframework/cloud/gateway/filter/NettyWriteResponseFilter.java index 7ec6c9df..237c4af5 100644 --- a/spring-cloud-gateway-core/src/main/java/org/springframework/cloud/gateway/filter/NettyWriteResponseFilter.java +++ b/spring-cloud-gateway-core/src/main/java/org/springframework/cloud/gateway/filter/NettyWriteResponseFilter.java @@ -1,5 +1,5 @@ /* - * Copyright 2013-2017 the original author or authors. + * Copyright 2013-2018 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. @@ -17,20 +17,24 @@ package org.springframework.cloud.gateway.filter; +import java.util.List; + import org.apache.commons.logging.Log; import org.apache.commons.logging.LogFactory; -import org.springframework.core.Ordered; -import org.springframework.core.io.buffer.NettyDataBuffer; -import org.springframework.core.io.buffer.NettyDataBufferFactory; -import org.springframework.http.server.reactive.ServerHttpResponse; -import org.springframework.web.server.ServerWebExchange; - -import static org.springframework.cloud.gateway.support.ServerWebExchangeUtils.CLIENT_RESPONSE_ATTR; - import reactor.core.publisher.Flux; import reactor.core.publisher.Mono; import reactor.ipc.netty.http.client.HttpClientResponse; +import org.springframework.core.Ordered; +import org.springframework.core.io.buffer.NettyDataBuffer; +import org.springframework.core.io.buffer.NettyDataBufferFactory; +import org.springframework.http.MediaType; +import org.springframework.http.server.reactive.ServerHttpResponse; +import org.springframework.lang.Nullable; +import org.springframework.web.server.ServerWebExchange; + +import static org.springframework.cloud.gateway.support.ServerWebExchangeUtils.CLIENT_RESPONSE_ATTR; + /** * @author Spencer Gibb */ @@ -40,6 +44,12 @@ public class NettyWriteResponseFilter implements GlobalFilter, Ordered { public static final int WRITE_RESPONSE_FILTER_ORDER = -1; + private final List streamingMediaTypes; + + public NettyWriteResponseFilter(List streamingMediaTypes) { + this.streamingMediaTypes = streamingMediaTypes; + } + @Override public int getOrder() { return WRITE_RESPONSE_FILTER_ORDER; @@ -65,8 +75,17 @@ public class NettyWriteResponseFilter implements GlobalFilter, Ordered { .retain() //TODO: needed? .map(factory::wrap); - return response.writeWith(body); + MediaType contentType = response.getHeaders().getContentType(); + return (isStreamingMediaType(contentType) ? + response.writeAndFlushWith(body.map(Flux::just)) : response.writeWith(body)); })); } + //TODO: use framework if possible + //TODO: port to WebClientWriteResponseFilter + private boolean isStreamingMediaType(@Nullable MediaType contentType) { + return (contentType != null && this.streamingMediaTypes.stream() + .anyMatch(contentType::isCompatibleWith)); + } + } diff --git a/spring-cloud-gateway-core/src/test/java/org/springframework/cloud/gateway/test/sse/SseIntegrationTests.java b/spring-cloud-gateway-core/src/test/java/org/springframework/cloud/gateway/test/sse/SseIntegrationTests.java new file mode 100644 index 00000000..75d7b2af --- /dev/null +++ b/spring-cloud-gateway-core/src/test/java/org/springframework/cloud/gateway/test/sse/SseIntegrationTests.java @@ -0,0 +1,318 @@ +/* + * Copyright 2013-2018 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.cloud.gateway.test.sse; + +import java.time.Duration; + +import org.apache.commons.logging.Log; +import org.apache.commons.logging.LogFactory; +import org.junit.After; +import org.junit.Before; +import org.junit.Test; +import reactor.core.publisher.Flux; +import reactor.test.StepVerifier; + +import org.springframework.beans.factory.annotation.Value; +import org.springframework.boot.autoconfigure.EnableAutoConfiguration; +import org.springframework.boot.builder.SpringApplicationBuilder; +import org.springframework.cloud.gateway.route.RouteLocator; +import org.springframework.cloud.gateway.route.builder.RouteLocatorBuilder; +import org.springframework.cloud.gateway.test.PermitAllSecurityConfiguration; +import org.springframework.cloud.gateway.test.support.HttpServer; +import org.springframework.cloud.gateway.test.support.ReactorHttpServer; +import org.springframework.context.ConfigurableApplicationContext; +import org.springframework.context.annotation.AnnotationConfigApplicationContext; +import org.springframework.context.annotation.Bean; +import org.springframework.context.annotation.Configuration; +import org.springframework.context.annotation.Import; +import org.springframework.core.ParameterizedTypeReference; +import org.springframework.core.ResolvableType; +import org.springframework.core.env.ConfigurableEnvironment; +import org.springframework.http.codec.ServerSentEvent; +import org.springframework.http.server.reactive.HttpHandler; +import org.springframework.web.bind.annotation.RequestMapping; +import org.springframework.web.bind.annotation.RestController; +import org.springframework.web.reactive.DispatcherHandler; +import org.springframework.web.reactive.config.EnableWebFlux; +import org.springframework.web.reactive.function.client.WebClient; +import org.springframework.web.server.adapter.WebHttpHandlerBuilder; + +import static org.junit.Assert.assertEquals; +import static org.junit.Assert.assertNull; +import static org.springframework.core.ResolvableType.forClassWithGenerics; +import static org.springframework.http.MediaType.TEXT_EVENT_STREAM; +import static org.springframework.web.reactive.function.BodyExtractors.toFlux; + +/** + * @author Sebastien Deleuze + */ +public class SseIntegrationTests { + + protected Log logger = LogFactory.getLog(getClass()); + + protected int serverPort; + + public HttpServer server; + + private AnnotationConfigApplicationContext wac; + + private WebClient webClient; + + private ConfigurableApplicationContext gatewayContext; + + private int gatewayPort; + + @Before + public void setup() throws Exception { + this.server = new ReactorHttpServer(); + this.server.setHandler(createHttpHandler()); + this.server.afterPropertiesSet(); + this.server.start(); + + // Set dynamically chosen port + this.serverPort = this.server.getPort(); + logger.info("SSE Port: "+this.serverPort); + + + this.gatewayContext = new SpringApplicationBuilder(GatewayConfig.class) + .properties("sse.server.port:"+this.serverPort, "server.port=0", "spring.jmx.enabled=false") + .run(); + + ConfigurableEnvironment env = this.gatewayContext.getBean(ConfigurableEnvironment.class); + this.gatewayPort = new Integer(env.getProperty("local.server.port")); + + this.webClient = WebClient.create("http://localhost:" + this.gatewayPort + "/sse"); + + logger.info("Gateway Port: "+this.gatewayPort); + } + + @After + public void tearDown() throws Exception { + this.server.stop(); + this.serverPort = 0; + this.gatewayPort = 0; + this.gatewayContext.close(); + this.wac.close(); + } + + private HttpHandler createHttpHandler() { + this.wac = new AnnotationConfigApplicationContext(); + this.wac.register(TestConfiguration.class); + this.wac.refresh(); + + return WebHttpHandlerBuilder.webHandler(new DispatcherHandler(this.wac)).build(); + } + + @Test + public void sseAsString() { + Flux result = this.webClient.get() + .uri("/string") + .accept(TEXT_EVENT_STREAM) + .exchange() + .flatMapMany(response -> response.bodyToFlux(String.class)); + + StepVerifier.create(result) + .expectNext("foo 0") + .expectNext("foo 1") + .thenCancel() + .verify(Duration.ofSeconds(5L)); + } + + @Test + public void sseAsPerson() { + Flux result = this.webClient.get() + .uri("/person") + .accept(TEXT_EVENT_STREAM) + .exchange() + .flatMapMany(response -> response.bodyToFlux(Person.class)); + + StepVerifier.create(result) + .expectNext(new Person("foo 0")) + .expectNext(new Person("foo 1")) + .thenCancel() + .verify(Duration.ofSeconds(5L)); + } + + @Test + @SuppressWarnings("Duplicates") + public void sseAsEvent() { + ResolvableType type = forClassWithGenerics(ServerSentEvent.class, String.class); + Flux> result = this.webClient.get() + .uri("/event") + .accept(TEXT_EVENT_STREAM) + .exchange() + .flatMapMany(response -> response.body( + toFlux(new ParameterizedTypeReference>() {}))); + + StepVerifier.create(result) + .consumeNextWith( event -> { + assertEquals("0", event.id()); + assertEquals("foo", event.data()); + assertEquals("bar", event.comment()); + assertNull(event.event()); + assertNull(event.retry()); + }) + .consumeNextWith( event -> { + assertEquals("1", event.id()); + assertEquals("foo", event.data()); + assertEquals("bar", event.comment()); + assertNull(event.event()); + assertNull(event.retry()); + }) + .thenCancel() + .verify(Duration.ofSeconds(5L)); + } + + @Test + @SuppressWarnings("Duplicates") + public void sseAsEventWithoutAcceptHeader() { + Flux> result = this.webClient.get() + .uri("/event") + .accept(TEXT_EVENT_STREAM) + .exchange() + .flatMapMany(response -> response.body( + toFlux(new ParameterizedTypeReference>() {}))); + + StepVerifier.create(result) + .consumeNextWith( event -> { + assertEquals("0", event.id()); + assertEquals("foo", event.data()); + assertEquals("bar", event.comment()); + assertNull(event.event()); + assertNull(event.retry()); + }) + .consumeNextWith( event -> { + assertEquals("1", event.id()); + assertEquals("foo", event.data()); + assertEquals("bar", event.comment()); + assertNull(event.event()); + assertNull(event.retry()); + }) + .thenCancel() + .verify(Duration.ofSeconds(5L)); + } + + /** + * Return an interval stream of with n number of ticks and buffer the + * emissions to avoid back pressure failures (e.g. on slow CI server). + */ + public static Flux interval(Duration period, int count) { + return Flux.interval(period).take(count).onBackpressureBuffer(2); + } + + @RestController + @SuppressWarnings("unused") + static class SseController { + + private static final Flux INTERVAL = interval(Duration.ofMillis(100), 50); + + + @RequestMapping("/sse/string") + Flux string() { + return INTERVAL.map(l -> "foo " + l); + } + + @RequestMapping("/sse/person") + Flux person() { + return INTERVAL.map(l -> new Person("foo " + l)); + } + + @RequestMapping("/sse/event") + Flux> sse() { + return INTERVAL.map(l -> ServerSentEvent.builder("foo") + .id(Long.toString(l)) + .comment("bar") + .build()); + } + + } + + + @Configuration + @EnableWebFlux + @SuppressWarnings("unused") + static class TestConfiguration { + + @Bean + public SseController sseController() { + return new SseController(); + } + } + + @Configuration + @EnableAutoConfiguration + @Import(PermitAllSecurityConfiguration.class) + protected static class GatewayConfig { + + @Value("${sse.server.port}") + private int port; + + @Bean + public RouteLocator sseRouteLocator(RouteLocatorBuilder builder) { + return builder.routes() + .route("sse_route", r -> r.alwaysTrue() + .uri("http://localhost:"+this.port)) + .build(); + } + } + + + @SuppressWarnings("unused") + private static class Person { + + private String name; + + public Person() { + } + + public Person(String name) { + this.name = name; + } + + public String getName() { + return name; + } + + public void setName(String name) { + this.name = 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 !(this.name != null ? !this.name.equals(person.name) : person.name != null); + } + + @Override + public int hashCode() { + return this.name != null ? this.name.hashCode() : 0; + } + + @Override + public String toString() { + return "Person{name='" + this.name + '\'' + '}'; + } + } + +} \ No newline at end of file diff --git a/spring-cloud-gateway-core/src/test/java/org/springframework/cloud/gateway/test/websocket/AbstractHttpServer.java b/spring-cloud-gateway-core/src/test/java/org/springframework/cloud/gateway/test/support/AbstractHttpServer.java similarity index 96% rename from spring-cloud-gateway-core/src/test/java/org/springframework/cloud/gateway/test/websocket/AbstractHttpServer.java rename to spring-cloud-gateway-core/src/test/java/org/springframework/cloud/gateway/test/support/AbstractHttpServer.java index 4bd8c034..c5ab6191 100644 --- a/spring-cloud-gateway-core/src/test/java/org/springframework/cloud/gateway/test/websocket/AbstractHttpServer.java +++ b/spring-cloud-gateway-core/src/test/java/org/springframework/cloud/gateway/test/support/AbstractHttpServer.java @@ -1,5 +1,5 @@ /* - * Copyright 2002-2017 the original author or authors. + * Copyright 2013-2018 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. @@ -12,9 +12,10 @@ * 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.cloud.gateway.test.websocket; +package org.springframework.cloud.gateway.test.support; import java.util.LinkedHashMap; import java.util.Map; diff --git a/spring-cloud-gateway-core/src/test/java/org/springframework/cloud/gateway/test/websocket/HttpServer.java b/spring-cloud-gateway-core/src/test/java/org/springframework/cloud/gateway/test/support/HttpServer.java similarity index 89% rename from spring-cloud-gateway-core/src/test/java/org/springframework/cloud/gateway/test/websocket/HttpServer.java rename to spring-cloud-gateway-core/src/test/java/org/springframework/cloud/gateway/test/support/HttpServer.java index 6362813b..ba4a2ee4 100644 --- a/spring-cloud-gateway-core/src/test/java/org/springframework/cloud/gateway/test/websocket/HttpServer.java +++ b/spring-cloud-gateway-core/src/test/java/org/springframework/cloud/gateway/test/support/HttpServer.java @@ -1,5 +1,5 @@ /* - * Copyright 2002-2017 the original author or authors. + * Copyright 2013-2018 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. @@ -12,9 +12,10 @@ * 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.cloud.gateway.test.websocket; +package org.springframework.cloud.gateway.test.support; import org.springframework.beans.factory.InitializingBean; import org.springframework.context.Lifecycle; diff --git a/spring-cloud-gateway-core/src/test/java/org/springframework/cloud/gateway/test/websocket/ReactorHttpServer.java b/spring-cloud-gateway-core/src/test/java/org/springframework/cloud/gateway/test/support/ReactorHttpServer.java similarity index 94% rename from spring-cloud-gateway-core/src/test/java/org/springframework/cloud/gateway/test/websocket/ReactorHttpServer.java rename to spring-cloud-gateway-core/src/test/java/org/springframework/cloud/gateway/test/support/ReactorHttpServer.java index 8b386057..b45b6d19 100644 --- a/spring-cloud-gateway-core/src/test/java/org/springframework/cloud/gateway/test/websocket/ReactorHttpServer.java +++ b/spring-cloud-gateway-core/src/test/java/org/springframework/cloud/gateway/test/support/ReactorHttpServer.java @@ -1,5 +1,5 @@ /* - * Copyright 2002-2017 the original author or authors. + * Copyright 2013-2018 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. @@ -12,9 +12,10 @@ * 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.cloud.gateway.test.websocket; +package org.springframework.cloud.gateway.test.support; import java.util.concurrent.atomic.AtomicReference; diff --git a/spring-cloud-gateway-core/src/test/java/org/springframework/cloud/gateway/test/websocket/WebSocketIntegrationTests.java b/spring-cloud-gateway-core/src/test/java/org/springframework/cloud/gateway/test/websocket/WebSocketIntegrationTests.java index d4c90608..9d2104ea 100644 --- a/spring-cloud-gateway-core/src/test/java/org/springframework/cloud/gateway/test/websocket/WebSocketIntegrationTests.java +++ b/spring-cloud-gateway-core/src/test/java/org/springframework/cloud/gateway/test/websocket/WebSocketIntegrationTests.java @@ -1,5 +1,5 @@ /* - * Copyright 2002-2017 the original author or authors. + * Copyright 2013-2018 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. @@ -12,6 +12,7 @@ * 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.cloud.gateway.test.websocket; @@ -38,6 +39,8 @@ import org.springframework.boot.builder.SpringApplicationBuilder; import org.springframework.cloud.gateway.route.RouteLocator; import org.springframework.cloud.gateway.route.builder.RouteLocatorBuilder; import org.springframework.cloud.gateway.test.PermitAllSecurityConfiguration; +import org.springframework.cloud.gateway.test.support.HttpServer; +import org.springframework.cloud.gateway.test.support.ReactorHttpServer; import org.springframework.cloud.netflix.ribbon.RibbonClient; import org.springframework.cloud.netflix.ribbon.StaticServerList; import org.springframework.context.ConfigurableApplicationContext; @@ -115,7 +118,6 @@ public class WebSocketIntegrationTests { .properties("ws.server.port:"+this.serverPort, "server.port=0", "spring.jmx.enabled=false") .run(); - GatewayConfig config = this.gatewayContext.getBean(GatewayConfig.class); ConfigurableEnvironment env = this.gatewayContext.getBean(ConfigurableEnvironment.class); this.gatewayPort = new Integer(env.getProperty("local.server.port")); }