diff --git a/couchbase/pom.xml b/couchbase/pom.xml index 8e218002..af9490ed 100644 --- a/couchbase/pom.xml +++ b/couchbase/pom.xml @@ -17,6 +17,7 @@ example + reactive util diff --git a/couchbase/reactive/README.md b/couchbase/reactive/README.md new file mode 100644 index 00000000..299b26d7 --- /dev/null +++ b/couchbase/reactive/README.md @@ -0,0 +1,51 @@ +# Spring Data Couchbase 3.0 - Reactive examples + +This project contains samples of reactive data access features with Spring Data (Couchbase). + +## Reactive Template API usage with `RxJavaCouchbaseOperations` + +The main reactive Template API class is `RxJavaCouchbaseTemplate`, ideally used through its interface `RxJavaCouchbaseOperations`. It defines a basic set of reactive data access operations using [RxJava 1](https://github.com/ReactiveX/RxJava/tree/1.x) `Single` and `Observable` reactive types. + +```java +Airline airline = new Airline(); + +Observable single = operations.save(airline) + +Observable airlines = operations.findByView(ViewQuery.from("airlines", "all"), Airline.class); +``` + +The test cases in `RxJavaCouchbaseOperationsIntegrationTests` show basic Template API usage. +Reactive data access reads and converts individual elements while processing the stream. + + +## Reactive Repository support + +Spring Data Couchbase provides reactive repository support with Project Reactor, RxJava 1 and RxJava 2 reactive types. The reactive API supports reactive type conversion between reactive types. + +```java +@N1qlPrimaryIndexed +@ViewIndexed(designDoc = "airlines") +public interface ReactiveAirlineRepository extends ReactiveCrudRepository { + + Mono findAirlineByIataCode(String code); + + @View(designDocument = "airlines", viewName = "all") + Flux findAllBy(); +} +``` + +```java +@N1qlPrimaryIndexed +@ViewIndexed(designDoc = "airlines") +public interface RxJava1AirlineRepository extends Repository { + + Single findAirlineByIataCode(String code); + + @View(designDocument = "airlines", viewName = "all") + Observable findAllBy(); + + Single findById(String id); + + Single save(Airline airline); +} +``` diff --git a/couchbase/reactive/pom.xml b/couchbase/reactive/pom.xml new file mode 100644 index 00000000..6911b08d --- /dev/null +++ b/couchbase/reactive/pom.xml @@ -0,0 +1,46 @@ + + 4.0.0 + + + org.springframework.data.examples + spring-data-couchbase-examples + 2.0.0.BUILD-SNAPSHOT + + + spring-data-couchbase-reactive + Spring Data Couchbase - Reactive features + + + + + org.springframework.boot + spring-boot-starter-data-couchbase + + + + io.projectreactor + reactor-core + + + + io.reactivex + rxjava-reactive-streams + + + + ${project.groupId} + spring-data-couchbase-example-utils + ${project.version} + test + + + + io.projectreactor + reactor-test + test + + + + + diff --git a/couchbase/reactive/src/main/java/example/springdata/couchbase/CouchbaseConfiguration.java b/couchbase/reactive/src/main/java/example/springdata/couchbase/CouchbaseConfiguration.java new file mode 100644 index 00000000..8ca84fa9 --- /dev/null +++ b/couchbase/reactive/src/main/java/example/springdata/couchbase/CouchbaseConfiguration.java @@ -0,0 +1,80 @@ +/* + * Copyright 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. + * 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.couchbase; + +import example.springdata.couchbase.model.Airline; +import lombok.RequiredArgsConstructor; + +import java.util.List; + +import javax.annotation.PostConstruct; + +import org.springframework.beans.factory.ObjectProvider; +import org.springframework.boot.autoconfigure.SpringBootApplication; +import org.springframework.boot.autoconfigure.data.couchbase.CouchbaseRepositoriesAutoConfiguration; +import org.springframework.context.annotation.Bean; +import org.springframework.data.couchbase.config.AbstractReactiveCouchbaseDataConfiguration; +import org.springframework.data.couchbase.config.BeanNames; +import org.springframework.data.couchbase.config.CouchbaseConfigurer; +import org.springframework.data.couchbase.core.CouchbaseOperations; +import org.springframework.data.couchbase.repository.config.EnableReactiveCouchbaseRepositories; +import org.springframework.data.couchbase.repository.support.IndexManager; + +import com.couchbase.client.java.query.N1qlQuery; + +/** + * Configuration class to configure reactive repositories. + * + * @author Mark Paluch + */ +@SpringBootApplication(exclude = CouchbaseRepositoriesAutoConfiguration.class) // see DATACOUCH-350 +@RequiredArgsConstructor +@EnableReactiveCouchbaseRepositories +public class CouchbaseConfiguration extends AbstractReactiveCouchbaseDataConfiguration { + + private final CouchbaseConfigurer couchbaseConfigurer; + private final ObjectProvider couchbaseOperationsProvider; + + @Override + protected CouchbaseConfigurer couchbaseConfigurer() { + return couchbaseConfigurer; + } + + /** + * Create an {@link IndexManager} that allows index creation. + * + * @return + */ + @Bean(name = BeanNames.COUCHBASE_INDEX_MANAGER) + public IndexManager indexManager() { + return new IndexManager(true, true, false); + } + + @PostConstruct + private void postConstruct() { + + // Need to post-process travel data to add _class attribute + + CouchbaseOperations couchbaseOperations = couchbaseOperationsProvider.getIfUnique(); + List airlinesWithoutClassAttribute = couchbaseOperations.findByN1QL(N1qlQuery.simple( // + "SELECT META(`travel-sample`).id AS _ID, META(`travel-sample`).cas AS _CAS, `travel-sample`.* " + // + "FROM `travel-sample` " + // + "WHERE type = \"airline\" AND _class IS MISSING;"), + Airline.class); + + airlinesWithoutClassAttribute.forEach(couchbaseOperations::save); + } +} diff --git a/couchbase/reactive/src/main/java/example/springdata/couchbase/model/Airline.java b/couchbase/reactive/src/main/java/example/springdata/couchbase/model/Airline.java new file mode 100644 index 00000000..176c2b95 --- /dev/null +++ b/couchbase/reactive/src/main/java/example/springdata/couchbase/model/Airline.java @@ -0,0 +1,48 @@ +/* + * Copyright 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. + * 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.couchbase.model; + +import lombok.Data; + +import org.springframework.data.couchbase.core.mapping.Document; + +import com.couchbase.client.java.repository.annotation.Field; +import com.couchbase.client.java.repository.annotation.Id; + +/** + * A domain object representing an Airline + * + * @author Chandana Kithalagama + * @author Mark Paluch + */ +@Data +@Document +public class Airline { + + @Id private String id; + + @Field private String type; + + @Field private String name; + + @Field("iata") private String iataCode; + + @Field private String icao; + + @Field private String callsign; + + @Field private String country; +} diff --git a/couchbase/reactive/src/main/java/example/springdata/couchbase/repository/ReactiveAirlineRepository.java b/couchbase/reactive/src/main/java/example/springdata/couchbase/repository/ReactiveAirlineRepository.java new file mode 100644 index 00000000..b4268ed9 --- /dev/null +++ b/couchbase/reactive/src/main/java/example/springdata/couchbase/repository/ReactiveAirlineRepository.java @@ -0,0 +1,51 @@ +/* + * Copyright 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. + * 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.couchbase.repository; + +import example.springdata.couchbase.model.Airline; +import reactor.core.publisher.Flux; +import reactor.core.publisher.Mono; + +import org.springframework.data.couchbase.core.query.N1qlPrimaryIndexed; +import org.springframework.data.couchbase.core.query.View; +import org.springframework.data.couchbase.core.query.ViewIndexed; +import org.springframework.data.repository.reactive.ReactiveCrudRepository; + +/** + * Repository interface to manage {@link Airline} instances. + * + * @author Mark Paluch + */ +@N1qlPrimaryIndexed +@ViewIndexed(designDoc = "airlines") +public interface ReactiveAirlineRepository extends ReactiveCrudRepository { + + /** + * Derived query selecting by {@code iataCode}. + * + * @param code + * @return + */ + Mono findAirlineByIataCode(String code); + + /** + * Query method using {@code airlines/all} view. + * + * @return + */ + @View(designDocument = "airlines", viewName = "all") + Flux findAllBy(); +} diff --git a/couchbase/reactive/src/main/java/example/springdata/couchbase/repository/RxJava1AirlineRepository.java b/couchbase/reactive/src/main/java/example/springdata/couchbase/repository/RxJava1AirlineRepository.java new file mode 100644 index 00000000..e7ee6946 --- /dev/null +++ b/couchbase/reactive/src/main/java/example/springdata/couchbase/repository/RxJava1AirlineRepository.java @@ -0,0 +1,69 @@ +/* + * Copyright 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. + * 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.couchbase.repository; + +import example.springdata.couchbase.model.Airline; +import rx.Observable; +import rx.Single; + +import org.springframework.data.couchbase.core.query.N1qlPrimaryIndexed; +import org.springframework.data.couchbase.core.query.View; +import org.springframework.data.couchbase.core.query.ViewIndexed; +import org.springframework.data.repository.Repository; + +/** + * Repository interface to manage {@link Airline} instances. + * + * @author Mark Paluch + */ +@N1qlPrimaryIndexed +@ViewIndexed(designDoc = "airlines") +public interface RxJava1AirlineRepository extends Repository { + + /** + * Derived query selecting by {@code iataCode}. + * + * @param code + * @return + */ + Single findAirlineByIataCode(String code); + + /** + * Query method using {@code airlines/all} view. + * + * @return + */ + @View(designDocument = "airlines", viewName = "all") + Observable findAllBy(); + + /** + * Overloaded {@link org.springframework.data.repository.reactive.ReactiveCrudRepository#findById(Object)} method + * returning a RxJava 1 {@link Single}. + * + * @param id + * @return + */ + Single findById(String id); + + /** + * Overloaded {@link org.springframework.data.repository.reactive.ReactiveCrudRepository#save(Object)} method + * returning a RxJava 1 {@link Single}. + * + * @param airline + * @return + */ + Single save(Airline airline); +} diff --git a/couchbase/reactive/src/main/resources/application.properties b/couchbase/reactive/src/main/resources/application.properties new file mode 100644 index 00000000..2d12d619 --- /dev/null +++ b/couchbase/reactive/src/main/resources/application.properties @@ -0,0 +1,9 @@ +spring.couchbase.bucket.name=travel-sample +spring.couchbase.bootstrap-hosts=localhost + +# Required for Couchbase 5 +spring.couchbase.bucket.password=password + +# Increased timeout to fit slower environments like TravisCI +spring.couchbase.env.timeouts.view=15000 +spring.couchbase.env.timeouts.query=15000 diff --git a/couchbase/reactive/src/test/java/example/springdata/couchbase/repository/ReactiveAirlineRepositoryIntegrationTests.java b/couchbase/reactive/src/test/java/example/springdata/couchbase/repository/ReactiveAirlineRepositoryIntegrationTests.java new file mode 100644 index 00000000..f405c6e3 --- /dev/null +++ b/couchbase/reactive/src/test/java/example/springdata/couchbase/repository/ReactiveAirlineRepositoryIntegrationTests.java @@ -0,0 +1,122 @@ +/* + * Copyright 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. + * 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.couchbase.repository; + +import static org.assertj.core.api.Assertions.*; + +import example.springdata.couchbase.model.Airline; +import example.springdata.couchbase.util.CouchbaseAvailableRule; +import reactor.core.publisher.Mono; +import reactor.test.StepVerifier; + +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.couchbase.core.CouchbaseOperations; +import org.springframework.test.context.junit4.SpringRunner; + +/** + * Integration tests showing basic CRUD operations through {@link ReactiveAirlineRepository} + * + * @author Mark Paluch + */ +@RunWith(SpringRunner.class) +@SpringBootTest +public class ReactiveAirlineRepositoryIntegrationTests { + + @ClassRule // + public static CouchbaseAvailableRule COUCHBASE = CouchbaseAvailableRule.onLocalhost(); + + @Autowired ReactiveAirlineRepository airlineRepository; + + @Autowired CouchbaseOperations couchbaseOperations; + + @Before + public void before() { + + Airline toDelete = couchbaseOperations.findById("LH", Airline.class); + + if (toDelete != null) { + couchbaseOperations.remove(toDelete); + } + } + + /** + * The derived query executes a N1QL query emitting a single element. + */ + @Test + public void shouldFindAirlineN1ql() { + + StepVerifier.create(airlineRepository.findAirlineByIataCode("TQ")).assertNext(it -> { + + assertThat(it.getCallsign()).isEqualTo("TXW"); + }).verifyComplete(); + } + + /** + * The derived query executes a N1QL query and the emitted element is used to invoke + * {@link org.springframework.data.repository.reactive.ReactiveCrudRepository#findById(Object)} for an Id-based + * lookup. Queries without a result do not emit a value. + */ + @Test + public void shouldFindById() { + + Mono airline = airlineRepository.findAirlineByIataCode("TQ") // + .map(Airline::getId) // + .flatMap(airlineRepository::findById); + + StepVerifier.create(airline).assertNext(it -> { + + assertThat(it.getCallsign()).isEqualTo("TXW"); + }).verifyComplete(); + + StepVerifier.create(airlineRepository.findById("unknown")).verifyComplete(); + } + + /** + * Find all {@link Airline}s applying the {@code airlines/all} view. + */ + @Test + public void shouldFindByView() { + StepVerifier.create(airlineRepository.findAllBy()).expectNextCount(187).verifyComplete(); + } + + /** + * Created elements are emitted by the + * {@link org.springframework.data.repository.reactive.ReactiveCrudRepository#save(Object)} method. + */ + @Test + public void shouldCreateAirline() { + + Airline airline = new Airline(); + + airline.setId("LH"); + airline.setIataCode("LH"); + airline.setIcao("DLH"); + airline.setCallsign("Lufthansa"); + airline.setName("Lufthansa"); + airline.setCountry("Germany"); + + Mono airlineMono = airlineRepository.save(airline) // + .map(Airline::getId) // + .flatMap(airlineRepository::findById); + + StepVerifier.create(airlineMono).expectNext(airline).verifyComplete(); + } +} diff --git a/couchbase/reactive/src/test/java/example/springdata/couchbase/repository/RxJava1AirlineRepositoryIntegrationTests.java b/couchbase/reactive/src/test/java/example/springdata/couchbase/repository/RxJava1AirlineRepositoryIntegrationTests.java new file mode 100644 index 00000000..a9b4749c --- /dev/null +++ b/couchbase/reactive/src/test/java/example/springdata/couchbase/repository/RxJava1AirlineRepositoryIntegrationTests.java @@ -0,0 +1,125 @@ +/* + * Copyright 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. + * 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.couchbase.repository; + +import static org.assertj.core.api.Assertions.*; + +import example.springdata.couchbase.model.Airline; +import example.springdata.couchbase.util.CouchbaseAvailableRule; +import rx.Single; +import rx.observers.AssertableSubscriber; + +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.couchbase.core.CouchbaseOperations; +import org.springframework.test.context.junit4.SpringRunner; + +/** + * Integration tests showing basic CRUD operations through {@link RxJava1AirlineRepository} + * + * @author Mark Paluch + */ +@RunWith(SpringRunner.class) +@SpringBootTest +public class RxJava1AirlineRepositoryIntegrationTests { + + @ClassRule // + public static CouchbaseAvailableRule COUCHBASE = CouchbaseAvailableRule.onLocalhost(); + + @Autowired RxJava1AirlineRepository airlineRepository; + + @Autowired CouchbaseOperations couchbaseOperations; + + @Before + public void before() { + + Airline toDelete = couchbaseOperations.findById("LH", Airline.class); + + if (toDelete != null) { + couchbaseOperations.remove(toDelete); + } + } + + /** + * The derived query executes a N1QL query emitting a single element. + */ + @Test + public void shouldFindAirlineN1ql() { + + AssertableSubscriber subscriber = airlineRepository.findAirlineByIataCode("TQ") // + .test() // + .awaitTerminalEvent() // + .assertCompleted(); + + assertThat(subscriber.getValueCount()).isEqualTo(1); + assertThat(subscriber.getOnNextEvents().get(0).getCallsign()).isEqualTo("TXW"); + } + + /** + * The derived query executes a N1QL query and the emitted element is used to invoke + * {@link org.springframework.data.repository.reactive.ReactiveCrudRepository#findById(Object)} for an Id-based + * lookup. Queries without a result do not emit a value. + */ + @Test + public void shouldFindById() { + + Single airline = airlineRepository.findAirlineByIataCode("TQ") // + .map(Airline::getId) // + .flatMap(airlineRepository::findById); + + AssertableSubscriber subscriber = airline.test().awaitTerminalEvent().assertCompleted(); + + assertThat(subscriber.getValueCount()).isEqualTo(1); + assertThat(subscriber.getOnNextEvents().get(0).getCallsign()).isEqualTo("TXW"); + + airlineRepository.findById("unknown").test().awaitTerminalEvent().assertNoValues(); + } + + /** + * Find all {@link Airline}s applying the {@code airlines/all} view. + */ + @Test + public void shouldFindByView() { + airlineRepository.findAllBy().test().awaitTerminalEvent().assertValueCount(187); + } + + /** + * Created elements are emitted by the + * {@link org.springframework.data.repository.reactive.ReactiveCrudRepository#save(Object)} method. + */ + @Test + public void shouldCreateAirline() { + + Airline airline = new Airline(); + + airline.setId("LH"); + airline.setIataCode("LH"); + airline.setIcao("DLH"); + airline.setCallsign("Lufthansa"); + airline.setName("Lufthansa"); + airline.setCountry("Germany"); + + Single single = airlineRepository.save(airline) // + .map(Airline::getId) // + .flatMap(airlineRepository::findById); + + single.test().awaitTerminalEvent().assertResult(airline); + } +} diff --git a/couchbase/reactive/src/test/java/example/springdata/couchbase/template/RxJavaCouchbaseOperationsIntegrationTests.java b/couchbase/reactive/src/test/java/example/springdata/couchbase/template/RxJavaCouchbaseOperationsIntegrationTests.java new file mode 100644 index 00000000..12acc2c7 --- /dev/null +++ b/couchbase/reactive/src/test/java/example/springdata/couchbase/template/RxJavaCouchbaseOperationsIntegrationTests.java @@ -0,0 +1,108 @@ +/* + * Copyright 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. + * 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.couchbase.template; + +import static org.assertj.core.api.Assertions.*; + +import example.springdata.couchbase.model.Airline; +import example.springdata.couchbase.util.CouchbaseAvailableRule; +import rx.Observable; +import rx.observers.AssertableSubscriber; + +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.couchbase.core.RxJavaCouchbaseOperations; +import org.springframework.test.context.junit4.SpringRunner; + +import com.couchbase.client.java.query.N1qlQuery; +import com.couchbase.client.java.view.ViewQuery; + +/** + * Integration tests showing basic CRUD operations through + * {@link org.springframework.data.couchbase.core.RxJavaCouchbaseOperations}. + * + * @author Mark Paluch + */ +@RunWith(SpringRunner.class) +@SpringBootTest +public class RxJavaCouchbaseOperationsIntegrationTests { + + @ClassRule // + public static CouchbaseAvailableRule COUCHBASE = CouchbaseAvailableRule.onLocalhost(); + + @Autowired RxJavaCouchbaseOperations operations; + + @Before + public void before() { + operations.findById("LH", Airline.class).flatMap(operations::remove).test().awaitTerminalEvent(); + } + + /** + * The derived query executes a N1QL query emitting a single element. + */ + @Test + public void shouldFindAirlineN1ql() { + + String n1ql = "SELECT META(`travel-sample`).id AS _ID, META(`travel-sample`).cas AS _CAS, `travel-sample`.* " + // + "FROM `travel-sample` " + // + "WHERE (`iata` = \"TQ\") AND `_class` = \"example.springdata.couchbase.model.Airline\""; + + AssertableSubscriber subscriber = operations.findByN1QL(N1qlQuery.simple(n1ql), Airline.class) // + .test() // + .awaitTerminalEvent() // + .assertCompleted(); + + assertThat(subscriber.getOnNextEvents()).hasSize(1); + assertThat(subscriber.getOnNextEvents().get(0).getCallsign()).isEqualTo("TXW"); + } + + /** + * Find all {@link Airline}s applying the {@code airlines/all} view. + */ + @Test + public void shouldFindByView() { + + Observable airlines = operations.findByView(ViewQuery.from("airlines", "all"), Airline.class); + + airlines.test().awaitTerminalEvent().assertValueCount(187); + } + + /** + * Created elements are emitted by {@link RxJavaCouchbaseOperations#save(Object)}. + */ + @Test + public void shouldCreateAirline() { + + Airline airline = new Airline(); + + airline.setId("LH"); + airline.setIataCode("LH"); + airline.setIcao("DLH"); + airline.setCallsign("Lufthansa"); + airline.setName("Lufthansa"); + airline.setCountry("Germany"); + + Observable single = operations.save(airline) // + .map(Airline::getId) // + .flatMap(id -> operations.findById(id, Airline.class)); + + single.test().awaitTerminalEvent().assertResult(airline); + } +}