diff --git a/applications/sink/pom.xml b/applications/sink/pom.xml index 515b7327..49ebe61e 100644 --- a/applications/sink/pom.xml +++ b/applications/sink/pom.xml @@ -32,5 +32,6 @@ twitter-message-sink wavefront-sink pgcopy-sink + rsocket-sink diff --git a/applications/sink/rsocket-sink/README.adoc b/applications/sink/rsocket-sink/README.adoc new file mode 100644 index 00000000..fe5359b9 --- /dev/null +++ b/applications/sink/rsocket-sink/README.adoc @@ -0,0 +1,18 @@ +//tag::ref-doc[] += RSocket Sink + +RSocket sink to send data using RSocket protocols' fire and forget strategy. + +== Options + +The **$$rsocket$$** $$sink$$ has the following options: + +//tag::configuration-properties[] +$$rsocket.consumer.host$$:: $$RSocket host.$$ *($$String$$, default: `$$localhost$$`)* +$$rsocket.consumer.port$$:: $$RSocket port.$$ *($$Integer$$, default: `$$7000$$`)* +$$rsocket.consumer.route$$:: $$Route used for RSocket.$$ *($$String$$, default: `$$$$`)* +$$rsocket.consumer.uri$$:: $$URI that can be used for websocket based transport.$$ *($$URI$$, default: `$$$$`)* +//end::configuration-properties[] + + +//end::ref-doc[] diff --git a/applications/sink/rsocket-sink/pom.xml b/applications/sink/rsocket-sink/pom.xml new file mode 100644 index 00000000..4593ad1c --- /dev/null +++ b/applications/sink/rsocket-sink/pom.xml @@ -0,0 +1,93 @@ + + + 4.0.0 + rsocket-sink + 3.0.0-SNAPSHOT + rsocket-sink + RSocket sink apps + jar + + + org.springframework.cloud.stream.app + stream-applications-core + 3.0.0-SNAPSHOT + + + + + + org.springframework.cloud.fn + rsocket-consumer + + + org.springframework.boot + spring-boot-starter-test + test + + + io.projectreactor + reactor-test + test + + + org.springframework.boot + spring-boot-starter-rsocket + test + + + org.springframework.boot + spring-boot-starter-test + test + + + + + + + org.springframework.cloud + spring-cloud-dataflow-apps-docs-plugin + + + org.springframework.cloud + spring-cloud-dataflow-apps-generator-plugin + + + rsocket + sink + ${project.version} + org.springframework.cloud.fn.consumer.rsocket.RsocketConsumerConfiguration.class + + + + org.springframework.cloud.fn + rsocket-consumer + + + + + + + + + + + + + true + + spring-snapshots + Spring Snapshots + https://repo.spring.io/libs-snapshot-local + + + + false + + spring-milestones + Spring Milestones + https://repo.spring.io/libs-milestone-local + + + + diff --git a/applications/sink/rsocket-sink/src/main/resources/META-INF/dataflow-configuration-metadata-whitelist.properties b/applications/sink/rsocket-sink/src/main/resources/META-INF/dataflow-configuration-metadata-whitelist.properties new file mode 100644 index 00000000..1fd7a382 --- /dev/null +++ b/applications/sink/rsocket-sink/src/main/resources/META-INF/dataflow-configuration-metadata-whitelist.properties @@ -0,0 +1,2 @@ +configuration-properties.classes=org.springframework.cloud.fn.consumer.rsocket.RsocketConsumerProperties + diff --git a/applications/sink/rsocket-sink/src/test/java/org/springframework/cloud/stream/app/rsocket/sink/RSocketSinkTests.java b/applications/sink/rsocket-sink/src/test/java/org/springframework/cloud/stream/app/rsocket/sink/RSocketSinkTests.java new file mode 100644 index 00000000..185166e9 --- /dev/null +++ b/applications/sink/rsocket-sink/src/test/java/org/springframework/cloud/stream/app/rsocket/sink/RSocketSinkTests.java @@ -0,0 +1,113 @@ +/* + * 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.stream.app.rsocket.sink; + +import java.util.function.Function; + +import org.junit.jupiter.api.BeforeAll; +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.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.cloud.fn.consumer.rsocket.RsocketConsumerConfiguration; +import org.springframework.cloud.stream.binder.test.InputDestination; +import org.springframework.cloud.stream.binder.test.TestChannelBinderConfiguration; +import org.springframework.context.ApplicationContext; +import org.springframework.context.annotation.Import; +import org.springframework.integration.support.MessageBuilder; +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; + +/** + * @author Soby Chacko + */ +@SpringBootTest(properties = {"spring.rsocket.server.port=0"}, classes = RSocketSinkTests.RSocketserverApplication.class) +@DirtiesContext +public class RSocketSinkTests { + + private static ApplicationContextRunner applicationContextRunner; + + @BeforeAll + static void setup() { + applicationContextRunner = new ApplicationContextRunner() + .withUserConfiguration(TestChannelBinderConfiguration.getCompleteConfiguration(RsocketSinkTestApplication.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( + "spring.cloud.function.definition=rsocketConsumer", + "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(); + + final StepVerifier stepVerifier = StepVerifier.create(RSocketserverApplication.fireForgetPayloads) + .expectNext("Hello RSocket") + .thenCancel() + .verifyLater(); + + final Message message = MessageBuilder.withPayload("Hello RSocket").build(); + InputDestination source = context.getBean(InputDestination.class); + source.send(message); + + stepVerifier.verify(); + }); + } + + @EnableAutoConfiguration + @SpringBootConfiguration + @Controller + static class RSocketserverApplication { + static final ReplayProcessor fireForgetPayloads = ReplayProcessor.create(); + + @MessageMapping("test-route") + void someMethod(String payload) { + this.fireForgetPayloads.onNext(payload); + } + } + + @EnableAutoConfiguration + @SpringBootConfiguration + @Import(RsocketConsumerConfiguration.class) + static class RsocketSinkTestApplication { + + } +} diff --git a/applications/source/twitter-stream-source/README.adoc b/applications/source/twitter-stream-source/README.adoc index 72a74510..11af5b77 100644 --- a/applications/source/twitter-stream-source/README.adoc +++ b/applications/source/twitter-stream-source/README.adoc @@ -6,7 +6,7 @@ Real-time Tweet streaming https://developer.twitter.com/en/docs/tweets/filter-re * The `Filter API` returns public statuses that match one or more filter predicates. Multiple parameters allows using a single connection to the Streaming API. TIP: The `track`, `follow`, and `locations` fields are combined with an *OR* operator! -Queries with `track=foo` and `follow=1234` returns Tweets matching `foo` *OR* created by user `1234`. +Queries with `track=foo` and `follow=1234` returns Tweets matching `test` *OR* created by user `1234`. * The `Sample API` returns a small random sample of all public statuses. The Tweets returned by the default access level are the same, so if two different clients connect to this endpoint, they will see the same Tweets. diff --git a/functions/consumer/rsocket-consumer/README.adoc b/functions/consumer/rsocket-consumer/README.adoc new file mode 100644 index 00000000..1aba1e70 --- /dev/null +++ b/functions/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/functions/consumer/rsocket-consumer/pom.xml b/functions/consumer/rsocket-consumer/pom.xml new file mode 100644 index 00000000..098fd30e --- /dev/null +++ b/functions/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/functions/consumer/rsocket-consumer/src/main/java/org/springframework/cloud/fn/consumer/rsocket/RsocketConsumerConfiguration.java b/functions/consumer/rsocket-consumer/src/main/java/org/springframework/cloud/fn/consumer/rsocket/RsocketConsumerConfiguration.java new file mode 100644 index 00000000..727dd5e6 --- /dev/null +++ b/functions/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/functions/consumer/rsocket-consumer/src/main/java/org/springframework/cloud/fn/consumer/rsocket/RsocketConsumerProperties.java b/functions/consumer/rsocket-consumer/src/main/java/org/springframework/cloud/fn/consumer/rsocket/RsocketConsumerProperties.java new file mode 100644 index 00000000..f2f36fc7 --- /dev/null +++ b/functions/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/functions/consumer/rsocket-consumer/src/test/java/org/springframework/cloud/fn/consumer/rsocket/RsocketConsumerTests.java b/functions/consumer/rsocket-consumer/src/test/java/org/springframework/cloud/fn/consumer/rsocket/RsocketConsumerTests.java new file mode 100644 index 00000000..561f7134 --- /dev/null +++ b/functions/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/functions/function-dependencies/pom.xml b/functions/function-dependencies/pom.xml index 8a5a57ee..81c4f0ec 100644 --- a/functions/function-dependencies/pom.xml +++ b/functions/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/functions/pom.xml b/functions/pom.xml index 2e3608a7..0cea3b09 100644 --- a/functions/pom.xml +++ b/functions/pom.xml @@ -71,6 +71,7 @@ consumer/s3-consumer consumer/twitter-consumer consumer/wavefront-consumer + consumer/rsocket-consumer function/aggregator-function function/filter-function