Updates for boot RSocket changes

removes pinned reactor version fixes gh-1114
2019-06-28 12:26:59 -04:00
parent b687f49201
commit e61028a8b7
7 changed files with 65 additions and 31 deletions
--- a/spring-cloud-gateway-rsocket/pom.xml
+++ b/spring-cloud-gateway-rsocket/pom.xml
@@ -30,22 +30,6 @@
 	<name>Spring Cloud Gateway RSocket</name>
 	<description>Spring Cloud Gateway RSocket</description>

-	<properties>
-		<reactor.version>Californium-SR5</reactor.version>
-	</properties>
-
-	<dependencyManagement>
-		<dependencies>
-			<dependency>
-				<groupId>io.projectreactor</groupId>
-				<artifactId>reactor-bom</artifactId>
-				<version>${reactor.version}</version>
-				<type>pom</type>
-				<scope>import</scope>
-			</dependency>
-		</dependencies>
-	</dependencyManagement>
-
 	<dependencies>
 		<dependency>
 			<groupId>org.springframework.boot</groupId>
--- 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
@@ -21,9 +21,13 @@ import java.util.List;
 import io.micrometer.core.instrument.MeterRegistry;
 import io.rsocket.RSocket;

+import org.springframework.boot.autoconfigure.AutoConfigureAfter;
 import org.springframework.boot.autoconfigure.condition.ConditionalOnClass;
 import org.springframework.boot.autoconfigure.condition.ConditionalOnProperty;
+import org.springframework.boot.autoconfigure.rsocket.RSocketServerAutoConfiguration;
 import org.springframework.boot.context.properties.EnableConfigurationProperties;
+import org.springframework.boot.rsocket.server.RSocketServerBootstrap;
+import org.springframework.boot.rsocket.server.RSocketServerFactory;
 import org.springframework.cloud.gateway.rsocket.core.GatewayRSocket;
 import org.springframework.cloud.gateway.rsocket.core.GatewayServerRSocketFactoryCustomizer;
 import org.springframework.cloud.gateway.rsocket.registry.Registry;
@@ -46,6 +50,7 @@ import org.springframework.core.env.Environment;
 		matchIfMissing = true)
@EnableConfigurationProperties
@ConditionalOnClass(RSocket.class)
+@AutoConfigureAfter(RSocketServerAutoConfiguration.class)
 public class GatewayRSocketAutoConfiguration {

 	@Bean
@@ -102,4 +107,11 @@ public class GatewayRSocketAutoConfiguration {
 		return new GatewayServerRSocketFactoryCustomizer(properties, meterRegistry);
 	}

+	@Bean
+	public RSocketServerBootstrap rSocketServerBootstrap(
+			RSocketServerFactory rSocketServerFactory,
+			GatewaySocketAcceptor gatewaySocketAcceptor) {
+		return new RSocketServerBootstrap(rSocketServerFactory, gatewaySocketAcceptor);
+	}
+
 }
--- a/spring-cloud-gateway-rsocket/src/main/java/org/springframework/cloud/gateway/rsocket/core/GatewayServerRSocketFactoryCustomizer.java
+++ b/spring-cloud-gateway-rsocket/src/main/java/org/springframework/cloud/gateway/rsocket/core/GatewayServerRSocketFactoryCustomizer.java
@@ -64,7 +64,7 @@ public class GatewayServerRSocketFactoryCustomizer

 	@Override
 	public ServerRSocketFactory apply(ServerRSocketFactory factory) {
-		serverInterceptors.forEach(factory::addServerPlugin);
+		serverInterceptors.forEach(factory::addResponderPlugin);

 		List<String> micrometerTags = properties.getMicrometerTags();
 		Tag[] tags = Tags.of(micrometerTags.toArray(new String[] {}))
--- a/spring-cloud-gateway-rsocket/src/main/java/org/springframework/cloud/gateway/rsocket/socketacceptor/GatewaySocketAcceptor.java
+++ b/spring-cloud-gateway-rsocket/src/main/java/org/springframework/cloud/gateway/rsocket/socketacceptor/GatewaySocketAcceptor.java
@@ -26,6 +26,8 @@ import io.micrometer.core.instrument.Tags;
 import io.rsocket.ConnectionSetupPayload;
 import io.rsocket.RSocket;
 import io.rsocket.SocketAcceptor;
+import org.apache.commons.logging.Log;
+import org.apache.commons.logging.LogFactory;
 import reactor.core.publisher.Mono;

 import org.springframework.cloud.gateway.rsocket.autoconfigure.GatewayRSocketProperties;
@@ -35,6 +37,8 @@ import org.springframework.cloud.gateway.rsocket.support.Metadata;

 public class GatewaySocketAcceptor implements SocketAcceptor {

+	private static final Log log = LogFactory.getLog(GatewaySocketAcceptor.class);
+
 	private final SocketAcceptorFilterChain filterChain;

 	private final GatewayRSocket.Factory rSocketFactory;
@@ -55,7 +59,9 @@ public class GatewaySocketAcceptor implements SocketAcceptor {
 	@Override
 	@SuppressWarnings("Duplicates")
 	public Mono<RSocket> accept(ConnectionSetupPayload setup, RSocket sendingSocket) {
-
+		if (log.isTraceEnabled()) {
+			log.trace("accept()");
+		}
 		// decorate GatewayRSocket with metrics
 		// current gateway id, type requester, service name (from metadata), service id

--- 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
@@ -16,11 +16,15 @@

 package org.springframework.cloud.gateway.rsocket.autoconfigure;

+import io.rsocket.SocketAcceptor;
 import org.junit.Test;

 import org.springframework.boot.actuate.autoconfigure.metrics.CompositeMeterRegistryAutoConfiguration;
 import org.springframework.boot.actuate.autoconfigure.metrics.MetricsAutoConfiguration;
 import org.springframework.boot.autoconfigure.AutoConfigurations;
+import org.springframework.boot.rsocket.server.RSocketServer;
+import org.springframework.boot.rsocket.server.RSocketServerBootstrap;
+import org.springframework.boot.rsocket.server.RSocketServerFactory;
 import org.springframework.boot.test.context.runner.ReactiveWebApplicationContextRunner;
 import org.springframework.cloud.gateway.rsocket.core.GatewayServerRSocketFactoryCustomizer;
 import org.springframework.cloud.gateway.rsocket.registry.Registry;
@@ -29,14 +33,19 @@ import org.springframework.cloud.gateway.rsocket.registry.RegistrySocketAcceptor
 import org.springframework.cloud.gateway.rsocket.socketacceptor.GatewaySocketAcceptor;
 import org.springframework.cloud.gateway.rsocket.socketacceptor.SocketAcceptorPredicate;
 import org.springframework.cloud.gateway.rsocket.socketacceptor.SocketAcceptorPredicateFilter;
+import org.springframework.context.annotation.Bean;
+import org.springframework.context.annotation.Configuration;

 import static org.assertj.core.api.Assertions.assertThat;
+import static org.mockito.ArgumentMatchers.any;
+import static org.mockito.Mockito.mock;
+import static org.mockito.Mockito.when;

 public class GatewayRSocketAutoConfigurationTests {

 	@Test
 	public void gatewayRSocketConfigured() {
-		new ReactiveWebApplicationContextRunner()
+		new ReactiveWebApplicationContextRunner().withUserConfiguration(MyConfig.class)
 				.withConfiguration(
 						AutoConfigurations.of(GatewayRSocketAutoConfiguration.class,
 								CompositeMeterRegistryAutoConfiguration.class,
@@ -48,7 +57,21 @@ public class GatewayRSocketAutoConfigurationTests {
 						.hasSingleBean(GatewayRSocketProperties.class)
 						.hasSingleBean(GatewaySocketAcceptor.class)
 						.hasSingleBean(SocketAcceptorPredicateFilter.class)
+						.hasSingleBean(RSocketServerBootstrap.class)
 						.doesNotHaveBean(SocketAcceptorPredicate.class));
 	}

+	@Configuration
+	protected static class MyConfig {
+
+		@Bean
+		RSocketServerFactory rSocketServerFactory() {
+			RSocketServerFactory serverFactory = mock(RSocketServerFactory.class);
+			when(serverFactory.create(any(SocketAcceptor.class)))
+					.thenReturn(mock(RSocketServer.class));
+			return serverFactory;
+		}
+
+	}
+
 }
--- a/spring-cloud-gateway-rsocket/src/test/java/org/springframework/cloud/gateway/rsocket/core/GatewayRSocketIntegrationTests.java
+++ b/spring-cloud-gateway-rsocket/src/test/java/org/springframework/cloud/gateway/rsocket/core/GatewayRSocketIntegrationTests.java
@@ -20,7 +20,6 @@ import java.time.Duration;

 import org.junit.AfterClass;
 import org.junit.BeforeClass;
-import org.junit.Ignore;
 import org.junit.Test;
 import org.junit.runner.RunWith;
 import reactor.test.StepVerifier;
@@ -38,7 +37,10 @@ import org.springframework.util.SocketUtils;
 import static org.assertj.core.api.Assertions.assertThat;

@RunWith(SpringRunner.class)
-@SpringBootTest(classes = PingPongApp.class, properties = { "ping.take=5" },
+@SpringBootTest(classes = PingPongApp.class, properties = { "ping.take=10",
+		// TODO: remove after
+		// https://github.com/spring-cloud/spring-cloud-gateway/issues/1140
+		"spring.main.allow-bean-definition-overriding=true" },
 		webEnvironment = WebEnvironment.RANDOM_PORT)
 public class GatewayRSocketIntegrationTests {

@@ -70,7 +72,6 @@ public class GatewayRSocketIntegrationTests {
 		System.clearProperty("spring.rsocket.server.port");
 	}

-	@Ignore // FIXME: 2.2.0 gh-1114
 	@Test
 	public void contextLoads() {
 		// @formatter:off
@@ -78,8 +79,7 @@ public class GatewayRSocketIntegrationTests {
 				.expectSubscription()
 				.then(() -> server.stop())
 				.thenConsumeWhile(s -> true)
-				//.expectComplete()
-				.thenCancel()  // https://github.com/rsocket/rsocket-java/issues/613
+				.expectComplete()
 				.verify(Duration.ofSeconds(20));
 		// @formatter:on

--- 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
@@ -28,6 +28,7 @@ import io.netty.buffer.ByteBufUtil;
 import io.rsocket.Payload;
 import io.rsocket.RSocket;
 import io.rsocket.RSocketFactory;
+import io.rsocket.frame.decoder.PayloadDecoder;
 import io.rsocket.micrometer.MicrometerRSocketInterceptor;
 import io.rsocket.transport.netty.client.TcpClientTransport;
 import io.rsocket.util.DefaultPayload;
@@ -35,12 +36,13 @@ import io.rsocket.util.RSocketProxy;
 import lombok.extern.slf4j.Slf4j;
 import org.reactivestreams.Publisher;
 import reactor.core.publisher.Flux;
+import reactor.core.publisher.Hooks;
 import reactor.core.publisher.Mono;

 import org.springframework.beans.factory.annotation.Autowired;
-import org.springframework.boot.SpringApplication;
 import org.springframework.boot.autoconfigure.SpringBootApplication;
 import org.springframework.boot.autoconfigure.condition.ConditionalOnProperty;
+import org.springframework.boot.builder.SpringApplicationBuilder;
 import org.springframework.boot.context.event.ApplicationReadyEvent;
 import org.springframework.cloud.gateway.rsocket.core.GatewayExchange;
 import org.springframework.cloud.gateway.rsocket.core.GatewayFilter;
@@ -81,7 +83,12 @@ public class PingPongApp {
 	}

 	public static void main(String[] args) {
-		SpringApplication.run(PingPongApp.class, args);
+		Hooks.onOperatorDebug();
+		new SpringApplicationBuilder(PingPongApp.class)
+				// TODO: remove after
+				// https://github.com/spring-cloud/spring-cloud-gateway/issues/1140
+				.properties("spring.main.allow-bean-definition-overriding=true")
+				.run(args);
 	}

 	static String reply(String in) {
@@ -134,11 +141,11 @@ public class PingPongApp {
 					meterRegistry, Tag.of("component", "ping"));
 			ByteBuf announcementMetadata = Metadata.from("ping").with("id", "ping" + id)
 					.encode();
-			pongFlux = RSocketFactory.connect()
+			pongFlux = RSocketFactory.connect().frameDecoder(PayloadDecoder.ZERO_COPY)
 					.metadataMimeType(Metadata.ROUTING_MIME_TYPE)
 					.setupPayload(
 							DefaultPayload.create(EMPTY_BUFFER, announcementMetadata))
-					.addClientPlugin(interceptor)
+					.addRequesterPlugin(interceptor)
 					.transport(TcpClientTransport.create(gatewayPort)) // proxy
 					.start().flatMapMany(socket -> doPing(take, socket));

@@ -151,10 +158,12 @@ public class PingPongApp {
 						ByteBuf data = ByteBufUtil.writeUtf8(ByteBufAllocator.DEFAULT,
 								"ping" + id);
 						ByteBuf routingMetadata = Metadata.from("pong").encode();
+						log.debug("Sending ping" + id);
 						return DefaultPayload.create(data, routingMetadata);
 						// onBackpressue is needed in case pong is not available yet
-					}).onBackpressureDrop(payload -> log
-							.debug("Dropped payload " + payload.getDataUtf8())))
+					}).log("doPing")
+							.onBackpressureDrop(payload -> log
+									.debug("Dropped payload " + payload.getDataUtf8())))
 					.map(Payload::getDataUtf8).doOnNext(str -> {
 						int received = pongsReceived.incrementAndGet();
 						log.info("received " + str + "(" + received + ") in Ping" + id);
@@ -209,7 +218,7 @@ public class PingPongApp {
 			RSocketFactory.connect().metadataMimeType(Metadata.ROUTING_MIME_TYPE)
 					.setupPayload(
 							DefaultPayload.create(EMPTY_BUFFER, announcementMetadata))
-					.addClientPlugin(interceptor).acceptor(this::accept)
+					.addRequesterPlugin(interceptor).acceptor(this::accept)
 					.transport(TcpClientTransport.create(gatewayPort)) // proxy
 					.start().block();
 		}