Support for @transactional for blocking and reactive Transactionmanager.

Closes 1145.

Author: mikereiche

pom.xml

@@ -45,7 +45,6 @@
         <version>${couchbase-transactions}</version>
     </dependency>
 
-
         <dependency>
             <groupId>org.springframework</groupId>
             <artifactId>spring-context-support</artifactId>

src/main/java/com/couchbase/transactions/AttemptContextReactiveAccessor.java

@@ -0,0 +1,29 @@
+/*
+/*
+ * Copyright 2021 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
+ *
+ *        https://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 com.couchbase.transactions;
+
+/**
+ * To access the AttemptContextReactive held by AttemptContext
+ *
+ * @author Michael Reiche
+ */
+public class AttemptContextReactiveAccessor {
+
+  public static AttemptContextReactive getACR(AttemptContext attemptContext) {
+    return attemptContext.ctx();
+  }
+}

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

@@ -18,12 +18,18 @@
 
 import java.io.Closeable;
 
+import com.couchbase.transactions.AttemptContextReactive;
+import com.couchbase.transactions.Transactions;
+import com.couchbase.transactions.config.TransactionConfig;
 import org.springframework.dao.support.PersistenceExceptionTranslator;
 
 import com.couchbase.client.java.Bucket;
 import com.couchbase.client.java.Cluster;
 import com.couchbase.client.java.Collection;
 import com.couchbase.client.java.Scope;
+import org.springframework.data.couchbase.transaction.ClientSession;
+import org.springframework.data.couchbase.transaction.ClientSessionOptions;
+import org.springframework.data.couchbase.transaction.CouchbaseStuffHandle;
 
 /**
  * The {@link CouchbaseClientFactory} is the main way to get access to the managed SDK instance and resources.
@@ -73,4 +79,10 @@ public interface CouchbaseClientFactory extends Closeable {
 	 */
 	PersistenceExceptionTranslator getExceptionTranslator();
 
+	ClientSession getSession(ClientSessionOptions options, Transactions transactions,
+																				TransactionConfig config , AttemptContextReactive atr);
+
+  CouchbaseClientFactory with(CouchbaseStuffHandle txOp);
+
+	CouchbaseStuffHandle getTransactionalOperator();
 }

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

@@ -0,0 +1,119 @@
+/*
+ * Copyright 2016-2021 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
+ *
+ *      https://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;
+
+import com.couchbase.client.java.Bucket;
+import com.couchbase.client.java.Cluster;
+import com.couchbase.client.java.ClusterInterface;
+import com.couchbase.client.java.Collection;
+import com.couchbase.client.java.Scope;
+import com.couchbase.transactions.AttemptContextReactive;
+import com.couchbase.transactions.Transactions;
+import com.couchbase.transactions.config.TransactionConfig;
+import org.springframework.data.couchbase.transaction.ClientSession;
+import org.springframework.data.couchbase.transaction.ClientSessionOptions;
+import org.springframework.data.couchbase.transaction.CouchbaseStuffHandle;
+import reactor.core.publisher.Mono;
+
+import org.springframework.dao.support.PersistenceExceptionTranslator;
+
+import java.io.IOException;
+
+
+/**
+ * Interface for factories creating reactive {@link Cluster} instances.
+ *
+ * @author Mark Paluch
+ * @author Christoph Strobl
+ * @author Mathieu Ouellet
+ * @since 2.0
+ */
+public interface ReactiveCouchbaseClientFactory /*extends CodecRegistryProvider*/ {
+	
+	/**
+	 * Provides access to the managed SDK {@link Cluster} reference.
+	 */
+	//Cluster getCluster();
+
+	Mono<ClusterInterface> getCluster();
+
+	/**
+	 * Provides access to the managed SDK {@link Bucket} reference.
+	 */
+	Mono<Bucket> getBucket();
+
+	/**
+	 * Provides access to the managed SDK {@link Scope} reference.
+	 */
+	//Scope getScope();
+
+	Mono<Scope> getScope();
+
+	/**
+	 * Provides access to a collection (identified by its name) in managed SDK {@link Scope} reference.
+	 *
+	 * @param name the name of the collection. If null is passed in, the default collection is assumed.
+	 */
+	//Collection getCollection(String name);
+
+	Mono<Collection> getCollection(String name);
+	/**
+	 * Provides access to the default collection.
+	 */
+	Mono<Collection> getDefaultCollection();
+
+	/**
+	 * Returns a new {@link CouchbaseClientFactory} set to the scope given as an argument.
+	 *
+	 * @param scopeName the name of the scope to use for all collection access.
+	 * @return a new client factory, bound to the other scope.
+	 */
+	ReactiveCouchbaseClientFactory withScope(String scopeName);
+
+	/**
+	 * The exception translator used on the factory.
+	 */
+	PersistenceExceptionTranslator getExceptionTranslator();
+
+	Mono<ClientSession> getSession(ClientSessionOptions options, Transactions transactions, TransactionConfig config);
+
+	String getBucketName();
+
+	String getScopeName();
+
+	void close() throws IOException;
+
+	Mono<ClientSession> getSession(ClientSessionOptions options);
+
+	ClientSession getSession(ClientSessionOptions options, Transactions transactions, TransactionConfig config,
+													 AttemptContextReactive atr);
+
+	/*
+	 * (non-Javadoc)
+	 * @see org.springframework.data.mongodb.ReactiveMongoDatabaseFactory#withSession(com.mongodb.session.ClientSession)
+	 */
+	ReactiveCouchbaseClientFactory withSession(ClientSession session);
+
+	/*
+	 * (non-Javadoc)
+	 * @see org.springframework.data.mongodb.ReactiveMongoDatabaseFactory#isTransactionActive()
+	 */
+	boolean isTransactionActive();
+
+	CouchbaseStuffHandle getTransactionalOperator();
+
+	ReactiveCouchbaseClientFactory with(CouchbaseStuffHandle txOp);
+}

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

@@ -15,10 +15,23 @@
  */
 package org.springframework.data.couchbase;
 
+import java.time.Duration;
+import java.time.temporal.ChronoUnit;
+import java.util.Optional;
+import java.util.UUID;
 import java.util.function.Supplier;
 
+import com.couchbase.transactions.AttemptContext;
+import com.couchbase.transactions.TransactionContext;
+import com.couchbase.transactions.config.MergedTransactionConfig;
+import com.couchbase.transactions.config.PerTransactionConfig;
+import com.couchbase.transactions.config.PerTransactionConfigBuilder;
 import org.springframework.dao.support.PersistenceExceptionTranslator;
 import org.springframework.data.couchbase.core.CouchbaseExceptionTranslator;
+import org.springframework.data.couchbase.transaction.ClientSession;
+import org.springframework.data.couchbase.transaction.ClientSessionImpl;
+import org.springframework.data.couchbase.transaction.ClientSessionOptions;
+import org.springframework.data.couchbase.transaction.CouchbaseStuffHandle;
 
 import com.couchbase.client.core.env.Authenticator;
 import com.couchbase.client.core.env.OwnedSupplier;
@@ -29,6 +42,9 @@
 import com.couchbase.client.java.Collection;
 import com.couchbase.client.java.Scope;
 import com.couchbase.client.java.env.ClusterEnvironment;
+import com.couchbase.transactions.AttemptContextReactive;
+import com.couchbase.transactions.Transactions;
+import com.couchbase.transactions.config.TransactionConfig;
 
 /**
  * The default implementation of a {@link CouchbaseClientFactory}.
@@ -42,6 +58,7 @@ public class SimpleCouchbaseClientFactory implements CouchbaseClientFactory {
 	private final Bucket bucket;
 	private final Scope scope;
 	private final PersistenceExceptionTranslator exceptionTranslator;
+	private final CouchbaseStuffHandle transactionalOperator;
 
 	public SimpleCouchbaseClientFactory(final String connectionString, final Authenticator authenticator,
 			final String bucketName) {
@@ -68,10 +85,16 @@ public SimpleCouchbaseClientFactory(final Cluster cluster, final String bucketNa
 
 	private SimpleCouchbaseClientFactory(final Supplier<Cluster> cluster, final String bucketName,
 			final String scopeName) {
+		this(cluster, bucketName, scopeName, null);
+	}
+
+	private SimpleCouchbaseClientFactory(final Supplier<Cluster> cluster, final String bucketName, final String scopeName,
+			final CouchbaseStuffHandle transactionalOperator) {
 		this.cluster = cluster;
 		this.bucket = cluster.get().bucket(bucketName);
 		this.scope = scopeName == null ? bucket.defaultScope() : bucket.scope(scopeName);
 		this.exceptionTranslator = new CouchbaseExceptionTranslator();
+		this.transactionalOperator = transactionalOperator;
 	}
 
 	@Override
@@ -116,11 +139,51 @@ public PersistenceExceptionTranslator getExceptionTranslator() {
 		return exceptionTranslator;
 	}
 
+	@Override
+	public ClientSession getSession(ClientSessionOptions options, Transactions transactions, TransactionConfig config,
+			AttemptContextReactive atr) {
+
+		AttemptContext at = AttemptContext.from( atr != null ? atr : transactions.reactive().newAttemptContextReactive());
+
+		return new ClientSessionImpl(this, transactions, config, at);
+	}
+
+	/* copied from AttemptContextReactive - needs to have cleanup() and createAttemptContext() public
+
+	  public AttemptContextReactive newAttemptContextReactive(Transactions transactions, TransactionConfig config){
+		PerTransactionConfig perConfig = PerTransactionConfigBuilder.create().build();
+		MergedTransactionConfig merged = new MergedTransactionConfig(config, Optional.of(perConfig));
+
+		TransactionContext overall = new TransactionContext(
+				transactions.reactive().cleanup().clusterData().cluster().environment().requestTracer(),
+				transactions.reactive().cleanup().clusterData().cluster().environment().eventBus(),
+				UUID.randomUUID().toString(), now(), Duration.ZERO, merged);
+
+		String txnId = UUID.randomUUID().toString();
+		//overall.LOGGER.info(configDebug(config, perConfig));
+		return transactions.reactive().createAttemptContext(overall, merged, txnId);
+	}
+	*/
+
+	@Override
+	public CouchbaseClientFactory with(CouchbaseStuffHandle txOp) {
+		return new SimpleCouchbaseClientFactory(cluster, bucket.name(), scope.name(), txOp);
+	}
+
+	@Override
+	public CouchbaseStuffHandle getTransactionalOperator() {
+		return (CouchbaseStuffHandle) transactionalOperator;
+	}
+
 	@Override
 	public void close() {
 		if (cluster instanceof OwnedSupplier) {
 			cluster.get().disconnect();
 		}
 	}
 
+	private static Duration now() {
+		return Duration.of(System.nanoTime(), ChronoUnit.NANOS);
+	}
+
 }