SpringCloud/stream-applications
1
0
Fork 0
Code Issues Packages Projects Releases Wiki Activity

Add true Consumer and tests (#99)

Browse Source 
* Add true Consumer and tests

* Fixed some bugs

* Update README

* Replace block() with subscribe()
This commit is contained in:
David Turanski
2020-08-11 13:08:28 -04:00
committed by GitHub
parent a68c7c96f6
commit f23cc84172
7 changed files with 104 additions and 15 deletions
Download Patch File Download Diff File Expand all files Collapse all files

12
applications/sink/mongodb-sink/README.adoc
View File

@@ -20,6 +20,18 @@ The **$$mongodb$$** $$sink$$ has the following options:
//tag::configuration-properties[]
$$mongodb.consumer.collection$$:: $$The MongoDB collection to store data.$$ *($$String$$, default: `$$<none>$$`)*
$$mongodb.consumer.collection-expression$$:: $$The SpEL expression to evaluate MongoDB collection.$$ *($$Expression$$, default: `$$<none>$$`)*
$$spring.data.mongodb.authentication-database$$:: $$Authentication database name.$$ *($$String$$, default: `$$<none>$$`)*
$$spring.data.mongodb.auto-index-creation$$:: $$Whether to enable auto-index creation.$$ *($$Boolean$$, default: `$$<none>$$`)*
$$spring.data.mongodb.database$$:: $$Database name.$$ *($$String$$, default: `$$<none>$$`)*
$$spring.data.mongodb.field-naming-strategy$$:: $$Fully qualified name of the FieldNamingStrategy to use.$$ *($$Class<?>$$, default: `$$<none>$$`)*
$$spring.data.mongodb.grid-fs-database$$:: $$GridFS database name.$$ *($$String$$, default: `$$<none>$$`)*
$$spring.data.mongodb.host$$:: $$Mongo server host. Cannot be set with URI.$$ *($$String$$, default: `$$<none>$$`)*
$$spring.data.mongodb.password$$:: $$Login password of the mongo server. Cannot be set with URI.$$ *($$Character[]$$, default: `$$<none>$$`)*
$$spring.data.mongodb.port$$:: $$Mongo server port. Cannot be set with URI.$$ *($$Integer$$, default: `$$<none>$$`)*
$$spring.data.mongodb.replica-set-name$$:: $$Required replica set name for the cluster. Cannot be set with URI.$$ *($$String$$, default: `$$<none>$$`)*
$$spring.data.mongodb.uri$$:: $$Mongo database URI. Cannot be set with host, port, credentials and replica set name.$$ *($$String$$, default: `$$mongodb://localhost/test$$`)*
$$spring.data.mongodb.username$$:: $$Login user of the mongo server. Cannot be set with URI.$$ *($$String$$, default: `$$<none>$$`)*
$$spring.data.mongodb.uuid-representation$$:: $$Representation to use when converting a UUID to a BSON binary value.$$ *($$UuidRepresentation$$, default: `$$java-legacy$$`, possible values: `UNSPECIFIED`,`STANDARD`,`C_SHARP_LEGACY`,`JAVA_LEGACY`,`PYTHON_LEGACY`)*
//end::configuration-properties[]
//end::ref-doc[]

15
applications/sink/mongodb-sink/pom.xml
View File

@@ -16,14 +16,24 @@
</parent>
<dependencies>
<dependency>
<groupId>org.springframework.cloud.fn</groupId>
<artifactId>mongodb-consumer</artifactId>
</dependency>
<dependency>
<groupId>org.springframework.boot</groupId>
<artifactId>spring-boot-starter-test</artifactId>
<scope>test</scope>
</dependency>
<dependency>
<groupId>org.springframework.cloud.fn</groupId>
<artifactId>mongodb-consumer</artifactId>
<groupId>io.projectreactor</groupId>
<artifactId>reactor-test</artifactId>
<scope>test</scope>
</dependency>
<dependency>
<groupId>de.flapdoodle.embed</groupId>
<artifactId>de.flapdoodle.embed.mongo</artifactId>
<scope>test</scope>
</dependency>
</dependencies>
@@ -43,6 +53,7 @@
<version>${project.version}</version>
<configClass>org.springframework.cloud.fn.consumer.mongo.MongoDbConsumerConfiguration.class
</configClass>
<functionDefinition>mongodbConsumer</functionDefinition>
</generatedApp>
<dependencies>
<dependency>

3
applications/sink/mongodb-sink/src/main/resources/META-INF/dataflow-configuration-metadata-whitelist.properties
View File

@@ -1 +1,2 @@
configuration-properties.classes=org.springframework.cloud.fn.consumer.mongo.MongoDbConsumerProperties
configuration-properties.classes=org.springframework.cloud.fn.consumer.mongo.MongoDbConsumerProperties,\
org.springframework.boot.autoconfigure.mongo.MongoProperties

56
applications/sink/mongodb-sink/src/main/test/org/springframework/cloud/stream/app/sink/mongodb/MongodbSinkTests.java Normal file
View File

@@ -0,0 +1,56 @@
/*
* 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.sink.mongodb;
import org.bson.Document;
import org.junit.jupiter.api.Test;
import reactor.test.StepVerifier;
import org.springframework.boot.autoconfigure.SpringBootApplication;
import org.springframework.cloud.fn.consumer.mongo.MongoDbConsumerConfiguration;
import org.springframework.cloud.stream.binder.test.InputDestination;
import org.springframework.cloud.stream.binder.test.TestChannelBinderConfiguration;
import org.springframework.context.annotation.Import;
import org.springframework.data.mongodb.core.ReactiveMongoTemplate;
import org.springframework.messaging.support.GenericMessage;
public class MongodbSinkTests {
@Test
void testInsert() {
TestChannelBinderConfiguration.applicationContextRunner(TestApp.class)
.withPropertyValues(
"spring.data.mongodb.port=0",
"spring.data.mongodb.database=test",
"mongodb.consumer.collection=demo",
"spring.cloud.function.definition=mongodbConsumer")
.run(context -> {
ReactiveMongoTemplate template = context.getBean(ReactiveMongoTemplate.class);
InputDestination inputDestination = context.getBean(InputDestination.class);
inputDestination.send(new GenericMessage("{\"foo\":\"bar\"}".getBytes()));
StepVerifier.create(template.findAll(Document.class, "demo"))
.assertNext(document -> document.containsKey("foo"))
.thenCancel()
.verify();
});
}
@SpringBootApplication
@Import(MongoDbConsumerConfiguration.class)
static class TestApp {
}
}

10
functions/consumer/mongodb-consumer/README.adoc
View File

@@ -4,11 +4,13 @@ A consumer that allows you to insert records into MongoDB.
## Beans for injection
You can import `MongoDbConsumerConfiguration` in the application and then inject the following bean.
You can import `MongoDbConsumerConfiguration` in the application and then inject one of the following beans.
`Function<Message<?>, Mono<Void>> mongodbConsumer`
`Function<Message<?>, Mono<Void>> mongodbConsumerFunction` - Allows you to subscribe.
You can use `mongodbConsumer` as a qualifier when injecting.
`Consumer<Message<?>> mongodbConsumer` - Wraps the function as a Consumer with no-op subscriber.
You can use `mongodbConsumer` or `mongodbConsumerFunction` as a qualifier when injecting.
The return value from the function can be ignored as this is used as a consumer to send records to MongoDB.
@@ -18,7 +20,7 @@ All configuration properties are prefixed with `mongodb.consumer`.
For more information on the various options available, please see link:src/main/java/org/springframework/cloud/fn/consumer/mongo/MongoDBConsumerProperties.java[MongoDBConsumerProperties].
## Tests
## Examples
See this link:src/test/java/org/springframework/cloud/fn/consumer/mongo/MongoDBConsumerApplicationTests.java[test suite] for the various ways, this consumer is used.

8
functions/consumer/mongodb-consumer/src/main/java/org/springframework/cloud/fn/consumer/mongo/MongoDbConsumerConfiguration.java
View File

@@ -16,6 +16,7 @@
package org.springframework.cloud.fn.consumer.mongo;
import java.util.function.Consumer;
import java.util.function.Function;
import reactor.core.publisher.Mono;
@@ -52,7 +53,12 @@ public class MongoDbConsumerConfiguration {
}
@Bean
public Function<Message<?>, Mono<Void>> mongodbConsumer(ReactiveMessageHandler mongoConsumerMessageHandler) {
public Consumer<Message<?>> mongodbConsumer(Function<Message<?>, Mono<Void>> mongodbConsumerFunction) {
return message -> mongodbConsumerFunction.apply(message).subscribe();
}
@Bean
public Function<Message<?>, Mono<Void>> mongodbConsumerFunction(ReactiveMessageHandler mongoConsumerMessageHandler) {
return mongoConsumerMessageHandler::handleMessage;
}

15
functions/consumer/mongodb-consumer/src/test/java/org/springframework/cloud/fn/consumer/mongo/MongoDbConsumerApplicationTests.java
View File

@@ -20,12 +20,11 @@ import java.time.Duration;
import java.util.Comparator;
import java.util.HashMap;
import java.util.Map;
import java.util.function.Function;
import java.util.function.Consumer;
import org.bson.Document;
import org.junit.jupiter.api.Test;
import reactor.core.publisher.Flux;
import reactor.core.publisher.Mono;
import reactor.test.StepVerifier;
import org.springframework.beans.factory.annotation.Autowired;
@@ -42,14 +41,14 @@ import static org.assertj.core.api.Assertions.assertThat;
*/
@SpringBootTest(properties = {
"spring.data.mongodb.port=0",
"mongodb.consumer.collection=testing"})
"mongodb.consumer.collection=testing" })
class MongoDbConsumerApplicationTests {
@Autowired
private MongoDbConsumerProperties properties;
@Autowired
private Function<Message<?>, Mono<Void>> mongoDbConsumer;
private Consumer<Message<?>> mongodbConsumer;
@Autowired
private ReactiveMongoTemplate mongoTemplate;
@@ -66,10 +65,12 @@ class MongoDbConsumerApplicationTests {
Flux<Message<?>> messages = Flux.just(
new GenericMessage<>(data1),
new GenericMessage<>(data2),
new GenericMessage<>("{\"my_data\": \"THE DATA\"}")
);
new GenericMessage<>("{\"my_data\": \"THE DATA\"}"));
messages.flatMap(mongoDbConsumer::apply).blockLast(Duration.ofSeconds(10));
messages.map(message -> {
mongodbConsumer.accept(message);
return message;
}).blockLast(Duration.ofSeconds(10));
StepVerifier.create(this.mongoTemplate.findAll(Document.class, properties.getCollection())
.sort(Comparator.comparing(d -> d.get("_id").toString())))