diff --git a/spring-cloud-gateway-rsocket/pom.xml b/spring-cloud-gateway-rsocket/pom.xml index 0be259da..b9d11e09 100644 --- a/spring-cloud-gateway-rsocket/pom.xml +++ b/spring-cloud-gateway-rsocket/pom.xml @@ -30,22 +30,6 @@ Spring Cloud Gateway RSocket Spring Cloud Gateway RSocket - - Californium-SR5 - - - - - - io.projectreactor - reactor-bom - ${reactor.version} - pom - import - - - - 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 54a57c63..c60245bd 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 @@ -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); + } + } diff --git 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 index e79c8060..b8574d45 100644 --- 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 micrometerTags = properties.getMicrometerTags(); Tag[] tags = Tags.of(micrometerTags.toArray(new String[] {})) diff --git 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 index 397b3856..0ed9b0aa 100644 --- 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 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 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 470b397f..55040727 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 @@ -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; + } + + } + } diff --git 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 index c70d4eba..91dffbfd 100644 --- 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 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 9df8e4ba..757c4e43 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 @@ -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(); }