SpringCloud/spring-cloud-gateway
1
0
Fork 0
Code Issues Packages Projects Releases Wiki Activity

Move to use new boot RSocket starter and infrastructure.

Browse Source
This commit is contained in:
Spencer Gibb
2019-04-15 15:44:47 -04:00
parent 312bf61b3e
commit df1181ce27
9 changed files with 128 additions and 254 deletions
Download Patch File Download Diff File Expand all files Collapse all files

10
spring-cloud-gateway-dependencies/pom.xml
View File

@@ -48,21 +48,11 @@
<artifactId>spring-cloud-starter-gateway</artifactId>
<version>${project.version}</version>
</dependency>
<dependency>
<groupId>io.rsocket</groupId>
<artifactId>rsocket-core</artifactId>
<version>${rsocket.version}</version>
</dependency>
<dependency>
<groupId>io.rsocket</groupId>
<artifactId>rsocket-micrometer</artifactId>
<version>${rsocket.version}</version>
</dependency>
<dependency>
<groupId>io.rsocket</groupId>
<artifactId>rsocket-transport-netty</artifactId>
<version>${rsocket.version}</version>
</dependency>
</dependencies>
</dependencyManagement>

3
spring-cloud-gateway-rsocket/pom.xml
View File

@@ -49,8 +49,7 @@
<dependencies>
<dependency>
<groupId>org.springframework.boot</groupId>
<artifactId>spring-boot-starter-webflux</artifactId>
<!-- TODO: optional -->
<artifactId>spring-boot-starter-rsocket</artifactId>
</dependency>
<dependency>
<groupId>org.springframework.boot</groupId>

6
spring-cloud-gateway-rsocket/src/main/java/org/springframework/cloud/gateway/rsocket/autoconfigure/GatewayRSocketAutoConfiguration.java
View File

@@ -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);
}
}

94
spring-cloud-gateway-rsocket/src/main/java/org/springframework/cloud/gateway/rsocket/autoconfigure/GatewayRSocketProperties.java
View File

@@ -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<String> 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<String> getMicrometerTags() {
return micrometerTags;
}
public void setMicrometerTags(List<String> 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<String> 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<String> getMicrometerTags() {
return micrometerTags;
}
public void setMicrometerTags(List<String> 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
}
}

153
spring-cloud-gateway-rsocket/src/main/java/org/springframework/cloud/gateway/rsocket/server/GatewayRSocketServer.java
View File

@@ -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<RSocketInterceptor> 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<String> 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();
}
}

85
spring-cloud-gateway-rsocket/src/main/java/org/springframework/cloud/gateway/rsocket/server/GatewayServerRSocketFactoryCustomizer.java Normal file
View File

@@ -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<RSocketInterceptor> 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<String> 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
}
}

4
spring-cloud-gateway-rsocket/src/test/java/org/springframework/cloud/gateway/rsocket/autoconfigure/GatewayRSocketAutoConfigurationTests.java
View File

@@ -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)

19
spring-cloud-gateway-rsocket/src/test/java/org/springframework/cloud/gateway/rsocket/server/GatewayRSocketIntegrationTests.java
View File

@@ -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();
}
}

8
spring-cloud-gateway-rsocket/src/test/java/org/springframework/cloud/gateway/rsocket/test/PingPongApp.java
View File

@@ -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")