#215 - Add reactive examples for MongoDB, Apache Cassandra and Redis.

cassandra/reactive/src/main/java/example/springdata/cassandra/people/ApplicationConfiguration.java

@@ -0,0 +1,41 @@
+/*
+ * Copyright 2016 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
+ *
+ *      http://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 example.springdata.cassandra.people;
+
+import org.springframework.boot.autoconfigure.SpringBootApplication;
+import org.springframework.data.cassandra.config.SchemaAction;
+import org.springframework.data.cassandra.config.java.AbstractReactiveCassandraConfiguration;
+import org.springframework.data.cassandra.repository.config.EnableReactiveCassandraRepositories;
+
+/**
+ * Simple configuration for reactive Cassandra support.
+ *
+ * @author Mark Paluch
+ */
+@SpringBootApplication
+@EnableReactiveCassandraRepositories
+class ApplicationConfiguration extends AbstractReactiveCassandraConfiguration {
+
+	@Override
+	protected String getKeyspaceName() {
+		return "example";
+	}
+
+	@Override
+	public SchemaAction getSchemaAction() {
+		return SchemaAction.RECREATE;
+	}
+}

cassandra/reactive/src/main/java/example/springdata/cassandra/people/Person.java

@@ -0,0 +1,40 @@
+/*
+ * Copyright 2016 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
+ *
+ *      http://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 example.springdata.cassandra.people;
+
+import lombok.AllArgsConstructor;
+import lombok.Data;
+import lombok.NoArgsConstructor;
+
+import org.springframework.cassandra.core.PrimaryKeyType;
+import org.springframework.data.cassandra.mapping.PrimaryKeyColumn;
+import org.springframework.data.cassandra.mapping.Table;
+
+/**
+ * An entity to represent a Person.
+ *
+ * @author Mark Paluch
+ */
+@Data
+@AllArgsConstructor
+@NoArgsConstructor
+@Table
+public class Person {
+
+	@PrimaryKeyColumn(type = PrimaryKeyType.CLUSTERED, ordinal = 2) private String firstname;
+	@PrimaryKeyColumn(type = PrimaryKeyType.PARTITIONED, ordinal = 1) private String lastname;
+	private int age;
+}

cassandra/reactive/src/main/java/example/springdata/cassandra/people/ReactivePersonRepository.java

@@ -0,0 +1,66 @@
+/*
+ * Copyright 2016 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
+ *
+ *      http://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 example.springdata.cassandra.people;
+
+import reactor.core.publisher.Flux;
+import reactor.core.publisher.Mono;
+
+import org.springframework.data.cassandra.repository.Query;
+import org.springframework.data.repository.reactive.ReactiveCrudRepository;
+
+/**
+ * Repository interface to manage {@link Person} instances.
+ *
+ * @author Mark Paluch
+ */
+public interface ReactivePersonRepository extends ReactiveCrudRepository<Person, String> {
+
+	/**
+	 * Derived query selecting by {@code lastname}.
+	 * 
+	 * @param lastname
+	 * @return
+	 */
+	Flux<Person> findByLastname(String lastname);
+
+	/**
+	 * String query selecting one entity.
+	 * 
+	 * @param lastname
+	 * @return
+	 */
+	@Query("SELECT * FROM person WHERE firstname = ?0 and lastname  = ?1")
+	Mono<Person> findByFirstnameInAndLastname(String firstname, String lastname);
+
+	/**
+	 * Derived query selecting by {@code lastname}. {@code lastname} uses deferred resolution that does not require
+	 * blocking to obtain the parameter value.
+	 *
+	 * @param lastname
+	 * @return
+	 */
+	Flux<Person> findByLastname(Mono<String> lastname);
+
+	/**
+	 * Derived query selecting by {@code firstname} and {@code lastname}. {@code firstname} uses deferred resolution that
+	 * does not require blocking to obtain the parameter value.
+	 *
+	 * @param firstname
+	 * @param lastname
+	 * @return
+	 */
+	Mono<Person> findByFirstnameAndLastname(Mono<String> firstname, String lastname);
+}

cassandra/reactive/src/main/java/example/springdata/cassandra/people/RxJava1PersonRepository.java

@@ -0,0 +1,66 @@
+/*
+ * Copyright 2016 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
+ *
+ *      http://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 example.springdata.cassandra.people;
+
+import rx.Observable;
+import rx.Single;
+
+import org.springframework.data.cassandra.repository.Query;
+import org.springframework.data.repository.reactive.RxJava1CrudRepository;
+
+/**
+ * Repository interface to manage {@link Person} instances.
+ *
+ * @author Mark Paluch
+ */
+public interface RxJava1PersonRepository extends RxJava1CrudRepository<Person, String> {
+
+	/**
+	 * Derived query selecting by {@code lastname}.
+	 *
+	 * @param lastname
+	 * @return
+	 */
+	Observable<Person> findByLastname(String lastname);
+
+	/**
+	 * String query selecting one entity.
+	 *
+	 * @param lastname
+	 * @return
+	 */
+	@Query("SELECT * FROM person WHERE firstname = ?0 and lastname  = ?1")
+	Single<Person> findByFirstnameAndLastname(String firstname, String lastname);
+
+	/**
+	 * Derived query selecting by {@code lastname}. {@code lastname} uses deferred resolution that does not require
+	 * blocking to obtain the parameter value.
+	 *
+	 * @param lastname
+	 * @return
+	 */
+	Observable<Person> findByLastname(Single<String> lastname);
+
+	/**
+	 * Derived query selecting by {@code firstname} and {@code lastname}. {@code firstname} uses deferred resolution that
+	 * does not require blocking to obtain the parameter value.
+	 *
+	 * @param firstname
+	 * @param lastname
+	 * @return
+	 */
+	Single<Person> findByFirstnameAndLastname(Single<String> firstname, String lastname);
+}

cassandra/reactive/src/test/java/example/springdata/cassandra/people/ReactiveCassandraTemplateIntegrationTest.java

@@ -0,0 +1,103 @@
+/*
+ * Copyright 2016 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
+ *
+ *      http://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 example.springdata.cassandra.people;
+
+import static com.datastax.driver.core.querybuilder.QueryBuilder.*;
+import static org.assertj.core.api.Assertions.*;
+
+import example.springdata.cassandra.util.RequiresCassandraKeyspace;
+import reactor.core.publisher.Flux;
+import rx.RxReactiveStreams;
+
+import java.util.concurrent.CountDownLatch;
+
+import org.junit.Before;
+import org.junit.ClassRule;
+import org.junit.Test;
+import org.junit.runner.RunWith;
+import org.springframework.beans.factory.annotation.Autowired;
+import org.springframework.boot.test.context.SpringBootTest;
+import org.springframework.data.cassandra.core.ReactiveCassandraTemplate;
+import org.springframework.test.context.junit4.SpringRunner;
+
+/**
+ * Integration test for {@link ReactiveCassandraTemplate}.
+ *
+ * @author Mark Paluch
+ */
+@RunWith(SpringRunner.class)
+@SpringBootTest
+public class ReactiveCassandraTemplateIntegrationTest {
+
+	@ClassRule public final static RequiresCassandraKeyspace CASSANDRA_KEYSPACE = RequiresCassandraKeyspace.onLocalhost();
+
+	@Autowired ReactiveCassandraTemplate template;
+
+	/**
+	 * Truncate table and insert some rows.
+	 */
+	@Before
+	public void setUp() {
+
+		template.truncate(Person.class) //
+				.thenMany(template.insert(Flux.just(new Person("Walter", "White", 50), //
+						new Person("Skyler", "White", 45), //
+						new Person("Saul", "Goodman", 42), //
+						new Person("Jesse", "Pinkman", 27)))) //
+				.then() //
+				.block();
+	}
+
+	/**
+	 * This sample performs a count, inserts data and performs a count again using reactive operator chaining. It prints
+	 * the two counts ({@code 4} and {@code 6}) to the console.
+	 */
+	@Test
+	public void shouldInsertAndCountData() throws Exception {
+
+		CountDownLatch countDownLatch = new CountDownLatch(1);
+
+		template.count(Person.class) //
+				.doOnNext(System.out::println) //
+				.thenMany(template.insert(Flux.just(new Person("Hank", "Schrader", 43), //
+						new Person("Mike", "Ehrmantraut", 62)))) //
+				.last() //
+				.flatMap(v -> template.count(Person.class)) //
+				.doOnNext(System.out::println) //
+				.doOnComplete(countDownLatch::countDown) //
+				.doOnError(throwable -> countDownLatch.countDown()) //
+				.subscribe();
+
+		countDownLatch.await();
+	}
+
+	/**
+	 * Note that the all object conversions are performed before the results are printed to the console.
+	 */
+	@Test
+	public void convertReactorTypesToRxJava1() throws Exception {
+
+		Flux<Person> flux = template.select(select().from("person").where(eq("lastname", "White")), Person.class);
+
+		long count = RxReactiveStreams.toObservable(flux) //
+				.count() //
+				.toSingle() //
+				.toBlocking() //
+				.value(); //
+
+		assertThat(count).isEqualTo(2);
+	}
+}

cassandra/reactive/src/test/java/example/springdata/cassandra/people/ReactivePersonRepositoryIntegrationTest.java

@@ -0,0 +1,149 @@
+/*
+ * Copyright 2016 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
+ *
+ *      http://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 example.springdata.cassandra.people;
+
+import static org.assertj.core.api.Assertions.*;
+
+import reactor.core.publisher.Flux;
+import reactor.core.publisher.Mono;
+
+import java.util.List;
+import java.util.concurrent.CountDownLatch;
+
+import org.junit.Before;
+import org.junit.Test;
+import org.junit.runner.RunWith;
+import org.springframework.beans.factory.annotation.Autowired;
+import org.springframework.boot.test.context.SpringBootTest;
+import org.springframework.test.context.junit4.SpringRunner;
+
+/**
+ * Integration test for {@link ReactivePersonRepository} using Project Reactor types and operators.
+ *
+ * @author Mark Paluch
+ */
+@RunWith(SpringRunner.class)
+@SpringBootTest
+public class ReactivePersonRepositoryIntegrationTest {
+
+	@Autowired ReactivePersonRepository repository;
+
+	/**
+	 * Clear table and insert some rows.
+	 */
+	@Before
+	public void setUp() {
+
+		repository.deleteAll() //
+				.thenMany(repository.save(Flux.just(new Person("Walter", "White", 50), //
+						new Person("Skyler", "White", 45), //
+						new Person("Saul", "Goodman", 42), //
+						new Person("Jesse", "Pinkman", 27))))
+				.then() //
+				.block();
+	}
+
+	/**
+	 * This sample performs a count, inserts data and performs a count again using reactive operator chaining.
+	 */
+	@Test
+	public void shouldInsertAndCountData() throws Exception {
+
+		CountDownLatch countDownLatch = new CountDownLatch(1);
+
+		repository.count() //
+				.doOnNext(System.out::println) //
+				.thenMany(repository.save(Flux.just(new Person("Hank", "Schrader", 43), //
+						new Person("Mike", "Ehrmantraut", 62)))) //
+				.last() //
+				.flatMap(v -> repository.count()) //
+				.doOnNext(System.out::println) //
+				.doOnComplete(countDownLatch::countDown) //
+				.doOnError(throwable -> countDownLatch.countDown()) //
+				.subscribe();
+
+		countDownLatch.await();
+	}
+
+	/**
+	 * Result set {@link com.datastax.driver.core.Row}s are converted to entities as they are emitted. Reactive pull and
+	 * prefetch define the amount of fetched records.
+	 */
+	@Test
+	public void shouldPerformConversionBeforeResultProcessing() throws Exception {
+
+		CountDownLatch countDownLatch = new CountDownLatch(1);
+
+		repository.findAll() //
+				.doOnNext(System.out::println) //
+				.doOnComplete(countDownLatch::countDown) //
+				.doOnError(throwable -> countDownLatch.countDown()) //
+				.subscribe();
+
+		countDownLatch.await();
+	}
+
+	/**
+	 * Fetch data using query derivation.
+	 */
+	@Test
+	public void shouldQueryDataWithQueryDerivation() {
+
+		List<Person> whites = repository.findByLastname("White") //
+				.collectList() //
+				.block();
+
+		assertThat(whites).hasSize(2);
+	}
+
+	/**
+	 * Fetch data using a string query.
+	 */
+	@Test
+	public void shouldQueryDataWithStringQuery() {
+
+		Person heisenberg = repository.findByFirstnameInAndLastname("Walter", "White") //
+				.block();
+
+		assertThat(heisenberg).isNotNull();
+	}
+
+	/**
+	 * Fetch data using query derivation.
+	 */
+	@Test
+	public void shouldQueryDataWithDeferredQueryDerivation() {
+
+		List<Person> whites = repository.findByLastname(Mono.just("White")) //
+				.collectList() //
+				.block();
+
+		assertThat(whites).hasSize(2);
+	}
+
+	/**
+	 * Fetch data using query derivation and deferred parameter resolution.
+	 */
+	@Test
+	public void shouldQueryDataWithMixedDeferredQueryDerivation() {
+
+		Person heisenberg = repository.findByFirstnameAndLastname(Mono.just("Walter"), "White") //
+				.block();
+
+		assertThat(heisenberg).isNotNull();
+	}
+
+}

cassandra/reactive/src/test/java/example/springdata/cassandra/people/RxJava1PersonRepositoryIntegrationTest.java

@@ -0,0 +1,160 @@
+/*
+ * Copyright 2016 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
+ *
+ *      http://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 example.springdata.cassandra.people;
+
+import static org.assertj.core.api.Assertions.*;
+
+import example.springdata.cassandra.util.RequiresCassandraKeyspace;
+import rx.Completable;
+import rx.Observable;
+import rx.Single;
+
+import java.util.List;
+import java.util.concurrent.CountDownLatch;
+
+import org.junit.Before;
+import org.junit.ClassRule;
+import org.junit.Test;
+import org.junit.runner.RunWith;
+import org.springframework.beans.factory.annotation.Autowired;
+import org.springframework.boot.test.context.SpringBootTest;
+import org.springframework.data.cassandra.core.ReactiveCassandraOperations;
+import org.springframework.test.context.junit4.SpringRunner;
+
+/**
+ * Integration test for {@link RxJava1PersonRepository} using RxJava1 types. Note that
+ * {@link ReactiveCassandraOperations} is only available using Project Reactor types as the native Template API
+ * implementation does not come in multiple reactive flavors.
+ *
+ * @author Mark Paluch
+ */
+@RunWith(SpringRunner.class)
+@SpringBootTest
+public class RxJava1PersonRepositoryIntegrationTest {
+
+	@ClassRule public final static RequiresCassandraKeyspace CASSANDRA_KEYSPACE = RequiresCassandraKeyspace.onLocalhost();
+
+	@Autowired RxJava1PersonRepository repository;
+	@Autowired ReactiveCassandraOperations operations;
+
+	@Before
+	public void setUp() throws Exception {
+
+		Completable deleteAll = repository.deleteAll();
+
+		Observable<Person> save = repository.save(Observable.just(new Person("Walter", "White", 50), //
+				new Person("Skyler", "White", 45), //
+				new Person("Saul", "Goodman", 42), //
+				new Person("Jesse", "Pinkman", 27)));
+
+		deleteAll.andThen(save).toBlocking().last();
+	}
+
+	/**
+	 * This sample performs a count, inserts data and performs a count again using reactive operator chaining.
+	 */
+	@Test
+	public void shouldInsertAndCountData() throws Exception {
+
+		CountDownLatch countDownLatch = new CountDownLatch(1);
+
+		repository.count() //
+				.doOnSuccess(System.out::println) //
+				.toObservable() //
+				.switchMap(count -> repository.save(Observable.just(new Person("Hank", "Schrader", 43), //
+						new Person("Mike", "Ehrmantraut", 62)))) //
+				.last() //
+				.toSingle() //
+				.flatMap(v -> repository.count()) //
+				.doOnSuccess(System.out::println) //
+				.doAfterTerminate(countDownLatch::countDown) //
+				.doOnError(throwable -> countDownLatch.countDown()) //
+				.subscribe();
+
+		countDownLatch.await();
+	}
+
+	/**
+	 * Result set {@link com.datastax.driver.core.Row}s are converted to entities as they are emitted. Reactive pull and
+	 * prefetch define the amount of fetched records.
+	 */
+	@Test
+	public void shouldPerformConversionBeforeResultProcessing() throws Exception {
+
+		CountDownLatch countDownLatch = new CountDownLatch(1);
+
+		repository.findAll() //
+				.doOnNext(System.out::println) //
+				.doOnCompleted(countDownLatch::countDown) //
+				.doOnError(throwable -> countDownLatch.countDown()) //
+				.subscribe();
+
+		countDownLatch.await();
+	}
+
+	/**
+	 * Fetch data using query derivation.
+	 */
+	@Test
+	public void shouldQueryDataWithQueryDerivation() {
+
+		List<Person> whites = repository.findByLastname("White") //
+				.toList() //
+				.toBlocking() //
+				.last();
+
+		assertThat(whites).hasSize(2);
+	}
+
+	/**
+	 * Fetch data using a string query.
+	 */
+	@Test
+	public void shouldQueryDataWithStringQuery() {
+
+		Person heisenberg = repository.findByFirstnameAndLastname("Walter", "White") //
+				.toBlocking() //
+				.value();
+
+		assertThat(heisenberg).isNotNull();
+	}
+
+	/**
+	 * Fetch data using query derivation.
+	 */
+	@Test
+	public void shouldQueryDataWithDeferredQueryDerivation() {
+
+		List<Person> whites = repository.findByLastname(Single.just("White")) //
+				.toList() //
+				.toBlocking() //
+				.single();
+
+		assertThat(whites).hasSize(2);
+	}
+
+	/**
+	 * Fetch data using query derivation and deferred parameter resolution.
+	 */
+	@Test
+	public void shouldQueryDataWithMixedDeferredQueryDerivation() {
+
+		Person heisenberg = repository.findByFirstnameAndLastname(Single.just("Walter"), "White") //
+				.toBlocking().value();
+
+		assertThat(heisenberg).isNotNull();
+	}
+}

cassandra/reactive/src/test/java/example/springdata/cassandra/people/package-info.java

@@ -0,0 +1,5 @@
+/**
+ * Package showing usage of Spring Data Cassandra Reactive Repositories and reactive Cassandra template.
+ */
+package example.springdata.cassandra.people;
+

cassandra/reactive/src/test/resources/application.properties

@@ -0,0 +1,2 @@
+logging.level.org.springframework.data.cassandra=INFO
+