Add support for Couchbase Transactions.

Closes #1145.

Author: mikereiche

pom.xml

@@ -21,6 +21,7 @@
         <couchbase>3.2.1</couchbase>
         <couchbase.osgi>3.2.1</couchbase.osgi>
         <springdata.commons>2.6.0-SNAPSHOT</springdata.commons>
+        <couchbase-transactions>1.2.1-SNAPSHOT</couchbase-transactions>
         <java-module-name>spring.data.couchbase</java-module-name>
     </properties>
 
@@ -37,6 +38,14 @@
     </dependencyManagement>
 
     <dependencies>
+
+    <dependency>
+        <groupId>com.couchbase.client</groupId>
+        <artifactId>couchbase-transactions</artifactId>
+        <version>${couchbase-transactions}</version>
+    </dependency>
+
+
         <dependency>
             <groupId>org.springframework</groupId>
             <artifactId>spring-context-support</artifactId>

src/main/java/org/springframework/data/couchbase/SimpleCouchbaseClientFactory.java

@@ -97,9 +97,9 @@ public Scope getScope() {
 	@Override
 	public Collection getCollection(final String collectionName) {
 		final Scope scope = getScope();
-		if (collectionName == null) {
+		if (collectionName == null || CollectionIdentifier.DEFAULT_COLLECTION.equals(collectionName)) {
 			if (!scope.name().equals(CollectionIdentifier.DEFAULT_SCOPE)) {
-				throw new IllegalStateException("A collectionName must be provided if a non-default scope is used!");
+				throw new IllegalStateException("A collectionName must be provided if a non-default scope is used");
 			}
 			return getBucket().defaultCollection();
 		}

src/main/java/org/springframework/data/couchbase/config/AbstractCouchbaseConfiguration.java

@@ -17,7 +17,10 @@
 package org.springframework.data.couchbase.config;
 
 import static com.couchbase.client.java.ClusterOptions.clusterOptions;
+import static org.springframework.data.couchbase.config.BeanNames.COUCHBASE_MAPPING_CONTEXT;
+import static org.springframework.data.couchbase.config.BeanNames.COUCHBASE_TRANSACTIONS;
 
+import java.time.Duration;
 import java.util.Collections;
 import java.util.HashSet;
 import java.util.Set;
@@ -46,6 +49,7 @@
 import org.springframework.util.ClassUtils;
 import org.springframework.util.StringUtils;
 
+import com.couchbase.client.core.cnc.Event;
 import com.couchbase.client.core.deps.com.fasterxml.jackson.databind.DeserializationFeature;
 import com.couchbase.client.core.encryption.CryptoManager;
 import com.couchbase.client.core.env.Authenticator;
@@ -57,6 +61,10 @@
 import com.couchbase.client.java.env.ClusterEnvironment;
 import com.couchbase.client.java.json.JacksonTransformers;
 import com.couchbase.client.java.json.JsonValueModule;
+import com.couchbase.transactions.TransactionDurabilityLevel;
+import com.couchbase.transactions.Transactions;
+import com.couchbase.transactions.config.TransactionConfig;
+import com.couchbase.transactions.config.TransactionConfigBuilder;
 import com.fasterxml.jackson.databind.ObjectMapper;
 
 /**
@@ -122,7 +130,7 @@ protected Authenticator authenticator() {
 	 * @param couchbaseCluster the cluster reference from the SDK.
 	 * @return the initialized factory.
 	 */
-	@Bean
+	@Bean(name = BeanNames.COUCHBASE_CLIENT_FACTORY)
 	public CouchbaseClientFactory couchbaseClientFactory(final Cluster couchbaseCluster) {
 		return new SimpleCouchbaseClientFactory(couchbaseCluster, getBucketName(), getScopeName());
 	}
@@ -283,7 +291,7 @@ public TranslationService couchbaseTranslationService() {
 	 *
 	 * @throws Exception on Bean construction failure.
 	 */
-	@Bean
+	@Bean(COUCHBASE_MAPPING_CONTEXT)
 	public CouchbaseMappingContext couchbaseMappingContext(CustomConversions customConversions) throws Exception {
 		CouchbaseMappingContext mappingContext = new CouchbaseMappingContext();
 		mappingContext.setInitialEntitySet(getInitialEntitySet());
@@ -312,6 +320,16 @@ public ObjectMapper couchbaseObjectMapper() {
 		return mapper;
 	}
 
+	@Bean(COUCHBASE_TRANSACTIONS)
+	public Transactions getTransactions(Cluster cluster) {
+		return Transactions.create(cluster, getTransactionConfig());
+	}
+
+	TransactionConfig getTransactionConfig() {
+		return TransactionConfigBuilder.create().logDirectly(Event.Severity.INFO).logOnFailure(true, Event.Severity.ERROR)
+				.expirationTime(Duration.ofMinutes(10)).durabilityLevel(TransactionDurabilityLevel.NONE).build();
+	}
+
 	/**
 	 * Configure whether to automatically create indices for domain types by deriving the from the entity or not.
 	 */

src/main/java/org/springframework/data/couchbase/config/BeanNames.java

@@ -34,6 +34,8 @@ public class BeanNames {
 
 	public static final String COUCHBASE_CUSTOM_CONVERSIONS = "couchbaseCustomConversions";
 
+	public static final String COUCHBASE_TRANSACTIONS = "couchbaseTransactions";
+
 	/**
 	 * The name for the bean that stores custom mapping between repositories and their backing couchbaseOperations.
 	 */
@@ -59,4 +61,7 @@ public class BeanNames {
 	 * The name for the bean that will handle reactive audit trail marking of entities.
 	 */
 	public static final String REACTIVE_COUCHBASE_AUDITING_HANDLER = "reactiveCouchbaseAuditingHandler";
+
+	public static final String COUCHBASE_CLIENT_FACTORY = "couchbaseClientFactory";
+
 }

src/main/java/org/springframework/data/couchbase/core/CouchbaseTemplateSupport.java

@@ -17,6 +17,8 @@
 
 package org.springframework.data.couchbase.core;
 
+import java.util.List;
+
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 import org.springframework.beans.BeansException;
@@ -34,11 +36,14 @@
 import org.springframework.data.couchbase.core.mapping.event.BeforeConvertEvent;
 import org.springframework.data.couchbase.core.mapping.event.BeforeSaveEvent;
 import org.springframework.data.couchbase.core.mapping.event.CouchbaseMappingEvent;
+import org.springframework.data.couchbase.core.query.N1qlJoin;
 import org.springframework.data.couchbase.repository.support.MappingCouchbaseEntityInformation;
 import org.springframework.data.mapping.PersistentPropertyAccessor;
+import org.springframework.data.mapping.PropertyHandler;
 import org.springframework.data.mapping.callback.EntityCallbacks;
 import org.springframework.data.mapping.context.MappingContext;
 import org.springframework.data.mapping.model.ConvertingPropertyAccessor;
+import org.springframework.data.util.TypeInformation;
 import org.springframework.util.Assert;
 
 /**
@@ -218,4 +223,8 @@ protected <T> T maybeCallAfterConvert(T object, CouchbaseDocument document, Stri
 		return object;
 	}
 
+	public CouchbaseTemplate getTemplate() {
+		return template;
+	}
+
 }

src/main/java/org/springframework/data/couchbase/core/ExecutableFindByIdOperation.java

@@ -18,14 +18,16 @@
 import java.time.Duration;
 import java.util.Collection;
 
-import org.springframework.data.couchbase.core.support.OneAndAllId;
 import org.springframework.data.couchbase.core.support.InCollection;
+import org.springframework.data.couchbase.core.support.InScope;
+import org.springframework.data.couchbase.core.support.OneAndAllId;
 import org.springframework.data.couchbase.core.support.WithGetOptions;
 import org.springframework.data.couchbase.core.support.WithProjectionId;
-import org.springframework.data.couchbase.core.support.InScope;
+import org.springframework.data.couchbase.core.support.WithTransaction;
 
 import com.couchbase.client.java.kv.GetOptions;
 import org.springframework.data.couchbase.core.support.WithExpiry;
+import com.couchbase.transactions.AttemptContextReactive;
 
 /**
  * Get Operations
@@ -132,11 +134,28 @@ interface FindByIdWithExpiry<T> extends FindByIdWithProjection<T>, WithExpiry<T>
 		FindByIdWithProjection<T> withExpiry(Duration expiry);
 	}
 
+	/**
+	 * Provide attempt context
+	 *
+	 * @param <T> the entity type to use for the results
+	 */
+	interface FindByIdWithTransaction<T> extends FindByIdWithExpiry<T>, WithTransaction<T> {
+		/**
+		 * Finds the distinct values for a specified {@literal field} across a single collection
+		 *
+		 * @param txCtx Must not be {@literal null}.
+		 * @return new instance of {@link ExecutableFindById}.
+		 * @throws IllegalArgumentException if field is {@literal null}.
+		 */
+		@Override
+		FindByIdWithProjection<T> transaction(AttemptContextReactive txCtx);
+	}
+
 	/**
 	 * Provides methods for constructing query operations in a fluent way.
 	 *
 	 * @param <T> the entity type to use for the results
 	 */
-	interface ExecutableFindById<T> extends FindByIdWithExpiry<T> {}
+	interface ExecutableFindById<T> extends FindByIdWithTransaction<T> {}
 
 }

src/main/java/org/springframework/data/couchbase/core/ExecutableFindByIdOperationSupport.java

@@ -24,6 +24,7 @@
 import org.springframework.util.Assert;
 
 import com.couchbase.client.java.kv.GetOptions;
+import com.couchbase.transactions.AttemptContextReactive;
 
 public class ExecutableFindByIdOperationSupport implements ExecutableFindByIdOperation {
 
@@ -35,7 +36,7 @@ public class ExecutableFindByIdOperationSupport implements ExecutableFindByIdOpe
 
 	@Override
 	public <T> ExecutableFindById<T> findById(Class<T> domainType) {
-		return new ExecutableFindByIdSupport<>(template, domainType, null, null, null, null, null);
+		return new ExecutableFindByIdSupport<>(template, domainType, null, null, null, null, null, null);
 	}
 
 	static class ExecutableFindByIdSupport<T> implements ExecutableFindById<T> {
@@ -47,19 +48,21 @@ static class ExecutableFindByIdSupport<T> implements ExecutableFindById<T> {
 		private final GetOptions options;
 		private final List<String> fields;
 		private final Duration expiry;
+		private final AttemptContextReactive txCtx;
 		private final ReactiveFindByIdSupport<T> reactiveSupport;
 
 		ExecutableFindByIdSupport(CouchbaseTemplate template, Class<T> domainType, String scope, String collection,
-				GetOptions options, List<String> fields, Duration expiry) {
+				GetOptions options, List<String> fields, Duration expiry, AttemptContextReactive txCtx) {
 			this.template = template;
 			this.domainType = domainType;
 			this.scope = scope;
 			this.collection = collection;
 			this.options = options;
 			this.fields = fields;
 			this.expiry = expiry;
+			this.txCtx = txCtx;
 			this.reactiveSupport = new ReactiveFindByIdSupport<>(template.reactive(), domainType, scope, collection, options,
-					fields, expiry, new NonReactiveSupportWrapper(template.support()));
+					fields, expiry, txCtx, new NonReactiveSupportWrapper(template.support()));
 		}
 
 		@Override
@@ -75,29 +78,35 @@ public Collection<? extends T> all(final Collection<String> ids) {
 		@Override
 		public TerminatingFindById<T> withOptions(final GetOptions options) {
 			Assert.notNull(options, "Options must not be null.");
-			return new ExecutableFindByIdSupport<>(template, domainType, scope, collection, options, fields, expiry);
+			return new ExecutableFindByIdSupport<>(template, domainType, scope, collection, options, fields, expiry, txCtx);
 		}
 
 		@Override
 		public FindByIdWithOptions<T> inCollection(final String collection) {
-			return new ExecutableFindByIdSupport<>(template, domainType, scope, collection, options, fields, expiry);
+			return new ExecutableFindByIdSupport<>(template, domainType, scope, collection, options, fields, expiry, txCtx);
 		}
 
 		@Override
 		public FindByIdInCollection<T> inScope(final String scope) {
-			return new ExecutableFindByIdSupport<>(template, domainType, scope, collection, options, fields, expiry);
+			return new ExecutableFindByIdSupport<>(template, domainType, scope, collection, options, fields, expiry, txCtx);
 		}
 
 		@Override
 		public FindByIdInScope<T> project(String... fields) {
 			Assert.notEmpty(fields, "Fields must not be null.");
-			return new ExecutableFindByIdSupport<>(template, domainType, scope, collection, options, Arrays.asList(fields), expiry);
+			return new ExecutableFindByIdSupport<>(template, domainType, scope, collection, options, Arrays.asList(fields),
+					expiry, txCtx);
 		}
 
 		@Override
 		public FindByIdWithProjection<T> withExpiry(final Duration expiry) {
-			return new ExecutableFindByIdSupport<>(template, domainType, scope, collection, options, fields,
-					expiry);
+			return new ExecutableFindByIdSupport<>(template, domainType, scope, collection, options, fields, expiry, txCtx);
+		}
+
+		@Override
+		public FindByIdWithExpiry<T> transaction(AttemptContextReactive txCtx) {
+			Assert.notNull(txCtx, "txCtx must not be null.");
+			return new ExecutableFindByIdSupport<>(template, domainType, scope, collection, options, fields, expiry, txCtx);
 		}
 
 	}

src/main/java/org/springframework/data/couchbase/core/ExecutableFindByQueryOperation.java

@@ -29,10 +29,12 @@
 import org.springframework.data.couchbase.core.support.WithDistinct;
 import org.springframework.data.couchbase.core.support.WithQuery;
 import org.springframework.data.couchbase.core.support.WithQueryOptions;
+import org.springframework.data.couchbase.core.support.WithTransaction;
 import org.springframework.lang.Nullable;
 
 import com.couchbase.client.java.query.QueryOptions;
 import com.couchbase.client.java.query.QueryScanConsistency;
+import com.couchbase.transactions.AttemptContextReactive;
 
 /**
  * Query Operations
@@ -287,14 +289,32 @@ interface FindByQueryWithDistinct<T> extends FindByQueryWithProjecting<T>, WithD
 		 * @throws IllegalArgumentException if field is {@literal null}.
 		 */
 		@Override
-		FindByQueryWithProjection<T> distinct(String[] distinctFields);
+		FindByQueryWithProjecting<T> distinct(String[] distinctFields);
+	}
+
+	/**
+	 * Fluent method to specify transaction
+	 *
+	 * @param <T> the entity type to use for the results.
+	 */
+	interface FindByQueryWithTransaction<T> extends FindByQueryWithDistinct<T>, WithTransaction<T> {
+
+		/**
+		 * Finds the distinct values for a specified {@literal field} across a single collection
+		 *
+		 * @param txCtx Must not be {@literal null}.
+		 * @return new instance of {@link ExecutableFindByQuery}.
+		 * @throws IllegalArgumentException if field is {@literal null}.
+		 */
+		@Override
+		FindByQueryWithDistinct<T> transaction(AttemptContextReactive txCtx);
 	}
 
 	/**
 	 * Provides methods for constructing query operations in a fluent way.
 	 *
 	 * @param <T> the entity type to use for the results
 	 */
-	interface ExecutableFindByQuery<T> extends FindByQueryWithDistinct<T> {}
+	interface ExecutableFindByQuery<T> extends FindByQueryWithTransaction<T> {}
 
 }