Add ReactiveTransactionWrapper/TransactionWrapper and a bunch of cleanup.

Author: mikereiche

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

@@ -22,6 +22,8 @@
 import java.time.temporal.ChronoUnit;
 import java.util.Optional;
 import java.util.UUID;
+import java.util.function.Consumer;
+import java.util.logging.Logger;
 
 import com.couchbase.client.core.annotation.Stability;
 import com.couchbase.client.core.transaction.CoreTransactionAttemptContext;
@@ -33,6 +35,7 @@
 import com.couchbase.client.core.transaction.support.AttemptState;
 import com.couchbase.client.java.codec.JsonSerializer;
 import reactor.core.publisher.Mono;
+import reactor.util.annotation.Nullable;
 
 /**
  * To access the ReactiveTransactionAttemptContext held by TransactionAttemptContext
@@ -66,13 +69,7 @@ public static ReactiveTransactionAttemptContext reactive(TransactionAttemptConte
 		} catch (Throwable err) {
 			throw new RuntimeException(err);
 		}
-		try {
-			Field field = TransactionAttemptContext.class.getDeclaredField("internal");
-			field.setAccessible(true);
-			return new ReactiveTransactionAttemptContext((CoreTransactionAttemptContext) field.get(atr), serializer);
-		} catch (Throwable err) {
-			throw new RuntimeException(err);
-		}
+		return new ReactiveTransactionAttemptContext(getCore(atr), serializer);
 	}
 
 	public static TransactionAttemptContext blocking(ReactiveTransactionAttemptContext atr) {
@@ -84,35 +81,22 @@ public static TransactionAttemptContext blocking(ReactiveTransactionAttemptConte
 		} catch (Throwable err) {
 			throw new RuntimeException(err);
 		}
-		try {
-			Field field = ReactiveTransactionAttemptContext.class.getDeclaredField("internal");
-			field.setAccessible(true);
-			return new TransactionAttemptContext((CoreTransactionAttemptContext) field.get(atr), serializer);
-		} catch (Throwable err) {
-			throw new RuntimeException(err);
-		}
+			return new TransactionAttemptContext(getCore(atr), serializer);
 	}
 
 	public static CoreTransactionLogger getLogger(ReactiveTransactionAttemptContext attemptContextReactive) {
-		// todo gp needed?
-		return null;
-		// return attemptContextReactive;
+		return attemptContextReactive.logger();
+	}
+
+	public static CoreTransactionLogger getLogger(TransactionAttemptContext attemptContextReactive) {
+		return attemptContextReactive.logger();
 	}
 
 	// todo gp needed?
 	@Stability.Internal
 	public static CoreTransactionAttemptContext newCoreTranactionAttemptContext(ReactiveTransactions transactions) {
-		// PerTransactionConfig perConfig = PerTransactionConfigBuilder.create().build();
-		// MergedTransactionConfig merged = new MergedTransactionConfig(transactions.config(), Optional.of(perConfig));
-		//
-		// TransactionContext overall = new TransactionContext(
-		// transactions.cleanup().clusterData().cluster().environment().requestTracer(),
-		// transactions.cleanup().clusterData().cluster().environment().eventBus(),
-		// UUID.randomUUID().toString(), now(), Duration.ZERO, merged);
 
 		String txnId = UUID.randomUUID().toString();
-		// overall.LOGGER.info(configDebug(transactions.config(), perConfig));
-
 		CoreTransactionsReactive coreTransactionsReactive;
 		try {
 			Field field = ReactiveTransactions.class.getDeclaredField("internal");
@@ -122,27 +106,19 @@ public static CoreTransactionAttemptContext newCoreTranactionAttemptContext(Reac
 			throw new RuntimeException(err);
 		}
 
-		CoreTransactionOptions perConfig = new CoreTransactionOptions(Optional.empty(),
-				Optional.empty(),
-				Optional.empty(),
-				Optional.of(Duration.ofMinutes(10)),
-				Optional.empty(),
-				Optional.empty());
+		CoreTransactionOptions perConfig = new CoreTransactionOptions(Optional.empty(), Optional.empty(), Optional.empty(),
+				Optional.of(Duration.ofMinutes(10)), Optional.empty(), Optional.empty());
 
 		CoreMergedTransactionConfig merged = new CoreMergedTransactionConfig(coreTransactionsReactive.config(),
 				Optional.ofNullable(perConfig));
 		CoreTransactionContext overall = new CoreTransactionContext(
 				coreTransactionsReactive.core().context().environment().requestTracer(),
 				coreTransactionsReactive.core().context().environment().eventBus(), UUID.randomUUID().toString(), merged,
 				coreTransactionsReactive.core().transactionsCleanup());
-		// overall.LOGGER.info(configDebug(config, perConfig, cleanup.clusterData().cluster().core()));
 
 		CoreTransactionAttemptContext coreTransactionAttemptContext = coreTransactionsReactive.createAttemptContext(overall,
 				merged, txnId);
 		return coreTransactionAttemptContext;
-		// ReactiveTransactionAttemptContext reactiveTransactionAttemptContext = new ReactiveTransactionAttemptContext(
-		// coreTransactionAttemptContext, null);
-		// return reactiveTransactionAttemptContext;
 	}
 
 	private static Duration now() {
@@ -168,15 +144,13 @@ public static CoreTransactionAttemptContext getCore(ReactiveTransactionAttemptCo
 	}
 
 	public static CoreTransactionAttemptContext getCore(TransactionAttemptContext atr) {
-		CoreTransactionAttemptContext coreTransactionsReactive;
 		try {
 			Field field = TransactionAttemptContext.class.getDeclaredField("internal");
 			field.setAccessible(true);
-			coreTransactionsReactive = (CoreTransactionAttemptContext) field.get(atr);
+			return (CoreTransactionAttemptContext) field.get(atr);
 		} catch (Throwable err) {
 			throw new RuntimeException(err);
 		}
-		return coreTransactionsReactive;
 	}
 
 	public static Mono<Void> implicitCommit(ReactiveTransactionAttemptContext atr, boolean b) {
@@ -186,17 +160,17 @@ public static Mono<Void> implicitCommit(ReactiveTransactionAttemptContext atr, b
 			// CoreTransactionAttemptContext.class.getDeclaredMethod("implicitCommit", Boolean.class);
 			Method[] methods = CoreTransactionAttemptContext.class.getDeclaredMethods();
 			Method method = null;
-			for(Method m:methods){
-				if( m.getName().equals("implicitCommit")){
+			for (Method m : methods) {
+				if (m.getName().equals("implicitCommit")) {
 					method = m;
 					break;
 				}
 			}
-			if(method == null){
+			if (method == null) {
 				throw new RuntimeException("did not find implicitCommit method");
 			}
 			method.setAccessible(true);
-			return (Mono<Void>)method.invoke(coreTransactionsReactive, b);
+			return (Mono<Void>) method.invoke(coreTransactionsReactive, b);
 		} catch (Throwable err) {
 			throw new RuntimeException(err);
 		}
@@ -214,10 +188,30 @@ public static AttemptState getState(ReactiveTransactionAttemptContext atr) {
 		}
 	}
 
-	public static ReactiveTransactionAttemptContext createReactiveTransactionAttemptContext(CoreTransactionAttemptContext core, JsonSerializer jsonSerializer) {
+	public static ReactiveTransactionAttemptContext createReactiveTransactionAttemptContext(
+			CoreTransactionAttemptContext core, JsonSerializer jsonSerializer) {
 		return new ReactiveTransactionAttemptContext(core, jsonSerializer);
 	}
 
+	public static CoreTransactionsReactive getCoreTransactionsReactive(ReactiveTransactions transactions) {
+		try {
+			Field field = ReactiveTransactions.class.getDeclaredField("internal");
+			field.setAccessible(true);
+			return (CoreTransactionsReactive) field.get(transactions);
+		} catch (Throwable err) {
+			throw new RuntimeException(err);
+		}
+	}
+
+	public static TransactionAttemptContext newTransactionAttemptContext(CoreTransactionAttemptContext ctx,
+			JsonSerializer jsonSerializer) {
+		return new TransactionAttemptContext(ctx, jsonSerializer);
+	}
+
+	public static TransactionResult run(Transactions transactions, Consumer<TransactionAttemptContext> transactionLogic, CoreTransactionOptions coreTransactionOptions) {
+		return reactive(transactions).runBlocking(transactionLogic, coreTransactionOptions);
+	}
+
 	// todo gp if needed let's expose in the SDK
 	// static private String configDebug(TransactionConfig config, PerTransactionConfig perConfig) {
 	// StringBuilder sb = new StringBuilder();

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

@@ -32,7 +32,6 @@
 
 import java.io.IOException;
 
-
 /**
  * Interface for factories creating reactive {@link Cluster} instances.
  *
@@ -42,12 +41,17 @@
  * @since 2.0
  */
 public interface ReactiveCouchbaseClientFactory /*extends CodecRegistryProvider*/ {
-	
+
 	/**
 	 * Provides access to the managed SDK {@link Cluster} reference.
 	 */
 	Mono<ClusterInterface> getCluster();
 
+	/**
+	 * Provides access to the managed SDK {@link Cluster} reference.
+	 */
+	ClusterInterface getBlockingCluster();
+
 	/**
 	 * Provides access to the managed SDK {@link Bucket} reference.
 	 */
@@ -58,13 +62,25 @@ public interface ReactiveCouchbaseClientFactory /*extends CodecRegistryProvider*
 	 */
 	Mono<Scope> getScope();
 
+	/**
+	 * Provides access to the managed SDK {@link Scope} reference without block()
+	 */
+	 Scope getBlockingScope(String scopeName);
+
 	/**
 	 * 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.
 	 */
 	Mono<Collection> getCollection(String name);
 
+	/**
+	 * Provides access to a collection (identified by its name) without block()
+	 *
+	 * @param name the name of the collection. If null is passed in, the default collection is assumed.
+	 */
+	 Collection getBlockingCollection(String collectionName);
+
 	/**
 	 * Provides access to the default collection.
 	 */
@@ -91,7 +107,8 @@ public interface ReactiveCouchbaseClientFactory /*extends CodecRegistryProvider*
 
 	void close() throws IOException;
 
-	ReactiveCouchbaseResourceHolder getTransactionResources(TransactionOptions options, CoreTransactionAttemptContext ctx);
+	ReactiveCouchbaseResourceHolder getTransactionResources(TransactionOptions options,
+			CoreTransactionAttemptContext ctx);
 
 	/*
 	 * (non-Javadoc)

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

@@ -29,6 +29,7 @@
 
 public class SimpleReactiveCouchbaseClientFactory implements ReactiveCouchbaseClientFactory {
 	final Mono<ClusterInterface> cluster;
+	final ClusterInterface theCluster;
 	final String bucketName;
 	final String scopeName;
 	final PersistenceExceptionTranslator exceptionTranslator;
@@ -39,6 +40,7 @@ public class SimpleReactiveCouchbaseClientFactory implements ReactiveCouchbaseCl
 	public SimpleReactiveCouchbaseClientFactory(Cluster cluster, String bucketName, String scopeName,
 			CouchbaseTransactionalOperator transactionalOperator) {
 		this.cluster = Mono.just(cluster);
+		this.theCluster = cluster;
 		this.bucketName = bucketName;
 		this.scopeName = scopeName;
 		this.exceptionTranslator = new CouchbaseExceptionTranslator();
@@ -56,6 +58,12 @@ public Mono<ClusterInterface> getCluster() {
 		return cluster;
 	}
 
+
+	@Override
+	public ClusterInterface getBlockingCluster() {
+		return theCluster;
+	}
+
 	@Override
 	public Mono<Bucket> getBucket() {
 		return cluster.map((c) -> c.bucket(bucketName));
@@ -71,6 +79,11 @@ public Mono<Scope> getScope() {
 		return cluster.map((c) -> c.bucket(bucketName).scope(scopeName != null ? scopeName : DEFAULT_SCOPE));
 	}
 
+	@Override
+	public Scope getBlockingScope(String scopeName) {
+		return theCluster.bucket(bucketName).scope(scopeName != null ? scopeName : (this.scopeName != null ? this.scopeName : DEFAULT_SCOPE));
+	}
+
 	@Override
 	public String getScopeName() {
 		return scopeName;
@@ -92,6 +105,22 @@ public Mono<Collection> getCollection(String collectionName) {
 		return getScope().map((s) -> s.collection(collectionName != null ? collectionName : DEFAULT_COLLECTION));
 	}
 
+	@Override
+	public Collection getBlockingCollection(String collectionName) {
+		if (getScopeName() != null && !DEFAULT_SCOPE.equals(getScopeName())) {
+			if (collectionName == null || DEFAULT_COLLECTION.equals(collectionName)) {
+				throw new IllegalStateException("A collectionName must be provided if a non-default scope is used.");
+			}
+		}
+		if (getScopeName() == null || DEFAULT_SCOPE.equals(getScopeName())) {
+			if (collectionName != null && !DEFAULT_COLLECTION.equals(collectionName)) {
+				throw new IllegalStateException(
+						"A collectionName must be null or " + DEFAULT_COLLECTION + " if scope is null or " + DEFAULT_SCOPE);
+			}
+		}
+		return theCluster.bucket(bucketName).scope(scopeName != null ? scopeName : DEFAULT_SCOPE).collection(collectionName != null ? collectionName : DEFAULT_COLLECTION);
+	}
+
 	@Override
 	public Mono<Collection> getDefaultCollection() {
 		if (getScopeName() != null && DEFAULT_SCOPE.equals(getScopeName())) {
@@ -204,6 +233,11 @@ public Mono<ClusterInterface> getCluster() throws DataAccessException {
 			return delegate.getCluster().map(this::decorateDatabase);
 		}
 
+		@Override
+		public ClusterInterface getBlockingCluster() throws DataAccessException {
+			return decorateDatabase(delegate.getBlockingCluster());
+		}
+
 		@Override
 		public Mono<Bucket> getBucket() {
 			return delegate.getBucket();
@@ -219,6 +253,15 @@ public Mono<Collection> getCollection(String name) {
 			return delegate.getCollection(name);
 		}
 
+		@Override
+		public Collection getBlockingCollection(String collectionName) {
+			return delegate.getBlockingCollection(collectionName);
+		}
+
+		@Override
+		public Scope getBlockingScope(String scopeName) {
+			return delegate.getBlockingScope(scopeName);
+		}
 		@Override
 		public Mono<Collection> getDefaultCollection() {
 			return delegate.getDefaultCollection();

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

@@ -75,7 +75,7 @@ public Mono<Boolean> one(final String id) {
 			LOG.trace("existsById {}", pArgs);
 			return Mono.just(id)
 					.flatMap(docId -> template.getCouchbaseClientFactory().withScope(pArgs.getScope())
-							.getCollection(pArgs.getCollection()).block().reactive().exists(id, buildOptions(pArgs.getOptions()))
+							.getBlockingCollection(pArgs.getCollection()).reactive().exists(id, buildOptions(pArgs.getOptions()))
 							.map(ExistsResult::exists))
 					.onErrorMap(throwable -> {
 						if (throwable instanceof RuntimeException) {

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

@@ -109,7 +109,7 @@ public Mono<T> first() {
 		public Flux<T> all() {
 			return Flux.defer(() -> {
 				String statement = assembleEntityQuery(false);
-				return template.getCouchbaseClientFactory().getCluster().block().reactive()
+				return template.getCouchbaseClientFactory().getBlockingCluster().reactive()
 						.analyticsQuery(statement, buildAnalyticsOptions()).onErrorMap(throwable -> {
 							if (throwable instanceof RuntimeException) {
 								return template.potentiallyConvertRuntimeException((RuntimeException) throwable);
@@ -142,7 +142,7 @@ public Flux<T> all() {
 		public Mono<Long> count() {
 			return Mono.defer(() -> {
 				String statement = assembleEntityQuery(true);
-				return template.getCouchbaseClientFactory().getCluster().block().reactive()
+				return template.getCouchbaseClientFactory().getBlockingCluster().reactive()
 						.analyticsQuery(statement, buildAnalyticsOptions()).onErrorMap(throwable -> {
 							if (throwable instanceof RuntimeException) {
 								return template.potentiallyConvertRuntimeException((RuntimeException) throwable);

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

@@ -92,7 +92,7 @@ public Mono<T> one(final String id) {
 			LOG.trace("findById {}", pArgs);
 
 			ReactiveCollection rc = template.getCouchbaseClientFactory().withScope(pArgs.getScope())
-					.getCollection(pArgs.getCollection()).block().reactive();
+					.getBlockingCollection(pArgs.getCollection()).reactive();
 
 			// this will get me a template with a session holding tx
 			Mono<ReactiveCouchbaseTemplate> tmpl = template.doGetTemplate();
@@ -118,16 +118,6 @@ public Mono<T> one(final String id) {
 						// todo gp no cas // todo mr - it's required by replace().one when comparing to internal.cas(). it's gone
 						// todo gp if we need this of course needs to be exposed nicely
 						Long cas = result.cas();
-						/*
-						try {
-							Method method = TransactionGetResult.class.getDeclaredMethod("internal");
-							method.setAccessible(true);
-							CoreTransactionGetResult internal = (CoreTransactionGetResult) method.invoke(result);
-							cas = internal.cas();
-						} catch (Throwable err) {
-							throw new RuntimeException(err);
-						}
-*/
 						return support.decodeEntity(id, new String(result.contentAsBytes()), cas, domainType, pArgs.getScope(),
 								pArgs.getCollection(), new TransactionResultHolder(result), null).doOnNext(out -> {
 									// todo gp is this safe? are we on the right thread?