DATAMONGO-1176 - Add support for reactive data access.

[mp911de]

We now support reactive data access with Spring Data MongoDB using MongoDB's ReactiveStreams driver. Blocking and reactive data access require two drivers hence two connections are required that require configuration of each driver with their according MongoTemplate/ReactiveMongoTemplate.

ReactiveMongoTemplate uses Project Reactor wrapper types Mono and Flux to implement ReactiveMongoTemplate and repository support. Reactive template supports common operations such as:

• Index Operation
• Collection retrieval/creation
• Various find methods (by Id, by Query)
• Geospatial queries
• Insert/Save/Update, findAndModify and findAndRemove methods
• Exists and Count projections
• Tailing (Infinite Streams)
• Reactive Callback methods

Not supported:

• Aggregations
• Map/Reduce

Person person = new Person("Dave", 25);

template.insert(person) //
    .then(template.updateFirst(new Query(Criteria.where("age").is(25)), new Update().set("firstName", "Sven"),
            Person.class)) //
    .flatMap(p -> template.find(new Query(Criteria.where("age").is(25)), Person.class))
    .subscribeWith(TestSubscriber.create()) //
    .await() //
    .assertValuesWith(result -> {
        assertThat(result.getFirstName(), is(equalTo("Sven")));
    });

Reactive Repository support is built on top of ReactiveMongoTemplate using ReactiveMongoRepository as the store-specific base repository. Reactive repositories are enabled by using @EnableReactiveMongoRepositories on a @Configuration class to opt-in for reactive support. Reactive repositories can be composed of a reactive base interface such as

• ReactiveCrudRepository
• ReactiveSortingRepository
• RxJava1CrudRepository
• RxJava1SortingRepository

and are identified as reactive repository if one method uses a reactive wrapper type (such as Flux or Observable). If a reactive repository is discovered, it's not implemented by the blocking repository support but with the reactive repository factory. Blocking methods are not (yet) synchronized when using a reactive repository so each repository method must use a reactive wrapper result type. Reactive repository support with Spring Data allows using RxJava1 and Project Reactor types to declare repository methods. Reactive wrapper types are internally converted so the composition library choice on repository level is left up to the user.

Reactive MongoDB repository feature coverage is similar to blocking repository support:

• Query Methods using String queries and Query Derivation
• Geospatial queries (Flux<T> and Flux<GeoResult<T>> without GeoPage/GeoResults support)
• Delete Queries
• JSON based query methods and field restriction
• Full-text queries
• Projections
• Infinite Streams

@Configuration
@EnableReactiveMongoRepositories
class ApplicationConfig extends AbstractReactiveMongoConfiguration {

  @Override
  protected String getDatabaseName() {
    return "e-store";
  }

  @Override
  public MongoClient mongoClient() {
    return MongoClients.create();
  }

  @Override
  protected String getMappingBasePackage() {
    return "com.springdata.mongodb"
  }
}

public interface PersonRepository extends ReactiveSortingRepository<Person, String> {

  Flux<Person> findByFirstname(String firstname);

  Flux<Person> findByFirstname(Publisher<String> firstname);

  Flux<Person> findByFirstnameOrderByLastname(String firstname, Pageable pageable);

  Mono<Person> findByFirstnameAndLastname(String firstname, String lastname);

  Flux<Person> findByLocationNear(Point location, Distance distance);

  Flux<GeoResult<Person>> findByLocationNear(Point location, Distance min, Distance max);

  @InfiniteStream
  Flux<Person> findByFirstname(String firstname);
}

public interface PersonRepository extends RxJava1SortingRepository<Person, String> {

  Observable<Person> findByFirstname(String firstname);

  Observable<Person> findByFirstname(Single<String> firstname);

  Observable<Person> findByFirstnameOrderByLastname(String firstname, Pageable pageable);

  Single<Person> findByFirstnameAndLastname(String firstname, String lastname);

  Observable<Person> findByLocationNear(Point location, Distance distance);

  Observable<GeoResult<Person>> findByLocationNear(Point location, Distance min, Distance max);

  @InfiniteStream
  Observable<Person> findByFirstname(String firstname);
}

---

Related ticket: DATAMONGO-1444
Dependencies:

• DATACMNS-836
• DATACMNS-836 - Add components to support reactive repositories. spring-data-commons#177
• DATAMONGO-1176

[mp911de]

This one should check that T is not a collection/Publisher. Otherwise, it's very easy to pass in objects that are not intended for saving (save(Flux.just(…)).