diff --git a/spring-cloud-gateway-dependencies/pom.xml b/spring-cloud-gateway-dependencies/pom.xml index 7e0d12ed..1bd76f42 100644 --- a/spring-cloud-gateway-dependencies/pom.xml +++ b/spring-cloud-gateway-dependencies/pom.xml @@ -48,21 +48,11 @@ spring-cloud-starter-gateway ${project.version} - - io.rsocket - rsocket-core - ${rsocket.version} - io.rsocket rsocket-micrometer ${rsocket.version} - - io.rsocket - rsocket-transport-netty - ${rsocket.version} - diff --git a/spring-cloud-gateway-rsocket/pom.xml b/spring-cloud-gateway-rsocket/pom.xml index b4acbf68..0be259da 100644 --- a/spring-cloud-gateway-rsocket/pom.xml +++ b/spring-cloud-gateway-rsocket/pom.xml @@ -49,8 +49,7 @@ org.springframework.boot - spring-boot-starter-webflux - + spring-boot-starter-rsocket org.springframework.boot diff --git a/spring-cloud-gateway-rsocket/src/main/java/org/springframework/cloud/gateway/rsocket/autoconfigure/GatewayRSocketAutoConfiguration.java b/spring-cloud-gateway-rsocket/src/main/java/org/springframework/cloud/gateway/rsocket/autoconfigure/GatewayRSocketAutoConfiguration.java index 25f23d6b..21f856da 100644 --- a/spring-cloud-gateway-rsocket/src/main/java/org/springframework/cloud/gateway/rsocket/autoconfigure/GatewayRSocketAutoConfiguration.java +++ b/spring-cloud-gateway-rsocket/src/main/java/org/springframework/cloud/gateway/rsocket/autoconfigure/GatewayRSocketAutoConfiguration.java @@ -29,7 +29,7 @@ import org.springframework.cloud.gateway.rsocket.registry.RegistryRoutes; import org.springframework.cloud.gateway.rsocket.registry.RegistrySocketAcceptorFilter; import org.springframework.cloud.gateway.rsocket.route.Routes; import org.springframework.cloud.gateway.rsocket.server.GatewayRSocket; -import org.springframework.cloud.gateway.rsocket.server.GatewayRSocketServer; +import org.springframework.cloud.gateway.rsocket.server.GatewayServerRSocketFactoryCustomizer; import org.springframework.cloud.gateway.rsocket.socketacceptor.GatewaySocketAcceptor; import org.springframework.cloud.gateway.rsocket.socketacceptor.SocketAcceptorFilter; import org.springframework.cloud.gateway.rsocket.socketacceptor.SocketAcceptorPredicate; @@ -96,9 +96,9 @@ public class GatewayRSocketAutoConfiguration { } @Bean - public GatewayRSocketServer gatewayApp(GatewaySocketAcceptor socketAcceptor, + public GatewayServerRSocketFactoryCustomizer gatewayServerRSocketFactoryCustomizer( GatewayRSocketProperties properties, MeterRegistry meterRegistry) { - return new GatewayRSocketServer(properties, socketAcceptor, meterRegistry); + return new GatewayServerRSocketFactoryCustomizer(properties, meterRegistry); } } diff --git a/spring-cloud-gateway-rsocket/src/main/java/org/springframework/cloud/gateway/rsocket/autoconfigure/GatewayRSocketProperties.java b/spring-cloud-gateway-rsocket/src/main/java/org/springframework/cloud/gateway/rsocket/autoconfigure/GatewayRSocketProperties.java index 0630cd4e..a086f6a6 100644 --- a/spring-cloud-gateway-rsocket/src/main/java/org/springframework/cloud/gateway/rsocket/autoconfigure/GatewayRSocketProperties.java +++ b/spring-cloud-gateway-rsocket/src/main/java/org/springframework/cloud/gateway/rsocket/autoconfigure/GatewayRSocketProperties.java @@ -32,7 +32,15 @@ public class GatewayRSocketProperties { private String id = "gateway"; // TODO: + UUID? - private final Server server = new Server(); + /** + * Tag names and values to be supplied to Micrometer Interceptor. + */ + private List micrometerTags = new ArrayList<>(); + + public GatewayRSocketProperties() { + micrometerTags.add("component"); + micrometerTags.add("gateway"); + } public boolean isEnabled() { return enabled; @@ -50,81 +58,23 @@ public class GatewayRSocketProperties { this.id = id; } - public Server getServer() { - return server; + public List getMicrometerTags() { + return micrometerTags; + } + + public void setMicrometerTags(List micrometerTags) { + this.micrometerTags = micrometerTags; } @Override public String toString() { - return new ToStringCreator(this).append("enabled", enabled).append("id", id) - .append("server", server).toString(); - } - - /** - * Server properties. - */ - public static class Server { - - // TODO: other transports - public enum TransportType { - - /** - * TCP Transport type. - */ - TCP - - } - - /** - * Tag names and values to be supplied to Micrometer Interceptor. - */ - private List micrometerTags = new ArrayList<>(); - - /** - * Server port. - */ - private int port = 7002; // TODO: different default port? - - public Server() { - micrometerTags.add("component"); - micrometerTags.add("gateway"); - } - - /** - * Server transport type. Defaults to TCP. - */ - private TransportType transport = TransportType.TCP; - - public List getMicrometerTags() { - return micrometerTags; - } - - public void setMicrometerTags(List micrometerTags) { - this.micrometerTags = micrometerTags; - } - - public int getPort() { - return port; - } - - public void setPort(int port) { - this.port = port; - } - - public TransportType getTransport() { - return transport; - } - - public void setTransport(TransportType transport) { - this.transport = transport; - } - - @Override - public String toString() { - return new ToStringCreator(this).append("micrometerTags", micrometerTags) - .append("port", port).append("transport", transport).toString(); - } - + // @formatter:off + return new ToStringCreator(this) + .append("enabled", enabled) + .append("id", id) + .append("micrometerTags", micrometerTags) + .toString(); + // @formatter:on } } diff --git a/spring-cloud-gateway-rsocket/src/main/java/org/springframework/cloud/gateway/rsocket/server/GatewayRSocketServer.java b/spring-cloud-gateway-rsocket/src/main/java/org/springframework/cloud/gateway/rsocket/server/GatewayRSocketServer.java deleted file mode 100644 index 38129fdc..00000000 --- a/spring-cloud-gateway-rsocket/src/main/java/org/springframework/cloud/gateway/rsocket/server/GatewayRSocketServer.java +++ /dev/null @@ -1,153 +0,0 @@ -/* - * Copyright 2018-2019 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 - * - * https://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.rsocket.server; - -import java.util.Arrays; -import java.util.List; -import java.util.concurrent.atomic.AtomicBoolean; -import java.util.stream.Collectors; - -import io.micrometer.core.instrument.MeterRegistry; -import io.micrometer.core.instrument.Tag; -import io.micrometer.core.instrument.Tags; -import io.rsocket.RSocketFactory; -import io.rsocket.SocketAcceptor; -import io.rsocket.micrometer.MicrometerDuplexConnectionInterceptor; -import io.rsocket.plugins.RSocketInterceptor; -import io.rsocket.transport.netty.server.CloseableChannel; -import io.rsocket.transport.netty.server.TcpServerTransport; -import org.apache.commons.logging.Log; -import org.apache.commons.logging.LogFactory; - -import org.springframework.cloud.gateway.rsocket.autoconfigure.GatewayRSocketProperties; -import org.springframework.cloud.gateway.rsocket.autoconfigure.GatewayRSocketProperties.Server.TransportType; -import org.springframework.context.SmartLifecycle; -import org.springframework.core.Ordered; -import org.springframework.util.Assert; - -public class GatewayRSocketServer implements Ordered, SmartLifecycle { - - private static final Log log = LogFactory.getLog(GatewayRSocketServer.class); - - private static final RSocketInterceptor[] EMPTY_INTERCEPTORS = new RSocketInterceptor[0]; - - private final GatewayRSocketProperties properties; - - private final SocketAcceptor socketAcceptor; - - private final List serverInterceptors; - - private final AtomicBoolean running = new AtomicBoolean(); - - private CloseableChannel closeableChannel; - - private final MeterRegistry meterRegistry; - - public GatewayRSocketServer(GatewayRSocketProperties properties, - SocketAcceptor socketAcceptor, MeterRegistry meterRegistry) { - this(properties, socketAcceptor, meterRegistry, EMPTY_INTERCEPTORS); - } - - public GatewayRSocketServer(GatewayRSocketProperties properties, - SocketAcceptor socketAcceptor, MeterRegistry meterRegistry, - RSocketInterceptor... interceptors) { - Assert.notNull(properties, "properties may not be null"); - Assert.notNull(socketAcceptor, "socketAcceptor may not be null"); - Assert.notNull(meterRegistry, "meterRegistry may not be null"); - Assert.notNull(interceptors, "interceptors may not be null"); - this.properties = properties; - this.socketAcceptor = socketAcceptor; - this.meterRegistry = meterRegistry; - this.serverInterceptors = Arrays.asList(interceptors); - } - - @Override - public int getOrder() { - // return 0; - return HIGHEST_PRECEDENCE; - } - - @Override - public void start() { - if (running.compareAndSet(false, true)) { - startServer(); - } - } - - @Override - public void stop() { - if (running.compareAndSet(true, false)) { - if (log.isInfoEnabled()) { - log.info("Stopping Gateway RSocket Server"); - } - if (closeableChannel != null) { - closeableChannel.dispose(); - } - } - } - - @Override - public boolean isRunning() { - return running.get(); - } - - protected void startServer() { - GatewayRSocketProperties.Server server = properties.getServer(); - int port = server.getPort(); - - TransportType transportType = server.getTransport(); - TcpServerTransport transport; - switch (transportType) { - case TCP: - transport = TcpServerTransport.create(port); - break; - default: - throw new IllegalArgumentException( - "No support for server transport " + transportType); - } - - if (log.isInfoEnabled()) { - log.info("Starting Gateway RSocket Server on port: " + port + ", transport: " - + transportType); - } - - RSocketFactory.ServerRSocketFactory factory = RSocketFactory.receive(); - - serverInterceptors.forEach(factory::addServerPlugin); - - List micrometerTags = server.getMicrometerTags(); - Tag[] tags = Tags.of(micrometerTags.toArray(new String[] {})) - .and("gateway.id", properties.getId()).stream() - .collect(Collectors.toList()).toArray(new Tag[] {}); - - factory - // TODO: add as bean like serverInterceptors above - .addConnectionPlugin( - new MicrometerDuplexConnectionInterceptor(meterRegistry, tags)) - .errorConsumer(throwable -> { - if (log.isDebugEnabled()) { - log.debug("Error with connection", throwable); - } - }) // TODO: add configurable errorConsumer - .acceptor(this.socketAcceptor).transport(transport).start() - .map(closeableChannel -> { - this.closeableChannel = closeableChannel; - return closeableChannel; - }).subscribe(); - } - -} diff --git a/spring-cloud-gateway-rsocket/src/main/java/org/springframework/cloud/gateway/rsocket/server/GatewayServerRSocketFactoryCustomizer.java b/spring-cloud-gateway-rsocket/src/main/java/org/springframework/cloud/gateway/rsocket/server/GatewayServerRSocketFactoryCustomizer.java new file mode 100644 index 00000000..c627fb59 --- /dev/null +++ b/spring-cloud-gateway-rsocket/src/main/java/org/springframework/cloud/gateway/rsocket/server/GatewayServerRSocketFactoryCustomizer.java @@ -0,0 +1,85 @@ +/* + * Copyright 2018-2019 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 + * + * https://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.rsocket.server; + +import java.util.Arrays; +import java.util.List; +import java.util.stream.Collectors; + +import io.micrometer.core.instrument.MeterRegistry; +import io.micrometer.core.instrument.Tag; +import io.micrometer.core.instrument.Tags; +import io.rsocket.RSocketFactory.ServerRSocketFactory; +import io.rsocket.micrometer.MicrometerDuplexConnectionInterceptor; +import io.rsocket.plugins.RSocketInterceptor; +import org.apache.commons.logging.Log; +import org.apache.commons.logging.LogFactory; + +import org.springframework.boot.rsocket.server.ServerRSocketFactoryCustomizer; +import org.springframework.cloud.gateway.rsocket.autoconfigure.GatewayRSocketProperties; +import org.springframework.util.Assert; + +public class GatewayServerRSocketFactoryCustomizer + implements ServerRSocketFactoryCustomizer { + + private static final Log log = LogFactory + .getLog(GatewayServerRSocketFactoryCustomizer.class); + + private static final RSocketInterceptor[] EMPTY_INTERCEPTORS = new RSocketInterceptor[0]; + + private final GatewayRSocketProperties properties; + + private final List serverInterceptors; + + private final MeterRegistry meterRegistry; + + public GatewayServerRSocketFactoryCustomizer(GatewayRSocketProperties properties, + MeterRegistry meterRegistry) { + this(properties, meterRegistry, EMPTY_INTERCEPTORS); + } + + public GatewayServerRSocketFactoryCustomizer(GatewayRSocketProperties properties, + MeterRegistry meterRegistry, RSocketInterceptor... interceptors) { + Assert.notNull(properties, "properties may not be null"); + Assert.notNull(meterRegistry, "meterRegistry may not be null"); + Assert.notNull(interceptors, "interceptors may not be null"); + this.properties = properties; + this.meterRegistry = meterRegistry; + this.serverInterceptors = Arrays.asList(interceptors); + } + + @Override + public ServerRSocketFactory apply(ServerRSocketFactory factory) { + serverInterceptors.forEach(factory::addServerPlugin); + + List micrometerTags = properties.getMicrometerTags(); + Tag[] tags = Tags.of(micrometerTags.toArray(new String[] {})) + .and("gateway.id", properties.getId()).stream() + .collect(Collectors.toList()).toArray(new Tag[] {}); + + return factory + // TODO: add as bean like serverInterceptors above + .addConnectionPlugin( + new MicrometerDuplexConnectionInterceptor(meterRegistry, tags)) + .errorConsumer(throwable -> { + if (log.isDebugEnabled()) { + log.debug("Error with connection", throwable); + } + }); // TODO: add configurable errorConsumer + } + +} diff --git a/spring-cloud-gateway-rsocket/src/test/java/org/springframework/cloud/gateway/rsocket/autoconfigure/GatewayRSocketAutoConfigurationTests.java b/spring-cloud-gateway-rsocket/src/test/java/org/springframework/cloud/gateway/rsocket/autoconfigure/GatewayRSocketAutoConfigurationTests.java index 2e808d60..82dfeab2 100644 --- a/spring-cloud-gateway-rsocket/src/test/java/org/springframework/cloud/gateway/rsocket/autoconfigure/GatewayRSocketAutoConfigurationTests.java +++ b/spring-cloud-gateway-rsocket/src/test/java/org/springframework/cloud/gateway/rsocket/autoconfigure/GatewayRSocketAutoConfigurationTests.java @@ -25,7 +25,7 @@ import org.springframework.boot.test.context.runner.ReactiveWebApplicationContex import org.springframework.cloud.gateway.rsocket.registry.Registry; import org.springframework.cloud.gateway.rsocket.registry.RegistryRoutes; import org.springframework.cloud.gateway.rsocket.registry.RegistrySocketAcceptorFilter; -import org.springframework.cloud.gateway.rsocket.server.GatewayRSocketServer; +import org.springframework.cloud.gateway.rsocket.server.GatewayServerRSocketFactoryCustomizer; import org.springframework.cloud.gateway.rsocket.socketacceptor.GatewaySocketAcceptor; import org.springframework.cloud.gateway.rsocket.socketacceptor.SocketAcceptorPredicate; import org.springframework.cloud.gateway.rsocket.socketacceptor.SocketAcceptorPredicateFilter; @@ -44,7 +44,7 @@ public class GatewayRSocketAutoConfigurationTests { .run(context -> assertThat(context).hasSingleBean(Registry.class) .hasSingleBean(RegistryRoutes.class) .hasSingleBean(RegistrySocketAcceptorFilter.class) - .hasSingleBean(GatewayRSocketServer.class) + .hasSingleBean(GatewayServerRSocketFactoryCustomizer.class) .hasSingleBean(GatewayRSocketProperties.class) .hasSingleBean(GatewaySocketAcceptor.class) .hasSingleBean(SocketAcceptorPredicateFilter.class) diff --git a/spring-cloud-gateway-rsocket/src/test/java/org/springframework/cloud/gateway/rsocket/server/GatewayRSocketIntegrationTests.java b/spring-cloud-gateway-rsocket/src/test/java/org/springframework/cloud/gateway/rsocket/server/GatewayRSocketIntegrationTests.java index 5c3bac5b..4187071e 100644 --- a/spring-cloud-gateway-rsocket/src/test/java/org/springframework/cloud/gateway/rsocket/server/GatewayRSocketIntegrationTests.java +++ b/spring-cloud-gateway-rsocket/src/test/java/org/springframework/cloud/gateway/rsocket/server/GatewayRSocketIntegrationTests.java @@ -25,11 +25,13 @@ import org.junit.runner.RunWith; import reactor.test.StepVerifier; import org.springframework.beans.factory.annotation.Autowired; +import org.springframework.boot.autoconfigure.rsocket.RSocketProperties; +import org.springframework.boot.rsocket.netty.NettyRSocketBootstrap; import org.springframework.boot.test.context.SpringBootTest; import org.springframework.boot.test.context.SpringBootTest.WebEnvironment; -import org.springframework.cloud.gateway.rsocket.autoconfigure.GatewayRSocketProperties; import org.springframework.cloud.gateway.rsocket.test.PingPongApp; import org.springframework.test.context.junit4.SpringRunner; +import org.springframework.test.util.ReflectionTestUtils; import org.springframework.util.SocketUtils; import static org.assertj.core.api.Assertions.assertThat; @@ -48,24 +50,23 @@ public class GatewayRSocketIntegrationTests { private PingPongApp.Pong pong; @Autowired - private GatewayRSocketProperties properties; + private RSocketProperties properties; @Autowired private PingPongApp.MySocketAcceptorFilter mySocketAcceptorFilter; @Autowired - private GatewayRSocketServer server; + private NettyRSocketBootstrap server; @BeforeClass public static void init() { port = SocketUtils.findAvailableTcpPort(); - System.setProperty("spring.cloud.gateway.rsocket.server.port", - String.valueOf(port)); + System.setProperty("spring.rsocket.server.port", String.valueOf(port)); } @AfterClass public static void after() { - System.clearProperty("spring.cloud.gateway.rsocket.server.port"); + System.clearProperty("spring.rsocket.server.port"); } @Test @@ -82,9 +83,11 @@ public class GatewayRSocketIntegrationTests { assertThat(ping.getPongsReceived()).isGreaterThan(0); assertThat(pong.getPingsReceived()).isGreaterThan(0); - assertThat(properties.getServer().getPort()).isNotEqualTo(7002); + Object server = properties.getServer(); + Object port = ReflectionTestUtils.invokeGetterMethod(server, "port"); + assertThat(port).isNotEqualTo(7002); assertThat(mySocketAcceptorFilter.invoked()).isTrue(); - assertThat(server.isRunning()).isFalse(); + assertThat(this.server.isRunning()).isFalse(); } } diff --git a/spring-cloud-gateway-rsocket/src/test/java/org/springframework/cloud/gateway/rsocket/test/PingPongApp.java b/spring-cloud-gateway-rsocket/src/test/java/org/springframework/cloud/gateway/rsocket/test/PingPongApp.java index feb79fa6..e669b0fa 100644 --- a/spring-cloud-gateway-rsocket/src/test/java/org/springframework/cloud/gateway/rsocket/test/PingPongApp.java +++ b/spring-cloud-gateway-rsocket/src/test/java/org/springframework/cloud/gateway/rsocket/test/PingPongApp.java @@ -125,8 +125,8 @@ public class PingPongApp { log.info("Starting Ping" + id); ConfigurableEnvironment env = event.getApplicationContext().getEnvironment(); Integer take = env.getProperty("ping.take", Integer.class, null); - Integer gatewayPort = env.getProperty( - "spring.cloud.gateway.rsocket.server.port", Integer.class, 7002); + Integer gatewayPort = env.getProperty("spring.rsocket.server.port", + Integer.class, 7002); log.debug("ping.take: " + take); @@ -208,8 +208,8 @@ public class PingPongApp { e.printStackTrace(); } log.info("Starting Pong"); - Integer gatewayPort = env.getProperty( - "spring.cloud.gateway.rsocket.server.port", Integer.class, 7002); + Integer gatewayPort = env.getProperty("spring.rsocket.server.port", + Integer.class, 7002); MicrometerRSocketInterceptor interceptor = new MicrometerRSocketInterceptor( meterRegistry, Tag.of("component", "pong")); ByteBuf announcementMetadata = Metadata.from("pong").with("id", "pong1")