RSocket consumer/sink

* This RSocket consumer is using the fire-and-forget strategy of execution.
* Generating the corresponding sink.
* Addressing the original PR review comments from
  https://github.com/spring-cloud/stream-applications/pull/116

* Addressing PR review comments

* Fix README

* Addressing PR review

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<Flux<Message<?>>, Mono<Void>> 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.

consumer/rsocket-consumer/pom.xml

@@ -0,0 +1,40 @@
+<?xml version="1.0" encoding="UTF-8"?>
+<project xmlns="http://maven.apache.org/POM/4.0.0"
+		 xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
+		 xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/xsd/maven-4.0.0.xsd">
+	<modelVersion>4.0.0</modelVersion>
+	<artifactId>rsocket-consumer</artifactId>
+	<version>1.0.0-SNAPSHOT</version>
+	<name>rsocket-consumer</name>
+
+	<parent>
+		<groupId>org.springframework.cloud.fn</groupId>
+		<artifactId>spring-functions-parent</artifactId>
+		<version>1.0.0-SNAPSHOT</version>
+		<relativePath>../../spring-functions-parent</relativePath>
+	</parent>
+
+	<dependencies>
+		<dependency>
+			<groupId>org.springframework.boot</groupId>
+			<artifactId>spring-boot-starter-rsocket</artifactId>
+		</dependency>
+		<dependency>
+			<groupId>org.springframework.boot</groupId>
+			<artifactId>spring-boot-configuration-processor</artifactId>
+			<scope>provided</scope>
+		</dependency>
+		<dependency>
+			<groupId>org.springframework.boot</groupId>
+			<artifactId>spring-boot-starter-test</artifactId>
+			<scope>test</scope>
+		</dependency>
+		<dependency>
+			<groupId>io.projectreactor</groupId>
+			<artifactId>reactor-test</artifactId>
+			<scope>test</scope>
+		</dependency>
+
+	</dependencies>
+
+</project>

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<Flux<Message<?>>, Mono<Void>> rsocketConsumer(RSocketRequester.Builder builder,
+																RsocketConsumerProperties rsocketConsumerProperties) {
+		final Mono<RSocketRequester> 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();
+	}
+
+}

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;
+	}
+}

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<Flux<Message<?>>, Mono<Void>> 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<String> fireForgetPayloads = ReplayProcessor.create();
+
+		@MessageMapping("test-route")
+		void someMethod(String payload) {
+			this.fireForgetPayloads.onNext(payload);
+		}
+	}
+}
+
+

function-dependencies/pom.xml

@@ -110,6 +110,11 @@
                 <artifactId>analytics-consumer</artifactId>
                 <version>${project.version}</version>
             </dependency>
+            <dependency>
+                <groupId>org.springframework.cloud.fn</groupId>
+                <artifactId>rsocket-consumer</artifactId>
+                <version>${project.version}</version>
+            </dependency>
             <dependency>
                 <groupId>org.springframework.cloud.fn</groupId>
                 <artifactId>cassandra-consumer</artifactId>

pom.xml

@@ -71,6 +71,7 @@
 		<module>consumer/s3-consumer</module>
 		<module>consumer/twitter-consumer</module>
 		<module>consumer/wavefront-consumer</module>
+		<module>consumer/rsocket-consumer</module>
 
 		<module>function/aggregator-function</module>
 		<module>function/filter-function</module>