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
This commit is contained in:
Soby Chacko
@@ -32,5 +32,6 @@
|
||||
<module>twitter-message-sink</module>
|
||||
<module>wavefront-sink</module>
|
||||
<module>pgcopy-sink</module>
|
||||
<module>rsocket-sink</module>
|
||||
</modules>
|
||||
</project>
|
||||
|
||||
18
applications/sink/rsocket-sink/README.adoc
Normal file
18
applications/sink/rsocket-sink/README.adoc
Normal file
@@ -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: `$$<none>$$`)*
|
||||
$$rsocket.consumer.uri$$:: $$URI that can be used for websocket based transport.$$ *($$URI$$, default: `$$<none>$$`)*
|
||||
//end::configuration-properties[]
|
||||
|
||||
|
||||
//end::ref-doc[]
|
||||
93
applications/sink/rsocket-sink/pom.xml
Normal file
93
applications/sink/rsocket-sink/pom.xml
Normal file
@@ -0,0 +1,93 @@
|
||||
<?xml version="1.0" encoding="UTF-8"?>
|
||||
<project xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance" xmlns="http://maven.apache.org/POM/4.0.0"
|
||||
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-sink</artifactId>
|
||||
<version>3.0.0-SNAPSHOT</version>
|
||||
<name>rsocket-sink</name>
|
||||
<description>RSocket sink apps</description>
|
||||
<packaging>jar</packaging>
|
||||
|
||||
<parent>
|
||||
<groupId>org.springframework.cloud.stream.app</groupId>
|
||||
<artifactId>stream-applications-core</artifactId>
|
||||
<version>3.0.0-SNAPSHOT</version>
|
||||
<relativePath/>
|
||||
</parent>
|
||||
|
||||
<dependencies>
|
||||
<dependency>
|
||||
<groupId>org.springframework.cloud.fn</groupId>
|
||||
<artifactId>rsocket-consumer</artifactId>
|
||||
</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>
|
||||
<dependency>
|
||||
<groupId>org.springframework.boot</groupId>
|
||||
<artifactId>spring-boot-starter-rsocket</artifactId>
|
||||
<scope>test</scope>
|
||||
</dependency>
|
||||
<dependency>
|
||||
<groupId>org.springframework.boot</groupId>
|
||||
<artifactId>spring-boot-starter-test</artifactId>
|
||||
<scope>test</scope>
|
||||
</dependency>
|
||||
</dependencies>
|
||||
|
||||
<build>
|
||||
<plugins>
|
||||
<plugin>
|
||||
<groupId>org.springframework.cloud</groupId>
|
||||
<artifactId>spring-cloud-dataflow-apps-docs-plugin</artifactId>
|
||||
</plugin>
|
||||
<plugin>
|
||||
<groupId>org.springframework.cloud</groupId>
|
||||
<artifactId>spring-cloud-dataflow-apps-generator-plugin</artifactId>
|
||||
<configuration>
|
||||
<application>
|
||||
<name>rsocket</name>
|
||||
<type>sink</type>
|
||||
<version>${project.version}</version>
|
||||
<configClass>org.springframework.cloud.fn.consumer.rsocket.RsocketConsumerConfiguration.class</configClass>
|
||||
<maven>
|
||||
<dependencies>
|
||||
<dependency>
|
||||
<groupId>org.springframework.cloud.fn</groupId>
|
||||
<artifactId>rsocket-consumer</artifactId>
|
||||
</dependency>
|
||||
</dependencies>
|
||||
</maven>
|
||||
</application>
|
||||
</configuration>
|
||||
</plugin>
|
||||
</plugins>
|
||||
</build>
|
||||
|
||||
<repositories>
|
||||
<repository>
|
||||
<snapshots>
|
||||
<enabled>true</enabled>
|
||||
</snapshots>
|
||||
<id>spring-snapshots</id>
|
||||
<name>Spring Snapshots</name>
|
||||
<url>https://repo.spring.io/libs-snapshot-local</url>
|
||||
</repository>
|
||||
<repository>
|
||||
<snapshots>
|
||||
<enabled>false</enabled>
|
||||
</snapshots>
|
||||
<id>spring-milestones</id>
|
||||
<name>Spring Milestones</name>
|
||||
<url>https://repo.spring.io/libs-milestone-local</url>
|
||||
</repository>
|
||||
</repositories>
|
||||
|
||||
</project>
|
||||
@@ -0,0 +1,2 @@
|
||||
configuration-properties.classes=org.springframework.cloud.fn.consumer.rsocket.RsocketConsumerProperties
|
||||
|
||||
@@ -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<Flux<Message<?>>, Mono<Void>> 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<String> 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<String> fireForgetPayloads = ReplayProcessor.create();
|
||||
|
||||
@MessageMapping("test-route")
|
||||
void someMethod(String payload) {
|
||||
this.fireForgetPayloads.onNext(payload);
|
||||
}
|
||||
}
|
||||
|
||||
@EnableAutoConfiguration
|
||||
@SpringBootConfiguration
|
||||
@Import(RsocketConsumerConfiguration.class)
|
||||
static class RsocketSinkTestApplication {
|
||||
|
||||
}
|
||||
}
|
||||
@@ -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.
|
||||
|
||||
26
functions/consumer/rsocket-consumer/README.adoc
Normal file
26
functions/consumer/rsocket-consumer/README.adoc
Normal file
@@ -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.
|
||||
40
functions/consumer/rsocket-consumer/pom.xml
Normal file
40
functions/consumer/rsocket-consumer/pom.xml
Normal file
@@ -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>
|
||||
@@ -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();
|
||||
}
|
||||
|
||||
}
|
||||
@@ -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;
|
||||
}
|
||||
}
|
||||
@@ -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);
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
|
||||
@@ -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>
|
||||
|
||||
@@ -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>
|
||||
|
||||
Reference in New Issue
Block a user