> getInitialEntitySet() throws ClassNotFoundException;
+}
diff --git a/src/main/java/org/springframework/data/couchbase/core/CouchbaseOperations.java b/src/main/java/org/springframework/data/couchbase/core/CouchbaseOperations.java
index 89a77dfc..df26a170 100644
--- a/src/main/java/org/springframework/data/couchbase/core/CouchbaseOperations.java
+++ b/src/main/java/org/springframework/data/couchbase/core/CouchbaseOperations.java
@@ -1,5 +1,5 @@
/*
- * Copyright 2012-2015 the original author or authors
+ * Copyright 2012-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.
@@ -37,6 +37,7 @@ import org.springframework.data.couchbase.core.convert.CouchbaseConverter;
import org.springframework.data.couchbase.core.convert.translation.TranslationService;
import org.springframework.data.couchbase.core.query.Consistency;
+
/**
* Defines common operations on the Couchbase data source, most commonly implemented by {@link CouchbaseTemplate}.
*
@@ -45,9 +46,6 @@ import org.springframework.data.couchbase.core.query.Consistency;
*/
public interface CouchbaseOperations {
- String SELECT_ID = "_ID";
- String SELECT_CAS = "_CAS";
-
/**
* Save the given object.
*
@@ -247,10 +245,7 @@ public interface CouchbaseOperations {
/**
* Query the N1QL Service for JSON data of type T. Enough data to construct the full
- * entity is expected to be selected, including the metadata {@value #SELECT_ID} and
- * {@value #SELECT_CAS} (document id and cas, obtained through N1QL's
- * "{@code META(bucket).id AS} {@value #SELECT_ID}" and
- * "{@code META(bucket).cas AS} {@value #SELECT_CAS}").
+ * entity is expected to be selected, including the metadata (document id and cas), obtained through N1QL's query.
* This is done via a {@link N1qlQuery} that contains a {@link Statement} and possibly
* additional query parameters ({@link N1qlParams}) and placeholder values if the
* statement contains placeholders.
diff --git a/src/main/java/org/springframework/data/couchbase/core/CouchbaseTemplate.java b/src/main/java/org/springframework/data/couchbase/core/CouchbaseTemplate.java
index 4e01d1e1..ec2b5578 100644
--- a/src/main/java/org/springframework/data/couchbase/core/CouchbaseTemplate.java
+++ b/src/main/java/org/springframework/data/couchbase/core/CouchbaseTemplate.java
@@ -16,6 +16,7 @@
package org.springframework.data.couchbase.core;
+
import java.util.ArrayList;
import java.util.Collection;
import java.util.Collections;
@@ -77,6 +78,9 @@ import org.springframework.data.mapping.PersistentPropertyAccessor;
import org.springframework.data.mapping.context.MappingContext;
import org.springframework.data.mapping.model.ConvertingPropertyAccessor;
+import static org.springframework.data.couchbase.core.support.TemplateUtils.SELECT_ID;
+import static org.springframework.data.couchbase.core.support.TemplateUtils.SELECT_CAS;
+
/**
* @author Michael Nitschinger
* @author Oliver Gierke
diff --git a/src/main/java/org/springframework/data/couchbase/core/RxJavaCouchbaseOperations.java b/src/main/java/org/springframework/data/couchbase/core/RxJavaCouchbaseOperations.java
new file mode 100644
index 00000000..26545bd0
--- /dev/null
+++ b/src/main/java/org/springframework/data/couchbase/core/RxJavaCouchbaseOperations.java
@@ -0,0 +1,87 @@
+/*
+ * 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 org.springframework.data.couchbase.core;
+
+import java.util.Collection;
+
+import com.couchbase.client.java.Bucket;
+import com.couchbase.client.java.PersistTo;
+import com.couchbase.client.java.ReplicateTo;
+import com.couchbase.client.java.cluster.ClusterInfo;
+import com.couchbase.client.java.query.AsyncN1qlQueryResult;
+import com.couchbase.client.java.query.N1qlQuery;
+import com.couchbase.client.java.view.AsyncSpatialViewResult;
+import com.couchbase.client.java.view.AsyncViewResult;
+import com.couchbase.client.java.view.SpatialViewQuery;
+import com.couchbase.client.java.view.ViewQuery;
+import org.springframework.data.couchbase.core.convert.CouchbaseConverter;
+import org.springframework.data.couchbase.core.query.Consistency;
+import rx.Observable;
+
+/**
+ * @author Subhashni Balakrishnan
+ * @since 3.0
+ */
+public interface RxJavaCouchbaseOperations {
+
+ Observable save(T objectToSave);
+
+ Observable save(Iterable batchToSave);
+
+ Observable save(T objectToSave, PersistTo persistTo, ReplicateTo replicateTo);
+
+ Observable save(Iterable batchToSave, PersistTo persistTo, ReplicateTo replicateTo);
+
+ Observable remove(T objectToRemove);
+
+ Observable remove(T objectToRemove, PersistTo persistTo, ReplicateTo replicateTo);
+
+ Observable remove(Iterable batchToRemove);
+
+ Observable remove(Iterable batchToRemove, PersistTo persistTo, ReplicateTo replicateTo);
+
+ Observable exists(String id);
+
+ Observable findById(String id, Class entityClass);
+
+ Observable queryN1QL(N1qlQuery n1ql);
+
+ Observable queryView(ViewQuery query);
+
+ Observable querySpatialView(SpatialViewQuery query);
+
+ Observable findByView(ViewQuery query, Class entityClass);
+
+ Observable findByN1QL(N1qlQuery n1ql, Class entityClass);
+
+ Observable findBySpatialView(SpatialViewQuery query, Class entityClass);
+
+ Observable findByN1QLProjection(N1qlQuery n1ql, Class fragmentClass);
+
+ Consistency getDefaultConsistency();
+
+ /**
+ * Returns the linked {@link Bucket} to this template.
+ *
+ * @return the client used for the template.
+ */
+ Bucket getCouchbaseBucket();
+
+ CouchbaseConverter getConverter();
+
+ ClusterInfo getCouchbaseClusterInfo();
+
+}
diff --git a/src/main/java/org/springframework/data/couchbase/core/RxJavaCouchbaseTemplate.java b/src/main/java/org/springframework/data/couchbase/core/RxJavaCouchbaseTemplate.java
new file mode 100644
index 00000000..264eec33
--- /dev/null
+++ b/src/main/java/org/springframework/data/couchbase/core/RxJavaCouchbaseTemplate.java
@@ -0,0 +1,366 @@
+/*
+ * 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 org.springframework.data.couchbase.core;
+
+import static org.springframework.data.couchbase.core.CouchbaseTemplate.ensureNotIterable;
+
+import java.util.Collection;
+
+import com.couchbase.client.java.AsyncBucket;
+import com.couchbase.client.java.Bucket;
+import com.couchbase.client.java.PersistTo;
+import com.couchbase.client.java.ReplicateTo;
+import com.couchbase.client.java.cluster.ClusterInfo;
+import com.couchbase.client.java.document.Document;
+import com.couchbase.client.java.document.RawJsonDocument;
+import com.couchbase.client.java.document.json.JsonObject;
+import com.couchbase.client.java.query.*;
+import com.couchbase.client.java.view.*;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+import org.springframework.data.couchbase.core.convert.CouchbaseConverter;
+import org.springframework.data.couchbase.core.convert.MappingCouchbaseConverter;
+import org.springframework.data.couchbase.core.convert.translation.JacksonTranslationService;
+import org.springframework.data.couchbase.core.convert.translation.TranslationService;
+import org.springframework.data.couchbase.core.mapping.*;
+import org.springframework.data.couchbase.core.query.Consistency;
+import org.springframework.data.couchbase.core.support.TemplateUtils;
+import org.springframework.data.mapping.PersistentPropertyAccessor;
+import org.springframework.data.mapping.context.MappingContext;
+import org.springframework.data.mapping.model.ConvertingPropertyAccessor;
+import rx.Observable;
+
+/**
+ * RxJavaCouchbaseTemplate implements operations using rxjava1 observables
+ * @author Subhashni Balakrishnan
+ * @since 3.0
+ */
+public class RxJavaCouchbaseTemplate implements RxJavaCouchbaseOperations {
+ private static final Logger LOGGER = LoggerFactory.getLogger(RxJavaCouchbaseTemplate.class);
+
+ private Bucket syncClient;
+ private AsyncBucket client;
+ private final ClusterInfo clusterInfo;
+ private final CouchbaseConverter converter;
+ private final TranslationService translationService;
+ protected final MappingContext, CouchbasePersistentProperty> mappingContext;
+ private Consistency configuredConsistency = Consistency.DEFAULT_CONSISTENCY;
+
+ private static final WriteResultChecking DEFAULT_WRITE_RESULT_CHECKING = WriteResultChecking.NONE;
+ private WriteResultChecking writeResultChecking = DEFAULT_WRITE_RESULT_CHECKING;
+
+
+ public Observable save(T objectToSave) {
+ return doPersist(objectToSave, PersistTo.NONE, ReplicateTo.NONE);
+ }
+
+ public Observable save(Iterable batchToSave) {
+ return Observable.from(batchToSave)
+ .flatMap(object -> save(object));
+ }
+
+ public Observable save(T objectToSave, PersistTo persistTo, ReplicateTo replicateTo) {
+ return doPersist(objectToSave, persistTo, replicateTo);
+ }
+
+ public Observable save(Iterable batchToSave, PersistTo persistTo, ReplicateTo replicateTo) {
+ return Observable.from(batchToSave)
+ .flatMap(object -> save(object, persistTo, replicateTo));
+ }
+
+ public Observable remove(T objectToRemove) {
+ return doRemove(objectToRemove, PersistTo.NONE, ReplicateTo.NONE);
+ }
+
+ public Observable remove(Iterable batchToRemove) {
+ return Observable.from(batchToRemove)
+ .flatMap(object -> remove(object));
+ }
+
+ public Observable remove(T objectToRemove, PersistTo persistTo, ReplicateTo replicateTo) {
+ return doRemove(objectToRemove, persistTo, replicateTo);
+ }
+
+ public Observable remove(Iterable batchToRemove, PersistTo persistTo, ReplicateTo replicateTo) {
+ return Observable.from(batchToRemove)
+ .flatMap(object -> remove(object, persistTo, replicateTo));
+ }
+
+
+ public RxJavaCouchbaseTemplate(final ClusterInfo clusterInfo, final Bucket client) {
+ this(clusterInfo, client, null, null);
+ }
+
+ public RxJavaCouchbaseTemplate(final ClusterInfo clusterInfo, final Bucket client, final TranslationService translationService) {
+ this(clusterInfo, client, null, translationService);
+ }
+
+
+ public void setWriteResultChecking(WriteResultChecking writeResultChecking) {
+ this.writeResultChecking = writeResultChecking == null ? DEFAULT_WRITE_RESULT_CHECKING : writeResultChecking;
+ }
+
+ public RxJavaCouchbaseTemplate(final ClusterInfo clusterInfo, final Bucket client,
+ final CouchbaseConverter converter,
+ final TranslationService translationService) {
+ this.syncClient = client;
+ this.clusterInfo = clusterInfo;
+ this.client = client.async();
+ this.converter = converter == null ? getDefaultConverter() : converter;
+ this.translationService = translationService == null ? getDefaultTranslationService() : translationService;
+ this.mappingContext = this.converter.getMappingContext();
+ }
+
+ private RawJsonDocument encodeAndWrap(final CouchbaseDocument source, Long version) {
+ String encodedContent = translationService.encode(source);
+ if (version == null) {
+ return RawJsonDocument.create(source.getId(), source.getExpiration(), encodedContent);
+ } else {
+ return RawJsonDocument.create(source.getId(), source.getExpiration(), encodedContent, version);
+ }
+ }
+
+
+ private TranslationService getDefaultTranslationService() {
+ JacksonTranslationService t = new JacksonTranslationService();
+ t.afterPropertiesSet();
+ return t;
+ }
+
+
+ private CouchbaseConverter getDefaultConverter() {
+ MappingCouchbaseConverter c = new MappingCouchbaseConverter(new CouchbaseMappingContext());
+ c.afterPropertiesSet();
+ return c;
+ }
+
+ private final ConvertingPropertyAccessor getPropertyAccessor(Object source) {
+ CouchbasePersistentEntity entity = mappingContext.getPersistentEntity(source.getClass());
+ PersistentPropertyAccessor accessor = entity.getPropertyAccessor(source);
+
+ return new ConvertingPropertyAccessor(accessor, converter.getConversionService());
+ }
+
+
+ private Observable doPersist(T objectToPersist, final PersistTo persistTo, final ReplicateTo replicateTo) {
+ ensureNotIterable(objectToPersist);
+
+ final ConvertingPropertyAccessor accessor = getPropertyAccessor(objectToPersist);
+ final CouchbasePersistentEntity persistentEntity = mappingContext.getPersistentEntity(objectToPersist.getClass());
+ final CouchbasePersistentProperty versionProperty = persistentEntity.getVersionProperty();
+ final Long version = versionProperty != null ? accessor.getProperty(versionProperty, Long.class) : null;
+
+ final CouchbaseDocument converted = new CouchbaseDocument();
+ converter.write(objectToPersist, converted);
+ RawJsonDocument doc = encodeAndWrap(converted, version);
+ return client.upsert(doc, persistTo, replicateTo)
+ .flatMap(rawJsonDocument -> Observable.just(objectToPersist))
+ .doOnError(e -> TemplateUtils.translateError(e));
+ }
+
+ private Observable doRemove(T objectToRemove, final PersistTo persistTo, final ReplicateTo replicateTo) {
+ ensureNotIterable(objectToRemove);
+ if(objectToRemove instanceof String) {
+ return client.remove((String) objectToRemove, persistTo, replicateTo)
+ .flatMap(rawJsonDocument -> Observable.just(objectToRemove))
+ .doOnError(e -> TemplateUtils.translateError(e));
+ } else {
+ final ConvertingPropertyAccessor accessor = getPropertyAccessor(objectToRemove);
+ final CouchbasePersistentEntity persistentEntity = mappingContext.getPersistentEntity(objectToRemove.getClass());
+ final CouchbasePersistentProperty versionProperty = persistentEntity.getVersionProperty();
+ final Long version = versionProperty != null ? accessor.getProperty(versionProperty, Long.class) : null;
+
+ final CouchbaseDocument converted = new CouchbaseDocument();
+ converter.write(objectToRemove, converted);
+ RawJsonDocument doc = encodeAndWrap(converted, version);
+ return client.remove(doc, persistTo, replicateTo)
+ .flatMap(rawJsonDocument -> Observable.just(objectToRemove))
+ .doOnError(e -> TemplateUtils.translateError(e));
+ }
+ }
+
+
+ @Override
+ public Observable exists(String id) {
+ return client.exists(id)
+ .doOnError(e -> TemplateUtils.translateError(e));
+ }
+
+ @Override
+ public Observable queryN1QL(N1qlQuery query) {
+ return client.query(query)
+ .doOnError(e -> TemplateUtils.translateError(e));
+ }
+
+ @Override
+ public Observable queryView(ViewQuery query) {
+ return client.query(query)
+ .doOnError(e -> TemplateUtils.translateError(e));
+ }
+
+ @Override
+ public Observable querySpatialView(SpatialViewQuery query){
+ return client.query(query)
+ .doOnError(e -> TemplateUtils.translateError(e));
+ }
+
+ @Override
+ public Observable findById(String id, Class entityClass) {
+ final CouchbasePersistentEntity entity = mappingContext.getPersistentEntity(entityClass);
+ if (entity.isTouchOnRead()) {
+ return client.getAndTouch(id, entity.getExpiry(), RawJsonDocument.class)
+ .switchIfEmpty(Observable.just(null))
+ .map(doc -> mapToEntity(id, doc, entityClass))
+ .doOnError(e -> TemplateUtils.translateError(e));
+ } else {
+ return client.get(id, RawJsonDocument.class)
+ .switchIfEmpty(Observable.just(null))
+ .map(doc -> mapToEntity(id, doc, entityClass))
+ .doOnError(e -> TemplateUtils.translateError(e));
+ }
+ }
+
+ @Override
+ public Observable findByView(ViewQuery query, Class entityClass) {
+ if (!query.isIncludeDocs() || !query.includeDocsTarget().equals(RawJsonDocument.class)) {
+ if (query.isOrderRetained()) {
+ query.includeDocsOrdered(RawJsonDocument.class);
+ } else {
+ query.includeDocs(RawJsonDocument.class);
+ }
+ }
+ //we'll always map the document to the entity, hence reduce never makes sense.
+ query.reduce(false);
+
+ return queryView(query)
+ .flatMap(asyncViewResult -> asyncViewResult.error()
+ .flatMap(error -> Observable.error(new CouchbaseQueryExecutionException("Unable to execute view query due to error:" + error.toString())))
+ .switchIfEmpty(asyncViewResult.rows()))
+ .map(row -> {
+ AsyncViewRow asyncViewRow = (AsyncViewRow) row;
+ return asyncViewRow.document(RawJsonDocument.class)
+ .map(doc -> mapToEntity(doc.id(), doc, entityClass)).toBlocking().single();
+ })
+ .doOnError(throwable -> Observable.error(new CouchbaseQueryExecutionException("Unable to execute view query", throwable)));
+ }
+
+
+ @Override
+ public Observable findByN1QL(N1qlQuery query, Class entityClass) {
+ return queryN1QL(query)
+ .flatMap(asyncN1qlQueryResult -> asyncN1qlQueryResult.errors()
+ .flatMap(error -> Observable.error(new CouchbaseQueryExecutionException("Unable to execute n1ql query due to error:" + error.toString())))
+ .switchIfEmpty(asyncN1qlQueryResult.rows()))
+ .map(row -> {
+ JsonObject json = ((AsyncN1qlQueryRow)row).value();
+ String id = json.getString(TemplateUtils.SELECT_ID);
+ Long cas = json.getLong(TemplateUtils.SELECT_CAS);
+ if (id == null || cas == null) {
+ throw new CouchbaseQueryExecutionException("Unable to retrieve enough metadata for N1QL to entity mapping, " +
+ "have you selected " + TemplateUtils.SELECT_ID + " and " + TemplateUtils.SELECT_CAS + "?");
+ }
+ json = json.removeKey(TemplateUtils.SELECT_ID).removeKey(TemplateUtils.SELECT_CAS);
+ RawJsonDocument entityDoc = RawJsonDocument.create(id, json.toString(), cas);
+ T decoded = mapToEntity(id, entityDoc, entityClass);
+ return decoded;
+ })
+ .doOnError(throwable -> Observable.error(new CouchbaseQueryExecutionException("Unable to execute n1ql query", throwable)));
+ }
+
+ @Override
+ public Observable findBySpatialView(SpatialViewQuery query, Class entityClass) {
+ return querySpatialView(query)
+ .flatMap(spatialViewResult -> spatialViewResult.error()
+ .flatMap(error -> Observable.error(new CouchbaseQueryExecutionException("Unable to execute spatial view query due to error:" + error.toString())))
+ .switchIfEmpty(spatialViewResult.rows()))
+ .map(row -> {
+ AsyncSpatialViewRow asyncSpatialViewRow = (AsyncSpatialViewRow) row;
+ return asyncSpatialViewRow.document(RawJsonDocument.class)
+ .map(doc -> mapToEntity(doc.id(), doc, entityClass))
+ .toBlocking().single();
+ })
+ .doOnError(throwable -> Observable.error(new CouchbaseQueryExecutionException("Unable to execute spatial view query", throwable)));
+ }
+
+ @Override
+ public Observable findByN1QLProjection(N1qlQuery query, Class entityClass) {
+ return queryN1QL(query)
+ .flatMap(asyncN1qlQueryResult -> asyncN1qlQueryResult.errors()
+ .flatMap(error -> Observable.error(new CouchbaseQueryExecutionException("Unable to execute n1ql query due to error:" + error.toString())))
+ .switchIfEmpty(asyncN1qlQueryResult.rows()))
+ .map(row -> {
+ JsonObject json = ((AsyncN1qlQueryRow)row).value();
+ T decoded = translationService.decodeFragment(json.toString(), entityClass);
+ return decoded;
+ })
+ .doOnError(throwable -> Observable.error(new CouchbaseQueryExecutionException("Unable to execute n1ql query", throwable)));
+ }
+
+
+ @Override
+ public Consistency getDefaultConsistency() {
+ return configuredConsistency;
+ }
+
+
+ public void setDefaultConsistency(Consistency consistency) {
+ this.configuredConsistency = consistency;
+ }
+
+ @Override
+ public CouchbaseConverter getConverter() {
+ return this.converter;
+ }
+
+
+ private T mapToEntity(String id, Document data, Class entityClass) {
+ if (data == null) {
+ return null;
+ }
+
+ final CouchbaseDocument converted = new CouchbaseDocument(id);
+ Object readEntity = converter.read(entityClass, (CouchbaseDocument) decodeAndUnwrap(data, converted));
+
+ final ConvertingPropertyAccessor accessor = getPropertyAccessor(readEntity);
+ CouchbasePersistentEntity persistentEntity = mappingContext.getPersistentEntity(readEntity.getClass());
+ if (persistentEntity.hasVersionProperty()) {
+ accessor.setProperty(persistentEntity.getVersionProperty(), data.cas());
+ }
+
+ return (T) readEntity;
+ }
+
+ /**
+ * Decode a {@link Document Document<String>} containing a JSON string
+ * into a {@link CouchbaseStorable}
+ */
+ private CouchbaseStorable decodeAndUnwrap(final Document source, final CouchbaseStorable target) {
+ //TODO at some point the necessity of CouchbaseStorable should be re-evaluated
+ return translationService.decode(source.content(), target);
+ }
+
+ @Override
+ public Bucket getCouchbaseBucket() {
+ return this.syncClient;
+ }
+
+ @Override
+ public ClusterInfo getCouchbaseClusterInfo() {
+ return this.clusterInfo;
+ }
+
+}
diff --git a/src/main/java/org/springframework/data/couchbase/core/support/TemplateUtils.java b/src/main/java/org/springframework/data/couchbase/core/support/TemplateUtils.java
new file mode 100644
index 00000000..d7dba745
--- /dev/null
+++ b/src/main/java/org/springframework/data/couchbase/core/support/TemplateUtils.java
@@ -0,0 +1,52 @@
+/*
+ * 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 org.springframework.data.couchbase.core.support;
+
+import java.util.concurrent.ExecutionException;
+import java.util.concurrent.TimeoutException;
+import org.springframework.dao.QueryTimeoutException;
+import org.springframework.dao.support.PersistenceExceptionTranslator;
+import org.springframework.data.couchbase.core.CouchbaseExceptionTranslator;
+import org.springframework.data.couchbase.core.OperationInterruptedException;
+import rx.Observable;
+
+/**
+ * @author Subhashni Balakrishnan
+ * @since 3.0
+ */
+public class TemplateUtils {
+ public static final String SELECT_ID = "_ID";
+ public static final String SELECT_CAS = "_CAS";
+ private static PersistenceExceptionTranslator exceptionTranslator = new CouchbaseExceptionTranslator();
+
+
+ public static Observable translateError(Throwable e) {
+ if (e instanceof RuntimeException) {
+ return Observable.error(exceptionTranslator.translateExceptionIfPossible((RuntimeException) e));
+ }
+ else if(e instanceof TimeoutException) {
+ return Observable.error(new QueryTimeoutException(e.getMessage(), e));
+ }
+ else if(e instanceof InterruptedException) {
+ return Observable.error(new OperationInterruptedException(e.getMessage(), e));
+ }
+ else if(e instanceof ExecutionException) {
+ return Observable.error(new OperationInterruptedException(e.getMessage(), e));
+ } else {
+ return Observable.error(e);
+ }
+ }
+}
diff --git a/src/main/java/org/springframework/data/couchbase/repository/ReactiveCouchbaseRepository.java b/src/main/java/org/springframework/data/couchbase/repository/ReactiveCouchbaseRepository.java
new file mode 100644
index 00000000..5481d7a7
--- /dev/null
+++ b/src/main/java/org/springframework/data/couchbase/repository/ReactiveCouchbaseRepository.java
@@ -0,0 +1,32 @@
+/*
+ * 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 org.springframework.data.couchbase.repository;
+
+import java.io.Serializable;
+
+import org.springframework.data.couchbase.core.RxJavaCouchbaseOperations;
+import org.springframework.data.repository.reactive.ReactiveCrudRepository;
+
+/**
+ * @author Subhashni Balakrishnan
+ * @since 3.0
+ */
+public interface ReactiveCouchbaseRepository extends ReactiveCrudRepository {
+ /**
+ * @return a reference to the underlying {@link RxJavaCouchbaseOperations operation template}.
+ */
+ RxJavaCouchbaseOperations getCouchbaseOperations();
+}
diff --git a/src/main/java/org/springframework/data/couchbase/repository/ReactiveCouchbaseSortingRepository.java b/src/main/java/org/springframework/data/couchbase/repository/ReactiveCouchbaseSortingRepository.java
new file mode 100644
index 00000000..b8040276
--- /dev/null
+++ b/src/main/java/org/springframework/data/couchbase/repository/ReactiveCouchbaseSortingRepository.java
@@ -0,0 +1,29 @@
+/*
+ * 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 org.springframework.data.couchbase.repository;
+
+import java.io.Serializable;
+import org.springframework.data.repository.reactive.ReactiveSortingRepository;
+
+/**
+ * Couchbase specific {@link org.springframework.data.repository.Repository} interface that is
+ * a {@link ReactiveSortingRepository}.
+ *
+ * @author Subhashni Balakrishnan
+ */
+public interface ReactiveCouchbaseSortingRepository
+ extends ReactiveCouchbaseRepository, ReactiveSortingRepository {
+}
diff --git a/src/main/java/org/springframework/data/couchbase/repository/cdi/ReactiveCouchbaseRepositoryBean.java b/src/main/java/org/springframework/data/couchbase/repository/cdi/ReactiveCouchbaseRepositoryBean.java
new file mode 100644
index 00000000..ab1981a6
--- /dev/null
+++ b/src/main/java/org/springframework/data/couchbase/repository/cdi/ReactiveCouchbaseRepositoryBean.java
@@ -0,0 +1,75 @@
+/*
+ * Copyright 2014 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 org.springframework.data.couchbase.repository.cdi;
+
+import javax.enterprise.context.spi.CreationalContext;
+import javax.enterprise.inject.spi.Bean;
+import javax.enterprise.inject.spi.BeanManager;
+import java.lang.annotation.Annotation;
+import java.util.Set;
+
+import org.springframework.data.couchbase.core.RxJavaCouchbaseOperations;
+import org.springframework.data.couchbase.repository.config.ReactiveRepositoryOperationsMapping;
+import org.springframework.data.couchbase.repository.support.IndexManager;
+import org.springframework.data.couchbase.repository.support.ReactiveCouchbaseRepositoryFactory;
+import org.springframework.data.repository.cdi.CdiRepositoryBean;
+import org.springframework.data.repository.config.CustomRepositoryImplementationDetector;
+import org.springframework.util.Assert;
+
+/**
+ * A bean which represents a Couchbase repository.
+ * @author Subhashni Balakrishnan
+ */
+public class ReactiveCouchbaseRepositoryBean extends CdiRepositoryBean {
+
+ private final Bean reactiveCouchbaseOperationsBean;
+
+ /**
+ * Creates a new {@link ReactiveCouchbaseRepositoryBean}.
+ *
+ * @param reactiveOperations must not be {@literal null}.
+ * @param qualifiers must not be {@literal null}.
+ * @param repositoryType must not be {@literal null}.
+ * @param beanManager must not be {@literal null}.
+ * @param detector detector for the custom {@link org.springframework.data.repository.Repository} implementations
+ * {@link org.springframework.data.repository.config.CustomRepositoryImplementationDetector}, can be {@literal null}.
+ */
+ public ReactiveCouchbaseRepositoryBean(Bean reactiveOperations, Set qualifiers, Class repositoryType,
+ BeanManager beanManager, CustomRepositoryImplementationDetector detector) {
+ super(qualifiers, repositoryType, beanManager, detector);
+
+ Assert.notNull(reactiveOperations, "Cannot create repository with 'null' for ReactiveCouchbaseOperations.");
+ this.reactiveCouchbaseOperationsBean = reactiveOperations;
+ }
+
+ /*
+ * (non-Javadoc)
+ * @see org.springframework.data.repository.cdi.CdiRepositoryBean#create(javax.enterprise.context.spi.CreationalContext, java.lang.Class, java.lang.Object)
+ */
+ @Override
+ protected T create(CreationalContext creationalContext, Class repositoryType, Object customImplementation) {
+ RxJavaCouchbaseOperations reactiveCouchbaseOperations = getDependencyInstance(reactiveCouchbaseOperationsBean, RxJavaCouchbaseOperations.class);
+ ReactiveRepositoryOperationsMapping reactiveCouchbaseOperationsMapping = new ReactiveRepositoryOperationsMapping(reactiveCouchbaseOperations);
+ IndexManager indexManager = new IndexManager();
+ return new ReactiveCouchbaseRepositoryFactory(reactiveCouchbaseOperationsMapping, indexManager).getRepository(repositoryType, customImplementation);
+ }
+
+ @Override
+ public Class getScope() {
+ return reactiveCouchbaseOperationsBean.getScope();
+ }
+}
diff --git a/src/main/java/org/springframework/data/couchbase/repository/cdi/ReactiveCouchbaseRepositoryExtension.java b/src/main/java/org/springframework/data/couchbase/repository/cdi/ReactiveCouchbaseRepositoryExtension.java
new file mode 100644
index 00000000..29eedfc8
--- /dev/null
+++ b/src/main/java/org/springframework/data/couchbase/repository/cdi/ReactiveCouchbaseRepositoryExtension.java
@@ -0,0 +1,100 @@
+/*
+ * 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 org.springframework.data.couchbase.repository.cdi;
+
+import javax.enterprise.event.Observes;
+import javax.enterprise.inject.UnsatisfiedResolutionException;
+import javax.enterprise.inject.spi.AfterBeanDiscovery;
+import javax.enterprise.inject.spi.Bean;
+import javax.enterprise.inject.spi.BeanManager;
+import javax.enterprise.inject.spi.ProcessBean;
+import java.lang.annotation.Annotation;
+import java.lang.reflect.Type;
+import java.util.HashMap;
+import java.util.Map;
+import java.util.Set;
+
+import org.springframework.data.couchbase.core.CouchbaseOperations;
+import org.springframework.data.couchbase.core.RxJavaCouchbaseOperations;
+import org.springframework.data.repository.cdi.CdiRepositoryBean;
+import org.springframework.data.repository.cdi.CdiRepositoryExtensionSupport;
+
+/**
+ * A portable CDI extension which registers beans for Spring Data Couchbase repositories.
+ * @author Mark Paluch
+ */
+public class ReactiveCouchbaseRepositoryExtension extends CdiRepositoryExtensionSupport{
+
+ private final Map, Bean> reactiveCouchbaseOperationsMap = new HashMap, Bean>();
+
+ /**
+ * Implementation of a an observer which checks for CouchbaseOperations beans and stores them in {@link #reactiveCouchbaseOperationsMap} for
+ * later association with corresponding repository beans.
+ *
+ * @param The type.
+ * @param processBean The annotated type as defined by CDI.
+ */
+ @SuppressWarnings("unchecked")
+ void processBean(@Observes ProcessBean processBean) {
+ Bean bean = processBean.getBean();
+ for (Type type : bean.getTypes()) {
+ if (type instanceof Class && CouchbaseOperations.class.isAssignableFrom((Class) type)) {
+ reactiveCouchbaseOperationsMap.put(bean.getQualifiers(), ((Bean) bean));
+ }
+ }
+ }
+
+ /**
+ * Implementation of a an observer which registers beans to the CDI container for the detected Spring Data
+ * repositories.
+ *
+ * The repository beans are associated to the CouchbaseOperations using their qualifiers.
+ *
+ * @param beanManager The BeanManager instance.
+ */
+ void afterBeanDiscovery(@Observes AfterBeanDiscovery afterBeanDiscovery, BeanManager beanManager) {
+ for (Map.Entry, Set> entry : getRepositoryTypes()) {
+
+ Class repositoryType = entry.getKey();
+ Set qualifiers = entry.getValue();
+
+ CdiRepositoryBean repositoryBean = createRepositoryBean(repositoryType, qualifiers, beanManager);
+ afterBeanDiscovery.addBean(repositoryBean);
+ registerBean(repositoryBean);
+ }
+ }
+
+ /**
+ * Creates a {@link Bean}.
+ *
+ * @param The type of the repository.
+ * @param repositoryType The class representing the repository.
+ * @param beanManager The BeanManager instance.
+ * @return The bean.
+ */
+ private CdiRepositoryBean createRepositoryBean(Class repositoryType, Set qualifiers, BeanManager beanManager) {
+
+ Bean reactiveCouchbaseOperationsBean = this.reactiveCouchbaseOperationsMap.get(qualifiers);
+
+ if (reactiveCouchbaseOperationsBean == null) {
+ throw new UnsatisfiedResolutionException(String.format("Unable to resolve a bean for '%s' with qualifiers %s.",
+ CouchbaseOperations.class.getName(), qualifiers));
+ }
+
+ return new ReactiveCouchbaseRepositoryBean(reactiveCouchbaseOperationsBean, qualifiers, repositoryType, beanManager, getCustomImplementationDetector());
+ }
+}
diff --git a/src/main/java/org/springframework/data/couchbase/repository/config/EnableReactiveCouchbaseRepositories.java b/src/main/java/org/springframework/data/couchbase/repository/config/EnableReactiveCouchbaseRepositories.java
new file mode 100644
index 00000000..1a523001
--- /dev/null
+++ b/src/main/java/org/springframework/data/couchbase/repository/config/EnableReactiveCouchbaseRepositories.java
@@ -0,0 +1,117 @@
+/*
+ * 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 org.springframework.data.couchbase.repository.config;
+
+import java.lang.annotation.*;
+
+import org.springframework.beans.factory.FactoryBean;
+import org.springframework.context.annotation.ComponentScan;
+import org.springframework.context.annotation.Import;
+import org.springframework.data.couchbase.config.BeanNames;
+import org.springframework.data.couchbase.repository.support.ReactiveCouchbaseRepositoryFactoryBean;
+import org.springframework.data.repository.config.DefaultRepositoryBaseClass;
+
+/**
+ * Annotation to activate reactive couchbase repositories. If no base package is configured through either {@link #value()},
+ * {@link #basePackages()} or {@link #basePackageClasses()} it will trigger scanning of the package of annotated class.
+ *
+ * @author Subhashni Balakrishnan
+ * @since 3.0
+ */
+@Target(ElementType.TYPE)
+@Retention(RetentionPolicy.RUNTIME)
+@Documented
+@Inherited
+@Import(ReactiveCouchbaseRepositoriesRegistrar.class)
+public @interface EnableReactiveCouchbaseRepositories {
+
+ /**
+ * Alias for the {@link #basePackages()} attribute. Allows for more concise annotation declarations e.g.:
+ * {@code @EnableCouchbaseRepositories("org.my.pkg")} instead of {@code @EnableCouchbaseRepositories(basePackages="org.my.pkg")}.
+ */
+ String[] value() default {};
+
+ /**
+ * Base packages to scan for annotated components. {@link #value()} is an alias for (and mutually exclusive with) this
+ * attribute. Use {@link #basePackageClasses()} for a type-safe alternative to String-based package names.
+ */
+ String[] basePackages() default {};
+
+ /**
+ * Type-safe alternative to {@link #basePackages()} for specifying the packages to scan for annotated components. The
+ * package of each class specified will be scanned. Consider creating a special no-op marker class or interface in
+ * each package that serves no purpose other than being referenced by this attribute.
+ */
+ Class[] basePackageClasses() default {};
+
+ /**
+ * Specifies which types are eligible for component scanning. Further narrows the set of candidate components from
+ * everything in {@link #basePackages()} to everything in the base packages that matches the given filter or filters.
+ */
+ ComponentScan.Filter[] includeFilters() default {};
+
+ /**
+ * Specifies which types are not eligible for component scanning.
+ */
+ ComponentScan.Filter[] excludeFilters() default {};
+
+ /**
+ * Returns the postfix to be used when looking up custom repository implementations. Defaults to {@literal Impl}. So
+ * for a repository named {@code PersonRepository} the corresponding implementation class will be looked up scanning
+ * for {@code PersonRepositoryImpl}.
+ *
+ * @return
+ */
+ String repositoryImplementationPostfix() default "";
+
+ /**
+ * Configures the location of where to find the Spring Data named queries properties file. Will default to
+ * {@code META-INFO/couchbase-named-queries.properties}.
+ *
+ * @return
+ */
+ String namedQueriesLocation() default "";
+
+ /**
+ * Configure the repository base class to be used to create repository proxies for this particular configuration.
+ *
+ * @return
+ */
+ Class repositoryBaseClass() default DefaultRepositoryBaseClass.class;
+
+
+ /**
+ * Returns the {@link FactoryBean} class to be used for each repository instance. Defaults to
+ * {@link ReactiveCouchbaseRepositoryFactoryBean}.
+ *
+ * @return
+ */
+ Class repositoryFactoryBeanClass() default ReactiveCouchbaseRepositoryFactoryBean.class;
+
+ /**
+ * Configures whether nested repository-interfaces (e.g. defined as inner classes) should be discovered by the
+ * repositories infrastructure.
+ */
+ boolean considerNestedRepositories() default false;
+
+ /**
+ * Configures the name of the {@link ReactiveCouchbaseTemplate} bean to be used by default with the repositories detected.
+ *
+ * @return
+ */
+ String couchbaseTemplateRef() default BeanNames.REACTIVE_COUCHBASE_TEMPLATE;
+
+}
diff --git a/src/main/java/org/springframework/data/couchbase/repository/config/ReactiveCouchbaseRepositoriesRegistrar.java b/src/main/java/org/springframework/data/couchbase/repository/config/ReactiveCouchbaseRepositoriesRegistrar.java
new file mode 100644
index 00000000..6943675e
--- /dev/null
+++ b/src/main/java/org/springframework/data/couchbase/repository/config/ReactiveCouchbaseRepositoriesRegistrar.java
@@ -0,0 +1,46 @@
+/*
+ * 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 org.springframework.data.couchbase.repository.config;
+
+import java.lang.annotation.Annotation;
+
+import org.springframework.data.repository.config.RepositoryBeanDefinitionRegistrarSupport;
+import org.springframework.data.repository.config.RepositoryConfigurationExtension;
+
+/**
+ * @author Subhashni Balakrishnan
+ * @since 3.0
+ */
+public class ReactiveCouchbaseRepositoriesRegistrar extends RepositoryBeanDefinitionRegistrarSupport {
+
+ /*
+ * (non-Javadoc)
+ * @see org.springframework.data.repository.config.RepositoryBeanDefinitionRegistrarSupport#getAnnotation()
+ */
+ @Override
+ protected Class getAnnotation() {
+ return EnableReactiveCouchbaseRepositories.class;
+ }
+
+ /*
+ * (non-Javadoc)
+ * @see org.springframework.data.repository.config.RepositoryBeanDefinitionRegistrarSupport#getExtension()
+ */
+ @Override
+ protected RepositoryConfigurationExtension getExtension() {
+ return new ReactiveCouchbaseRepositoryConfigurationExtension();
+ }
+}
diff --git a/src/main/java/org/springframework/data/couchbase/repository/config/ReactiveCouchbaseRepositoryConfigurationExtension.java b/src/main/java/org/springframework/data/couchbase/repository/config/ReactiveCouchbaseRepositoryConfigurationExtension.java
new file mode 100644
index 00000000..aa9f2db7
--- /dev/null
+++ b/src/main/java/org/springframework/data/couchbase/repository/config/ReactiveCouchbaseRepositoryConfigurationExtension.java
@@ -0,0 +1,64 @@
+/*
+ * 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 org.springframework.data.couchbase.repository.config;
+
+import org.springframework.data.couchbase.repository.support.ReactiveCouchbaseRepositoryFactoryBean;
+import org.springframework.data.repository.config.RepositoryConfigurationExtensionSupport;
+import org.springframework.data.repository.config.XmlRepositoryConfigurationSource;
+import org.w3c.dom.Element;
+
+import org.springframework.beans.factory.support.BeanDefinitionBuilder;
+import org.springframework.data.config.ParsingUtils;
+import org.springframework.data.couchbase.config.BeanNames;
+import org.springframework.data.repository.config.AnnotationRepositoryConfigurationSource;
+
+/**
+ * @author Subhashni Balakrishnan
+ * @since 3.0
+ */
+public class ReactiveCouchbaseRepositoryConfigurationExtension extends RepositoryConfigurationExtensionSupport {
+
+ /** The reference property to use in xml configuration to specify the template to use with a repository. */
+ private static final String REACTIVE_COUCHBASE_TEMPLATE_REF = "reactive-couchbase-template-ref";
+
+ /** The reference property to use in xml configuration to specify the index manager bean to use with a repository. */
+ private static final String COUCHBASE_INDEX_MANAGER_REF = "couchbase-index-manager-ref";
+
+ @Override
+ protected String getModulePrefix() {
+ return "reactive-couchbase";
+ }
+
+ public String getRepositoryFactoryClassName() {
+ return ReactiveCouchbaseRepositoryFactoryBean.class.getName();
+ }
+
+ @Override
+ public void postProcess(final BeanDefinitionBuilder builder, final XmlRepositoryConfigurationSource config) {
+ Element element = config.getElement();
+ ParsingUtils.setPropertyReference(builder, element, REACTIVE_COUCHBASE_TEMPLATE_REF, "reactiveCouchbaseOperations");
+ ParsingUtils.setPropertyReference(builder, element, COUCHBASE_INDEX_MANAGER_REF, "indexManager");
+ }
+
+ @Override
+ public void postProcess(final BeanDefinitionBuilder builder, final AnnotationRepositoryConfigurationSource config) {
+ builder.addDependsOn(BeanNames.REACTIVE_COUCHBASE_OPERATIONS_MAPPING);
+ builder.addDependsOn(BeanNames.COUCHBASE_INDEX_MANAGER);
+ builder.addPropertyReference("couchbaseOperationsMapping", BeanNames.REACTIVE_COUCHBASE_OPERATIONS_MAPPING);
+ builder.addPropertyReference("indexManager", BeanNames.COUCHBASE_INDEX_MANAGER);
+ }
+
+}
diff --git a/src/main/java/org/springframework/data/couchbase/repository/config/ReactiveRepositoryOperationsMapping.java b/src/main/java/org/springframework/data/couchbase/repository/config/ReactiveRepositoryOperationsMapping.java
new file mode 100644
index 00000000..cde49cea
--- /dev/null
+++ b/src/main/java/org/springframework/data/couchbase/repository/config/ReactiveRepositoryOperationsMapping.java
@@ -0,0 +1,126 @@
+/*
+ * 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 org.springframework.data.couchbase.repository.config;
+
+import java.util.HashMap;
+import java.util.Map;
+
+import org.springframework.data.couchbase.core.CouchbaseOperations;
+import org.springframework.data.couchbase.core.RxJavaCouchbaseOperations;
+import org.springframework.data.couchbase.core.mapping.CouchbasePersistentEntity;
+import org.springframework.data.couchbase.core.mapping.CouchbasePersistentProperty;
+import org.springframework.data.mapping.context.MappingContext;
+import org.springframework.util.Assert;
+
+/**
+ * @author Subhashni Balakrishnan
+ * @since 3.0
+ */
+public class ReactiveRepositoryOperationsMapping {
+ private RxJavaCouchbaseOperations defaultOperations;
+ private Map byRepository = new HashMap();
+ private Map byEntity = new HashMap();
+
+ /**
+ * Creates a new mapping, setting the default fallback to use by otherwise non mapped repositories.
+ *
+ * @param defaultOperations the default fallback reactive couchbase operations.
+ */
+ public ReactiveRepositoryOperationsMapping(RxJavaCouchbaseOperations defaultOperations) {
+ Assert.notNull(defaultOperations);
+ this.defaultOperations = defaultOperations;
+ }
+
+ /**
+ * Change the default reactive couchbase operations in an existing mapping.
+ *
+ * @param aDefault the new default couchbase operations.
+ * @return the mapping, for chaining.
+ */
+ public ReactiveRepositoryOperationsMapping setDefault(RxJavaCouchbaseOperations aDefault) {
+ Assert.notNull(aDefault);
+ this.defaultOperations = aDefault;
+ return this;
+ }
+
+ /**
+ * Add a highest priority mapping that will associate a specific repository interface with a given
+ * {@link RxJavaCouchbaseOperations}.
+ *
+ * @param repositoryInterface the repository interface {@link Class}.
+ * @param operations the ReactiveCouchbaseOperations to use.
+ * @return the mapping, for chaining.
+ */
+ public ReactiveRepositoryOperationsMapping map(Class repositoryInterface, RxJavaCouchbaseOperations operations) {
+ byRepository.put(repositoryInterface.getName(), operations);
+ return this;
+ }
+
+ /**
+ * Add a middle priority mapping that will associate any un-mapped repository that deals with the given domain type
+ * Class with a given {@link CouchbaseOperations}.
+ *
+ * @param entityClass the domain type's {@link Class}.
+ * @param operations the CouchbaseOperations to use.
+ * @return the mapping, for chaining.
+ */
+ public ReactiveRepositoryOperationsMapping mapEntity(Class entityClass, RxJavaCouchbaseOperations operations) {
+ byEntity.put(entityClass.getName(), operations);
+ return this;
+ }
+
+ /**
+ * @return the configured default {@link RxJavaCouchbaseOperations}.
+ */
+ public RxJavaCouchbaseOperations getDefault() {
+ return defaultOperations;
+ }
+
+ /**
+ * Get the {@link MappingContext} to use in repositories. It is extracted from the default {@link RxJavaCouchbaseOperations}.
+ *
+ * @return the mapping context.
+ */
+ public MappingContext, CouchbasePersistentProperty> getMappingContext() {
+ return defaultOperations.getConverter().getMappingContext();
+ }
+
+ /**
+ * Given a repository interface and its domain type, resolves which {@link RxJavaCouchbaseOperations} it should be backed with.
+ *
+ * Starts by looking for a direct mapping to the interface, then a common mapping for the domain type, then falls back
+ * to the default CouchbaseOperations.
+ *
+ * @param repositoryInterface the repository's interface.
+ * @param domainType the repository's domain type / entity.
+ * @return the CouchbaseOperations to back the repository.
+ */
+ public RxJavaCouchbaseOperations resolve(Class repositoryInterface, Class domainType) {
+ RxJavaCouchbaseOperations result = byRepository.get(repositoryInterface.getName());
+ if (result != null) {
+ return result;
+ } else {
+ result = byEntity.get(domainType.getName());
+ if (result != null) {
+ return result;
+ } else {
+ return defaultOperations;
+ }
+ }
+ }
+}
+
diff --git a/src/main/java/org/springframework/data/couchbase/repository/query/ReactiveAbstractN1qlBasedQuery.java b/src/main/java/org/springframework/data/couchbase/repository/query/ReactiveAbstractN1qlBasedQuery.java
new file mode 100644
index 00000000..ca5ca868
--- /dev/null
+++ b/src/main/java/org/springframework/data/couchbase/repository/query/ReactiveAbstractN1qlBasedQuery.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 org.springframework.data.couchbase.repository.query;
+
+import java.util.Map;
+import com.couchbase.client.java.document.json.JsonValue;
+import com.couchbase.client.java.query.N1qlQuery;
+import com.couchbase.client.java.query.Statement;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+import org.springframework.data.couchbase.core.RxJavaCouchbaseOperations;
+import org.springframework.data.couchbase.repository.query.support.N1qlUtils;
+import org.springframework.data.repository.query.*;
+import org.springframework.data.repository.util.ReactiveWrapperConverters;
+import reactor.core.publisher.Flux;
+
+/**
+ * @author Subhashni Balakrishnan
+ * @since 3.0
+ */
+public abstract class ReactiveAbstractN1qlBasedQuery implements RepositoryQuery {
+ private static final Logger LOG = LoggerFactory.getLogger(ReactiveAbstractN1qlBasedQuery.class);
+
+ protected final CouchbaseQueryMethod queryMethod;
+ private final RxJavaCouchbaseOperations couchbaseOperations;
+
+ protected ReactiveAbstractN1qlBasedQuery(CouchbaseQueryMethod method, RxJavaCouchbaseOperations operations) {
+ this.queryMethod = method;
+ this.couchbaseOperations = operations;
+ }
+
+ protected abstract Statement getStatement(ParameterAccessor accessor, Object[] runtimeParameters, ReturnedType returnedType);
+
+ protected abstract JsonValue getPlaceholderValues(ParameterAccessor accessor);
+
+ @Override
+ public Object execute(Object[] parameters) {
+ ReactiveCouchbaseParameterAccessor accessor = new ReactiveCouchbaseParameterAccessor(queryMethod, parameters);
+ ResultProcessor processor = this.queryMethod.getResultProcessor().withDynamicProjection(accessor);
+ ReturnedType returnedType = processor.getReturnedType();
+
+ Class typeToRead = returnedType.getTypeToRead();
+ typeToRead = typeToRead == null ? returnedType.getDomainType() : typeToRead;
+
+ Statement statement = getStatement(accessor, parameters, returnedType);
+ JsonValue queryPlaceholderValues = getPlaceholderValues(accessor);
+
+ //prepare the final query
+ N1qlQuery query = N1qlUtils.buildQuery(statement, queryPlaceholderValues,
+ getCouchbaseOperations().getDefaultConsistency().n1qlConsistency());
+ return ReactiveWrapperConverters.toWrapper(
+ processor.processResult(executeDependingOnType(query, queryMethod, typeToRead)), Flux.class);
+ }
+
+
+ protected Object executeDependingOnType(N1qlQuery query,
+ QueryMethod queryMethod,
+ Class typeToRead) {
+
+ if (queryMethod.isModifyingQuery()) {
+ throw new UnsupportedOperationException("Modifying queries not yet supported");
+ }
+
+ if (queryMethod.isQueryForEntity()) {
+ return execute(query, typeToRead);
+ } else {
+ return executeSingleProjection(query);
+ }
+ }
+
+ private void logIfNecessary(N1qlQuery query) {
+ if (LOG.isDebugEnabled()) {
+ LOG.debug("Executing N1QL query: " + query.n1ql());
+ }
+ }
+
+ protected Object execute(N1qlQuery query, Class typeToRead) {
+ logIfNecessary(query);
+ return couchbaseOperations.findByN1QL(query, typeToRead);
+ }
+
+ protected Object executeSingleProjection(N1qlQuery query) {
+ logIfNecessary(query);
+ return couchbaseOperations.findByN1QLProjection(query, Map.class);
+ }
+
+ @Override
+ public CouchbaseQueryMethod getQueryMethod() {
+ return this.queryMethod;
+ }
+
+ protected RxJavaCouchbaseOperations getCouchbaseOperations() {
+ return this.couchbaseOperations;
+ }
+}
diff --git a/src/main/java/org/springframework/data/couchbase/repository/query/ReactiveCouchbaseParameterAccessor.java b/src/main/java/org/springframework/data/couchbase/repository/query/ReactiveCouchbaseParameterAccessor.java
new file mode 100644
index 00000000..a7bb7aab
--- /dev/null
+++ b/src/main/java/org/springframework/data/couchbase/repository/query/ReactiveCouchbaseParameterAccessor.java
@@ -0,0 +1,83 @@
+/*
+ * 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 org.springframework.data.couchbase.repository.query;
+
+
+import java.util.ArrayList;
+import java.util.List;
+
+import org.springframework.data.repository.query.ParametersParameterAccessor;
+import org.springframework.data.repository.util.ReactiveWrapperConverters;
+import org.springframework.data.repository.util.ReactiveWrappers;
+import reactor.core.publisher.Flux;
+import reactor.core.publisher.Mono;
+import reactor.core.publisher.MonoProcessor;
+
+/**
+ * Reactive {@link org.springframework.data.repository.query.ParametersParameterAccessor} implementation that subscribes
+ * to reactive parameter wrapper types upon creation. This class performs synchronization when accessing parameters.
+ *
+ * @author Subhashni Balakrishnan
+ * @since 3.0
+ */
+public class ReactiveCouchbaseParameterAccessor extends ParametersParameterAccessor {
+
+ private final Object[] values;
+ private final List> subscriptions;
+
+ public ReactiveCouchbaseParameterAccessor(CouchbaseQueryMethod method, Object[] values) {
+ super(method.getParameters(), values);
+ this.values = values;
+ this.subscriptions = new ArrayList<>(values.length);
+
+ for (int i = 0; i < values.length; i++) {
+
+ Object value = values[i];
+
+ if (value == null || !ReactiveWrappers.supports(value.getClass())) {
+ subscriptions.add(null);
+ continue;
+ }
+
+ if (ReactiveWrappers.isSingleValueType(value.getClass())) {
+ subscriptions.add(ReactiveWrapperConverters.toWrapper(value, Mono.class).subscribe());
+ } else {
+ subscriptions.add(ReactiveWrapperConverters.toWrapper(value, Flux.class).collectList().subscribe());
+ }
+ }
+ }
+
+ /* (non-Javadoc)
+ * @see org.springframework.data.repository.query.ParametersParameterAccessor#getValue(int)
+ */
+ @SuppressWarnings("unchecked")
+ @Override
+ protected T getValue(int index) {
+
+ if (subscriptions.get(index) != null) {
+ return (T) subscriptions.get(index).block();
+ }
+
+ return super.getValue(index);
+ }
+
+ /* (non-Javadoc)
+ * @see org.springframework.data.repository.query.ParametersParameterAccessor#getBindableValue(int)
+ */
+ public Object getBindableValue(int index) {
+ return getValue(getParameters().getBindableParameter(index).getIndex());
+ }
+}
diff --git a/src/main/java/org/springframework/data/couchbase/repository/query/ReactivePartTreeN1qlBasedQuery.java b/src/main/java/org/springframework/data/couchbase/repository/query/ReactivePartTreeN1qlBasedQuery.java
new file mode 100644
index 00000000..d17380e5
--- /dev/null
+++ b/src/main/java/org/springframework/data/couchbase/repository/query/ReactivePartTreeN1qlBasedQuery.java
@@ -0,0 +1,77 @@
+/*
+ * 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 org.springframework.data.couchbase.repository.query;
+
+import static com.couchbase.client.java.query.Select.select;
+import static com.couchbase.client.java.query.dsl.functions.AggregateFunctions.count;
+
+import com.couchbase.client.java.document.json.JsonArray;
+import com.couchbase.client.java.document.json.JsonValue;
+import com.couchbase.client.java.query.Statement;
+import com.couchbase.client.java.query.dsl.Expression;
+import com.couchbase.client.java.query.dsl.path.FromPath;
+import com.couchbase.client.java.query.dsl.path.LimitPath;
+import com.couchbase.client.java.query.dsl.path.WherePath;
+import org.springframework.data.couchbase.core.RxJavaCouchbaseOperations;
+import org.springframework.data.couchbase.repository.query.support.N1qlUtils;
+import org.springframework.data.repository.query.ParameterAccessor;
+import org.springframework.data.repository.query.RepositoryQuery;
+import org.springframework.data.repository.query.ReturnedType;
+import org.springframework.data.repository.query.parser.PartTree;
+
+/**
+ * A reactive {@link RepositoryQuery} for Couchbase, based on query derivation
+ *
+ * @author Subhashni Balakrishnan
+ * @since 3.0
+ */
+public class ReactivePartTreeN1qlBasedQuery extends ReactiveAbstractN1qlBasedQuery {
+
+ private final PartTree partTree;
+
+ public ReactivePartTreeN1qlBasedQuery(CouchbaseQueryMethod queryMethod, RxJavaCouchbaseOperations operations) {
+ super(queryMethod, operations);
+ this.partTree = new PartTree(queryMethod.getName(), queryMethod.getEntityInformation().getJavaType());
+ }
+
+ @Override
+ protected JsonValue getPlaceholderValues(ParameterAccessor accessor) {
+ return JsonArray.empty();
+ }
+
+ @Override
+ protected Statement getStatement(ParameterAccessor accessor, Object[] runtimeParameters, ReturnedType returnedType) {
+ String bucketName = getCouchbaseOperations().getCouchbaseBucket().name();
+ Expression bucket = N1qlUtils.escapedBucket(bucketName);
+
+ FromPath select;
+ if (partTree.isCountProjection()) {
+ select = select(count("*"));
+ } else {
+ select = N1qlUtils.createSelectClauseForEntity(bucketName, returnedType, this.getCouchbaseOperations().getConverter());
+ }
+ WherePath selectFrom = select.from(bucket);
+
+ N1qlQueryCreator queryCreator = new N1qlQueryCreator(partTree, accessor, selectFrom,
+ getCouchbaseOperations().getConverter(), getQueryMethod());
+ LimitPath selectFromWhereOrderBy = queryCreator.createQuery();
+ if (partTree.isLimiting()) {
+ return selectFromWhereOrderBy.limit(partTree.getMaxResults());
+ } else {
+ return selectFromWhereOrderBy;
+ }
+ }
+}
diff --git a/src/main/java/org/springframework/data/couchbase/repository/query/ReactiveSpatialViewBasedQuery.java b/src/main/java/org/springframework/data/couchbase/repository/query/ReactiveSpatialViewBasedQuery.java
new file mode 100644
index 00000000..840b3278
--- /dev/null
+++ b/src/main/java/org/springframework/data/couchbase/repository/query/ReactiveSpatialViewBasedQuery.java
@@ -0,0 +1,79 @@
+/*
+ * 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 org.springframework.data.couchbase.repository.query;
+
+import com.couchbase.client.java.view.SpatialViewQuery;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+import org.springframework.data.couchbase.core.RxJavaCouchbaseOperations;
+import org.springframework.data.repository.query.RepositoryQuery;
+import org.springframework.data.repository.query.parser.PartTree;
+import org.springframework.data.repository.util.ReactiveWrapperConverters;
+import reactor.core.publisher.Flux;
+
+/**
+ * A reactive {@link RepositoryQuery} for Couchbase, for spatial queries
+ *
+ * @author Subhashni Balakrishnan
+ * @since 3.0
+ */
+public class ReactiveSpatialViewBasedQuery implements RepositoryQuery {
+ private static final Logger LOG = LoggerFactory.getLogger(ReactiveSpatialViewBasedQuery.class);
+
+ private final CouchbaseQueryMethod method;
+ private final RxJavaCouchbaseOperations operations;
+
+ public ReactiveSpatialViewBasedQuery(CouchbaseQueryMethod method, RxJavaCouchbaseOperations operations) {
+ this.method = method;
+ this.operations = operations;
+ }
+
+ @Override
+ public Object execute(Object[] runtimeParams) {
+ String designDoc = method.getDimensionalAnnotation().designDocument();
+ String viewName = method.getDimensionalAnnotation().spatialViewName();
+ int dimensions = method.getDimensionalAnnotation().dimensions();
+ PartTree tree = new PartTree(method.getName(), method.getEntityInformation().getJavaType());
+
+ //prepare a spatial view query to be used as a base for the query creator
+ SpatialViewQuery baseSpatialQuery = SpatialViewQuery.from(designDoc, viewName)
+ .stale(operations.getDefaultConsistency().viewConsistency());
+
+ //use the SpatialViewQueryCreator to complete it
+ SpatialViewQueryCreator creator = new SpatialViewQueryCreator(dimensions,
+ tree, new ReactiveCouchbaseParameterAccessor(method, runtimeParams),
+ baseSpatialQuery, operations.getConverter());
+ SpatialViewQueryCreator.SpatialViewQueryWrapper finalQuery = creator.createQuery();
+
+ //execute the spatial query
+ return execute(finalQuery);
+ }
+
+ protected Object execute(SpatialViewQueryCreator.SpatialViewQueryWrapper query) {
+ if (LOG.isDebugEnabled()) {
+ LOG.debug("Executing spatial view query: " + query.getQuery().toString());
+ }
+
+ //TODO: eliminate false positives in geo query
+ return ReactiveWrapperConverters.toWrapper(operations.findBySpatialView(query.getQuery(),
+ method.getEntityInformation().getJavaType()), Flux.class);
+ }
+
+ @Override
+ public CouchbaseQueryMethod getQueryMethod() {
+ return method;
+ }
+}
diff --git a/src/main/java/org/springframework/data/couchbase/repository/query/ReactiveStringN1qlBasedQuery.java b/src/main/java/org/springframework/data/couchbase/repository/query/ReactiveStringN1qlBasedQuery.java
new file mode 100644
index 00000000..96505b97
--- /dev/null
+++ b/src/main/java/org/springframework/data/couchbase/repository/query/ReactiveStringN1qlBasedQuery.java
@@ -0,0 +1,81 @@
+/*
+ * 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 org.springframework.data.couchbase.repository.query;
+
+import com.couchbase.client.java.document.json.JsonValue;
+import com.couchbase.client.java.query.N1qlQuery;
+import com.couchbase.client.java.query.Statement;
+import org.springframework.data.couchbase.core.RxJavaCouchbaseOperations;
+import org.springframework.expression.EvaluationContext;
+import org.springframework.expression.spel.standard.SpelExpressionParser;
+
+import org.springframework.data.repository.query.EvaluationContextProvider;
+import org.springframework.data.repository.query.ParameterAccessor;
+import org.springframework.data.repository.query.RepositoryQuery;
+import org.springframework.data.repository.query.ReturnedType;
+
+/**
+ * A reactive StringN1qlBasedQuery {@link RepositoryQuery} for Couchbase, based on N1QL and a String statement.
+ *
+ * The statement can contain positional placeholders (eg. name = $1) that will map to the
+ * method's parameters, in the same order.
+ *
+ * The statement can also contain SpEL expressions enclosed in #{ and }.
+ *
+ *
+ * @author Subhashni Balakrishnan
+ * @since 3.0
+ */
+public class ReactiveStringN1qlBasedQuery extends ReactiveAbstractN1qlBasedQuery {
+
+ private final StringBasedN1qlQueryParser queryParser;
+ private final SpelExpressionParser parser;
+ private final EvaluationContextProvider evaluationContextProvider;
+
+ protected String getTypeField() {
+ return getCouchbaseOperations().getConverter().getTypeKey();
+ }
+
+ protected Class getTypeValue() {
+ return getQueryMethod().getEntityInformation().getJavaType();
+ }
+
+ public ReactiveStringN1qlBasedQuery(String statement,
+ CouchbaseQueryMethod queryMethod,
+ RxJavaCouchbaseOperations couchbaseOperations,
+ SpelExpressionParser spelParser,
+ final EvaluationContextProvider evaluationContextProvider) {
+ super(queryMethod, couchbaseOperations);
+
+ this.queryParser = new StringBasedN1qlQueryParser(statement, queryMethod,
+ getCouchbaseOperations().getCouchbaseBucket().name(), getTypeField(), getTypeValue());
+ this.parser = spelParser;
+ this.evaluationContextProvider = evaluationContextProvider;
+ }
+
+ @Override
+ protected JsonValue getPlaceholderValues(ParameterAccessor accessor) {
+ return this.queryParser.getPlaceholderValues(accessor);
+ }
+
+ @Override
+ public Statement getStatement(ParameterAccessor accessor, Object[] runtimeParameters, ReturnedType returnedType) {
+ EvaluationContext evaluationContext = evaluationContextProvider.getEvaluationContext(getQueryMethod().getParameters(), runtimeParameters);
+ String parsedStatement = queryParser.doParse(parser, evaluationContext, false);
+ return N1qlQuery.simple(parsedStatement).statement();
+ }
+
+}
diff --git a/src/main/java/org/springframework/data/couchbase/repository/query/ReactiveViewBasedCouchbaseQuery.java b/src/main/java/org/springframework/data/couchbase/repository/query/ReactiveViewBasedCouchbaseQuery.java
new file mode 100644
index 00000000..9a9dbbb1
--- /dev/null
+++ b/src/main/java/org/springframework/data/couchbase/repository/query/ReactiveViewBasedCouchbaseQuery.java
@@ -0,0 +1,158 @@
+/*
+ * 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 org.springframework.data.couchbase.repository.query;
+
+import com.couchbase.client.java.view.AsyncViewRow;
+import com.couchbase.client.java.view.ViewQuery;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import org.springframework.data.couchbase.core.CouchbaseQueryExecutionException;
+import org.springframework.data.couchbase.core.RxJavaCouchbaseOperations;
+import org.springframework.data.mapping.PropertyReferenceException;
+import org.springframework.data.repository.query.QueryMethod;
+import org.springframework.data.repository.query.RepositoryQuery;
+import org.springframework.data.repository.query.parser.PartTree;
+import org.springframework.data.repository.util.ReactiveWrapperConverters;
+import org.springframework.util.StringUtils;
+import reactor.core.publisher.Flux;
+import rx.Observable;
+
+/**
+ * Execute a reactive repository query through the View mechanism.
+ *
+ * @author Subhashni Balakrishnan
+ * @since 3.0
+ */
+public class ReactiveViewBasedCouchbaseQuery implements RepositoryQuery {
+
+ private static final Logger LOG = LoggerFactory.getLogger(ReactiveViewBasedCouchbaseQuery.class);
+
+ private final CouchbaseQueryMethod method;
+ private final RxJavaCouchbaseOperations operations;
+
+ public ReactiveViewBasedCouchbaseQuery(CouchbaseQueryMethod method, RxJavaCouchbaseOperations operations) {
+ this.method = method;
+ this.operations = operations;
+ }
+
+ @Override
+ public Object execute(Object[] runtimeParams) {
+ if (method.hasViewName()) { //only allow derivation on @View explicitly defining a viewName
+ return deriveAndExecute(runtimeParams);
+ } else {
+ return guessViewAndExecute();
+ }
+ }
+
+ protected Object guessViewAndExecute() {
+ String designDoc = designDocName(method);
+ String methodName = method.getName();
+ boolean isExplicitReduce = method.hasViewAnnotation() && method.getViewAnnotation().reduce();
+ boolean isReduce = methodName.startsWith("count") || isExplicitReduce;
+ String viewName = StringUtils.uncapitalize(methodName.replaceFirst("find|count", ""));
+
+ ViewQuery simpleQuery = ViewQuery.from(designDoc, viewName)
+ .stale(operations.getDefaultConsistency().viewConsistency());
+ if (isReduce) {
+ simpleQuery.reduce();
+ return executeReduce(simpleQuery, designDoc, viewName);
+ } else {
+ return execute(simpleQuery);
+ }
+ }
+
+ protected Object deriveAndExecute(Object[] runtimeParams) {
+ String designDoc = designDocName(method);
+ String viewName = method.getViewAnnotation().viewName();
+
+ //prepare a ViewQuery to be used as a base for the ViewQueryCreator
+ ViewQuery baseQuery = ViewQuery.from(designDoc, viewName)
+ .stale(operations.getDefaultConsistency().viewConsistency());
+
+ try {
+ PartTree tree = new PartTree(method.getName(), method.getEntityInformation().getJavaType());
+
+ //use a ViewQueryCreator to complete the base query
+ ViewQueryCreator creator = new ViewQueryCreator(tree, new ReactiveCouchbaseParameterAccessor(method, runtimeParams),
+ method.getViewAnnotation(), baseQuery, operations.getConverter());
+ ViewQueryCreator.DerivedViewQuery result = creator.createQuery();
+
+ if (result.isReduce) {
+ return executeReduce(result.builtQuery, designDoc, viewName);
+ } else {
+ //otherwise just execute the query
+ return execute(result.builtQuery);
+ }
+ } catch (PropertyReferenceException e) {
+ /*
+ For views, not including an attribute name in the method will result in returning
+ the whole set of results from the view.
+ This is detected by looking for PropertyReferenceExceptions that seem to complain
+ about a missing property that corresponds to the method name
+ */
+ if (e.getPropertyName().equals(method.getName())) {
+ return execute(baseQuery);
+ }
+ throw e;
+ }
+ }
+
+ protected Object execute(ViewQuery query) {
+ if (LOG.isDebugEnabled()) {
+ LOG.debug("Executing view query: " + query.toString());
+ }
+ return ReactiveWrapperConverters.toWrapper(operations.findByView(query, method.getEntityInformation().getJavaType()),
+ Flux.class);
+ }
+
+ protected Object executeReduce(ViewQuery query, String designDoc, String viewName) {
+ if (LOG.isDebugEnabled()) {
+ LOG.debug("Executing view reduced query: " + query.toString());
+ }
+ return ReactiveWrapperConverters.toWrapper(operations.queryView(query)
+ .flatMap(asyncViewResult -> asyncViewResult.error()
+ .flatMap(error -> Observable.error(new CouchbaseQueryExecutionException("Unable to execute reducing view "
+ + viewName +" in design document " + designDoc +
+ "due to error:" + error.toString())))
+ .switchIfEmpty(asyncViewResult.rows()))
+ .map(row -> {
+ AsyncViewRow asyncViewRow = (AsyncViewRow) row;
+ return asyncViewRow.value();
+ }).take(1), Flux.class);
+ }
+
+ @Override
+ public QueryMethod getQueryMethod() {
+ return method;
+ }
+
+ /**
+ * Returns the best-guess design document name.
+ *
+ * @return the design document name.
+ */
+ private static String designDocName(CouchbaseQueryMethod method) {
+ if (method.hasViewSpecification()) {
+ return method.getViewAnnotation().designDocument();
+ } else if (method.hasViewAnnotation()) {
+ return StringUtils.uncapitalize(method.getEntityInformation().getJavaType().getSimpleName());
+ } else {
+ throw new IllegalStateException("View-based query should only happen on a method with @View annotation");
+ }
+ }
+
+}
diff --git a/src/main/java/org/springframework/data/couchbase/repository/query/SpatialViewBasedQuery.java b/src/main/java/org/springframework/data/couchbase/repository/query/SpatialViewBasedQuery.java
index 4f8f7909..e3f0cf7c 100644
--- a/src/main/java/org/springframework/data/couchbase/repository/query/SpatialViewBasedQuery.java
+++ b/src/main/java/org/springframework/data/couchbase/repository/query/SpatialViewBasedQuery.java
@@ -16,26 +16,13 @@
package org.springframework.data.couchbase.repository.query;
-import java.beans.PropertyDescriptor;
-import java.util.ArrayList;
-import java.util.List;
-
import com.couchbase.client.java.view.SpatialViewQuery;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
-
-import org.springframework.beans.BeanWrapper;
-import org.springframework.beans.BeanWrapperImpl;
import org.springframework.data.couchbase.core.CouchbaseOperations;
-import org.springframework.data.couchbase.core.CouchbaseTemplate;
-import org.springframework.data.couchbase.core.mapping.CouchbasePersistentProperty;
import org.springframework.data.couchbase.core.query.Dimensional;
-import org.springframework.data.couchbase.repository.query.support.GeoUtils;
-import org.springframework.data.mapping.PersistentEntity;
-import org.springframework.data.mapping.PersistentProperty;
import org.springframework.data.repository.query.ParametersParameterAccessor;
import org.springframework.data.repository.query.RepositoryQuery;
-import org.springframework.data.repository.query.parser.Part;
import org.springframework.data.repository.query.parser.PartTree;
/**
diff --git a/src/main/java/org/springframework/data/couchbase/repository/query/StringBasedN1qlQueryParser.java b/src/main/java/org/springframework/data/couchbase/repository/query/StringBasedN1qlQueryParser.java
new file mode 100644
index 00000000..6a47a071
--- /dev/null
+++ b/src/main/java/org/springframework/data/couchbase/repository/query/StringBasedN1qlQueryParser.java
@@ -0,0 +1,266 @@
+package org.springframework.data.couchbase.repository.query;
+
+import static org.springframework.data.couchbase.core.support.TemplateUtils.*;
+
+import java.util.ArrayList;
+import java.util.List;
+import org.slf4j.Logger;
+import java.util.regex.Matcher;
+import java.util.regex.Pattern;
+
+import com.couchbase.client.java.document.json.JsonArray;
+import com.couchbase.client.java.document.json.JsonObject;
+import com.couchbase.client.java.document.json.JsonValue;
+import org.slf4j.LoggerFactory;
+import org.springframework.data.repository.query.Parameter;
+import org.springframework.data.repository.query.ParameterAccessor;
+import org.springframework.data.repository.query.QueryMethod;
+import org.springframework.expression.EvaluationContext;
+import org.springframework.expression.common.TemplateParserContext;
+import org.springframework.expression.spel.standard.SpelExpressionParser;
+
+/**
+ * @author Subhashni Balakrishnan
+ */
+public class StringBasedN1qlQueryParser {
+ private static final Logger LOGGER = LoggerFactory.getLogger(StringBasedN1qlQueryParser.class);
+
+
+ public static final String SPEL_PREFIX = "n1ql";
+ /**
+ * Use this variable in a SpEL expression in a {@link org.springframework.data.couchbase.core.query.Query @Query}
+ * annotation's inline statement. This will be replaced by the correct SELECT x FROM y clause needed
+ * for entity mapping. Eg. "#{{@value SPEL_SELECT_FROM_CLAUSE}} WHERE test = true".
+ * Note this only makes sense once, as the beginning of the statement.
+ */
+ public static final String SPEL_SELECT_FROM_CLAUSE = "#" + SPEL_PREFIX + ".selectEntity";
+
+ /**
+ * Use this variable in a SpEL expression in a {@link org.springframework.data.couchbase.core.query.Query @Query}
+ * annotation's inline statement. This will be replaced by the (escaped) bucket name corresponding to the repository's
+ * entity. Eg. "SELECT * FROM #{{@value SPEL_BUCKET}} LIMIT 3".
+ */
+ public static final String SPEL_BUCKET = "#" + SPEL_PREFIX + ".bucket";
+
+ /**
+ * Use this variable in a SpEL expression in a {@link org.springframework.data.couchbase.core.query.Query @Query}
+ * annotation's inline statement. This will be replaced by the fields allowing to construct the repository's entity
+ * (SELECT clause). Eg. "SELECT #{{@value SPEL_ENTITY}} FROM test".
+ */
+ public static final String SPEL_ENTITY = "#" + SPEL_PREFIX + ".fields";
+
+ /**
+ * Use this variable in a SpEL expression in a {@link org.springframework.data.couchbase.core.query.Query @Query}
+ * annotation's inline statement WHERE clause. This will be replaced by the expression allowing to only select
+ * documents matching the entity's class. Eg. "SELECT * FROM test WHERE test = true AND #{{@value SPEL_FILTER}}".
+ */
+ public static final String SPEL_FILTER = "#" + SPEL_PREFIX + ".filter";
+
+ /** regexp that detect $named placeholder (starts with a letter, then alphanum chars) */
+ public static final Pattern NAMED_PLACEHOLDER_PATTERN = Pattern.compile("\\W(\\$\\p{Alpha}\\p{Alnum}*)\\b");
+
+ /** regexp that detect positional placeholder ($ followed by digits only) */
+ public static final Pattern POSITIONAL_PLACEHOLDER_PATTERN = Pattern.compile("\\W(\\$\\p{Digit}+)\\b");
+
+ /** regexp that detects " and ' quote boundaries, ignoring escaped quotes */
+ public static final Pattern QUOTE_DETECTION_PATTERN = Pattern.compile("[\"'](?:[^\"'\\\\]*(?:\\\\.)?)*[\"']");
+
+
+ /** enumeration of all the combinations of placeholder types that could be found in a N1QL statement */
+ private enum PlaceholderType {
+ NAMED, POSITIONAL, NONE
+ }
+
+ private final String statement;
+ private final QueryMethod queryMethod;
+ private final PlaceholderType placeHolderType;
+ private final N1qlSpelValues statementContext;
+ private final N1qlSpelValues countContext;
+
+ public StringBasedN1qlQueryParser(String statement,
+ QueryMethod queryMethod,
+ String bucketName,
+ String typeField,
+ Class typeValue) {
+ this.statement = statement;
+ this.queryMethod = queryMethod;
+ this.placeHolderType = checkPlaceholders(statement);
+ this.statementContext = createN1qlSpelValues(bucketName, typeField, typeValue, false);
+ this.countContext = createN1qlSpelValues(bucketName, typeField, typeValue, true);
+
+ }
+
+ public static N1qlSpelValues createN1qlSpelValues(String bucketName, String typeField, Class typeValue, boolean isCount) {
+ String b = "`" + bucketName + "`";
+ String entity = "META(" + b + ").id AS " + SELECT_ID +
+ ", META(" + b + ").cas AS " + SELECT_CAS;
+ String count = "COUNT(*) AS " + CountFragment.COUNT_ALIAS;
+ String selectEntity;
+ if (isCount) {
+ selectEntity = "SELECT " + count + " FROM " + b;
+ } else {
+ selectEntity = "SELECT " + entity + ", " + b + ".* FROM " + b;
+ }
+
+ String typeSelection = "`" + typeField + "` = \"" + typeValue.getName() + "\"";
+
+ return new N1qlSpelValues(selectEntity, entity, b, typeSelection);
+ }
+
+ //this static method can be used to test the parsing behavior for Couchbase specific spel variables
+ //in isolation from the rest of the spel parser initialization chain.
+ public String doParse(SpelExpressionParser parser, EvaluationContext evaluationContext, boolean isCountQuery) {
+ org.springframework.expression.Expression parsedExpression = parser.parseExpression(this.getStatement(), new TemplateParserContext());
+ if (isCountQuery) {
+ evaluationContext.setVariable(SPEL_PREFIX, this.getCountContext());
+ } else {
+ evaluationContext.setVariable(SPEL_PREFIX, this.getStatementContext());
+ }
+ return parsedExpression.getValue(evaluationContext, String.class);
+ }
+
+ private PlaceholderType checkPlaceholders(String statement) {
+ Matcher quoteMatcher = QUOTE_DETECTION_PATTERN.matcher(statement);
+ Matcher positionMatcher = POSITIONAL_PLACEHOLDER_PATTERN.matcher(statement);
+ Matcher namedMatcher = NAMED_PLACEHOLDER_PATTERN.matcher(statement);
+
+ List quotes = new ArrayList();
+ while(quoteMatcher.find()) {
+ quotes.add(new int[] { quoteMatcher.start(), quoteMatcher.end() });
+ }
+
+ int posCount = 0;
+ int namedCount = 0;
+
+ while(positionMatcher.find()) {
+ String placeholder = positionMatcher.group(1);
+ //check not in quoted
+ if (checkNotQuoted(placeholder, positionMatcher.start(), positionMatcher.end(), quotes)) {
+ LOGGER.trace("{}: Found positional placeholder {}", this.queryMethod.getName(), placeholder);
+ posCount++;
+ }
+ }
+
+ while(namedMatcher.find()) {
+ String placeholder = namedMatcher.group(1);
+ //check not in quoted
+ if (checkNotQuoted(placeholder, namedMatcher.start(), namedMatcher.end(), quotes)) {
+ LOGGER.trace("{}: Found named placeholder {}", this.queryMethod.getName(), placeholder);
+ namedCount++;
+ }
+ }
+
+ if (posCount > 0 && namedCount > 0) {
+ throw new IllegalArgumentException("Using both named (" + namedCount + ") and positional (" + posCount +
+ ") placeholders is not supported, please choose one over the other in " + this.queryMethod.getName());
+ } else if (posCount > 0) {
+ return PlaceholderType.POSITIONAL;
+ } else if (namedCount > 0) {
+ return PlaceholderType.NAMED;
+ } else {
+ return PlaceholderType.NONE;
+ }
+ }
+
+ private boolean checkNotQuoted(String item, int start, int end, List quotes) {
+ for (int[] quote : quotes) {
+ if (quote[0] <= start && quote[1] >= end) {
+ LOGGER.trace("{}: potential placeholder {} is inside quotes, ignored", this.queryMethod.getName(), item);
+ return false;
+ }
+ }
+ return true;
+ }
+
+ private JsonValue getPositionalPlaceholderValues(ParameterAccessor accessor) {
+ JsonArray posValues = JsonArray.create();
+ for (Parameter parameter : this.queryMethod.getParameters().getBindableParameters()) {
+ posValues.add(accessor.getBindableValue(parameter.getIndex()));
+ }
+ return posValues;
+ }
+
+ private JsonObject getNamedPlaceholderValues(ParameterAccessor accessor) {
+ JsonObject namedValues = JsonObject.create();
+
+ for (Parameter parameter : this.queryMethod.getParameters().getBindableParameters()) {
+ String placeholder = parameter.getPlaceholder();
+ Object value = accessor.getBindableValue(parameter.getIndex());
+
+ if (placeholder != null && placeholder.charAt(0) == ':') {
+ placeholder = placeholder.replaceFirst(":", "");
+ namedValues.put(placeholder, value);
+ } else {
+ namedValues.put(parameter.getName(), value);
+ }
+ }
+ return namedValues;
+ }
+
+ protected JsonValue getPlaceholderValues(ParameterAccessor accessor) {
+ switch (this.placeHolderType) {
+ case NAMED:
+ return getNamedPlaceholderValues(accessor);
+ case POSITIONAL:
+ return getPositionalPlaceholderValues(accessor);
+ case NONE:
+ default:
+ return JsonArray.empty();
+ }
+ }
+
+ protected boolean useGeneratedCountQuery() {
+ return this.statement.contains(SPEL_SELECT_FROM_CLAUSE);
+ }
+
+ public N1qlSpelValues getCountContext() {
+ return this.countContext;
+ }
+
+ public N1qlSpelValues getStatementContext() {
+ return this.statementContext;
+ }
+
+ public String getStatement() {
+ return this.statement;
+ }
+
+ /**
+ * This class is exposed to SpEL parsing through the variable #{@value SPEL_PREFIX}.
+ * Use the attributes in your SpEL expressions: {@link #selectEntity}, {@link #fields}, {@link #bucket} and {@link #filter}.
+ */
+ public static final class N1qlSpelValues {
+
+ /**
+ * #{{@value SPEL_SELECT_FROM_CLAUSE}.
+ * selectEntity will be replaced by the SELECT-FROM clause corresponding to the entity. Use once at the beginning.
+ */
+ public final String selectEntity;
+
+ /**
+ * #{{@value SPEL_ENTITY}.
+ * fields will be replaced by the list of N1QL fields allowing to reconstruct the entity.
+ */
+ public final String fields;
+
+ /**
+ * #{{@value SPEL_BUCKET}.
+ * bucket will be replaced by (escaped) bucket name in which the entity is stored.
+ */
+ public final String bucket;
+
+ /**
+ * #{{@value SPEL_FILTER}.
+ * filter will be replaced by an expression allowing to select only entries matching the entity in a WHERE clause.
+ */
+ public final String filter;
+
+ public N1qlSpelValues(String selectClause, String entityFields, String bucket, String filter) {
+ this.selectEntity = selectClause;
+ this.fields = entityFields;
+ this.bucket = bucket;
+ this.filter = filter;
+ }
+ }
+
+}
diff --git a/src/main/java/org/springframework/data/couchbase/repository/query/StringN1qlBasedQuery.java b/src/main/java/org/springframework/data/couchbase/repository/query/StringN1qlBasedQuery.java
index cf02fdc0..382c8d98 100644
--- a/src/main/java/org/springframework/data/couchbase/repository/query/StringN1qlBasedQuery.java
+++ b/src/main/java/org/springframework/data/couchbase/repository/query/StringN1qlBasedQuery.java
@@ -16,28 +16,17 @@
package org.springframework.data.couchbase.repository.query;
-import java.util.ArrayList;
-import java.util.List;
-import java.util.regex.Matcher;
-import java.util.regex.Pattern;
-import com.couchbase.client.java.document.json.JsonArray;
-import com.couchbase.client.java.document.json.JsonObject;
import com.couchbase.client.java.document.json.JsonValue;
import com.couchbase.client.java.query.N1qlQuery;
import com.couchbase.client.java.query.Statement;
-import org.slf4j.Logger;
-import org.slf4j.LoggerFactory;
import org.springframework.data.couchbase.core.CouchbaseOperations;
import org.springframework.data.repository.query.EvaluationContextProvider;
-import org.springframework.data.repository.query.Parameter;
import org.springframework.data.repository.query.ParameterAccessor;
import org.springframework.data.repository.query.RepositoryQuery;
import org.springframework.data.repository.query.ReturnedType;
import org.springframework.expression.EvaluationContext;
-import org.springframework.expression.Expression;
-import org.springframework.expression.common.TemplateParserContext;
import org.springframework.expression.spel.standard.SpelExpressionParser;
/**
@@ -48,68 +37,18 @@ import org.springframework.expression.spel.standard.SpelExpressionParser;
*
* The statement can also contain SpEL expressions enclosed in #{ and }.
*
- * There are couchbase-provided variables included for the {@link #SPEL_BUCKET bucket namespace},
- * the {@link #SPEL_ENTITY ID and CAS fields} necessary for entity reconstruction
- * or a shortcut that covers {@link #SPEL_SELECT_FROM_CLAUSE SELECT AND FROM clauses},
- * along with a variable for {@link #SPEL_FILTER WHERE clause filtering} of the correct entity.
+ * There are couchbase-provided variables included for the {@link StringBasedN1qlQueryParser#SPEL_BUCKET bucket namespace},
+ * the {@link StringBasedN1qlQueryParser#SPEL_ENTITY ID and CAS fields} necessary for entity reconstruction
+ * or a shortcut that covers {@link StringBasedN1qlQueryParser#SPEL_SELECT_FROM_CLAUSE SELECT AND FROM clauses},
+ * along with a variable for {@link StringBasedN1qlQueryParser#SPEL_FILTER WHERE clause filtering} of the correct entity.
*
* @author Simon Baslé
* @author Subhashni Balakrishnan
*/
public class StringN1qlBasedQuery extends AbstractN1qlBasedQuery {
-
- private static final Logger LOGGER = LoggerFactory.getLogger(StringN1qlBasedQuery.class);
-
-
- public static final String SPEL_PREFIX = "n1ql";
-
- /**
- * Use this variable in a SpEL expression in a {@link org.springframework.data.couchbase.core.query.Query @Query}
- * annotation's inline statement. This will be replaced by the correct SELECT x FROM y clause needed
- * for entity mapping. Eg. "#{{@value StringN1qlBasedQuery#SPEL_SELECT_FROM_CLAUSE}} WHERE test = true".
- * Note this only makes sense once, as the beginning of the statement.
- */
- public static final String SPEL_SELECT_FROM_CLAUSE = "#" + SPEL_PREFIX + ".selectEntity";
-
- /**
- * Use this variable in a SpEL expression in a {@link org.springframework.data.couchbase.core.query.Query @Query}
- * annotation's inline statement. This will be replaced by the (escaped) bucket name corresponding to the repository's
- * entity. Eg. "SELECT * FROM #{{@value StringN1qlBasedQuery#SPEL_BUCKET}} LIMIT 3".
- */
- public static final String SPEL_BUCKET = "#" + SPEL_PREFIX + ".bucket";
-
- /**
- * Use this variable in a SpEL expression in a {@link org.springframework.data.couchbase.core.query.Query @Query}
- * annotation's inline statement. This will be replaced by the fields allowing to construct the repository's entity
- * (SELECT clause). Eg. "SELECT #{{@value StringN1qlBasedQuery#SPEL_ENTITY}} FROM test".
- */
- public static final String SPEL_ENTITY = "#" + SPEL_PREFIX + ".fields";
-
- /**
- * Use this variable in a SpEL expression in a {@link org.springframework.data.couchbase.core.query.Query @Query}
- * annotation's inline statement WHERE clause. This will be replaced by the expression allowing to only select
- * documents matching the entity's class. Eg. "SELECT * FROM test WHERE test = true AND #{{@value StringN1qlBasedQuery#SPEL_FILTER}}".
- */
- public static final String SPEL_FILTER = "#" + SPEL_PREFIX + ".filter";
-
- /** regexp that detect $named placeholder (starts with a letter, then alphanum chars) */
- private static final Pattern NAMED_PLACEHOLDER_PATTERN = Pattern.compile("\\W(\\$\\p{Alpha}\\p{Alnum}*)\\b");
- /** regexp that detect positional placeholder ($ followed by digits only) */
- private static final Pattern POSITIONAL_PLACEHOLDER_PATTERN = Pattern.compile("\\W(\\$\\p{Digit}+)\\b");
- /** regexp that detects " and ' quote boundaries, ignoring escaped quotes */
- private static final Pattern QUOTE_DETECTION_PATTERN = Pattern.compile("[\"'](?:[^\"'\\\\]*(?:\\\\.)?)*[\"']");
-
- /** enumeration of all the combinations of placeholder types that could be found in a N1QL statement */
- private enum PlaceholderType {
- NAMED, POSITIONAL, NONE
- }
-
- private final String originalStatement;
- private final PlaceholderType placeHolderType;
private final SpelExpressionParser parser;
private final EvaluationContextProvider evaluationContextProvider;
- private final N1qlSpelValues countContext;
- private final N1qlSpelValues statementContext;
+ private final StringBasedN1qlQueryParser queryParser;
protected String getTypeField() {
return getCouchbaseOperations().getConverter().getTypeKey();
@@ -122,201 +61,34 @@ public class StringN1qlBasedQuery extends AbstractN1qlBasedQuery {
public StringN1qlBasedQuery(String statement, CouchbaseQueryMethod queryMethod, CouchbaseOperations couchbaseOperations,
SpelExpressionParser spelParser, final EvaluationContextProvider evaluationContextProvider) {
super(queryMethod, couchbaseOperations);
-
- this.originalStatement = statement;
- this.placeHolderType = checkPlaceholders(statement);
-
+ this.queryParser = new StringBasedN1qlQueryParser(statement, queryMethod,
+ getCouchbaseOperations().getCouchbaseBucket().name(), getTypeField(), getTypeValue());
this.parser = spelParser;
this.evaluationContextProvider = evaluationContextProvider;
-
- this.statementContext = createN1qlSpelValues(getCouchbaseOperations().getCouchbaseBucket().name(), getTypeField(), getTypeValue(), false);
- this.countContext = createN1qlSpelValues(getCouchbaseOperations().getCouchbaseBucket().name(), getTypeField(), getTypeValue(), true);
- }
-
- private PlaceholderType checkPlaceholders(String statement) {
- Matcher quoteMatcher = QUOTE_DETECTION_PATTERN.matcher(statement);
- Matcher positionMatcher = POSITIONAL_PLACEHOLDER_PATTERN.matcher(statement);
- Matcher namedMatcher = NAMED_PLACEHOLDER_PATTERN.matcher(statement);
-
- List quotes = new ArrayList();
- while(quoteMatcher.find()) {
- quotes.add(new int[] { quoteMatcher.start(), quoteMatcher.end() });
- }
-
- int posCount = 0;
- int namedCount = 0;
-
- while(positionMatcher.find()) {
- String placeholder = positionMatcher.group(1);
- //check not in quoted
- if (checkNotQuoted(placeholder, positionMatcher.start(), positionMatcher.end(), quotes)) {
- LOGGER.trace("{}: Found positional placeholder {}", getQueryMethod().getName(), placeholder);
- posCount++;
- }
- }
-
- while(namedMatcher.find()) {
- String placeholder = namedMatcher.group(1);
- //check not in quoted
- if (checkNotQuoted(placeholder, namedMatcher.start(), namedMatcher.end(), quotes)) {
- LOGGER.trace("{}: Found named placeholder {}", getQueryMethod().getName(), placeholder);
- namedCount++;
- }
- }
-
- if (posCount > 0 && namedCount > 0) {
- throw new IllegalArgumentException("Using both named (" + namedCount + ") and positional (" + posCount +
- ") placeholders is not supported, please choose one over the other in " + queryMethod.getName());
- } else if (posCount > 0) {
- return PlaceholderType.POSITIONAL;
- } else if (namedCount > 0) {
- return PlaceholderType.NAMED;
- } else {
- return PlaceholderType.NONE;
- }
- }
-
- private boolean checkNotQuoted(String item, int start, int end, List quotes) {
- for (int[] quote : quotes) {
- if (quote[0] <= start && quote[1] >= end) {
- LOGGER.trace("{}: potential placeholder {} is inside quotes, ignored", queryMethod.getName(), item);
- return false;
- }
- }
- return true;
- }
-
- public static N1qlSpelValues createN1qlSpelValues(String bucketName, String typeField, Class typeValue, boolean isCount) {
- String b = "`" + bucketName + "`";
- String entity = "META(" + b + ").id AS " + CouchbaseOperations.SELECT_ID +
- ", META(" + b + ").cas AS " + CouchbaseOperations.SELECT_CAS;
- String count = "COUNT(*) AS " + CountFragment.COUNT_ALIAS;
- String selectEntity;
- if (isCount) {
- selectEntity = "SELECT " + count + " FROM " + b;
- } else {
- selectEntity = "SELECT " + entity + ", " + b + ".* FROM " + b;
- }
- String typeSelection = "`" + typeField + "` = \"" + typeValue.getName() + "\"";
-
- return new N1qlSpelValues(selectEntity, entity, b, typeSelection);
- }
-
- /**
- * Parse the statement to detect SPEL blocks (delimited by #{ and })
- * and replace said expression blocks with their corresponding values.
- *
- * @param statement the full statement into which SpEL expressions should be parsed and replaced.
- * @param runtimeParameters the parameters passed into the method at runtime.
- * @return the statement with the SpEL interpreted and replaced by its values.
- */
- protected String parseSpel(String statement, boolean isCount, Object[] runtimeParameters) {
- EvaluationContext evaluationContext = evaluationContextProvider.getEvaluationContext(getQueryMethod().getParameters(), runtimeParameters);
- N1qlSpelValues n1qlSpelValues = this.statementContext;
- if (isCount) {
- n1qlSpelValues = this.countContext;
- }
- return doParse(statement, parser, evaluationContext, n1qlSpelValues);
- }
-
- //this static method can be used to test the parsing behavior for Couchbase specific spel variables
- //in isolation from the rest of the spel parser initialization chain.
- protected static String doParse(String statement, SpelExpressionParser parser, EvaluationContext evaluationContext, N1qlSpelValues n1qlSpelValues) {
- Expression parsedExpression = parser.parseExpression(statement, new TemplateParserContext());
- evaluationContext.setVariable(SPEL_PREFIX, n1qlSpelValues);
- return parsedExpression.getValue(evaluationContext, String.class);
}
@Override
protected JsonValue getPlaceholderValues(ParameterAccessor accessor) {
- switch (this.placeHolderType) {
- case NAMED:
- return getNamedPlaceholderValues(accessor);
- case POSITIONAL:
- return getPositionalPlaceholderValues(accessor);
- case NONE:
- default:
- return JsonArray.empty();
- }
- }
-
- private JsonValue getPositionalPlaceholderValues(ParameterAccessor accessor) {
- JsonArray posValues = JsonArray.create();
- for (Parameter parameter : getQueryMethod().getParameters().getBindableParameters()) {
- posValues.add(accessor.getBindableValue(parameter.getIndex()));
- }
- return posValues;
- }
-
- private JsonObject getNamedPlaceholderValues(ParameterAccessor accessor) {
- JsonObject namedValues = JsonObject.create();
-
- for (Parameter parameter : getQueryMethod().getParameters().getBindableParameters()) {
- String placeholder = parameter.getPlaceholder();
- Object value = accessor.getBindableValue(parameter.getIndex());
-
- if (placeholder != null && placeholder.charAt(0) == ':') {
- placeholder = placeholder.replaceFirst(":", "");
- namedValues.put(placeholder, value);
- } else {
- namedValues.put(parameter.getName(), value);
- }
- }
- return namedValues;
+ return this.queryParser.getPlaceholderValues(accessor);
}
@Override
public Statement getStatement(ParameterAccessor accessor, Object[] runtimeParameters, ReturnedType returnedType) {
- String parsedStatement = parseSpel(this.originalStatement, false, runtimeParameters);
+ EvaluationContext evaluationContext = evaluationContextProvider.getEvaluationContext(getQueryMethod().getParameters(), runtimeParameters);
+ String parsedStatement = this.queryParser.doParse(parser, evaluationContext, false);
return N1qlQuery.simple(parsedStatement).statement();
}
@Override
protected Statement getCount(ParameterAccessor accessor, Object[] runtimeParameters) {
- String parsedCountStatement = parseSpel(this.originalStatement, true, runtimeParameters);
- return N1qlQuery.simple(parsedCountStatement).statement();
+ EvaluationContext evaluationContext = evaluationContextProvider.getEvaluationContext(getQueryMethod().getParameters(), runtimeParameters);
+ String parsedStatement = this.queryParser.doParse(parser, evaluationContext, true);
+ return N1qlQuery.simple(parsedStatement).statement();
}
@Override
protected boolean useGeneratedCountQuery() {
- return this.originalStatement.contains(SPEL_SELECT_FROM_CLAUSE);
+ return this.queryParser.useGeneratedCountQuery();
}
- /**
- * This class is exposed to SpEL parsing through the variable #{@value StringN1qlBasedQuery#SPEL_PREFIX}.
- * Use the attributes in your SpEL expressions: {@link #selectEntity}, {@link #fields}, {@link #bucket} and {@link #filter}.
- */
- public static final class N1qlSpelValues {
-
- /**
- * #{{@value org.springframework.data.couchbase.repository.query.StringN1qlBasedQuery#SPEL_PREFIX}.
- * selectEntity will be replaced by the SELECT-FROM clause corresponding to the entity. Use once at the beginning.
- */
- public final String selectEntity;
-
- /**
- * #{{@value org.springframework.data.couchbase.repository.query.StringN1qlBasedQuery#SPEL_PREFIX}.
- * fields will be replaced by the list of N1QL fields allowing to reconstruct the entity.
- */
- public final String fields;
-
- /**
- * #{{@value org.springframework.data.couchbase.repository.query.StringN1qlBasedQuery#SPEL_PREFIX}.
- * bucket will be replaced by (escaped) bucket name in which the entity is stored.
- */
- public final String bucket;
-
- /**
- * #{{@value org.springframework.data.couchbase.repository.query.StringN1qlBasedQuery#SPEL_PREFIX}.
- * filter will be replaced by an expression allowing to select only entries matching the entity in a WHERE clause.
- */
- public final String filter;
-
- public N1qlSpelValues(String selectClause, String entityFields, String bucket, String filter) {
- this.selectEntity = selectClause;
- this.fields = entityFields;
- this.bucket = bucket;
- this.filter = filter;
- }
- }
}
diff --git a/src/main/java/org/springframework/data/couchbase/repository/query/support/N1qlUtils.java b/src/main/java/org/springframework/data/couchbase/repository/query/support/N1qlUtils.java
index 8ee7c768..0d0fd4b9 100644
--- a/src/main/java/org/springframework/data/couchbase/repository/query/support/N1qlUtils.java
+++ b/src/main/java/org/springframework/data/couchbase/repository/query/support/N1qlUtils.java
@@ -1,5 +1,5 @@
/*
- * Copyright 2012-2015 the original author or authors
+ * Copyright 2012-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.
@@ -24,11 +24,19 @@ import static com.couchbase.client.java.query.dsl.Expression.x;
import static com.couchbase.client.java.query.dsl.functions.AggregateFunctions.count;
import static com.couchbase.client.java.query.dsl.functions.MetaFunctions.meta;
import static com.couchbase.client.java.query.dsl.functions.StringFunctions.lower;
+import static org.springframework.data.couchbase.core.support.TemplateUtils.*;
import java.util.ArrayList;
import java.util.List;
+import java.util.regex.Pattern;
+import com.couchbase.client.java.document.json.JsonArray;
+import com.couchbase.client.java.document.json.JsonObject;
+import com.couchbase.client.java.document.json.JsonValue;
+import com.couchbase.client.java.query.N1qlParams;
+import com.couchbase.client.java.query.N1qlQuery;
import com.couchbase.client.java.query.Statement;
+import com.couchbase.client.java.query.consistency.ScanConsistency;
import com.couchbase.client.java.query.dsl.Expression;
import com.couchbase.client.java.query.dsl.functions.TypeFunctions;
import com.couchbase.client.java.query.dsl.path.FromPath;
@@ -41,11 +49,15 @@ import org.springframework.data.couchbase.core.mapping.CouchbasePersistentEntity
import org.springframework.data.couchbase.core.mapping.CouchbasePersistentProperty;
import org.springframework.data.couchbase.repository.query.CouchbaseEntityInformation;
import org.springframework.data.couchbase.repository.query.CountFragment;
+import org.springframework.data.couchbase.repository.query.StringN1qlBasedQuery;
import org.springframework.data.domain.Sort;
import org.springframework.data.mapping.PropertyPath;
import org.springframework.data.mapping.context.PersistentPropertyPath;
import org.springframework.data.repository.core.EntityMetadata;
import org.springframework.data.repository.query.ReturnedType;
+import org.springframework.expression.EvaluationContext;
+import org.springframework.expression.common.TemplateParserContext;
+import org.springframework.expression.spel.standard.SpelExpressionParser;
/**
* Utility class to deal with constructing well formed N1QL queries around Spring Data entities, so that
@@ -87,8 +99,8 @@ public class N1qlUtils {
*/
public static FromPath createSelectClauseForEntity(String bucketName, ReturnedType returnedType, CouchbaseConverter converter) {
Expression bucket = escapedBucket(bucketName);
- Expression metaId = path(meta(bucket), "id").as(CouchbaseOperations.SELECT_ID);
- Expression metaCas = path(meta(bucket), "cas").as(CouchbaseOperations.SELECT_CAS);
+ Expression metaId = path(meta(bucket), "id").as(SELECT_ID);
+ Expression metaCas = path(meta(bucket), "cas").as(SELECT_CAS);
List expList = new ArrayList();
expList.add(metaId);
expList.add(metaCas);
@@ -208,4 +220,28 @@ public class N1qlUtils {
public static Statement createCountQueryForEntity(String bucketName, CouchbaseConverter converter, CouchbaseEntityInformation entityInformation) {
return select(count("*").as(CountFragment.COUNT_ALIAS)).from(escapedBucket(bucketName)).where(createWhereFilterForEntity(null, converter, entityInformation));
}
+
+ /**
+ * Creates N1QLQuery object from the statement, query placeholder values and scan consistency
+ *
+ * @param statement
+ * @param queryPlaceholderValues
+ * @param scanConsistency
+ * @return
+ */
+ public static N1qlQuery buildQuery(Statement statement, JsonValue queryPlaceholderValues, ScanConsistency scanConsistency) {
+ N1qlParams n1qlParams = N1qlParams.build().consistency(scanConsistency);
+ N1qlQuery query;
+
+ if (queryPlaceholderValues instanceof JsonObject && !((JsonObject) queryPlaceholderValues).isEmpty()) {
+ query = N1qlQuery.parameterized(statement, (JsonObject) queryPlaceholderValues, n1qlParams);
+ } else if (queryPlaceholderValues instanceof JsonArray && !((JsonArray) queryPlaceholderValues).isEmpty()) {
+ query = N1qlQuery.parameterized(statement, (JsonArray) queryPlaceholderValues, n1qlParams);
+ } else {
+ query = N1qlQuery.simple(statement, n1qlParams);
+ }
+ return query;
+ }
+
+
}
diff --git a/src/main/java/org/springframework/data/couchbase/repository/support/IndexManager.java b/src/main/java/org/springframework/data/couchbase/repository/support/IndexManager.java
index 5619817b..11ed153c 100644
--- a/src/main/java/org/springframework/data/couchbase/repository/support/IndexManager.java
+++ b/src/main/java/org/springframework/data/couchbase/repository/support/IndexManager.java
@@ -21,6 +21,7 @@ import static com.couchbase.client.java.query.dsl.Expression.x;
import java.util.Collections;
+import com.couchbase.client.java.Bucket;
import com.couchbase.client.java.bucket.BucketManager;
import com.couchbase.client.java.document.json.JsonObject;
import com.couchbase.client.java.query.AsyncN1qlQueryResult;
@@ -31,6 +32,7 @@ import com.couchbase.client.java.view.DefaultView;
import com.couchbase.client.java.view.DesignDocument;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
+import org.springframework.data.couchbase.core.RxJavaCouchbaseOperations;
import rx.Observable;
import rx.exceptions.CompositeException;
import rx.functions.Action1;
@@ -134,15 +136,15 @@ public class IndexManager {
Observable n1qlSecondaryAsync = Observable.empty();
if (viewIndexed != null && !ignoreViews) {
- viewAsync = buildAllView(viewIndexed, metadata, couchbaseOperations);
+ viewAsync = buildAllView(viewIndexed, metadata, couchbaseOperations.getCouchbaseBucket(), couchbaseOperations.getConverter().getTypeKey());
}
if (n1qlPrimaryIndexed != null && !ignoreN1qlPrimary) {
- n1qlPrimaryAsync = buildN1qlPrimary(metadata, couchbaseOperations);
+ n1qlPrimaryAsync = buildN1qlPrimary(metadata, couchbaseOperations.getCouchbaseBucket());
}
if (n1qlSecondaryIndexed != null && !ignoreN1qlSecondary) {
- n1qlSecondaryAsync = buildN1qlSecondary(n1qlSecondaryIndexed, metadata, couchbaseOperations);
+ n1qlSecondaryAsync = buildN1qlSecondary(n1qlSecondaryIndexed, metadata, couchbaseOperations.getCouchbaseBucket(), couchbaseOperations.getConverter().getTypeKey());
}
//trigger the builds, wait for the last one, throw CompositeException if errors
@@ -151,14 +153,54 @@ public class IndexManager {
.lastOrDefault(null);
}
- private Observable buildN1qlPrimary(final RepositoryInformation metadata, CouchbaseOperations couchbaseOperations) {
- final String bucketName = couchbaseOperations.getCouchbaseBucket().name();
+ /**
+ * Build the relevant indexes according to the provided annotation and repository metadata, in parallel but blocking
+ * until all relevant indexes are created. Existing indexes will be detected and skipped.
+ *
+ * Note that this IndexManager could be configured to ignore some of the annotation types.
+ * In case of multiple errors, a {@link CompositeException} can be raised with up to 3 causes (one per type of index).
+ *
+ * @param metadata the repository's metadata (allowing to find out the type of entity stored, the key under which type
+ * information is stored, etc...).
+ * @param viewIndexed the annotation for creation of a View-based index.
+ * @param n1qlPrimaryIndexed the annotation for creation of a N1QL-based primary index (generic).
+ * @param n1qlSecondaryIndexed the annotation for creation of a N1QL-based secondary index (specific to the repository
+ * stored entity).
+ * @param rxjava1CouchbaseOperations the template to use for index creation.
+ * @throws CompositeException when several errors (for multiple index types) have been raised.
+ */
+ public void buildIndexes(RepositoryInformation metadata, ViewIndexed viewIndexed, N1qlPrimaryIndexed n1qlPrimaryIndexed,
+ N1qlSecondaryIndexed n1qlSecondaryIndexed, RxJavaCouchbaseOperations rxjava1CouchbaseOperations) {
+ Observable viewAsync = Observable.empty();
+ Observable n1qlPrimaryAsync = Observable.empty();
+ Observable n1qlSecondaryAsync = Observable.empty();
+
+ if (viewIndexed != null && !ignoreViews) {
+ viewAsync = buildAllView(viewIndexed, metadata, rxjava1CouchbaseOperations.getCouchbaseBucket(), rxjava1CouchbaseOperations.getConverter().getTypeKey());
+ }
+
+ if (n1qlPrimaryIndexed != null && !ignoreN1qlPrimary) {
+ n1qlPrimaryAsync = buildN1qlPrimary(metadata, rxjava1CouchbaseOperations.getCouchbaseBucket());
+ }
+
+ if (n1qlSecondaryIndexed != null && !ignoreN1qlSecondary) {
+ n1qlSecondaryAsync = buildN1qlSecondary(n1qlSecondaryIndexed, metadata, rxjava1CouchbaseOperations.getCouchbaseBucket(), rxjava1CouchbaseOperations.getConverter().getTypeKey());
+ }
+
+ //trigger the builds, wait for the last one, throw CompositeException if errors
+ Observable.mergeDelayError(viewAsync, n1qlPrimaryAsync, n1qlSecondaryAsync)
+ .toBlocking()
+ .lastOrDefault(null);
+ }
+
+ private Observable buildN1qlPrimary(final RepositoryInformation metadata, Bucket bucket) {
+ final String bucketName = bucket.name();
Statement createPrimary = Index.createPrimaryIndex()
.on(bucketName)
.using(IndexType.GSI);
LOGGER.debug("Creating N1QL primary index for repository {}", metadata.getRepositoryInterface().getSimpleName());
- return couchbaseOperations.getCouchbaseBucket().async().query(createPrimary)
+ return bucket.async().query(createPrimary)
.flatMap(new Func1>() {
@Override
public Observable call(AsyncN1qlQueryResult asyncN1qlQueryResult) {
@@ -184,10 +226,9 @@ public class IndexManager {
});
}
- private Observable buildN1qlSecondary(N1qlSecondaryIndexed config, final RepositoryInformation metadata, CouchbaseOperations couchbaseOperations) {
- final String bucketName = couchbaseOperations.getCouchbaseBucket().name();
+ private Observable buildN1qlSecondary(N1qlSecondaryIndexed config, final RepositoryInformation metadata, Bucket bucket, String typeKey) {
+ final String bucketName = bucket.name();
final String indexName = config.indexName();
- String typeKey = couchbaseOperations.getConverter().getTypeKey();
final String type = metadata.getDomainType().getName();
Statement createIndex = Index.createIndex(indexName)
@@ -196,7 +237,7 @@ public class IndexManager {
.using(IndexType.GSI);
LOGGER.debug("Creating N1QL secondary index for repository {}", metadata.getRepositoryInterface().getSimpleName());
- return couchbaseOperations.getCouchbaseBucket().async().query(createIndex)
+ return bucket.async().query(createIndex)
.flatMap(new Func1>() {
@Override
public Observable call(AsyncN1qlQueryResult asyncN1qlQueryResult) {
@@ -222,15 +263,14 @@ public class IndexManager {
});
}
- private Observable buildAllView(ViewIndexed config, final RepositoryInformation metadata, CouchbaseOperations couchbaseOperations) {
+ private Observable buildAllView(ViewIndexed config, final RepositoryInformation metadata, Bucket bucket, String typeKey) {
if (config == null) return Observable.empty();
LOGGER.debug("Creating View index index for repository {}", metadata.getRepositoryInterface().getSimpleName());
- BucketManager manager = couchbaseOperations.getCouchbaseBucket().bucketManager();
+ BucketManager manager = bucket.bucketManager();
String viewName = config.viewName();
String mapFunction = config.mapFunction();
if (mapFunction.isEmpty()) {
- String typeKey = couchbaseOperations.getConverter().getTypeKey();
String type = metadata.getDomainType().getName();
mapFunction = String.format(TEMPLATE_MAP_FUNCTION, typeKey, type);
diff --git a/src/main/java/org/springframework/data/couchbase/repository/support/ReactiveCouchbaseRepositoryFactory.java b/src/main/java/org/springframework/data/couchbase/repository/support/ReactiveCouchbaseRepositoryFactory.java
new file mode 100644
index 00000000..909a9402
--- /dev/null
+++ b/src/main/java/org/springframework/data/couchbase/repository/support/ReactiveCouchbaseRepositoryFactory.java
@@ -0,0 +1,237 @@
+/*
+ * 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 org.springframework.data.couchbase.repository.support;
+
+import java.io.Serializable;
+import java.lang.reflect.Method;
+
+import com.couchbase.client.java.util.features.CouchbaseFeature;
+import org.springframework.core.annotation.AnnotationUtils;
+import org.springframework.data.couchbase.core.RxJavaCouchbaseOperations;
+import org.springframework.data.couchbase.core.UnsupportedCouchbaseFeatureException;
+import org.springframework.data.couchbase.core.mapping.CouchbasePersistentEntity;
+import org.springframework.data.couchbase.core.mapping.CouchbasePersistentProperty;
+import org.springframework.data.couchbase.core.query.*;
+import org.springframework.data.couchbase.repository.config.ReactiveRepositoryOperationsMapping;
+import org.springframework.data.couchbase.repository.query.*;
+import org.springframework.data.mapping.context.MappingContext;
+import org.springframework.data.mapping.model.MappingException;
+import org.springframework.data.projection.ProjectionFactory;
+import org.springframework.data.repository.core.NamedQueries;
+import org.springframework.data.repository.core.RepositoryInformation;
+import org.springframework.data.repository.core.RepositoryMetadata;
+import org.springframework.data.repository.core.support.ReactiveRepositoryFactorySupport;
+import org.springframework.data.repository.query.EvaluationContextProvider;
+import org.springframework.data.repository.query.QueryLookupStrategy;
+import org.springframework.data.repository.query.RepositoryQuery;
+import org.springframework.expression.spel.standard.SpelExpressionParser;
+import org.springframework.util.Assert;
+
+/**
+ * @Subhashni Balakrishnan
+ * @since 3.0
+ */
+public class ReactiveCouchbaseRepositoryFactory extends ReactiveRepositoryFactorySupport {
+ private static final SpelExpressionParser SPEL_PARSER = new SpelExpressionParser();
+
+ /**
+ * Holds the reference to the template.
+ */
+ private final ReactiveRepositoryOperationsMapping couchbaseOperationsMapping;
+
+ /**
+ * Holds the reference to the {@link IndexManager}.
+ */
+ private final IndexManager indexManager;
+
+ /**
+ * Holds the mapping context.
+ */
+ private final MappingContext, CouchbasePersistentProperty> mappingContext;
+
+ /**
+ * Holds a custom ViewPostProcessor..
+ */
+ private final ViewPostProcessor viewPostProcessor;
+
+ /**
+ * Create a new factory.
+ *
+ * @param couchbaseOperationsMapping the template for the underlying actions.
+ */
+ public ReactiveCouchbaseRepositoryFactory(final ReactiveRepositoryOperationsMapping couchbaseOperationsMapping,
+ final IndexManager indexManager) {
+ Assert.notNull(couchbaseOperationsMapping);
+ Assert.notNull(indexManager);
+
+ this.couchbaseOperationsMapping = couchbaseOperationsMapping;
+ this.indexManager = indexManager;
+ mappingContext = this.couchbaseOperationsMapping.getMappingContext();
+ viewPostProcessor = ViewPostProcessor.INSTANCE;
+
+ addRepositoryProxyPostProcessor(viewPostProcessor);
+ }
+
+ /**
+ * Returns entity information based on the domain class.
+ *
+ * @param domainClass the class for the entity.
+ * @param the value type
+ * @param the id type.
+ *
+ * @return entity information for that domain class.
+ */
+ @Override
+ public CouchbaseEntityInformation getEntityInformation(final Class domainClass) {
+ CouchbasePersistentEntity entity = mappingContext.getPersistentEntity(domainClass);
+
+ if (entity == null) {
+ throw new MappingException(String.format("Could not lookup mapping metadata for domain class %s!", domainClass.getName()));
+ }
+
+ return new MappingCouchbaseEntityInformation((CouchbasePersistentEntity) entity);
+ }
+
+ /**
+ * Returns a new Repository based on the metadata. Two categories of repositories can be instantiated:
+ * {@link SimpleReactiveCouchbaseRepository} and {@link ReactiveN1qlCouchbaseRepository}.
+ *
+ * This method performs feature checks to decide which of the two categories can be instantiated (eg. is N1QL available?).
+ * Instantiation is done via reflection, see {@link #getRepositoryBaseClass(RepositoryMetadata)}.
+ *
+ * @param metadata the repository metadata.
+ *
+ * @return a new created repository.
+ */
+ @Override
+ protected final Object getTargetRepository(final RepositoryInformation metadata) {
+ RxJavaCouchbaseOperations couchbaseOperations = couchbaseOperationsMapping.resolve(metadata.getRepositoryInterface(),
+ metadata.getDomainType());
+ boolean isN1qlAvailable = couchbaseOperations.getCouchbaseClusterInfo().checkAvailable(CouchbaseFeature.N1QL);
+
+ ViewIndexed viewIndexed = AnnotationUtils.findAnnotation(metadata.getRepositoryInterface(), ViewIndexed.class);
+ N1qlPrimaryIndexed n1qlPrimaryIndexed = AnnotationUtils.findAnnotation(metadata.getRepositoryInterface(), N1qlPrimaryIndexed.class);
+ N1qlSecondaryIndexed n1qlSecondaryIndexed = AnnotationUtils.findAnnotation(metadata.getRepositoryInterface(), N1qlSecondaryIndexed.class);
+
+ checkFeatures(metadata, isN1qlAvailable, n1qlPrimaryIndexed, n1qlSecondaryIndexed);
+
+ indexManager.buildIndexes(metadata, viewIndexed, n1qlPrimaryIndexed, n1qlSecondaryIndexed, couchbaseOperations);
+
+ CouchbaseEntityInformation entityInformation = getEntityInformation(metadata.getDomainType());
+ SimpleReactiveCouchbaseRepository repo = getTargetRepositoryViaReflection(metadata, entityInformation, couchbaseOperations);
+ repo.setViewMetadataProvider(viewPostProcessor.getViewMetadataProvider());
+ return repo;
+ }
+
+ private void checkFeatures(RepositoryInformation metadata, boolean isN1qlAvailable,
+ N1qlPrimaryIndexed n1qlPrimaryIndexed, N1qlSecondaryIndexed n1qlSecondaryIndexed) {
+ //paging repo will always need N1QL, also check if the repository requires a N1QL index
+ boolean needsN1ql = metadata.isPagingRepository() || n1qlPrimaryIndexed != null || n1qlSecondaryIndexed != null;
+
+ //for other repos, they might also need N1QL if they don't have only @View methods
+ if (!needsN1ql) {
+ for (Method method : metadata.getQueryMethods()) {
+
+ boolean hasN1ql = AnnotationUtils.findAnnotation(method, Query.class) != null;
+ boolean hasView = AnnotationUtils.findAnnotation(method, View.class) != null;
+
+ if (hasN1ql || !hasView) {
+ needsN1ql = true;
+ break;
+ }
+ }
+ }
+
+ if (needsN1ql && !isN1qlAvailable) {
+ throw new UnsupportedCouchbaseFeatureException("Repository uses N1QL", CouchbaseFeature.N1QL);
+ }
+ }
+
+ /**
+ * Returns the base class for the repository being constructed. Two categories of repositories can be produced by
+ * this factory: {@link SimpleReactiveCouchbaseRepository} and {@link ReactiveN1qlCouchbaseRepository}. This method checks if N1QL
+ * is available to choose between the two, but the actual concrete class is determined respectively by
+ * {@link #getSimpleBaseClass(RepositoryMetadata)} and {@link #getN1qlBaseClass(RepositoryMetadata)}.
+ *
+ * Override these methods if you want to change the base class for all your repositories.
+ *
+ * @param repositoryMetadata metadata for the repository.
+ *
+ * @return the base class.
+ */
+ @Override
+ protected final Class getRepositoryBaseClass(final RepositoryMetadata repositoryMetadata) {
+ RxJavaCouchbaseOperations couchbaseOperations = couchbaseOperationsMapping.resolve(repositoryMetadata.getRepositoryInterface(),
+ repositoryMetadata.getDomainType());
+ boolean isN1qlAvailable = couchbaseOperations.getCouchbaseClusterInfo().checkAvailable(CouchbaseFeature.N1QL);
+ if (isN1qlAvailable) {
+ return getN1qlBaseClass(repositoryMetadata);
+ }
+ return getSimpleBaseClass(repositoryMetadata);
+ }
+
+ protected Class getN1qlBaseClass(final RepositoryMetadata repositoryMetadata) {
+ return ReactiveN1qlCouchbaseRepository.class;
+ }
+
+ protected Class getSimpleBaseClass(final RepositoryMetadata repositoryMetadata) {
+ return SimpleReactiveCouchbaseRepository.class;
+ }
+
+ @Override
+ protected QueryLookupStrategy getQueryLookupStrategy(QueryLookupStrategy.Key key, EvaluationContextProvider contextProvider) {
+ return new ReactiveCouchbaseRepositoryFactory.CouchbaseQueryLookupStrategy(contextProvider);
+ }
+
+ /**
+ * Strategy to lookup Couchbase queries implementation to be used.
+ */
+ private class CouchbaseQueryLookupStrategy implements QueryLookupStrategy {
+
+ private final EvaluationContextProvider evaluationContextProvider;
+
+ public CouchbaseQueryLookupStrategy(EvaluationContextProvider evaluationContextProvider) {
+ this.evaluationContextProvider = evaluationContextProvider;
+ }
+
+ @Override
+ public RepositoryQuery resolveQuery(Method method, RepositoryMetadata metadata, ProjectionFactory factory, NamedQueries namedQueries) {
+ RxJavaCouchbaseOperations couchbaseOperations = couchbaseOperationsMapping.resolve(metadata.getRepositoryInterface(),
+ metadata.getDomainType());
+
+ CouchbaseQueryMethod queryMethod = new CouchbaseQueryMethod(method, metadata, factory, mappingContext);
+ String namedQueryName = queryMethod.getNamedQueryName();
+
+ if (queryMethod.hasDimensionalAnnotation()) {
+ return new ReactiveSpatialViewBasedQuery(queryMethod, couchbaseOperations);
+ } else if (queryMethod.hasViewAnnotation()) {
+ return new ReactiveViewBasedCouchbaseQuery(queryMethod, couchbaseOperations);
+ } else if (queryMethod.hasN1qlAnnotation()) {
+ if (queryMethod.hasInlineN1qlQuery()) {
+ return new ReactiveStringN1qlBasedQuery(queryMethod.getInlineN1qlQuery(), queryMethod, couchbaseOperations,
+ SPEL_PARSER, evaluationContextProvider);
+ } else if (namedQueries.hasQuery(namedQueryName)) {
+ String namedQuery = namedQueries.getQuery(namedQueryName);
+ return new ReactiveStringN1qlBasedQuery(namedQuery, queryMethod, couchbaseOperations,
+ SPEL_PARSER, evaluationContextProvider);
+ } //otherwise will do default, queryDerivation
+ }
+ return new ReactivePartTreeN1qlBasedQuery(queryMethod, couchbaseOperations);
+ }
+ }
+
+
+}
diff --git a/src/main/java/org/springframework/data/couchbase/repository/support/ReactiveCouchbaseRepositoryFactoryBean.java b/src/main/java/org/springframework/data/couchbase/repository/support/ReactiveCouchbaseRepositoryFactoryBean.java
new file mode 100644
index 00000000..60c5a00f
--- /dev/null
+++ b/src/main/java/org/springframework/data/couchbase/repository/support/ReactiveCouchbaseRepositoryFactoryBean.java
@@ -0,0 +1,107 @@
+/*
+ * 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 org.springframework.data.couchbase.repository.support;
+
+import java.io.Serializable;
+
+import org.springframework.data.couchbase.core.RxJavaCouchbaseOperations;
+import org.springframework.data.couchbase.repository.config.ReactiveRepositoryOperationsMapping;
+import org.springframework.data.repository.Repository;
+import org.springframework.data.repository.core.support.RepositoryFactoryBeanSupport;
+import org.springframework.data.repository.core.support.RepositoryFactorySupport;
+import org.springframework.util.Assert;
+
+/**
+ * @author Subhashni Balakrishnan
+ * @since 3.0
+ */
+public class ReactiveCouchbaseRepositoryFactoryBean, S, ID extends Serializable>
+ extends RepositoryFactoryBeanSupport {
+
+ /**
+ * Contains the reference to the template.
+ */
+ private ReactiveRepositoryOperationsMapping couchbaseOperationsMapping;
+
+ /**
+ * Contains the reference to the IndexManager.
+ */
+ private IndexManager indexManager;
+
+ /**
+ * Creates a new {@link CouchbaseRepositoryFactoryBean} for the given repository interface.
+ *
+ * @param repositoryInterface must not be {@literal null}.
+ */
+ public ReactiveCouchbaseRepositoryFactoryBean(Class repositoryInterface) {
+ super(repositoryInterface);
+ }
+
+ /**
+ * Set the template reference.
+ *
+ * @param couchbaseOperationsMapping the reference to the operations template.
+ */
+ public void setCouchbaseOperations(final RxJavaCouchbaseOperations couchbaseOperationsMapping) {
+ setCouchbaseOperationsMapping(new ReactiveRepositoryOperationsMapping(couchbaseOperationsMapping));
+ }
+
+ public void setCouchbaseOperationsMapping(final ReactiveRepositoryOperationsMapping couchbaseOperationsMapping) {
+ this.couchbaseOperationsMapping = couchbaseOperationsMapping;
+ setMappingContext(couchbaseOperationsMapping.getMappingContext());
+ }
+
+ /**
+ * Set the IndexManager reference.
+ *
+ * @param indexManager the IndexManager to use.
+ */
+ public void setIndexManager(final IndexManager indexManager) {
+ this.indexManager = indexManager;
+ }
+
+ /**
+ * Returns a factory instance.
+ *
+ * @return the factory instance.
+ */
+ @Override
+ protected RepositoryFactorySupport createRepositoryFactory() {
+ return getFactoryInstance(couchbaseOperationsMapping, indexManager);
+ }
+
+ /**
+ * Get the factory instance for the operations.
+ *
+ * @param couchbaseOperationsMapping the reference to the template.
+ * @param indexManager the reference to the {@link IndexManager}.
+ * @return the factory instance.
+ */
+ protected ReactiveCouchbaseRepositoryFactory getFactoryInstance(final ReactiveRepositoryOperationsMapping couchbaseOperationsMapping,
+ IndexManager indexManager) {
+ return new ReactiveCouchbaseRepositoryFactory(couchbaseOperationsMapping, indexManager);
+ }
+
+ /**
+ * Make sure that the dependencies are set and not null.
+ */
+ @Override
+ public void afterPropertiesSet() {
+ super.afterPropertiesSet();
+ Assert.notNull(couchbaseOperationsMapping, "operationsMapping must not be null!");
+ Assert.notNull(indexManager, "indexManager must not be null!");
+ }
+}
diff --git a/src/main/java/org/springframework/data/couchbase/repository/support/ReactiveN1qlCouchbaseRepository.java b/src/main/java/org/springframework/data/couchbase/repository/support/ReactiveN1qlCouchbaseRepository.java
new file mode 100644
index 00000000..52dec019
--- /dev/null
+++ b/src/main/java/org/springframework/data/couchbase/repository/support/ReactiveN1qlCouchbaseRepository.java
@@ -0,0 +1,67 @@
+/*
+ * 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 org.springframework.data.couchbase.repository.support;
+
+import java.io.Serializable;
+
+import com.couchbase.client.java.query.N1qlParams;
+import com.couchbase.client.java.query.N1qlQuery;
+import com.couchbase.client.java.query.Statement;
+import com.couchbase.client.java.query.consistency.ScanConsistency;
+import com.couchbase.client.java.query.dsl.Expression;
+import com.couchbase.client.java.query.dsl.path.WherePath;
+import org.springframework.data.couchbase.core.RxJavaCouchbaseOperations;
+import org.springframework.data.couchbase.repository.ReactiveCouchbaseSortingRepository;
+import org.springframework.data.couchbase.repository.query.CouchbaseEntityInformation;
+import org.springframework.data.couchbase.repository.query.support.N1qlUtils;
+import org.springframework.data.domain.Sort;
+import org.springframework.util.Assert;
+import reactor.core.publisher.Flux;
+
+/**
+ * @author Subhashni Balakrishnan
+ * @since 3.0
+ */
+public class ReactiveN1qlCouchbaseRepository
+ extends SimpleReactiveCouchbaseRepository
+ implements ReactiveCouchbaseSortingRepository {
+
+ public ReactiveN1qlCouchbaseRepository(CouchbaseEntityInformation metadata, RxJavaCouchbaseOperations operations) {
+ super(metadata, operations);
+ }
+
+ @SuppressWarnings("unchecked")
+ @Override
+ public Flux findAll(Sort sort) {
+ Assert.notNull(sort);
+
+ //prepare elements of the query
+ WherePath selectFrom = N1qlUtils.createSelectFromForEntity(getCouchbaseOperations().getCouchbaseBucket().name());
+ Expression whereCriteria = N1qlUtils.createWhereFilterForEntity(null, getCouchbaseOperations().getConverter(),
+ getEntityInformation());
+
+ //apply the sort
+ com.couchbase.client.java.query.dsl.Sort[] orderings = N1qlUtils.createSort(sort, getCouchbaseOperations().getConverter());
+ Statement st = selectFrom.where(whereCriteria).orderBy(orderings);
+
+ //fire the query
+ ScanConsistency consistency = getCouchbaseOperations().getDefaultConsistency().n1qlConsistency();
+ N1qlQuery query = N1qlQuery.simple(st, N1qlParams.build().consistency(consistency));
+ return mapFlux(getCouchbaseOperations().findByN1QL(query, getEntityInformation().getJavaType()));
+ }
+
+
+}
diff --git a/src/main/java/org/springframework/data/couchbase/repository/support/SimpleReactiveCouchbaseRepository.java b/src/main/java/org/springframework/data/couchbase/repository/support/SimpleReactiveCouchbaseRepository.java
new file mode 100644
index 00000000..c78e9a7c
--- /dev/null
+++ b/src/main/java/org/springframework/data/couchbase/repository/support/SimpleReactiveCouchbaseRepository.java
@@ -0,0 +1,313 @@
+/*
+ * 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 org.springframework.data.couchbase.repository.support;
+
+import java.io.Serializable;
+
+import com.couchbase.client.java.document.json.JsonArray;
+import com.couchbase.client.java.error.DocumentDoesNotExistException;
+import com.couchbase.client.java.view.AsyncViewResult;
+import com.couchbase.client.java.view.AsyncViewRow;
+import com.couchbase.client.java.view.ViewQuery;
+
+import org.reactivestreams.Publisher;
+import org.springframework.data.couchbase.core.RxJavaCouchbaseOperations;
+import org.springframework.data.couchbase.core.query.View;
+import org.springframework.data.couchbase.repository.ReactiveCouchbaseRepository;
+import org.springframework.data.couchbase.repository.query.CouchbaseEntityInformation;
+import org.springframework.data.repository.util.ReactiveWrapperConverters;
+import org.springframework.util.Assert;
+import org.springframework.util.StringUtils;
+import rx.Single;
+import rx.Observable;
+import reactor.core.publisher.Flux;
+import reactor.core.publisher.Mono;
+
+/**
+ * Reactive repository base implementation for Couchbase.
+ *
+ * @author Subhashni Balakrishnan
+ * @since 3.0
+ */
+public class SimpleReactiveCouchbaseRepository implements ReactiveCouchbaseRepository {
+
+ /**
+ * Holds the reference to the {@link org.springframework.data.couchbase.core.RxJavaCouchbaseTemplate}.
+ */
+ private final RxJavaCouchbaseOperations operations;
+
+ /**
+ * Contains information about the entity being used in this repository.
+ */
+ private final CouchbaseEntityInformation entityInformation;
+
+ /**
+ * Custom ViewMetadataProvider.
+ */
+ private ViewMetadataProvider viewMetadataProvider;
+
+ /**
+ * Create a new Repository.
+ *
+ * @param metadata the Metadata for the entity.
+ * @param operations the reference to the reactive template used.
+ */
+ public SimpleReactiveCouchbaseRepository(final CouchbaseEntityInformation metadata,
+ final RxJavaCouchbaseOperations operations) {
+ Assert.notNull(operations);
+ Assert.notNull(metadata);
+
+ this.entityInformation = metadata;
+ this.operations = operations;
+ }
+
+ /**
+ * Configures a custom {@link ViewMetadataProvider} to be used to detect {@link View}s to be applied to queries.
+ *
+ * @param viewMetadataProvider that is used to lookup any annotated View on a query method.
+ */
+ public void setViewMetadataProvider(final ViewMetadataProvider viewMetadataProvider) {
+ this.viewMetadataProvider = viewMetadataProvider;
+ }
+
+ protected Mono mapMono(Single single) {
+ return ReactiveWrapperConverters.toWrapper(single , Mono.class);
+ }
+
+ protected Flux mapFlux(Observable observable) {
+ return ReactiveWrapperConverters.toWrapper(observable, Flux.class);
+ }
+
+ @SuppressWarnings("unchecked")
+ public Mono save(S entity) {
+ Assert.notNull(entity, "Entity must not be null!");
+ return mapMono(operations.save(entity).toSingle());
+ }
+
+ @SuppressWarnings("unchecked")
+ @Override
+ public