#265 - Add reactive examples.

Original pull request: #275.

couchbase/pom.xml

@@ -17,6 +17,7 @@
 
 	<modules>
 		<module>example</module>
+		<module>reactive</module>
 		<module>util</module>
 	</modules>
 

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<Airline> single = operations.save(airline)
+
+Observable<Airline> 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<Airline, String> {
+
+    Mono<Airline> findAirlineByIataCode(String code);
+
+    @View(designDocument = "airlines", viewName = "all")
+    Flux<Airline> findAllBy();
+}
+```
+
+```java
+@N1qlPrimaryIndexed
+@ViewIndexed(designDoc = "airlines")
+public interface RxJava1AirlineRepository extends Repository<Airline, String> {
+
+    Single<Airline> findAirlineByIataCode(String code);
+
+    @View(designDocument = "airlines", viewName = "all")
+    Observable<Airline> findAllBy();
+
+    Single<Airline> findById(String id);
+
+    Single<Airline> save(Airline airline);
+}
+```

couchbase/reactive/pom.xml

@@ -0,0 +1,46 @@
+<project xmlns="http://maven.apache.org/POM/4.0.0" xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
+		 xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/xsd/maven-4.0.0.xsd">
+	<modelVersion>4.0.0</modelVersion>
+
+	<parent>
+		<groupId>org.springframework.data.examples</groupId>
+		<artifactId>spring-data-couchbase-examples</artifactId>
+		<version>2.0.0.BUILD-SNAPSHOT</version>
+	</parent>
+
+	<artifactId>spring-data-couchbase-reactive</artifactId>
+	<name>Spring Data Couchbase - Reactive features</name>
+
+	<dependencies>
+
+		<dependency>
+			<groupId>org.springframework.boot</groupId>
+			<artifactId>spring-boot-starter-data-couchbase</artifactId>
+		</dependency>
+
+		<dependency>
+			<groupId>io.projectreactor</groupId>
+			<artifactId>reactor-core</artifactId>
+		</dependency>
+
+		<dependency>
+			<groupId>io.reactivex</groupId>
+			<artifactId>rxjava-reactive-streams</artifactId>
+		</dependency>
+
+		<dependency>
+			<groupId>${project.groupId}</groupId>
+			<artifactId>spring-data-couchbase-example-utils</artifactId>
+			<version>${project.version}</version>
+			<scope>test</scope>
+		</dependency>
+
+		<dependency>
+			<groupId>io.projectreactor</groupId>
+			<artifactId>reactor-test</artifactId>
+			<scope>test</scope>
+		</dependency>
+
+	</dependencies>
+
+</project>

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<CouchbaseOperations> 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<Airline> 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);
+	}
+}

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;
+}

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<Airline, String> {
+
+	/**
+	 * Derived query selecting by {@code iataCode}.
+	 *
+	 * @param code
+	 * @return
+	 */
+	Mono<Airline> findAirlineByIataCode(String code);
+
+	/**
+	 * Query method using {@code airlines/all} view.
+	 *
+	 * @return
+	 */
+	@View(designDocument = "airlines", viewName = "all")
+	Flux<Airline> findAllBy();
+}

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<Airline, String> {
+
+	/**
+	 * Derived query selecting by {@code iataCode}.
+	 *
+	 * @param code
+	 * @return
+	 */
+	Single<Airline> findAirlineByIataCode(String code);
+
+	/**
+	 * Query method using {@code airlines/all} view.
+	 *
+	 * @return
+	 */
+	@View(designDocument = "airlines", viewName = "all")
+	Observable<Airline> findAllBy();
+
+	/**
+	 * Overloaded {@link org.springframework.data.repository.reactive.ReactiveCrudRepository#findById(Object)} method
+	 * returning a RxJava 1 {@link Single}.
+	 *
+	 * @param id
+	 * @return
+	 */
+	Single<Airline> findById(String id);
+
+	/**
+	 * Overloaded {@link org.springframework.data.repository.reactive.ReactiveCrudRepository#save(Object)} method
+	 * returning a RxJava 1 {@link Single}.
+	 *
+	 * @param airline
+	 * @return
+	 */
+	Single<Airline> save(Airline airline);
+}

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

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> 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<Airline> airlineMono = airlineRepository.save(airline) //
+				.map(Airline::getId) //
+				.flatMap(airlineRepository::findById);
+
+		StepVerifier.create(airlineMono).expectNext(airline).verifyComplete();
+	}
+}

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<Airline> 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> airline = airlineRepository.findAirlineByIataCode("TQ") //
+				.map(Airline::getId) //
+				.flatMap(airlineRepository::findById);
+
+		AssertableSubscriber<Airline> 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<Airline> single = airlineRepository.save(airline) //
+				.map(Airline::getId) //
+				.flatMap(airlineRepository::findById);
+
+		single.test().awaitTerminalEvent().assertResult(airline);
+	}
+}

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<Airline> 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<Airline> 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<Airline> single = operations.save(airline) //
+				.map(Airline::getId) //
+				.flatMap(id -> operations.findById(id, Airline.class));
+
+		single.test().awaitTerminalEvent().assertResult(airline);
+	}
+}