Spring/spring-data-examples
1
0
Fork 0
Code Issues Packages Projects Releases Wiki Activity

Switch MongoDB samples to records.

Browse Source 
See #606.
This commit is contained in:
Mark Paluch
2021-04-28 15:18:59 +02:00
parent b44dc3dea1
commit e565f33872
69 changed files with 395 additions and 555 deletions
Download Patch File Download Diff File Expand all files Collapse all files

2
mongodb/change-streams/src/main/java/example/springdata/mongodb/CollectingMessageListener.java
View File

@@ -55,7 +55,7 @@ class CollectingMessageListener<S, T> implements MessageListener<S, T> {
void awaitNextMessages(int count) throws InterruptedException {
for (int i = 0; i < count; i++) {
for (var i = 0; i < count; i++) {
messages.take();
}
}

22
mongodb/change-streams/src/main/java/example/springdata/mongodb/Person.java
View File

@@ -15,9 +15,8 @@
*/
package example.springdata.mongodb;
import lombok.Data;
import org.bson.types.ObjectId;
import org.springframework.data.annotation.Id;
import org.springframework.data.mongodb.core.mapping.Field;
@@ -29,21 +28,8 @@ import org.springframework.data.mongodb.core.mapping.Field;
*
* @author Christoph Strobl
*/
@Data
class Person {
@Id ObjectId id;
@Field("first_name") private String firstname;
@Field("last_name") private String lastname;
private int age;
Person(String firstname, String lastname, int age) {
this.firstname = firstname;
this.lastname = lastname;
this.age = age;
record Person(@Id ObjectId id, @Field("first_name") String firstname, @Field("last_name") String lastname, int age) {
public Person(String firstname, String lastname, int age) {
this(null, firstname, lastname, age);
}
}

18
mongodb/change-streams/src/test/java/example/springdata/mongodb/ChangeStreamsTests.java
View File

@@ -21,9 +21,7 @@ import static org.springframework.data.mongodb.core.query.Criteria.*;
import static org.springframework.data.mongodb.core.query.Query.*;
import static org.springframework.data.mongodb.core.query.Update.*;
import com.mongodb.client.MongoClient;
import example.springdata.mongodb.util.EmbeddedMongo;
import reactor.core.publisher.Flux;
import reactor.test.StepVerifier;
import java.time.Duration;
@@ -32,13 +30,13 @@ import org.bson.Document;
import org.junit.ClassRule;
import org.junit.Test;
import org.junit.runner.RunWith;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.boot.autoconfigure.SpringBootApplication;
import org.springframework.boot.autoconfigure.mongo.embedded.EmbeddedMongoAutoConfiguration;
import org.springframework.boot.test.autoconfigure.data.mongo.DataMongoTest;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import org.springframework.data.mongodb.core.ChangeStreamEvent;
import org.springframework.data.mongodb.core.ChangeStreamOptions;
import org.springframework.data.mongodb.core.MongoTemplate;
import org.springframework.data.mongodb.core.ReactiveMongoOperations;
@@ -48,9 +46,9 @@ import org.springframework.data.mongodb.core.messaging.ChangeStreamRequest;
import org.springframework.data.mongodb.core.messaging.DefaultMessageListenerContainer;
import org.springframework.data.mongodb.core.messaging.Message;
import org.springframework.data.mongodb.core.messaging.MessageListenerContainer;
import org.springframework.data.mongodb.core.messaging.Subscription;
import org.springframework.test.context.junit4.SpringRunner;
import com.mongodb.client.MongoClient;
import com.mongodb.client.model.changestream.ChangeStreamDocument;
import com.mongodb.reactivestreams.client.MongoClients;
@@ -150,14 +148,14 @@ public class ChangeStreamsTests {
@Test
public void imperativeChangeEvents() throws InterruptedException {
CollectingMessageListener<ChangeStreamDocument<Document>, Person> messageListener = new CollectingMessageListener<>();
var messageListener = new CollectingMessageListener<ChangeStreamDocument<Document>, Person>();
ChangeStreamRequest<Person> request = ChangeStreamRequest.builder(messageListener) //
var request = ChangeStreamRequest.builder(messageListener) //
.collection("person") //
.filter(newAggregation(match(where("operationType").is("insert")))) // we are only interested in inserts
.build();
Subscription subscription = container.register(request, Person.class);
var subscription = container.register(request, Person.class);
subscription.await(Duration.ofMillis(200)); // wait till the subscription becomes active
template.save(gabriel);
@@ -168,7 +166,7 @@ public class ChangeStreamsTests {
assertThat(messageListener.messageCount()).isEqualTo(2); // first two insert events, so far so good
template.update(Person.class) //
.matching(query(where("id").is(ash.getId()))) //
.matching(query(where("id").is(ash.id()))) //
.apply(update("age", 40)) //
.first();
@@ -189,7 +187,7 @@ public class ChangeStreamsTests {
@Test
public void reactiveChangeEvents() {
Flux<ChangeStreamEvent<Person>> changeStream = reactiveTemplate.changeStream("person",
var changeStream = reactiveTemplate.changeStream("person",
ChangeStreamOptions.builder().filter(newAggregation(match(where("operationType").is("insert")))).build(),
Person.class);
@@ -207,7 +205,7 @@ public class ChangeStreamsTests {
.then(() -> {
StepVerifier.create(reactiveTemplate.update(Person.class) //
.matching(query(where("id").is(ash.getId()))) //
.matching(query(where("id").is(ash.id()))) //
.apply(update("age", 40)) //
.first()).expectNextCount(1).verifyComplete();
}).expectNoEvent(Duration.ofMillis(200)) // updates are skipped