diff --git a/consumer/rsocket-consumer/README.adoc b/consumer/rsocket-consumer/README.adoc new file mode 100644 index 00000000..1aba1e70 --- /dev/null +++ b/consumer/rsocket-consumer/README.adoc @@ -0,0 +1,26 @@ +# RSocket Consumer + +A consumer that allows you to communicate to an RSocket route using its fire and forget strategy of execution. +The consumer uses the RSocket support from https://docs.spring.io/spring/docs/current/spring-framework-reference/web-reactive.html#rsocket-requester[Spring Framework]. + +## Beans for injection + +You can import `RSocketConsumerConfiguration` in the application and then inject the following bean. + +`Function>, Mono> rsocketConsumer` + +You can use `rsocketConsumer` as a qualifier when injecting. + +## Configuration Options + +All configuration properties are prefixed with `rsocket.consumer`. + +For more information on the various options available, please see link:src/main/java/org/springframework/cloud/fn/consumer/rsocket/RsocketConsumerProperties.java[RsocketConsumerProperties]. + +## Examples + +See this link:src/test/java/org/springframework/cloud/fn/consumer/rsocket/RsocketConsumerTests.java[test suite] for learning more about this consumer. + +## Other usage + +See this https://github.com/spring-cloud/stream-applications/blob/master/applications/sink/rsocket-sink/README.adoc[README] where this consumer is used to create a Spring Cloud Stream based RSocket Sink application. \ No newline at end of file diff --git a/consumer/rsocket-consumer/pom.xml b/consumer/rsocket-consumer/pom.xml new file mode 100644 index 00000000..098fd30e --- /dev/null +++ b/consumer/rsocket-consumer/pom.xml @@ -0,0 +1,40 @@ + + + 4.0.0 + rsocket-consumer + 1.0.0-SNAPSHOT + rsocket-consumer + + + org.springframework.cloud.fn + spring-functions-parent + 1.0.0-SNAPSHOT + ../../spring-functions-parent + + + + + org.springframework.boot + spring-boot-starter-rsocket + + + org.springframework.boot + spring-boot-configuration-processor + provided + + + org.springframework.boot + spring-boot-starter-test + test + + + io.projectreactor + reactor-test + test + + + + + diff --git a/consumer/rsocket-consumer/src/main/java/org/springframework/cloud/fn/consumer/rsocket/RsocketConsumerConfiguration.java b/consumer/rsocket-consumer/src/main/java/org/springframework/cloud/fn/consumer/rsocket/RsocketConsumerConfiguration.java new file mode 100644 index 00000000..727dd5e6 --- /dev/null +++ b/consumer/rsocket-consumer/src/main/java/org/springframework/cloud/fn/consumer/rsocket/RsocketConsumerConfiguration.java @@ -0,0 +1,51 @@ +/* + * Copyright 2020-2020 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.fn.consumer.rsocket; + +import java.util.function.Function; + +import reactor.core.publisher.Flux; +import reactor.core.publisher.Mono; + +import org.springframework.boot.context.properties.EnableConfigurationProperties; +import org.springframework.context.annotation.Bean; +import org.springframework.context.annotation.Configuration; +import org.springframework.messaging.Message; +import org.springframework.messaging.rsocket.RSocketRequester; + +@Configuration(proxyBeanMethods = false) +@EnableConfigurationProperties(RsocketConsumerProperties.class) +public class RsocketConsumerConfiguration { + + @Bean + public Function>, Mono> rsocketConsumer(RSocketRequester.Builder builder, + RsocketConsumerProperties rsocketConsumerProperties) { + final Mono rSocketRequester = + rsocketConsumerProperties.getUri() != null ? builder.connectWebSocket(rsocketConsumerProperties.getUri()).cache() : + builder.connectTcp(rsocketConsumerProperties.getHost(), + rsocketConsumerProperties.getPort()).cache(); + + return input -> + input.flatMap(message -> + rSocketRequester + .flatMap(requester -> requester.route(rsocketConsumerProperties.getRoute()) + .data(message.getPayload()) + .send())) + .ignoreElements(); + } + +} diff --git a/consumer/rsocket-consumer/src/main/java/org/springframework/cloud/fn/consumer/rsocket/RsocketConsumerProperties.java b/consumer/rsocket-consumer/src/main/java/org/springframework/cloud/fn/consumer/rsocket/RsocketConsumerProperties.java new file mode 100644 index 00000000..f2f36fc7 --- /dev/null +++ b/consumer/rsocket-consumer/src/main/java/org/springframework/cloud/fn/consumer/rsocket/RsocketConsumerProperties.java @@ -0,0 +1,77 @@ +/* + * Copyright 2020-2020 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.fn.consumer.rsocket; + +import java.net.URI; + +import org.springframework.boot.context.properties.ConfigurationProperties; + +@ConfigurationProperties("rsocket.consumer") +public class RsocketConsumerProperties { + + /** + * RSocket host. + */ + private String host = "localhost"; + + /** + * RSocket port. + */ + private int port = 7000; + + /** + * URI that can be used for websocket based transport. + */ + private URI uri; + + /** + * Route used for RSocket. + */ + private String route; + + public String getHost() { + return this.host; + } + + public void setHost(String host) { + this.host = host; + } + + public int getPort() { + return this.port; + } + + public void setPort(int port) { + this.port = port; + } + + public String getRoute() { + return this.route; + } + + public void setRoute(String route) { + this.route = route; + } + + public URI getUri() { + return this.uri; + } + + public void setUri(URI uri) { + this.uri = uri; + } +} diff --git a/consumer/rsocket-consumer/src/test/java/org/springframework/cloud/fn/consumer/rsocket/RsocketConsumerTests.java b/consumer/rsocket-consumer/src/test/java/org/springframework/cloud/fn/consumer/rsocket/RsocketConsumerTests.java new file mode 100644 index 00000000..561f7134 --- /dev/null +++ b/consumer/rsocket-consumer/src/test/java/org/springframework/cloud/fn/consumer/rsocket/RsocketConsumerTests.java @@ -0,0 +1,91 @@ +/* + * Copyright 2020-2020 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.fn.consumer.rsocket; + +import java.util.function.Function; + +import org.junit.jupiter.api.Test; +import reactor.core.publisher.Flux; +import reactor.core.publisher.Mono; +import reactor.core.publisher.ReplayProcessor; +import reactor.test.StepVerifier; + +import org.springframework.beans.factory.annotation.Autowired; +import org.springframework.boot.SpringBootConfiguration; +import org.springframework.boot.autoconfigure.EnableAutoConfiguration; +import org.springframework.boot.autoconfigure.rsocket.RSocketRequesterAutoConfiguration; +import org.springframework.boot.autoconfigure.rsocket.RSocketStrategiesAutoConfiguration; +import org.springframework.boot.rsocket.context.RSocketServerBootstrap; +import org.springframework.boot.rsocket.server.RSocketServer; +import org.springframework.boot.test.context.SpringBootTest; +import org.springframework.boot.test.context.runner.ApplicationContextRunner; +import org.springframework.context.ApplicationContext; +import org.springframework.messaging.Message; +import org.springframework.messaging.handler.annotation.MessageMapping; +import org.springframework.messaging.support.GenericMessage; +import org.springframework.stereotype.Controller; +import org.springframework.test.annotation.DirtiesContext; +import org.springframework.test.util.ReflectionTestUtils; + +@SpringBootTest(properties = {"spring.rsocket.server.port=0"}) +@DirtiesContext +public class RsocketConsumerTests { + + private static ApplicationContextRunner applicationContextRunner = new ApplicationContextRunner() + .withUserConfiguration(RsocketConsumerConfiguration.class, RSocketRequesterAutoConfiguration.class, + RSocketStrategiesAutoConfiguration.class); + + @Autowired + ApplicationContext applicationContext; + + @Test + void testRsocketConsumer() { + + RSocketServerBootstrap serverBootstrap = applicationContext.getBean(RSocketServerBootstrap.class); + RSocketServer server = (RSocketServer) ReflectionTestUtils.getField(serverBootstrap, "server"); + final int port = server.address().getPort(); + + applicationContextRunner.withPropertyValues( + "rsocket.consumer.port=" + port, + "rsocket.consumer.route=test-route") + .run(context -> { + Function>, Mono> rsocketConsumer = context.getBean("rsocketConsumer", Function.class); + rsocketConsumer.apply(Flux.just(new GenericMessage<>("Hello RSocket"))) + .subscribe(); + + StepVerifier.create(RSocketserverApplication.fireForgetPayloads) + .expectNext("Hello RSocket") + .thenCancel() + .verify(); + }); + + } + + @EnableAutoConfiguration + @SpringBootConfiguration + @Controller + static class RSocketserverApplication { + static final ReplayProcessor fireForgetPayloads = ReplayProcessor.create(); + + @MessageMapping("test-route") + void someMethod(String payload) { + this.fireForgetPayloads.onNext(payload); + } + } +} + + diff --git a/function-dependencies/pom.xml b/function-dependencies/pom.xml index 8a5a57ee..81c4f0ec 100644 --- a/function-dependencies/pom.xml +++ b/function-dependencies/pom.xml @@ -110,6 +110,11 @@ analytics-consumer ${project.version} + + org.springframework.cloud.fn + rsocket-consumer + ${project.version} + org.springframework.cloud.fn cassandra-consumer diff --git a/pom.xml b/pom.xml index 2e3608a7..0cea3b09 100644 --- a/pom.xml +++ b/pom.xml @@ -71,6 +71,7 @@ consumer/s3-consumer consumer/twitter-consumer consumer/wavefront-consumer + consumer/rsocket-consumer function/aggregator-function function/filter-function