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

#297 - Move MongoDB example from RxJava1 to RxJava2.

Browse Source
This commit is contained in:
Christoph Strobl
2017-07-25 16:48:33 +02:00
committed by Oliver Gierke
parent b87336ba8f
commit 7a557bfd72
5 changed files with 47 additions and 47 deletions
Download Patch File Download Diff File Expand all files Collapse all files

14
mongodb/reactive/README.md
View File

@@ -27,7 +27,7 @@ Reactive data access reads and converts individual elements while processing the
## Reactive Repository support
Spring Data MongoDB provides reactive repository support with Project Reactor and RxJava 1 reactive types. The reactive API supports reactive type conversion between reactive types.
Spring Data MongoDB provides reactive repository support with Project Reactor and RxJava 2 reactive types. The reactive API supports reactive type conversion between reactive types.
```java
public interface ReactivePersonRepository extends ReactiveCrudRepository<Person, String> {
@@ -48,19 +48,19 @@ public interface ReactivePersonRepository extends ReactiveCrudRepository<Person,
```
```java
public interface RxJava1PersonRepository extends RxJava1CrudRepository<Person, String> {
public interface RxJava2PersonRepository extends RxJava2CrudRepository<Person, String> {
Observable<Person> findByLastname(String lastname);
Flowable<Person> findByLastname(String lastname);
@Query("{ 'firstname': ?0, 'lastname': ?1}")
Single<Person> findByFirstnameAndLastname(String firstname, String lastname);
Maybe<Person> findByFirstnameAndLastname(String firstname, String lastname);
// Accept parameter inside a reactive type for deferred execution
Observable<Person> findByLastname(Single<String> lastname);
Flowable<Person> findByLastname(Single<String> lastname);
Single<Person> findByFirstnameAndLastname(Single<String> firstname, String lastname);
Maybe<Person> findByFirstnameAndLastname(Single<String> firstname, String lastname);
@InfiniteStream // Use a tailable cursor
Observable<Person> findWithTailableCursorBy();
Flowable<Person> findWithTailableCursorBy();
}
```

3
mongodb/reactive/pom.xml
View File

@@ -23,8 +23,9 @@
</dependency>
<dependency>
<groupId>io.reactivex</groupId>
<groupId>io.reactivex.rxjava2</groupId>
<artifactId>rxjava</artifactId>
<version>2.1.1</version>
</dependency>
<dependency>

22
mongodb/reactive/src/main/java/example/springdata/mongodb/people/RxJava1PersonRepository.java → mongodb/reactive/src/main/java/example/springdata/mongodb/people/RxJava2PersonRepository.java
View File

@@ -1,5 +1,5 @@
/*
* Copyright 2016 the original author or authors.
* Copyright 2016-2017 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.
@@ -15,19 +15,21 @@
*/
package example.springdata.mongodb.people;
import rx.Observable;
import rx.Single;
import io.reactivex.Flowable;
import io.reactivex.Maybe;
import io.reactivex.Single;
import org.springframework.data.mongodb.repository.Query;
import org.springframework.data.mongodb.repository.Tailable;
import org.springframework.data.repository.reactive.RxJava1CrudRepository;
import org.springframework.data.repository.reactive.RxJava2CrudRepository;
/**
* Repository interface to manage {@link Person} instances.
*
* @author Mark Paluch
* @author Christoph Strobl
*/
public interface RxJava1PersonRepository extends RxJava1CrudRepository<Person, String> {
public interface RxJava2PersonRepository extends RxJava2CrudRepository<Person, String> {
/**
* Derived query selecting by {@code lastname}.
@@ -35,7 +37,7 @@ public interface RxJava1PersonRepository extends RxJava1CrudRepository<Person, S
* @param lastname
* @return
*/
Observable<Person> findByLastname(String lastname);
Flowable<Person> findByLastname(String lastname);
/**
* String query selecting one entity.
@@ -44,7 +46,7 @@ public interface RxJava1PersonRepository extends RxJava1CrudRepository<Person, S
* @return
*/
@Query("{ 'firstname': ?0, 'lastname': ?1}")
Single<Person> findByFirstnameAndLastname(String firstname, String lastname);
Maybe<Person> findByFirstnameAndLastname(String firstname, String lastname);
/**
* Derived query selecting by {@code lastname}. {@code lastname} uses deferred resolution that does not require
@@ -53,7 +55,7 @@ public interface RxJava1PersonRepository extends RxJava1CrudRepository<Person, S
* @param lastname
* @return
*/
Observable<Person> findByLastname(Single<String> lastname);
Flowable<Person> findByLastname(Single<String> lastname);
/**
* Derived query selecting by {@code firstname} and {@code lastname}. {@code firstname} uses deferred resolution which
@@ -63,7 +65,7 @@ public interface RxJava1PersonRepository extends RxJava1CrudRepository<Person, S
* @param lastname
* @return
*/
Single<Person> findByFirstnameAndLastname(Single<String> firstname, String lastname);
Maybe<Person> findByFirstnameAndLastname(Single<String> firstname, String lastname);
/**
* Use a tailable cursor to emit a stream of entities as new entities are written to the capped collection.
@@ -71,5 +73,5 @@ public interface RxJava1PersonRepository extends RxJava1CrudRepository<Person, S
* @return
*/
@Tailable
Observable<Person> findWithTailableCursorBy();
Flowable<Person> findWithTailableCursorBy();
}

2
mongodb/reactive/src/test/java/example/springdata/mongodb/people/ReactiveMongoTemplateIntegrationTest.java
View File

@@ -89,7 +89,7 @@ public class ReactiveMongoTemplateIntegrationTest {
* Note that the all object conversions are performed before the results are printed to the console.
*/
@Test
public void convertReactorTypesToRxJava1() throws Exception {
public void convertReactorTypesToRxJava2() throws Exception {
Flux<Person> flux = template.find(Query.query(Criteria.where("lastname").is("White")), Person.class);

53
mongodb/reactive/src/test/java/example/springdata/mongodb/people/RxJava1PersonRepositoryIntegrationTest.java → mongodb/reactive/src/test/java/example/springdata/mongodb/people/RxJava2PersonRepositoryIntegrationTest.java
View File

@@ -17,11 +17,11 @@ package example.springdata.mongodb.people;
import static org.assertj.core.api.Assertions.*;
import io.reactivex.Flowable;
import io.reactivex.Single;
import io.reactivex.disposables.Disposable;
import reactor.core.publisher.Flux;
import reactor.core.publisher.Mono;
import rx.Observable;
import rx.Single;
import rx.Subscription;
import java.util.List;
import java.util.concurrent.CountDownLatch;
@@ -36,18 +36,19 @@ import org.springframework.data.mongodb.core.ReactiveMongoOperations;
import org.springframework.test.context.junit4.SpringRunner;
/**
* Integration test for {@link RxJava1PersonRepository} using RxJava1 types. Note that {@link ReactiveMongoOperations}
* Integration test for {@link RxJava2PersonRepository} using RxJava2 types. Note that {@link ReactiveMongoOperations}
* is only available using Project Reactor types as the native Template API implementation does not come in multiple
* reactive flavors.
*
* @author Mark Paluch
* @author Jens Schauder
* @author Christoph Strobl
*/
@RunWith(SpringRunner.class)
@SpringBootTest
public class RxJava1PersonRepositoryIntegrationTest {
public class RxJava2PersonRepositoryIntegrationTest {
@Autowired RxJava1PersonRepository repository;
@Autowired RxJava2PersonRepository repository;
@Autowired ReactiveMongoOperations operations;
@Before
@@ -55,17 +56,16 @@ public class RxJava1PersonRepositoryIntegrationTest {
operations.collectionExists(Person.class) //
.flatMap(exists -> exists ? operations.dropCollection(Person.class) : Mono.just(exists)) //
.flatMap(o -> operations.createCollection(Person.class, CollectionOptions.empty().size(1024 * 1024).maxDocuments(100).capped())) //
.flatMap(o -> operations.createCollection(Person.class,
CollectionOptions.empty().size(1024 * 1024).maxDocuments(100).capped())) //
.then() //
.block();
repository
.saveAll(Observable.just(new Person("Walter", "White", 50), //
new Person("Skyler", "White", 45), //
new Person("Saul", "Goodman", 42), //
new Person("Jesse", "Pinkman", 27))) //
.toBlocking() //
.last();
repository.saveAll(Flowable.just(new Person("Walter", "White", 50), //
new Person("Skyler", "White", 45), //
new Person("Saul", "Goodman", 42), //
new Person("Jesse", "Pinkman", 27))) //
.blockingLast();
}
/**
@@ -76,14 +76,14 @@ public class RxJava1PersonRepositoryIntegrationTest {
CountDownLatch countDownLatch = new CountDownLatch(1);
Observable<Person> people = Observable.just(new Person("Hank", "Schrader", 43), //
Flowable<Person> people = Flowable.just(new Person("Hank", "Schrader", 43), //
new Person("Mike", "Ehrmantraut", 62));
repository.count() //
.doOnSuccess(System.out::println) //
.toObservable() //
.toFlowable() //
.switchMap(count -> repository.saveAll(people)) //
.last() //
.lastElement() //
.toSingle() //
.flatMap(v -> repository.count()) //
.doOnSuccess(System.out::println) //
@@ -104,7 +104,7 @@ public class RxJava1PersonRepositoryIntegrationTest {
repository.findAll() //
.doOnNext(System.out::println) //
.doOnCompleted(countDownLatch::countDown) //
.doOnComplete(countDownLatch::countDown) //
.doOnError(throwable -> countDownLatch.countDown()) //
.subscribe();
@@ -117,9 +117,9 @@ public class RxJava1PersonRepositoryIntegrationTest {
@Test
public void shouldStreamDataWithTailableCursor() throws Exception {
Subscription subscription = repository.findWithTailableCursorBy() //
Disposable subscription = repository.findWithTailableCursorBy() //
.doOnNext(System.out::println) //
.doOnCompleted(() -> System.out.println("Complete")) //
.doOnComplete(() -> System.out.println("Complete")) //
.doOnTerminate(() -> System.out.println("Terminated")) //
.subscribe();
@@ -131,7 +131,7 @@ public class RxJava1PersonRepositoryIntegrationTest {
repository.save(new Person("Mike", "Ehrmantraut", 62)).subscribe();
Thread.sleep(100);
subscription.unsubscribe();
subscription.dispose();
repository.save(new Person("Gus", "Fring", 53)).subscribe();
Thread.sleep(100);
@@ -145,8 +145,7 @@ public class RxJava1PersonRepositoryIntegrationTest {
List<Person> whites = repository.findByLastname("White") //
.toList() //
.toBlocking() //
.last();
.blockingGet();
assertThat(whites).hasSize(2);
}
@@ -158,8 +157,7 @@ public class RxJava1PersonRepositoryIntegrationTest {
public void shouldQueryDataWithStringQuery() {
Person heisenberg = repository.findByFirstnameAndLastname("Walter", "White") //
.toBlocking() //
.value();
.blockingGet();
assertThat(heisenberg).isNotNull();
}
@@ -172,8 +170,7 @@ public class RxJava1PersonRepositoryIntegrationTest {
List<Person> whites = repository.findByLastname(Single.just("White")) //
.toList() //
.toBlocking() //
.single();
.blockingGet();
assertThat(whites).hasSize(2);
}
@@ -185,7 +182,7 @@ public class RxJava1PersonRepositoryIntegrationTest {
public void shouldQueryDataWithMixedDeferredQueryDerivation() {
Person heisenberg = repository.findByFirstnameAndLastname(Single.just("Walter"), "White") //
.toBlocking().value();
.blockingGet();
assertThat(heisenberg).isNotNull();
}