SpringCloud/spring-functions-catalog
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 2a5490068d
commit 5d1786bd99
3 changed files with 21 additions and 12 deletions
Download Patch File Download Diff File Expand all files Collapse all files

10
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
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
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())))