diff --git a/.mvn/wrapper/maven-wrapper.jar b/.mvn/wrapper/maven-wrapper.jar
deleted file mode 100755
index 01e679973..000000000
Binary files a/.mvn/wrapper/maven-wrapper.jar and /dev/null differ
diff --git a/pom.xml b/pom.xml
index 6aac4a641..2a4ec9bca 100644
--- a/pom.xml
+++ b/pom.xml
@@ -18,8 +18,8 @@
- 3.2.5
- 3.2.5
+ 3.3.0
+ 3.3.02.7.0-SNAPSHOTspring.data.couchbase
@@ -37,6 +37,7 @@
+
org.springframeworkspring-context-support
@@ -180,7 +181,14 @@
com.squareup.okhttp3okhttp
- 4.4.0
+ 4.8.1
+ test
+
+
+
+ com.squareup.okhttp3
+ okhttp-tls
+ 4.8.1test
@@ -212,6 +220,18 @@
4.0.3test
+
+ org.testcontainers
+ testcontainers
+
+
+
+ ch.qos.logback
+ logback-classic
+ 1.2.5
+ compile
+
+
@@ -230,10 +250,6 @@
false
-
- jitpack.io
- https://jitpack.io
-
@@ -265,6 +281,7 @@
org.apache.maven.pluginsmaven-failsafe-plugin
+ false**/*IntegrationTest.java**/*IntegrationTests.java
diff --git a/src/main/java/com/couchbase/client/java/Cluster.java b/src/main/java/com/couchbase/client/java/Cluster.java
new file mode 100644
index 000000000..23c588033
--- /dev/null
+++ b/src/main/java/com/couchbase/client/java/Cluster.java
@@ -0,0 +1,589 @@
+/*
+ * Copyright (c) 2018 Couchbase, Inc.
+ *
+ * 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 com.couchbase.client.java;
+
+import com.couchbase.client.core.Core;
+import com.couchbase.client.core.diagnostics.ClusterState;
+import com.couchbase.client.core.diagnostics.DiagnosticsResult;
+import com.couchbase.client.core.annotation.Stability;
+import com.couchbase.client.core.diagnostics.PingResult;
+import com.couchbase.client.core.env.Authenticator;
+import com.couchbase.client.core.env.PasswordAuthenticator;
+import com.couchbase.client.core.env.SeedNode;
+import com.couchbase.client.core.error.CouchbaseException;
+import com.couchbase.client.core.error.TimeoutException;
+import com.couchbase.client.core.msg.search.SearchRequest;
+import com.couchbase.client.java.analytics.AnalyticsOptions;
+import com.couchbase.client.java.analytics.AnalyticsResult;
+import com.couchbase.client.java.diagnostics.DiagnosticsOptions;
+import com.couchbase.client.java.diagnostics.PingOptions;
+import com.couchbase.client.java.diagnostics.WaitUntilReadyOptions;
+import com.couchbase.client.java.env.ClusterEnvironment;
+import com.couchbase.client.java.manager.analytics.AnalyticsIndexManager;
+import com.couchbase.client.java.manager.bucket.BucketManager;
+import com.couchbase.client.java.manager.eventing.EventingFunctionManager;
+import com.couchbase.client.java.manager.query.QueryIndexManager;
+import com.couchbase.client.java.manager.search.SearchIndexManager;
+import com.couchbase.client.java.manager.user.UserManager;
+import com.couchbase.client.java.query.QueryOptions;
+import com.couchbase.client.java.query.QueryResult;
+import com.couchbase.client.java.search.SearchOptions;
+import com.couchbase.client.java.search.SearchQuery;
+import com.couchbase.client.java.search.result.SearchResult;
+import com.couchbase.client.java.transactions.Transactions;
+
+import java.time.Duration;
+import java.util.Map;
+import java.util.Set;
+import java.util.concurrent.ConcurrentHashMap;
+import java.util.function.Supplier;
+
+import static com.couchbase.client.core.util.Validators.notNull;
+import static com.couchbase.client.core.util.Validators.notNullOrEmpty;
+import static com.couchbase.client.java.AsyncCluster.extractClusterEnvironment;
+import static com.couchbase.client.java.AsyncCluster.seedNodesFromConnectionString;
+import static com.couchbase.client.java.AsyncUtils.block;
+import static com.couchbase.client.java.ClusterOptions.clusterOptions;
+import static com.couchbase.client.java.ReactiveCluster.DEFAULT_ANALYTICS_OPTIONS;
+import static com.couchbase.client.java.ReactiveCluster.DEFAULT_DIAGNOSTICS_OPTIONS;
+import static com.couchbase.client.java.ReactiveCluster.DEFAULT_QUERY_OPTIONS;
+import static com.couchbase.client.java.ReactiveCluster.DEFAULT_SEARCH_OPTIONS;
+
+/**
+ * The {@link Cluster} is the main entry point when connecting to a Couchbase cluster.
+ *
+ * Most likely you want to start out by using the {@link #connect(String, String, String)} entry point. For more
+ * advanced options you want to use the {@link #connect(String, ClusterOptions)} method. The entry point that allows
+ * overriding the seed nodes ({@link #connect(Set, ClusterOptions)} is only needed if you run a couchbase cluster
+ * at non-standard ports.
+ *
+ * See the individual connect methods for more information, but here is a snippet to get you off the ground quickly. It
+ * assumes you have Couchbase running locally and the "travel-sample" sample bucket loaded:
+ *
+ * //Connect and open a bucket
+ * Cluster cluster = Cluster.connect("127.0.0.1","Administrator","password");
+ * Bucket bucket = cluster.bucket("travel-sample");
+ * Collection collection = bucket.defaultCollection();
+ *
+ * // Perform a N1QL query
+ * QueryResult queryResult = cluster.query("select * from `travel-sample` limit 5");
+ * System.out.println(queryResult.rowsAsObject());
+ *
+ * // Perform a KV request and load a document
+ * GetResult getResult = collection.get("airline_10");
+ * System.out.println(getResult);
+ *
+ *
+ * When the application shuts down (or the SDK is not needed anymore), you are required to call {@link #disconnect()}.
+ * If you omit this step, the application will terminate (all spawned threads are daemon threads) but any operations
+ * or work in-flight will not be able to complete and lead to undesired side-effects. Note that disconnect will also
+ * shutdown all associated {@link Bucket buckets}.
+ *
+ * Cluster-level operations like {@link #query(String)} will not work unless at leas one bucket is opened against a
+ * pre 6.5 cluster. If you are using 6.5 or later, you can run cluster-level queries without opening a bucket. All
+ * of these operations are lazy, so the SDK will bootstrap in the background and service queries as quickly as possible.
+ * This also means that the first operations might be a bit slower until all sockets are opened in the background and
+ * the configuration is loaded. If you want to wait explicitly, you can utilize the {@link #waitUntilReady(Duration)}
+ * method before performing your first query.
+ *
+ * The SDK will only work against Couchbase Server 5.0 and later, because RBAC (role-based access control) is a first
+ * class concept since 3.0 and therefore required.
+ */
+// todo gpx as per discussion with miker - if required, ClusterInterface will be added to the SDK instead
+public class Cluster implements ClusterInterface {
+
+ /**
+ * Holds the underlying async cluster reference.
+ */
+ private final AsyncCluster asyncCluster;
+
+ /**
+ * Holds the adjacent reactive cluster reference.
+ */
+ private final ReactiveCluster reactiveCluster;
+
+ /**
+ * The search index manager manages search indexes.
+ */
+ private final SearchIndexManager searchIndexManager;
+
+ /**
+ * The user manager manages users and groups.
+ */
+ private final UserManager userManager;
+
+ /**
+ * The bucket manager manages buckets and allows to flush them.
+ */
+ private final BucketManager bucketManager;
+
+ /**
+ * Allows to manage query indexes.
+ */
+ private final QueryIndexManager queryIndexManager;
+
+ /**
+ * Allows to manage analytics indexes.
+ */
+ private final AnalyticsIndexManager analyticsIndexManager;
+
+ /**
+ * Allows to manage eventing functions.
+ */
+ private final EventingFunctionManager eventingFunctionManager;
+
+ /**
+ * Stores already opened buckets for reuse.
+ */
+ private final Map bucketCache = new ConcurrentHashMap<>();
+
+ /**
+ * Connect to a Couchbase cluster with a username and a password as credentials.
+ *
+ * This is the simplest (and recommended) method to connect to the cluster if you do not need to provide any
+ * custom options.
+ *
+ * The first argument (the connection string in its simplest form) is used to supply the hostnames of the cluster. In
+ * development it is OK to only pass in one hostname (or IP address), but in production we recommend passing in at
+ * least 3 nodes of the cluster (comma separated). The reason is that if one or more of the nodes are not reachable
+ * the client will still be able to bootstrap (and your application will become more resilient as a result).
+ *
+ * Here is how you specify one node to use for bootstrapping:
+ *
+ * Cluster cluster = Cluster.connect("127.0.0.1", "user", "password"); // ok during development
+ *
+ * This is what we recommend in production:
+ *
+ * Cluster cluster = Cluster.connect("host1,host2,host3", "user", "password"); // recommended in production
+ *
+ * It is important to understand that the SDK will only use the bootstrap ("seed nodes") host list to establish an
+ * initial contact with the cluster. Once the configuration is loaded this list is discarded and the client will
+ * connect to all nodes based on this configuration.
+ *
+ * This method will return immediately and the SDK will try to establish all the necessary resources and connections
+ * in the background. This means that depending on how fast it can be bootstrapped, the first couple cluster-level
+ * operations like {@link #query(String)} will take a bit longer. If you want to wait explicitly until those resources
+ * are available, you can use the {@link #waitUntilReady(Duration)} method before running any of them:
+ *
+ *
+ * @param connectionString connection string used to locate the Couchbase cluster.
+ * @param username the name of the user with appropriate permissions on the cluster.
+ * @param password the password of the user with appropriate permissions on the cluster.
+ * @return the instantiated {@link Cluster}.
+ */
+ public static Cluster connect(final String connectionString, final String username, final String password) {
+ return connect(connectionString, clusterOptions(PasswordAuthenticator.create(username, password)));
+ }
+
+ /**
+ * Connect to a Couchbase cluster with custom options.
+ *
+ * You likely want to use this over the simpler {@link #connect(String, String, String)} if:
+ *
+ *
A custom {@link ClusterEnvironment}
+ *
Or a custom {@link Authenticator}
+ *
+ * needs to be provided.
+ *
+ * A custom environment can be passed in like this:
+ *
+ * It is VERY important to shut down the environment when being passed in separately (as shown in
+ * the code sample above) and AFTER the cluster is disconnected. This will ensure an orderly shutdown
+ * and makes sure that no resources are left lingering.
+ *
+ * If you want to pass in a custom {@link Authenticator}, it is likely because you are setting up certificate-based
+ * authentication instead of using a username and a password directly. Remember to also enable TLS.
+ *
+ * This method will return immediately and the SDK will try to establish all the necessary resources and connections
+ * in the background. This means that depending on how fast it can be bootstrapped, the first couple cluster-level
+ * operations like {@link #query(String)} will take a bit longer. If you want to wait explicitly until those resources
+ * are available, you can use the {@link #waitUntilReady(Duration)} method before running any of them:
+ *
+ *
+ * @param connectionString connection string used to locate the Couchbase cluster.
+ * @param options custom options when creating the cluster.
+ * @return the instantiated {@link Cluster}.
+ */
+ public static Cluster connect(final String connectionString, final ClusterOptions options) {
+ notNullOrEmpty(connectionString, "ConnectionString");
+ notNull(options, "ClusterOptions");
+
+ final ClusterOptions.Built opts = options.build();
+ final Supplier environmentSupplier = extractClusterEnvironment(connectionString, opts);
+ return new Cluster(
+ environmentSupplier,
+ opts.authenticator(),
+ seedNodesFromConnectionString(connectionString, environmentSupplier.get())
+ );
+ }
+
+ /**
+ * Connect to a Couchbase cluster with a list of seed nodes and custom options.
+ *
+ * Note that you likely only want to use this method if you need to pass in custom ports for specific seed nodes
+ * during bootstrap. Otherwise we recommend relying on the simpler {@link #connect(String, String, String)} method
+ * instead.
+ *
+ * The following example shows how to bootstrap against a node with custom KV and management ports:
+ *
+ * @param seedNodes the seed nodes used to connect to the cluster.
+ * @param options custom options when creating the cluster.
+ * @return the instantiated {@link Cluster}.
+ */
+ public static Cluster connect(final Set seedNodes, final ClusterOptions options) {
+ notNullOrEmpty(seedNodes, "SeedNodes");
+ notNull(options, "ClusterOptions");
+
+ final ClusterOptions.Built opts = options.build();
+ return new Cluster(extractClusterEnvironment("", opts), opts.authenticator(), seedNodes);
+ }
+
+ /**
+ * Creates a new cluster from a {@link ClusterEnvironment}.
+ *
+ * @param environment the environment to use.
+ * @param authenticator the authenticator to use.
+ * @param seedNodes the seed nodes to bootstrap from.
+ */
+ private Cluster(final Supplier environment, final Authenticator authenticator,
+ final Set seedNodes) {
+ this.asyncCluster = new AsyncCluster(environment, authenticator, seedNodes);
+ this.reactiveCluster = new ReactiveCluster(asyncCluster);
+ this.searchIndexManager = new SearchIndexManager(asyncCluster.searchIndexes());
+ this.userManager = new UserManager(asyncCluster.users());
+ this.bucketManager = new BucketManager(asyncCluster.buckets());
+ this.queryIndexManager = new QueryIndexManager(asyncCluster.queryIndexes());
+ this.analyticsIndexManager = new AnalyticsIndexManager(this);
+ this.eventingFunctionManager = new EventingFunctionManager(asyncCluster.eventingFunctions());
+ }
+
+ /**
+ * Provides access to the related {@link AsyncCluster}.
+ *
+ * Note that the {@link AsyncCluster} is considered advanced API and should only be used to get the last drop
+ * of performance or if you are building higher-level abstractions on top. If in doubt, we recommend using the
+ * {@link #reactive()} API instead.
+ */
+ public AsyncCluster async() {
+ return asyncCluster;
+ }
+
+ /**
+ * Provides access to the related {@link ReactiveCluster}.
+ */
+ public ReactiveCluster reactive() {
+ return reactiveCluster;
+ }
+
+ /**
+ * Provides access to the underlying {@link Core}.
+ *
+ *
This is advanced and volatile API - it might change any time without notice. Use with care!
+ */
+ @Stability.Volatile
+ public Core core() {
+ return asyncCluster.core();
+ }
+
+ /**
+ * The user manager allows to manage users and groups.
+ */
+ public UserManager users() {
+ return userManager;
+ }
+
+ /**
+ * The bucket manager allows to perform administrative tasks on buckets and their resources.
+ */
+ public BucketManager buckets() {
+ return bucketManager;
+ }
+
+ /**
+ * The analytics index manager allows to modify and create indexes for the analytics service.
+ */
+ public AnalyticsIndexManager analyticsIndexes() {
+ return analyticsIndexManager;
+ }
+
+ /**
+ * The query index manager allows to modify and create indexes for the query service.
+ */
+ public QueryIndexManager queryIndexes() {
+ return queryIndexManager;
+ }
+
+ /**
+ * The search index manager allows to modify and create indexes for the search service.
+ */
+ public SearchIndexManager searchIndexes() {
+ return searchIndexManager;
+ }
+
+ /**
+ * Provides access to the eventing function management services.
+ */
+ @Stability.Uncommitted
+ public EventingFunctionManager eventingFunctions() {
+ return eventingFunctionManager;
+ }
+
+ /**
+ * Provides access to the used {@link ClusterEnvironment}.
+ */
+ public ClusterEnvironment environment() {
+ return asyncCluster.environment();
+ }
+
+ /**
+ * Performs a query against the query (N1QL) services.
+ *
+ * @param statement the N1QL query statement.
+ * @return the {@link QueryResult} once the response arrives successfully.
+ * @throws TimeoutException if the operation times out before getting a result.
+ * @throws CouchbaseException for all other error reasons (acts as a base type and catch-all).
+ */
+ public QueryResult query(final String statement) {
+ return query(statement, DEFAULT_QUERY_OPTIONS);
+ }
+
+ /**
+ * Performs a query against the query (N1QL) services with custom options.
+ *
+ * @param statement the N1QL query statement as a raw string.
+ * @param options the custom options for this query.
+ * @return the {@link QueryResult} once the response arrives successfully.
+ * @throws TimeoutException if the operation times out before getting a result.
+ * @throws CouchbaseException for all other error reasons (acts as a base type and catch-all).
+ */
+ public QueryResult query(final String statement, final QueryOptions options) {
+ return block(async().query(statement, options));
+ }
+
+ /**
+ * Performs an analytics query with default {@link AnalyticsOptions}.
+ *
+ * @param statement the query statement as a raw string.
+ * @return the {@link AnalyticsResult} once the response arrives successfully.
+ * @throws TimeoutException if the operation times out before getting a result.
+ * @throws CouchbaseException for all other error reasons (acts as a base type and catch-all).
+ */
+ public AnalyticsResult analyticsQuery(final String statement) {
+ return analyticsQuery(statement, DEFAULT_ANALYTICS_OPTIONS);
+ }
+
+ /**
+ * Performs an analytics query with custom {@link AnalyticsOptions}.
+ *
+ * @param statement the query statement as a raw string.
+ * @param options the custom options for this query.
+ * @return the {@link AnalyticsResult} once the response arrives successfully.
+ * @throws TimeoutException if the operation times out before getting a result.
+ * @throws CouchbaseException for all other error reasons (acts as a base type and catch-all).
+ */
+ public AnalyticsResult analyticsQuery(final String statement, final AnalyticsOptions options) {
+ return block(async().analyticsQuery(statement, options));
+ }
+
+ /**
+ * Performs a Full Text Search (FTS) query with default {@link SearchOptions}.
+ *
+ * @param query the query, in the form of a {@link SearchQuery}
+ * @return the {@link SearchRequest} once the response arrives successfully.
+ * @throws TimeoutException if the operation times out before getting a result.
+ * @throws CouchbaseException for all other error reasons (acts as a base type and catch-all).
+ */
+ public SearchResult searchQuery(final String indexName, final SearchQuery query) {
+ return searchQuery(indexName, query, DEFAULT_SEARCH_OPTIONS);
+ }
+
+ /**
+ * Performs a Full Text Search (FTS) query with custom {@link SearchOptions}.
+ *
+ * @param query the query, in the form of a {@link SearchQuery}
+ * @param options the custom options for this query.
+ * @return the {@link SearchRequest} once the response arrives successfully.
+ * @throws TimeoutException if the operation times out before getting a result.
+ * @throws CouchbaseException for all other error reasons (acts as a base type and catch-all).
+ */
+ public SearchResult searchQuery(final String indexName, final SearchQuery query, final SearchOptions options) {
+ return block(asyncCluster.searchQuery(indexName, query, options));
+ }
+
+ /**
+ * Opens a {@link Bucket} with the given name.
+ *
+ * @param bucketName the name of the bucket to open.
+ * @return a {@link Bucket} once opened.
+ */
+ public Bucket bucket(final String bucketName) {
+ return bucketCache.computeIfAbsent(bucketName, n -> new Bucket(asyncCluster.bucket(n)));
+ }
+
+ /**
+ * Performs a non-reversible disconnect of this {@link Cluster}.
+ *
+ * If this method is used, the default disconnect timeout on the environment is used. Please use the companion
+ * overload ({@link #disconnect(Duration)} if you want to provide a custom duration.
+ *
+ * If a custom {@link ClusterEnvironment} has been passed in during connect, it is VERY important to
+ * shut it down after calling this method. This will prevent any in-flight tasks to be stopped prematurely.
+ */
+ public void disconnect() {
+ block(asyncCluster.disconnect());
+ }
+
+ /**
+ * Performs a non-reversible disconnect of this {@link Cluster}.
+ *
+ * If a custom {@link ClusterEnvironment} has been passed in during connect, it is VERY important to
+ * shut it down after calling this method. This will prevent any in-flight tasks to be stopped prematurely.
+ *
+ * @param timeout allows to override the default disconnect duration.
+ */
+ public void disconnect(final Duration timeout) {
+ block(asyncCluster.disconnect(timeout));
+ }
+
+ /**
+ * Runs a diagnostic report on the current state of the cluster from the SDKs point of view.
+ *
+ * Please note that it does not perform any I/O to do this, it will only use the current known state of the cluster
+ * to assemble the report (so, if for example no N1QL query has been run the socket pool might be empty and as
+ * result not show up in the report).
+ *
+ * @return the {@link DiagnosticsResult} once complete.
+ */
+ public DiagnosticsResult diagnostics() {
+ return block(asyncCluster.diagnostics(DEFAULT_DIAGNOSTICS_OPTIONS));
+ }
+
+ /**
+ * Runs a diagnostic report with custom options on the current state of the cluster from the SDKs point of view.
+ *
+ * Please note that it does not perform any I/O to do this, it will only use the current known state of the cluster
+ * to assemble the report (so, if for example no N1QL query has been run the socket pool might be empty and as
+ * result not show up in the report).
+ *
+ * @param options options that allow to customize the report.
+ * @return the {@link DiagnosticsResult} once complete.
+ */
+ public DiagnosticsResult diagnostics(final DiagnosticsOptions options) {
+ return block(asyncCluster.diagnostics(options));
+ }
+
+ /**
+ * Performs application-level ping requests against services in the couchbase cluster.
+ *
+ * Note that this operation performs active I/O against services and endpoints to assess their health. If you do
+ * not wish to perform I/O, consider using the {@link #diagnostics()} instead. You can also combine the functionality
+ * of both APIs as needed, which is {@link #waitUntilReady(Duration)} is doing in its implementation as well.
+ *
+ * @return the {@link PingResult} once complete.
+ */
+ public PingResult ping() {
+ return block(asyncCluster.ping());
+ }
+
+ /**
+ * Performs application-level ping requests with custom options against services in the couchbase cluster.
+ *
+ * Note that this operation performs active I/O against services and endpoints to assess their health. If you do
+ * not wish to perform I/O, consider using the {@link #diagnostics(DiagnosticsOptions)} instead. You can also combine
+ * the functionality of both APIs as needed, which is {@link #waitUntilReady(Duration)} is doing in its
+ * implementation as well.
+ *
+ * @return the {@link PingResult} once complete.
+ */
+ public PingResult ping(final PingOptions options) {
+ return block(asyncCluster.ping(options));
+ }
+
+ /**
+ * Waits until the desired {@link ClusterState} is reached.
+ *
+ * This method will wait until either the cluster state is "online", or the timeout is reached. Since the SDK is
+ * bootstrapping lazily, this method allows to eagerly check during bootstrap if all of the services are online
+ * and usable before moving on.
+ *
+ * @param timeout the maximum time to wait until readiness.
+ */
+ public void waitUntilReady(final Duration timeout) {
+ block(asyncCluster.waitUntilReady(timeout));
+ }
+
+ /**
+ * Waits until the desired {@link ClusterState} is reached.
+ *
+ * This method will wait until either the cluster state is "online" by default, or the timeout is reached. Since the
+ * SDK is bootstrapping lazily, this method allows to eagerly check during bootstrap if all of the services are online
+ * and usable before moving on. You can tune the properties through {@link WaitUntilReadyOptions}.
+ *
+ * @param timeout the maximum time to wait until readiness.
+ * @param options the options to customize the readiness waiting.
+ */
+ public void waitUntilReady(final Duration timeout, final WaitUntilReadyOptions options) {
+ block(asyncCluster.waitUntilReady(timeout, options));
+ }
+
+ /**
+ * Allows access to transactions.
+ *
+ * @return the {@link Transactions} interface.
+ */
+ @Stability.Uncommitted
+ public Transactions transactions() {
+ return new Transactions(core(), environment().jsonSerializer());
+ }
+}
+
diff --git a/src/main/java/com/couchbase/client/java/ClusterInterface.java b/src/main/java/com/couchbase/client/java/ClusterInterface.java
new file mode 100644
index 000000000..872a6efdf
--- /dev/null
+++ b/src/main/java/com/couchbase/client/java/ClusterInterface.java
@@ -0,0 +1,111 @@
+/*
+ * Copyright (c) 2018 Couchbase, Inc.
+ *
+ * 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 com.couchbase.client.java;
+
+import com.couchbase.client.core.Core;
+import com.couchbase.client.core.annotation.Stability;
+import com.couchbase.client.core.diagnostics.DiagnosticsResult;
+import com.couchbase.client.core.diagnostics.PingResult;
+import com.couchbase.client.core.env.Authenticator;
+import com.couchbase.client.core.env.PasswordAuthenticator;
+import com.couchbase.client.core.env.SeedNode;
+import com.couchbase.client.java.analytics.AnalyticsOptions;
+//import com.couchbase.client.java.analytics.AnalyticsResult;
+import com.couchbase.client.java.diagnostics.DiagnosticsOptions;
+import com.couchbase.client.java.diagnostics.PingOptions;
+import com.couchbase.client.java.diagnostics.WaitUntilReadyOptions;
+import com.couchbase.client.java.env.ClusterEnvironment;
+import com.couchbase.client.java.manager.analytics.AnalyticsIndexManager;
+import com.couchbase.client.java.manager.bucket.BucketManager;
+import com.couchbase.client.java.manager.eventing.EventingFunctionManager;
+import com.couchbase.client.java.manager.query.QueryIndexManager;
+import com.couchbase.client.java.manager.search.SearchIndexManager;
+import com.couchbase.client.java.manager.user.UserManager;
+import com.couchbase.client.java.query.QueryOptions;
+import com.couchbase.client.java.query.QueryResult;
+import com.couchbase.client.java.search.SearchOptions;
+import com.couchbase.client.java.search.SearchQuery;
+import com.couchbase.client.java.search.result.SearchResult;
+import com.couchbase.client.java.transactions.Transactions;
+import org.springframework.data.couchbase.transaction.CouchbaseTransactionalOperator;
+
+import java.time.Duration;
+import java.util.Set;
+import java.util.function.Supplier;
+
+import static com.couchbase.client.core.util.Validators.notNull;
+import static com.couchbase.client.core.util.Validators.notNullOrEmpty;
+import static com.couchbase.client.java.AsyncCluster.extractClusterEnvironment;
+import static com.couchbase.client.java.AsyncCluster.seedNodesFromConnectionString;
+import static com.couchbase.client.java.ClusterOptions.clusterOptions;
+
+public interface ClusterInterface {
+
+ AsyncCluster async();
+
+ ReactiveCluster reactive();
+
+ @Stability.Volatile
+ Core core();
+
+ UserManager users();
+
+ BucketManager buckets();
+
+ AnalyticsIndexManager analyticsIndexes();
+
+ QueryIndexManager queryIndexes();
+
+ SearchIndexManager searchIndexes();
+
+ @Stability.Uncommitted
+ EventingFunctionManager eventingFunctions();
+
+ ClusterEnvironment environment();
+
+ QueryResult query(String statement);
+
+ QueryResult query(String statement, QueryOptions options);
+
+ //AnalyticsResult analyticsQuery(String statement);
+
+ // AnalyticsResult analyticsQuery(String statement, AnalyticsOptions options);
+
+ SearchResult searchQuery(String indexName, SearchQuery query);
+
+ SearchResult searchQuery(String indexName, SearchQuery query, SearchOptions options);
+
+ Bucket bucket(String bucketName);
+
+ void disconnect();
+
+ void disconnect(Duration timeout);
+
+ DiagnosticsResult diagnostics();
+
+ DiagnosticsResult diagnostics(DiagnosticsOptions options);
+
+ PingResult ping();
+
+ PingResult ping(PingOptions options);
+
+ void waitUntilReady(Duration timeout);
+
+ void waitUntilReady(Duration timeout, WaitUntilReadyOptions options);
+
+ Transactions transactions();
+}
diff --git a/src/main/java/com/couchbase/client/java/transactions/AttemptContextReactiveAccessor.java b/src/main/java/com/couchbase/client/java/transactions/AttemptContextReactiveAccessor.java
new file mode 100644
index 000000000..f0ffd69bd
--- /dev/null
+++ b/src/main/java/com/couchbase/client/java/transactions/AttemptContextReactiveAccessor.java
@@ -0,0 +1,223 @@
+/*
+/*
+ * 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.client.java.transactions;
+
+import java.lang.reflect.Field;
+import java.lang.reflect.Method;
+import java.time.Duration;
+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;
+import com.couchbase.client.core.transaction.CoreTransactionContext;
+import com.couchbase.client.core.transaction.CoreTransactionsReactive;
+import com.couchbase.client.core.transaction.config.CoreMergedTransactionConfig;
+import com.couchbase.client.core.transaction.config.CoreTransactionOptions;
+import com.couchbase.client.core.transaction.log.CoreTransactionLogger;
+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
+ *
+ * @author Michael Reiche
+ */
+public class AttemptContextReactiveAccessor {
+
+ public static ReactiveTransactions reactive(Transactions transactions) {
+ try {
+ Field field = Transactions.class.getDeclaredField("reactive");
+ field.setAccessible(true);
+ return (ReactiveTransactions) field.get(transactions);
+ } catch (Throwable err) {
+ throw new RuntimeException(err);
+ }
+ }
+
+ public static ReactiveTransactionAttemptContext reactive(TransactionAttemptContext atr) {
+ JsonSerializer serializer;
+ try {
+ Field field = TransactionAttemptContext.class.getDeclaredField("serializer");
+ field.setAccessible(true);
+ serializer = (JsonSerializer) field.get(atr);
+ } catch (Throwable err) {
+ throw new RuntimeException(err);
+ }
+ return new ReactiveTransactionAttemptContext(getCore(atr), serializer);
+ }
+
+ public static TransactionAttemptContext blocking(ReactiveTransactionAttemptContext atr) {
+ JsonSerializer serializer;
+ try {
+ Field field = ReactiveTransactionAttemptContext.class.getDeclaredField("serializer");
+ field.setAccessible(true);
+ serializer = (JsonSerializer) field.get(atr);
+ } catch (Throwable err) {
+ throw new RuntimeException(err);
+ }
+ return new TransactionAttemptContext(getCore(atr), serializer);
+ }
+
+ public static CoreTransactionLogger getLogger(ReactiveTransactionAttemptContext attemptContextReactive) {
+ return attemptContextReactive.logger();
+ }
+
+ public static CoreTransactionLogger getLogger(TransactionAttemptContext attemptContextReactive) {
+ return attemptContextReactive.logger();
+ }
+
+ // todo gp needed?
+ @Stability.Internal
+ public static CoreTransactionAttemptContext newCoreTranactionAttemptContext(ReactiveTransactions transactions) {
+
+ String txnId = UUID.randomUUID().toString();
+ CoreTransactionsReactive coreTransactionsReactive;
+ try {
+ Field field = ReactiveTransactions.class.getDeclaredField("internal");
+ field.setAccessible(true);
+ coreTransactionsReactive = (CoreTransactionsReactive) field.get(transactions);
+ } catch (Throwable err) {
+ throw new RuntimeException(err);
+ }
+
+ 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());
+
+ CoreTransactionAttemptContext coreTransactionAttemptContext = coreTransactionsReactive.createAttemptContext(overall,
+ merged, txnId);
+ return coreTransactionAttemptContext;
+ }
+
+ private static Duration now() {
+ return Duration.of(System.nanoTime(), ChronoUnit.NANOS);
+ }
+
+ public static ReactiveTransactionAttemptContext from(CoreTransactionAttemptContext coreTransactionAttemptContext,
+ JsonSerializer serializer) {
+ TransactionAttemptContext tac = new TransactionAttemptContext(coreTransactionAttemptContext, serializer);
+ return reactive(tac);
+ }
+
+ public static CoreTransactionAttemptContext getCore(ReactiveTransactionAttemptContext atr) {
+ CoreTransactionAttemptContext coreTransactionsReactive;
+ try {
+ Field field = ReactiveTransactionAttemptContext.class.getDeclaredField("internal");
+ field.setAccessible(true);
+ coreTransactionsReactive = (CoreTransactionAttemptContext) field.get(atr);
+ } catch (Throwable err) {
+ throw new RuntimeException(err);
+ }
+ return coreTransactionsReactive;
+ }
+
+ public static CoreTransactionAttemptContext getCore(TransactionAttemptContext atr) {
+ try {
+ Field field = TransactionAttemptContext.class.getDeclaredField("internal");
+ field.setAccessible(true);
+ return (CoreTransactionAttemptContext) field.get(atr);
+ } catch (Throwable err) {
+ throw new RuntimeException(err);
+ }
+ }
+
+ public static Mono implicitCommit(ReactiveTransactionAttemptContext atr, boolean b) {
+ CoreTransactionAttemptContext coreTransactionsReactive = getCore(atr);
+ try {
+ // getDeclaredMethod() does not find it (because of primitive arg?)
+ // CoreTransactionAttemptContext.class.getDeclaredMethod("implicitCommit", Boolean.class);
+ Method[] methods = CoreTransactionAttemptContext.class.getDeclaredMethods();
+ Method method = null;
+ for (Method m : methods) {
+ if (m.getName().equals("implicitCommit")) {
+ method = m;
+ break;
+ }
+ }
+ if (method == null) {
+ throw new RuntimeException("did not find implicitCommit method");
+ }
+ method.setAccessible(true);
+ return (Mono) method.invoke(coreTransactionsReactive, b);
+ } catch (Throwable err) {
+ throw new RuntimeException(err);
+ }
+
+ }
+
+ public static AttemptState getState(ReactiveTransactionAttemptContext atr) {
+ CoreTransactionAttemptContext coreTransactionsReactive = getCore(atr);
+ try {
+ Field field = CoreTransactionAttemptContext.class.getDeclaredField("state");
+ field.setAccessible(true);
+ return (AttemptState) field.get(coreTransactionsReactive);
+ } catch (Throwable err) {
+ throw new RuntimeException(err);
+ }
+ }
+
+ 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 transactionLogic, CoreTransactionOptions coreTransactionOptions) {
+ return reactive(transactions).runBlocking(transactionLogic, coreTransactionOptions);
+ }
+
+ CoreTransactionAttemptContext coreTransactionsReactive;
+ try {
+ Field field = TransactionAttemptContext.class.getDeclaredField("internal");
+ field.setAccessible(true);
+ coreTransactionsReactive = (CoreTransactionAttemptContext) field.get(atr);
+ } catch (Throwable err) {
+ throw new RuntimeException(err);
+ }
+ return coreTransactionsReactive;
+ }
+
+ public static ReactiveTransactionAttemptContext createReactiveTransactionAttemptContext(CoreTransactionAttemptContext core, JsonSerializer jsonSerializer) {
+ return new ReactiveTransactionAttemptContext(core, jsonSerializer);
+ }
+}
diff --git a/src/main/java/com/couchbase/transactions/TransactionsReactive.java b/src/main/java/com/couchbase/transactions/TransactionsReactive.java
new file mode 100644
index 000000000..352135ead
--- /dev/null
+++ b/src/main/java/com/couchbase/transactions/TransactionsReactive.java
@@ -0,0 +1,753 @@
+///*
+// * Copyright 2021 Couchbase, Inc.
+// *
+// * 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 com.couchbase.transactions;
+//
+//import com.couchbase.client.core.annotation.Stability;
+//import com.couchbase.client.core.cnc.EventBus;
+//import com.couchbase.client.core.retry.reactor.DefaultRetry;
+//import com.couchbase.client.core.retry.reactor.Jitter;
+//import com.couchbase.client.core.retry.reactor.RetryContext;
+//import com.couchbase.client.java.Cluster;
+//import com.couchbase.client.java.ReactiveCollection;
+//import com.couchbase.client.java.ReactiveScope;
+//import com.couchbase.client.java.json.JsonObject;
+//import com.couchbase.client.java.query.ReactiveQueryResult;
+//import com.couchbase.transactions.cleanup.ClusterData;
+//import com.couchbase.transactions.cleanup.TransactionsCleanup;
+//import com.couchbase.transactions.components.ATR;
+//import com.couchbase.transactions.components.ActiveTransactionRecord;
+//import com.couchbase.transactions.config.MergedTransactionConfig;
+//import com.couchbase.transactions.config.PerTransactionConfig;
+//import com.couchbase.transactions.config.PerTransactionConfigBuilder;
+//import com.couchbase.transactions.config.SingleQueryTransactionConfig;
+//import com.couchbase.transactions.config.SingleQueryTransactionConfigBuilder;
+//import com.couchbase.transactions.config.TransactionConfig;
+//import com.couchbase.transactions.deferred.TransactionSerializedContext;
+//import com.couchbase.transactions.error.TransactionCommitAmbiguous;
+//import com.couchbase.transactions.error.TransactionExpired;
+//import com.couchbase.transactions.error.TransactionFailedException;
+//import com.couchbase.transactions.error.internal.ErrorClasses;
+//import com.couchbase.transactions.error.external.TransactionOperationFailed;
+//import com.couchbase.transactions.forwards.Supported;
+//import com.couchbase.transactions.log.EventBusPersistedLogger;
+//import com.couchbase.transactions.log.PersistedLogWriter;
+//import com.couchbase.transactions.log.TransactionLogEvent;
+//import com.couchbase.transactions.support.AttemptContextFactory;
+//import com.couchbase.transactions.support.AttemptState;
+//import com.couchbase.transactions.support.OptionsWrapperUtil;
+//import com.couchbase.transactions.util.DebugUtil;
+//import reactor.core.publisher.Mono;
+//import reactor.core.scheduler.Schedulers;
+//
+//import java.time.Duration;
+//import java.time.temporal.ChronoUnit;
+//import java.util.Objects;
+//import java.util.Optional;
+//import java.util.UUID;
+//import java.util.concurrent.TimeUnit;
+//import java.util.concurrent.atomic.AtomicReference;
+//import java.util.function.Consumer;
+//import java.util.function.Function;
+//import java.util.function.Predicate;
+//
+//import static com.couchbase.transactions.error.internal.TransactionOperationFailedBuilder.createError;
+//import static com.couchbase.transactions.log.PersistedLogWriter.MAX_LOG_ENTRIES_DEFAULT;
+//import static com.couchbase.transactions.support.SpanWrapperUtil.DB_COUCHBASE_TRANSACTIONS;
+//
+///**
+// * An asynchronous version of {@link Transactions}, allowing transactions to be created and run in an asynchronous
+// * manner.
+// *
+// * The main method to run transactions is {@link TransactionsReactive#run}.
+// */
+//public class TransactionsReactive {
+// static final int MAX_ATTEMPTS = 1000;
+// private final TransactionsCleanup cleanup;
+// private final TransactionConfig config;
+// private AttemptContextFactory attemptContextFactory;
+// private EventBusPersistedLogger persistedLogger;
+//
+// /**
+// * This is package-private. Applications should create a {@link Transactions} object instead, and then call {@link
+// * Transactions#reactive}.
+// */
+// static TransactionsReactive create(Cluster cluster, TransactionConfig config) {
+// return new TransactionsReactive(cluster, config);
+// }
+//
+// private TransactionsReactive(Cluster cluster, TransactionConfig config) {
+// Objects.requireNonNull(cluster);
+// Objects.requireNonNull(config);
+//
+// ClusterData clusterData = new ClusterData(cluster);
+// this.config = config;
+// this.attemptContextFactory = config.attemptContextFactory();
+// MergedTransactionConfig merged = new MergedTransactionConfig(config, Optional.empty());
+// cleanup = new TransactionsCleanup(merged, clusterData);
+//
+// config.persistentLoggingCollection().ifPresent(collection -> {
+// PersistedLogWriter persistedLogWriter = new PersistedLogWriter(collection, MAX_LOG_ENTRIES_DEFAULT);
+// persistedLogger = new EventBusPersistedLogger(cluster.environment().eventBus(), persistedLogWriter, merged);
+// });
+// }
+//
+//
+// /**
+// * The main transactions 'engine', responsible for attempting the transaction logic as many times as required,
+// * until the transaction commits, is explicitly rolled back, or expires.
+// */
+// // TODO: changed from private to public. package-protected plus an accessor would be ok to
+// public Mono executeTransaction(MergedTransactionConfig config,
+// TransactionContext overall,
+// Mono transactionLogic) {
+// AtomicReference startTime = new AtomicReference<>();
+//
+// return Mono.just(overall)
+//
+// .subscribeOn(reactor.core.scheduler.Schedulers.elastic())
+//
+// .doOnSubscribe(v -> {
+// if (startTime.get() == null) startTime.set(System.nanoTime());
+// })
+//
+// // Where the magic happens: execute the app's transaction logic
+// // A ReactiveTransactionAttemptContext gets created in here. Rollback requires one of these (so it knows what
+// // to rollback), so only errors thrown inside this block can trigger rollback.
+// // So, expiry checks only get done inside this block.
+// .then(transactionLogic)
+//
+// .flatMap(this::executeImplicitCommit)
+//
+// // Track an attempt if non-error, and request that the attempt be cleaned up. Similar logic is also
+// // done in executeHandleErrorsPreRetry.
+// .doOnNext(ctx -> executeAddAttemptAndCleanupRequest(config, overall, ctx))
+//
+// // Track an attempt if error, and perform rollback if needed.
+// // All errors reaching here must be a `TransactionOperationFailed`.
+// .onErrorResume(err -> executeHandleErrorsPreRetry(config, overall, err))
+//
+// // This is the main place to retry txns. Feed all errors up to this centralised point.
+// // All errors reaching here must be a `TransactionOperationFailed`.
+// .retryWhen(executeCreateRetryWhen(overall))
+//
+// // If we're here, then we've hit an error that we don't want to retry.
+// // Either raise some derivative of TransactionFailedException to the app, or return an ReactiveTransactionAttemptContext
+// // to return success (some errors result in success, e.g. TRANSACTION_FAILED_POST_COMMIT)
+// // All errors reaching here must be an `ErrorWrapper`.
+// .onErrorResume(err -> executeHandleErrorsPostRetry(overall, err))
+//
+// .doOnError(err -> {
+// if (config.logOnFailure() && !config.logDirectly()) {
+// EventBus eventBus = cleanup.clusterData().cluster().environment().eventBus();
+// overall.LOGGER.logs().forEach(log -> {
+// eventBus.publish(new TransactionLogEvent(config.logOnFailureLevel(),
+// TransactionLogEvent.DEFAULT_CATEGORY, log.toString()));
+// });
+// }
+// })
+//
+// // If we get here, success
+// .doOnSuccess(v ->
+// overall.LOGGER.info("finished txn in %dus",
+// TimeUnit.NANOSECONDS.toMicros(System.nanoTime() - startTime.get()))
+// )
+//
+// // Safe to do single() as there will only ever be 1 result
+// .single()
+// .map(v -> createResultFromContext(overall));
+// }
+//
+// private reactor.util.retry.Retry executeCreateRetryWhen(TransactionContext overall) {
+// Predicate> predicate = context -> {
+// Throwable exception = context.exception();
+//
+// if (!(exception instanceof TransactionOperationFailed)) {
+// // A bug. Only TransactionOperationFailed is allowed to reach here.
+// throw new IllegalStateException("Non-TransactionOperationFailed '" + DebugUtil.dbg(exception) + "' received during retry, this is a bug", exception);
+// }
+//
+// TransactionOperationFailed e = (TransactionOperationFailed) exception;
+//
+// overall.LOGGER.info("TransactionOperationFailed retryTransaction=%s", e.retryTransaction());
+//
+// return e.retryTransaction();
+// };
+//
+// return DefaultRetry.create(predicate)
+//
+// .exponentialBackoff(Duration.of(1, ChronoUnit.MILLIS),
+// Duration.of(2, ChronoUnit.MILLIS))
+//
+// .doOnRetry(v -> overall.LOGGER.info("<>", "retrying transaction after backoff %dmillis", v.backoff().toMillis()))
+//
+// // Add some jitter so two txns don't livelock each other
+// .jitter(Jitter.random())
+//
+// // Really, this is a safety-guard. The txn will be aborted when it expires.
+// .retryMax(MAX_ATTEMPTS)
+//
+// .toReactorRetry();
+// }
+//
+// private Mono executeHandleErrorsPreRetry(MergedTransactionConfig config,
+// TransactionContext overall, Throwable err) {
+// if (!(err instanceof TransactionOperationFailed)) {
+// // A bug. Only TransactionOperationFailed is allowed to reach here.
+// overall.LOGGER.warn("<>", "received non-TransactionOperationFailed error %s, unable to rollback as don't have " +
+// "context", DebugUtil.dbg(err));
+// return Mono.error(new IllegalStateException("received non-TransactionOperationFailed error " + err.getClass().getName() + " in pre-retry", err));
+// }
+//
+// Mono autoRollback = Mono.empty();
+// Mono cleanupReq = Mono.empty();
+//
+// TransactionOperationFailed e = (TransactionOperationFailed) err;
+// ReactiveTransactionAttemptContext ctx = e.context();
+//
+// overall.LOGGER.info("<>", "finishing attempt off after error '%s'", e);
+//
+// if (e.autoRollbackAttempt()) {
+// // In queryMode we always ROLLBACK, as there is possibly delta table state to cleanup, and there may be an
+// // ATR - we don't know
+// if (ctx.state() == AttemptState.NOT_STARTED && !ctx.queryMode()) {
+// // This is a better way of doing [RETRY-ERR-NOATR] and likely means that the older logic for
+// // handling that won't trigger now
+// ctx.LOGGER.info(ctx.attemptId(), "told to auto-rollback but in NOT_STARTED state, so nothing to do - skipping rollback");
+// }
+// else {
+// ctx.LOGGER.info(ctx.attemptId(), "auto-rolling-back on error");
+//
+// autoRollback = ctx.rollbackInternal(false);
+// }
+// } else {
+// ctx.LOGGER.info(ctx.attemptId(), "has been told to skip auto-rollback");
+// }
+//
+// if (!config.runRegularAttemptsCleanupThread()) {
+// // Don't add a request to a queue that no-one will be processing
+// ctx.LOGGER.trace(ctx.attemptId(), "skipping addition of cleanup request on failure as regular cleanup disabled");
+// }
+// else {
+// cleanupReq = Mono.fromRunnable(() -> addCleanupRequestForContext(ctx));
+// }
+//
+// Mono addAttempt = Mono.fromRunnable(() -> {
+// TransactionAttempt ta = TransactionAttempt.createFromContext(ctx, Optional.of(err));
+// overall.addAttempt(ta);
+// ctx.LOGGER.info(ctx.attemptId(), "added attempt %s after error", ta);
+// });
+//
+// final Mono cleanupReqForLambda = cleanupReq;
+//
+// return autoRollback
+// // See [Primary Operations] section in design document
+// .onErrorResume(er -> {
+// overall.LOGGER.info("<>", "rollback failed with %s, raising original error but with retryTransaction turned off",
+// DebugUtil.dbg(er));
+//
+// // Still want to add attempt and cleanup request
+// return cleanupReqForLambda
+// .then(addAttempt)
+// .then(Mono.error(createError(e.context(), e.causingErrorClass())
+// .raiseException(e.toRaise())
+// .cause(e.getCause())
+// .build()));
+// })
+// .then(cleanupReqForLambda)
+// // Only want to add the attempt after doing the rollback, so the attempt has the correct state (hopefully
+// // ROLLED_BACK)
+// .then(addAttempt)
+// .then(Mono.defer(() -> {
+// if (e.retryTransaction() && overall.hasExpiredClientSide()) {
+// overall.LOGGER.info("<>", "original error planned to retry transaction, but it has subsequently expired");
+// return Mono.error(createError(ctx, ErrorClasses.FAIL_EXPIRY)
+// .doNotRollbackAttempt()
+// .raiseException(TransactionOperationFailed.FinalErrorToRaise.TRANSACTION_EXPIRED)
+// .build());
+// }
+// else {
+// // Raise the error up the stack so the logic later can decide whether to retry the transaction
+// overall.LOGGER.info("<>", "reraising original exception %s", DebugUtil.dbg(err));
+// return Mono.error(err);
+// }
+// }))
+// .doFinally(v -> ctx.span().failWith(e))
+// .thenReturn(ctx);
+// }
+//
+// private Mono executeHandleErrorsPostRetry(TransactionContext overall, Throwable err) {
+// if (!(err instanceof TransactionOperationFailed)) {
+// // A bug. Only TransactionOperationFailed is allowed to reach here.
+// return Mono.error(new IllegalStateException("Non-TransactionOperationFailed '" + DebugUtil.dbg(err) + "' received, this is a bug"));
+// }
+//
+// TransactionResult result = createResultFromContext(overall);
+// TransactionOperationFailed e = (TransactionOperationFailed) err;
+//
+// if (e.toRaise() == TransactionOperationFailed.FinalErrorToRaise.TRANSACTION_FAILED_POST_COMMIT) {
+// e.context().LOGGER.info(e.context().attemptId(), "converted TRANSACTION_FAILED_POST_COMMIT to success, unstagingComplete() will be false");
+//
+// return Mono.just(e.context());
+// }
+// else {
+// TransactionFailedException ret;
+//
+// switch (e.toRaise()) {
+// case TRANSACTION_EXPIRED: {
+// String msg = "Transaction has expired configured timeout of " + overall.expirationTime().toMillis() + "msecs. The transaction is not committed.";
+// ret = new TransactionExpired(e.getCause(), result, msg);
+// break;
+// }
+// case TRANSACTION_COMMIT_AMBIGUOUS: {
+// String msg = "It is ambiguous whether the transaction committed";
+// ret = new TransactionCommitAmbiguous(e.getCause(), result, msg);
+// break;
+// }
+// default:
+// ret = new TransactionFailedException(e.getCause(), result);
+// break;
+// }
+//
+// e.context().LOGGER.info(e.context().attemptId(), "converted TransactionOperationFailed %s to final error %s",
+// e.toRaise(), ret);
+//
+// return Mono.error(ret);
+// }
+// }
+//
+// private void executeAddAttemptAndCleanupRequest(MergedTransactionConfig config, TransactionContext overall,
+// ReactiveTransactionAttemptContext ctx) {
+// TransactionAttempt ta = TransactionAttempt.createFromContext(ctx, Optional.empty());
+// overall.addAttempt(ta);
+// ctx.LOGGER.info(ctx.attemptId(), "added attempt %s after success", ta);
+//
+// if (config.runRegularAttemptsCleanupThread()) {
+// addCleanupRequestForContext(ctx);
+// } else {
+// ctx.LOGGER.trace(ctx.attemptId(), "skipping addition of cleanup request on success");
+// }
+//
+// ctx.span().finish();
+// }
+//
+// private Mono executeImplicitCommit(ReactiveTransactionAttemptContext ctx) {
+// return Mono.defer(() -> {
+// // If app has not explicitly performed a commit, assume they want to do so anyway
+// if (!ctx.isDone()) {
+// if (ctx.serialized().isPresent()) {
+// return Mono.just(ctx);
+// } else {
+// ctx.LOGGER.trace(ctx.attemptId(), "doing implicit commit");
+//
+// return ctx.commit()
+// .then(Mono.just(ctx))
+// .onErrorResume(err -> Mono.error(TransactionOperationFailed.convertToOperationFailedIfNeeded(err, ctx)));
+// }
+// } else {
+// return Mono.just(ctx);
+// }
+// });
+// }
+//
+// // TODO: changed from package-protected to public (could have just used an accessor class in same package)
+// public ReactiveTransactionAttemptContext createAttemptContext(TransactionContext overall,
+// MergedTransactionConfig config,
+// String attemptId) {
+// // null only happens in testing with Mockito, harmless
+// if (overall != null) {
+// return attemptContextFactory.create(overall, config, attemptId, this, Optional.of(overall.span()));
+// } else {
+// return null;
+// }
+// }
+//
+// /**
+// * Runs the supplied transactional logic until success or failure.
+// *
+// * This is the asynchronous version of {@link Transactions#run}, so to cover the differences:
+// *
+// *
The transaction logic is supplied with a {@link ReactiveTransactionAttemptContext}, which contains asynchronous
+// * methods to allow it to read, mutate, insert and delete documents, as well as commit or rollback the
+// * transactions.
+// *
The transaction logic should run these methods as a Reactor chain.
+// *
The transaction logic should return a Mono{@literal <}Void{@literal >}. Any
+// * Flux or Mono can be converted to a Mono{@literal <}Void{@literal >} by
+// * calling .then() on it.
+// *
This method returns a Mono{@literal <}TransactionResult{@literal >}, which should be handled
+// * as a normal Reactor Mono.
+// *
+// *
+// * @param transactionLogic the application's transaction logic
+// * @param perConfig the configuration to use for this transaction
+// * @return there is no need to check the returned {@link TransactionResult}, as success is implied by the lack of a
+// * thrown exception. It contains information useful only for debugging and logging.
+// * @throws TransactionFailedException or a derived exception if the transaction fails to commit for any reason, possibly
+// * after multiple retries. The exception contains further details of the error. Not
+// */
+// public Mono run(Function> transactionLogic,
+// PerTransactionConfig perConfig) {
+// return Mono.defer(() -> {
+// MergedTransactionConfig merged = new MergedTransactionConfig(config, Optional.of(perConfig));
+//
+// TransactionContext overall =
+// new TransactionContext(cleanup.clusterData().cluster().environment().requestTracer(),
+// cleanup.clusterData().cluster().environment().eventBus(),
+// UUID.randomUUID().toString(),
+// now(),
+// Duration.ZERO,
+// merged);
+// AtomicReference startTime = new AtomicReference<>(0L);
+//
+// Mono ob = Mono.fromCallable(() -> {
+// String txnId = UUID.randomUUID().toString();
+// overall.LOGGER.info(configDebug(config, perConfig));
+// return createAttemptContext(overall, merged, txnId);
+// }).flatMap(ctx -> {
+// ctx.LOGGER.info("starting attempt %d/%s/%s",
+// overall.numAttempts(), ctx.transactionId(), ctx.attemptId());
+// Mono result = transactionLogic.apply(ctx);
+// return result
+// .onErrorResume(err -> {
+// ctx.LOGGER.info(ctx.attemptId(), "caught exception '%s' in async, rethrowing", err);
+// logElidedStacktrace(ctx, err);
+//
+// return Mono.error(TransactionOperationFailed.convertToOperationFailedIfNeeded(err, ctx));
+// })
+// .thenReturn(ctx);
+// }).doOnSubscribe(v -> startTime.set(System.nanoTime()))
+// .doOnNext(v -> v.LOGGER.trace(v.attemptId(), "finished attempt %d in %sms",
+// overall.numAttempts(), (System.nanoTime() - startTime.get()) / 1_000_000));
+//
+// return executeTransaction(merged, overall, ob)
+// .doOnNext(v -> overall.span().finish())
+// .doOnError(err -> overall.span().failWith(err));
+// });
+// }
+//
+// // Printing the stacktrace is expensive in terms of log noise, but has been a life saver on many debugging
+// // encounters. Strike a balance by eliding the more useless elements.
+// // TODO: changed from private to public
+// public void logElidedStacktrace(ReactiveTransactionAttemptContext ctx, Throwable err) {
+// DebugUtil.fetchElidedStacktrace(err, (s) -> ctx.LOGGER.info(ctx.attemptId(), " " + s.toString()));
+// }
+//
+// // TODO: changed from private to public
+// public static String configDebug(TransactionConfig config, PerTransactionConfig perConfig) {
+// StringBuilder sb = new StringBuilder();
+// sb.append("library version: ");
+// sb.append(TransactionsReactive.class.getPackage().getImplementationVersion());
+// sb.append(" config: ");
+// sb.append("atrs=");
+// sb.append(config.numAtrs());
+// sb.append(", metadataCollection=");
+// sb.append(config.metadataCollection());
+// sb.append(", expiry=");
+// sb.append(perConfig.expirationTime().orElse(config.transactionExpirationTime()).toMillis());
+// sb.append("msecs durability=");
+// sb.append(config.durabilityLevel());
+// sb.append(" per-txn config=");
+// sb.append(" durability=");
+// sb.append(perConfig.durabilityLevel());
+// sb.append(", supported=");
+// sb.append(Supported.SUPPORTED);
+// return sb.toString();
+// }
+//
+// /**
+// * Convenience overload that runs {@link TransactionsReactive#run} with a default PerTransactionConfig.
+// */
+// public Mono run(Function> transactionLogic) {
+// return run(transactionLogic, PerTransactionConfigBuilder.create().build());
+// }
+//
+// @Stability.Volatile
+// public Mono commit(TransactionSerializedContext serialized, PerTransactionConfig perConfig) {
+// return deferred(serialized,
+// perConfig,
+// // Nothing to actually do, just want the implicit commit
+// (ctx) -> Mono.empty());
+// }
+//
+// @Stability.Volatile
+// public Mono rollback(TransactionSerializedContext serialized, PerTransactionConfig perConfig) {
+// return deferred(serialized,
+// perConfig,
+// (ctx) -> ctx.rollback());
+// }
+//
+// @Stability.Volatile
+// private Mono deferred(TransactionSerializedContext serialized,
+// PerTransactionConfig perConfig,
+// Function> initial) {
+// MergedTransactionConfig merged = new MergedTransactionConfig(config, Optional.of(perConfig));
+// JsonObject hydrated = JsonObject.fromJson(serialized.encodeAsString());
+//
+// String atrBucket = hydrated.getString("atrBucket");
+// String atrScope = hydrated.getString("atrScope");
+// String atrCollectionName = hydrated.getString("atrCollection");
+// String atrId = hydrated.getString("atrId");
+// ReactiveCollection atrCollection = cleanup.clusterData()
+// .getBucketFromName(atrBucket)
+// .scope(atrScope)
+// .collection(atrCollectionName);
+//
+// return ActiveTransactionRecord.getAtr(atrCollection,
+// atrId,
+// OptionsWrapperUtil.kvTimeoutNonMutating(merged, atrCollection.core()),
+// null)
+//
+// .flatMap(atrOpt -> {
+// if (!atrOpt.isPresent()) {
+// return Mono.error(new IllegalStateException(String.format("ATR %s/%s could not be found",
+// atrBucket, atrId)));
+// }
+// else {
+// ATR atr = atrOpt.get();
+//
+// // Note startTimeServerMillis is written with ${Mutation.CAS} while currentTimeServer
+// // could have come from $vbucket.HLC and is hence one-second granularity. So, this is a
+// // somewhat imperfect comparison.
+// Duration currentTimeServer = Duration.ofNanos(atr.cas());
+// Duration startTimeServer = Duration.ofMillis(hydrated.getLong("startTimeServerMillis"));
+//
+// // This includes the time elapsed during the first part of the transaction, plus any time
+// // elapsed during the period the transaction was expired. Total time since the transaction
+// // began, basically.
+// Duration timeElapsed = currentTimeServer.minus(startTimeServer);
+//
+// TransactionContext overall =
+// new TransactionContext(cleanup.clusterData().cluster().environment().requestTracer(),
+// cleanup.clusterData().cluster().environment().eventBus(),
+// UUID.randomUUID().toString(),
+// Duration.ofNanos(System.nanoTime()),
+// timeElapsed,
+// merged);
+// AtomicReference startTime = new AtomicReference<>(0L);
+//
+// overall.LOGGER.info("elapsed time = %dmsecs (ATR start time %dmsecs, current ATR time %dmsecs)",
+// timeElapsed.toMillis(), startTimeServer.toMillis(), currentTimeServer.toMillis());
+//
+// Mono ob = Mono.defer(() -> {
+// ReactiveTransactionAttemptContext ctx = attemptContextFactory.createFrom(hydrated, overall, merged, this);
+// ctx.LOGGER.info("starting attempt %d/%s/%s",
+// overall.numAttempts(), ctx.transactionId(), ctx.attemptId());
+// ctx.LOGGER.info(configDebug(config, perConfig));
+//
+// return initial.apply(ctx)
+//
+// // TXNJ-50: Make sure we run user's blocking logic on a scheduler that can take it
+// .subscribeOn(Schedulers.elastic())
+//
+// .onErrorResume(err -> {
+// ctx.LOGGER.info(ctx.attemptId(), "caught exception '%s' in deferred, rethrowing",
+// err);
+//
+// logElidedStacktrace(ctx, err);
+//
+// return Mono.error(TransactionOperationFailed.convertToOperationFailedIfNeeded(err, ctx));
+// })
+//
+// .doOnSubscribe(v -> startTime.set(System.nanoTime()))
+//
+// .doOnNext(v -> {
+// ctx.LOGGER.trace(ctx.attemptId(), "finished attempt %d in %sms",
+// overall.numAttempts(), (System.nanoTime() - startTime.get()) / 1_000_000);
+// })
+//
+// .thenReturn(ctx);
+// });
+//
+// return executeTransaction(merged, overall, ob)
+// .doOnNext(v -> overall.span().attribute(DB_COUCHBASE_TRANSACTIONS + "retries", overall.numAttempts()).finish())
+// .doOnError(err -> overall.span().attribute(DB_COUCHBASE_TRANSACTIONS + "retries", overall.numAttempts()).failWith(err));
+// }
+// });
+// }
+//
+// Mono runBlocking(Consumer txnLogic, PerTransactionConfig perConfig) {
+// return Mono.defer(() -> {
+// MergedTransactionConfig merged = new MergedTransactionConfig(config, Optional.of(perConfig));
+// TransactionContext overall =
+// new TransactionContext(cleanup.clusterData().cluster().environment().requestTracer(),
+// cleanup.clusterData().cluster().environment().eventBus(),
+// UUID.randomUUID().toString(),
+// now(),
+// Duration.ZERO,
+// merged);
+// AtomicReference startTime = new AtomicReference<>(0L);
+// overall.LOGGER.info(configDebug(config, perConfig));
+//
+// Mono ob = Mono.defer(() -> {
+// String txnId = UUID.randomUUID().toString();
+// ReactiveTransactionAttemptContext ctx = createAttemptContext(overall, merged, txnId);
+// TransactionAttemptContext ctxBlocking = new TransactionAttemptContext(ctx);
+// ctx.LOGGER.info("starting attempt %d/%s/%s",
+// overall.numAttempts(), ctx.transactionId(), ctx.attemptId());
+//
+// return Mono.fromRunnable(() -> txnLogic.accept(ctxBlocking))
+//
+// // TXNJ-50: Make sure we run user's blocking logic on a scheduler that can take it
+// .subscribeOn(Schedulers.elastic())
+//
+// .onErrorResume(err -> {
+// ctx.LOGGER.info(ctx.attemptId(), "caught exception '%s' in runBlocking, rethrowing", err);
+//
+// logElidedStacktrace(ctx, err);
+//
+// return Mono.error(TransactionOperationFailed.convertToOperationFailedIfNeeded(err, ctx));
+// })
+//
+// .doOnSubscribe(v -> startTime.set(System.nanoTime()))
+//
+// .doOnNext(v -> {
+// ctx.LOGGER.trace(ctx.attemptId(), "finished attempt %d in %sms",
+// overall.numAttempts(), (System.nanoTime() - startTime.get()) / 1_000_000);
+// })
+//
+// .thenReturn(ctx);
+// });
+//
+// return executeTransaction(merged, overall, ob)
+// .doOnNext(v -> overall.span().attribute(DB_COUCHBASE_TRANSACTIONS + "retries", overall.numAttempts()).finish())
+// .doOnError(err -> overall.span().attribute(DB_COUCHBASE_TRANSACTIONS + "retries", overall.numAttempts()).failWith(err));
+// });
+// }
+//
+// public TransactionConfig config() {
+// return config;
+// }
+//
+// private static Duration now() {
+// return Duration.of(System.nanoTime(), ChronoUnit.NANOS);
+// }
+//
+// TransactionsCleanup cleanup() {
+// return cleanup;
+// }
+//
+// private void addCleanupRequestForContext(ReactiveTransactionAttemptContext ctx) {
+// // Whether the txn was successful or not, still want to clean it up
+// if (ctx.queryMode()) {
+// ctx.LOGGER.info(ctx.attemptId(), "Skipping cleanup request as in query mode");
+// }
+// else if (ctx.serialized().isPresent()) {
+// ctx.LOGGER.info(ctx.attemptId(), "Skipping cleanup request as deferred transaction");
+// }
+// else if (ctx.atrId().isPresent() && ctx.atrCollection().isPresent()) {
+// switch (ctx.state()) {
+// case NOT_STARTED:
+// case COMPLETED:
+// case ROLLED_BACK:
+// ctx.LOGGER.trace(ctx.attemptId(), "Skipping addition of cleanup request in state %s", ctx.state());
+// break;
+// default:
+// ctx.LOGGER.trace(ctx.attemptId(), "Adding cleanup request for %s/%s",
+// ctx.atrCollection().get().name(), ctx.atrId().get());
+//
+// cleanup.add(ctx.createCleanupRequest());
+// }
+// } else {
+// // No ATR entry to remove
+// ctx.LOGGER.trace(ctx.attemptId(), "Skipping cleanup request as no ATR entry to remove (due to no " +
+// "mutations)");
+// }
+// }
+//
+// private static TransactionResult createResultFromContext(TransactionContext overall) {
+// return new TransactionResult(overall.attempts(),
+// overall.LOGGER,
+// Duration.of(System.nanoTime() - overall.startTimeClient().toNanos(), ChronoUnit.NANOS),
+// overall.transactionId(),
+// overall.serialized());
+// }
+//
+// /**
+// * Performs a single query transaction, with default configuration.
+// *
+// * @param statement the statement to execute.
+// * @return a ReactiveSingleQueryTransactionResult
+// */
+// @Stability.Uncommitted
+// public Mono query(String statement) {
+// return query(null, statement, SingleQueryTransactionConfigBuilder.create().build());
+// }
+//
+// /**
+// * Performs a single query transaction, with a custom configuration.
+// *
+// * @param statement the statement to execute.
+// * @param queryOptions configuration options.
+// * @return a ReactiveSingleQueryTransactionResult
+// */
+// @Stability.Uncommitted
+// public Mono query(String statement, SingleQueryTransactionConfig queryOptions) {
+// return query(null, statement, queryOptions);
+// }
+//
+// /**
+// * Performs a single query transaction, with a scope context and default configuration.
+// *
+// * @param statement the statement to execute.
+// * @param scope the query will be executed in the context of this scope, so it can refer to a collection on this scope
+// * rather than needed to provide the full keyspace.
+// * @return a ReactiveSingleQueryTransactionResult
+// */
+// @Stability.Uncommitted
+// public Mono query(ReactiveScope scope, String statement) {
+// return query(scope, statement, SingleQueryTransactionConfigBuilder.create().build());
+// }
+//
+// /**
+// * Performs a single query transaction, with a scope context and custom configuration.
+// *
+// * @param statement the statement to execute.
+// * @param scope the query will be executed in the context of this scope, so it can refer to a collection on this scope
+// * rather than needed to provide the full keyspace.
+// * @param queryOptions configuration options.
+// * @return a ReactiveSingleQueryTransactionResult
+// */
+// @Stability.Uncommitted
+// public Mono query(ReactiveScope scope, String statement, SingleQueryTransactionConfig queryOptions) {
+// return Mono.defer(() -> {
+// AtomicReference queryResult = new AtomicReference<>();
+// return run((ctx) -> ctx.query(scope, statement, queryOptions.queryOptions(), true)
+// .doOnNext(qr -> queryResult.set(qr))
+// .then(), queryOptions.convert())
+// .map(result -> new ReactiveSingleQueryTransactionResult(result.log(), queryResult.get()));
+// });
+// }
+//
+// @Stability.Internal
+// @Deprecated // Prefer setting TransactionConfigBuilder#testFactories now
+// public void setAttemptContextFactory(AttemptContextFactory factory) {
+// this.attemptContextFactory = factory;
+// }
+// public ReactiveTransactionAttemptContext newAttemptContextReactive(){
+// PerTransactionConfig perConfig = PerTransactionConfigBuilder.create().build();
+// MergedTransactionConfig merged = new MergedTransactionConfig(config, Optional.of(perConfig));
+//
+// TransactionContext overall = new TransactionContext(
+// cleanup().clusterData().cluster().environment().requestTracer(),
+// cleanup().clusterData().cluster().environment().eventBus(),
+// UUID.randomUUID().toString(), now(), Duration.ZERO, merged);
+//
+// String txnId = UUID.randomUUID().toString();
+// overall.LOGGER.info(configDebug(config, perConfig));
+// return createAttemptContext(overall, merged, txnId);
+// }
+//
+//}
diff --git a/src/main/java/com/example/demo/CouchbaseTransactionManager.pre-core b/src/main/java/com/example/demo/CouchbaseTransactionManager.pre-core
new file mode 100644
index 000000000..ab9d84087
--- /dev/null
+++ b/src/main/java/com/example/demo/CouchbaseTransactionManager.pre-core
@@ -0,0 +1,201 @@
+package com.example.demo;
+
+import java.util.concurrent.atomic.AtomicReference;
+
+import com.couchbase.client.java.transactions.ReactiveTransactionAttemptContext;
+import com.couchbase.client.java.transactions.TransactionAttemptContext;
+import com.couchbase.client.java.transactions.AttemptContextReactiveAccessor;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+import org.springframework.beans.factory.DisposableBean;
+import org.springframework.data.couchbase.core.CouchbaseTemplate;
+import org.springframework.data.couchbase.transaction.ClientSession;
+import org.springframework.data.couchbase.transaction.ClientSessionImpl;
+import org.springframework.transaction.TransactionDefinition;
+import org.springframework.transaction.TransactionException;
+import org.springframework.transaction.support.AbstractPlatformTransactionManager;
+import org.springframework.transaction.support.CallbackPreferringPlatformTransactionManager;
+import org.springframework.transaction.support.DefaultTransactionStatus;
+import org.springframework.transaction.support.ResourceHolderSupport;
+import org.springframework.transaction.support.ResourceTransactionManager;
+import org.springframework.transaction.support.SmartTransactionObject;
+import org.springframework.transaction.support.TransactionCallback;
+import org.springframework.transaction.support.TransactionSynchronizationManager;
+import org.springframework.transaction.support.TransactionSynchronizationUtils;
+import org.springframework.util.Assert;
+
+// todo gp why is there separate CouchbaseCallbackTransactionManager if this class also extends CallbackPreferringPlatformTransactionManager?
+// todo gp there is another CouchbaseTransactionManager in another package, which is valid?
+public class CouchbaseTransactionManager extends AbstractPlatformTransactionManager
+ implements DisposableBean, ResourceTransactionManager, CallbackPreferringPlatformTransactionManager {
+
+ private static final Logger LOGGER = LoggerFactory.getLogger(CouchbaseTransactionManager.class);
+
+ private final CouchbaseTemplate template;
+
+ public CouchbaseTransactionManager(CouchbaseTemplate template) {
+ this.template = template;
+ }
+
+ public CouchbaseTransactionalTemplate template() {
+ return new CouchbaseTransactionalTemplate(template);
+ }
+
+ @Override
+ public T execute(TransactionDefinition definition, TransactionCallback callback) throws TransactionException {
+ final AtomicReference result = new AtomicReference<>();
+ // todo gp like CouchbaseCallbackTransactionManager, it needs access to CouchbaseClientFactory here (Cluster)
+// TransactionResult txnResult = transactions.run(attemptContext -> {
+//
+// if (TransactionSynchronizationManager.hasResource(template.getCouchbaseClientFactory())) {
+// ((CouchbaseResourceHolder) TransactionSynchronizationManager
+// .getResource(template.reactive().getCouchbaseClientFactory()))
+// .setAttemptContext(attemptContext);
+// } else {
+// TransactionSynchronizationManager.bindResource(
+// template.reactive().getCouchbaseClientFactory(),
+// new CouchbaseResourceHolder(attemptContext)
+// );
+// }
+//
+// try {
+// // Since we are on a different thread now transparently, at least make sure
+// // that the original method invocation is synchronized.
+// synchronized (this) {
+// result.set(callback.doInTransaction(null));
+// }
+// } catch (RuntimeException e) {
+// System.err.println("RuntimeException: "+e+" instanceof RuntimeException "+(e instanceof RuntimeException));
+// throw e;
+// } catch (Throwable e) {
+// System.err.println("RuntimeException: "+e+" instanceof "+(e instanceof Throwable));
+// throw new RuntimeException(e);
+// }
+// });
+
+// LOGGER.debug("Completed Couchbase Transaction with Result: " + txnResult);
+ return result.get();
+ }
+
+ @Override
+ protected CouchbaseTransactionObject doGetTransaction() throws TransactionException {
+ CouchbaseResourceHolder resourceHolder = (CouchbaseResourceHolder) TransactionSynchronizationManager
+ .getResource(template.getCouchbaseClientFactory());
+ return new CouchbaseTransactionObject(resourceHolder);
+ }
+
+ @Override
+ protected boolean isExistingTransaction(Object transaction) throws TransactionException {
+ return extractTransaction(transaction).hasResourceHolder();
+ }
+
+ @Override
+ protected void doBegin(Object transaction, TransactionDefinition definition) throws TransactionException {
+ LOGGER.debug("Beginning Couchbase Transaction with Definition {}", definition);
+ }
+
+ @Override
+ protected void doCommit(DefaultTransactionStatus status) throws TransactionException {
+ LOGGER.debug("Committing Couchbase Transaction with status {}", status);
+ }
+
+ @Override
+ protected void doRollback(DefaultTransactionStatus status) throws TransactionException {
+ LOGGER.warn("Rolling back Couchbase Transaction with status {}", status);
+ }
+
+ @Override
+ protected void doCleanupAfterCompletion(Object transaction) {
+ LOGGER.trace("Performing cleanup of Couchbase Transaction {}", transaction);
+ }
+
+ @Override
+ public void destroy() {
+ }
+
+ @Override
+ public Object getResourceFactory() {
+ return template.getCouchbaseClientFactory();
+ }
+
+ private static CouchbaseTransactionObject extractTransaction(Object transaction) {
+ Assert.isInstanceOf(CouchbaseTransactionObject.class, transaction,
+ () -> String.format("Expected to find a %s but it turned out to be %s.", CouchbaseTransactionObject.class,
+ transaction.getClass()));
+
+ return (CouchbaseTransactionObject) transaction;
+ }
+
+ public static class CouchbaseResourceHolder extends ResourceHolderSupport {
+
+ private volatile TransactionAttemptContext attemptContext;
+ private volatile ReactiveTransactionAttemptContext attemptContextReactive;
+ private volatile ClientSession session = new ClientSessionImpl();
+
+ public CouchbaseResourceHolder(TransactionAttemptContext attemptContext) {
+ this.attemptContext = attemptContext;
+ }
+
+ public TransactionAttemptContext getAttemptContext() {
+ return attemptContext;
+ }
+
+ public void setAttemptContext(TransactionAttemptContext attemptContext) {
+ this.attemptContext = attemptContext;
+ }
+
+ public ReactiveTransactionAttemptContext getAttemptContextReactive() {
+ return attemptContext!= null ? AttemptContextReactiveAccessor.getACR(attemptContext) : attemptContextReactive;
+ }
+ public void setAttemptContextReactive(ReactiveTransactionAttemptContext attemptContextReactive) {
+ this.attemptContextReactive = attemptContextReactive;
+ }
+
+ public ClientSession getSession() {
+ return session;
+ }
+
+ public void setSession(ClientSession session){
+ this.session = session;
+ }
+
+ @Override
+ public String toString() {
+ return "CouchbaseResourceHolder{" +
+ "attemptContext=" + attemptContext +
+ '}';
+ }
+
+ }
+
+ protected static class CouchbaseTransactionObject implements SmartTransactionObject {
+
+ final CouchbaseResourceHolder resourceHolder;
+
+ CouchbaseTransactionObject(CouchbaseResourceHolder resourceHolderIn) {
+ resourceHolder = resourceHolderIn;
+ }
+
+ @Override
+ public boolean isRollbackOnly() {
+ return resourceHolder != null && resourceHolder.isRollbackOnly();
+ }
+
+ @Override
+ public void flush() {
+ TransactionSynchronizationUtils.triggerFlush();
+ }
+
+ public boolean hasResourceHolder() {
+ return resourceHolder != null;
+ }
+
+ @Override
+ public String toString() {
+ return "CouchbaseTransactionObject{" +
+ "resourceHolder=" + resourceHolder +
+ '}';
+ }
+ }
+
+}
diff --git a/src/main/java/com/example/demo/CouchbaseTransactionalTemplate.java b/src/main/java/com/example/demo/CouchbaseTransactionalTemplate.java
new file mode 100644
index 000000000..c0098a9d8
--- /dev/null
+++ b/src/main/java/com/example/demo/CouchbaseTransactionalTemplate.java
@@ -0,0 +1,67 @@
+package com.example.demo;
+
+import com.couchbase.client.core.transaction.CoreTransactionAttemptContext;
+import com.couchbase.client.core.transaction.CoreTransactionGetResult;
+import com.couchbase.client.java.codec.Transcoder;
+import com.couchbase.client.java.transactions.ReactiveTransactionAttemptContext;
+import com.couchbase.client.java.transactions.TransactionAttemptContext;
+import com.couchbase.client.java.transactions.TransactionGetResult;
+import org.springframework.data.couchbase.core.CouchbaseTemplate;
+import org.springframework.data.couchbase.core.ReactiveCouchbaseTemplate;
+import org.springframework.data.couchbase.transaction.ReactiveCouchbaseResourceHolder;
+import org.springframework.transaction.support.TransactionSynchronizationManager;
+
+import static com.couchbase.client.java.transactions.internal.ConverterUtil.makeCollectionIdentifier;
+
+public class CouchbaseTransactionalTemplate {
+
+ private final CouchbaseTemplate template;
+
+ public CouchbaseTransactionalTemplate(CouchbaseTemplate template) {
+ this.template = template;
+ }
+
+ public SpringTransactionGetResult findById(String id, Class domainType) {
+ try {
+ CoreTransactionAttemptContext ctx = getContext();
+ CoreTransactionGetResult getResult = ctx.get( makeCollectionIdentifier(template.getCouchbaseClientFactory().getDefaultCollection().async()) , id).block();
+
+ T t = template.support().decodeEntity(id, new String(getResult.contentAsBytes()), getResult.cas(), domainType,
+ null, null, null);
+ return new SpringTransactionGetResult<>(t, getResult);
+ } catch (Exception e) {
+ e.printStackTrace();
+ throw e;
+ }
+
+ }
+
+ public void replaceById(CoreTransactionGetResult getResult, T entity) {
+ CoreTransactionAttemptContext ctx = getContext();
+ Transcoder transCoder = template.getCouchbaseClientFactory().getCluster().environment().transcoder();
+ Transcoder.EncodedValue encoded = transCoder.encode(template.support().encodeEntity(entity).export());
+ ctx.replace(getResult, encoded.encoded());
+ }
+
+ private CoreTransactionAttemptContext getContext() {
+ ReactiveCouchbaseResourceHolder resource = (ReactiveCouchbaseResourceHolder) TransactionSynchronizationManager
+ .getResource(template.getCouchbaseClientFactory());
+ CoreTransactionAttemptContext atr;
+ if (resource != null) {
+ atr = resource.getCore();
+ } else {
+ ReactiveCouchbaseResourceHolder holder = (ReactiveCouchbaseResourceHolder) TransactionSynchronizationManager
+ .getResource(template.getCouchbaseClientFactory().getCluster());
+ atr = holder.getCore();
+ }
+ return atr;
+ }
+
+
+ public static ReactiveCouchbaseResourceHolder getSession(ReactiveCouchbaseTemplate template) {
+ ReactiveCouchbaseResourceHolder resource = (ReactiveCouchbaseResourceHolder) TransactionSynchronizationManager
+ .getResource(template.getCouchbaseClientFactory());
+ return resource;
+ }
+
+}
diff --git a/src/main/java/com/example/demo/SpringTransactionGetResult.java b/src/main/java/com/example/demo/SpringTransactionGetResult.java
new file mode 100644
index 000000000..27ede4aaf
--- /dev/null
+++ b/src/main/java/com/example/demo/SpringTransactionGetResult.java
@@ -0,0 +1,31 @@
+package com.example.demo;
+
+import com.couchbase.client.core.transaction.CoreTransactionGetResult;
+import com.couchbase.client.java.transactions.TransactionGetResult;
+
+public class SpringTransactionGetResult {
+
+ private final T value;
+ private final CoreTransactionGetResult inner;
+
+ public SpringTransactionGetResult(T value, CoreTransactionGetResult inner) {
+ this.value = value;
+ this.inner = inner;
+ }
+
+ public T getValue() {
+ return value;
+ }
+
+ public CoreTransactionGetResult getInner() {
+ return inner;
+ }
+
+ @Override
+ public String toString() {
+ return "SpringTransactionGetResult{" +
+ "value=" + value +
+ ", inner=" + inner +
+ '}';
+ }
+}
diff --git a/src/main/java/org/springframework/data/couchbase/CouchbaseClientFactory.java b/src/main/java/org/springframework/data/couchbase/CouchbaseClientFactory.java
index 51781a7d8..70a6c9227 100644
--- a/src/main/java/org/springframework/data/couchbase/CouchbaseClientFactory.java
+++ b/src/main/java/org/springframework/data/couchbase/CouchbaseClientFactory.java
@@ -18,6 +18,9 @@
import java.io.Closeable;
+import com.couchbase.client.core.transaction.CoreTransactionAttemptContext;
+import com.couchbase.client.java.transactions.ReactiveTransactionAttemptContext;
+import com.couchbase.client.java.transactions.config.TransactionOptions;
import org.springframework.dao.support.PersistenceExceptionTranslator;
import com.couchbase.client.java.Bucket;
@@ -73,4 +76,10 @@ public interface CouchbaseClientFactory extends Closeable {
*/
PersistenceExceptionTranslator getExceptionTranslator();
+ CoreTransactionAttemptContext getCore(TransactionOptions options,
+ CoreTransactionAttemptContext atr);
+
+ //CouchbaseClientFactory with(CouchbaseTransactionalOperator txOp);
+
+ //CouchbaseTransactionalOperator getTransactionalOperator();
}
diff --git a/src/main/java/org/springframework/data/couchbase/ReactiveCouchbaseClientFactory.java b/src/main/java/org/springframework/data/couchbase/ReactiveCouchbaseClientFactory.java
new file mode 100644
index 000000000..13f0f9773
--- /dev/null
+++ b/src/main/java/org/springframework/data/couchbase/ReactiveCouchbaseClientFactory.java
@@ -0,0 +1,130 @@
+/*
+ * 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.core.transaction.CoreTransactionAttemptContext;
+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.client.java.transactions.ReactiveTransactionAttemptContext;
+import com.couchbase.client.java.transactions.config.TransactionOptions;
+import org.springframework.data.couchbase.transaction.CouchbaseTransactionalOperator;
+import org.springframework.data.couchbase.transaction.ReactiveCouchbaseResourceHolder;
+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.
+ */
+ Mono getCluster();
+
+ /**
+ * Provides access to the managed SDK {@link Cluster} reference.
+ */
+ ClusterInterface getBlockingCluster();
+
+ /**
+ * Provides access to the managed SDK {@link Bucket} reference.
+ */
+ Mono getBucket();
+
+ /**
+ * Provides access to the managed SDK {@link Scope} reference.
+ */
+ Mono 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 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.
+ */
+ Mono 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 getTransactionResources(TransactionOptions options);
+
+ String getBucketName();
+
+ String getScopeName();
+
+ void close() throws IOException;
+
+ ReactiveCouchbaseResourceHolder getTransactionResources(TransactionOptions options, CoreTransactionAttemptContext ctx);
+
+ /*
+ * (non-Javadoc)
+ * @see org.springframework.data.mongodb.ReactiveMongoDatabaseFactory#withSession(com.mongodb.session.ClientSession)
+ */
+ ReactiveCouchbaseClientFactory withCore(ReactiveCouchbaseResourceHolder core);
+
+ /*
+ * (non-Javadoc)
+ * @see org.springframework.data.mongodb.ReactiveMongoDatabaseFactory#with(com.mongodb.session.ClientSession)
+ */
+ ReactiveCouchbaseClientFactory with(CouchbaseTransactionalOperator txOp);
+
+ /*
+ * (non-Javadoc)
+ * @see org.springframework.data.mongodb.ReactiveMongoDatabaseFactory#isTransactionActive()
+ */
+ boolean isTransactionActive();
+
+ CouchbaseTransactionalOperator getTransactionalOperator();
+}
diff --git a/src/main/java/org/springframework/data/couchbase/SimpleCouchbaseClientFactory.java b/src/main/java/org/springframework/data/couchbase/SimpleCouchbaseClientFactory.java
index b97b57f95..f6d702447 100644
--- a/src/main/java/org/springframework/data/couchbase/SimpleCouchbaseClientFactory.java
+++ b/src/main/java/org/springframework/data/couchbase/SimpleCouchbaseClientFactory.java
@@ -15,8 +15,12 @@
*/
package org.springframework.data.couchbase;
+import java.time.Duration;
+import java.time.temporal.ChronoUnit;
import java.util.function.Supplier;
+import com.couchbase.client.core.transaction.CoreTransactionAttemptContext;
+import com.couchbase.client.java.transactions.config.TransactionOptions;
import org.springframework.dao.support.PersistenceExceptionTranslator;
import org.springframework.data.couchbase.core.CouchbaseExceptionTranslator;
@@ -28,7 +32,11 @@
import com.couchbase.client.java.ClusterOptions;
import com.couchbase.client.java.Collection;
import com.couchbase.client.java.Scope;
+import com.couchbase.client.java.codec.JsonSerializer;
import com.couchbase.client.java.env.ClusterEnvironment;
+import com.couchbase.client.java.transactions.AttemptContextReactiveAccessor;
+import com.couchbase.client.java.transactions.config.TransactionsCleanupConfig;
+import com.couchbase.client.java.transactions.config.TransactionsConfig;
/**
* The default implementation of a {@link CouchbaseClientFactory}.
@@ -42,32 +50,38 @@ public class SimpleCouchbaseClientFactory implements CouchbaseClientFactory {
private final Bucket bucket;
private final Scope scope;
private final PersistenceExceptionTranslator exceptionTranslator;
+ //private JsonSerializer serializer = null;
public SimpleCouchbaseClientFactory(final String connectionString, final Authenticator authenticator,
- final String bucketName) {
+ final String bucketName) {
this(connectionString, authenticator, bucketName, null);
}
public SimpleCouchbaseClientFactory(final String connectionString, final Authenticator authenticator,
- final String bucketName, final String scopeName) {
- this(new OwnedSupplier<>(Cluster.connect(connectionString, ClusterOptions.clusterOptions(authenticator))),
+ final String bucketName, final String scopeName) {
+ this(new OwnedSupplier<>(Cluster.connect(connectionString, ClusterOptions.clusterOptions(authenticator)
+ // todo gp disabling cleanupLostAttempts to simplify output during development
+ .environment(env -> env.transactionsConfig(
+ TransactionsConfig.cleanupConfig(TransactionsCleanupConfig.cleanupLostAttempts(false)))))),
bucketName, scopeName);
}
public SimpleCouchbaseClientFactory(final String connectionString, final Authenticator authenticator,
- final String bucketName, final String scopeName, final ClusterEnvironment environment) {
+ final String bucketName, final String scopeName, final ClusterEnvironment environment) {
this(
new OwnedSupplier<>(
Cluster.connect(connectionString, ClusterOptions.clusterOptions(authenticator).environment(environment))),
bucketName, scopeName);
+ //this.serializer = environment.jsonSerializer();
}
public SimpleCouchbaseClientFactory(final Cluster cluster, final String bucketName, final String scopeName) {
this(() -> cluster, bucketName, scopeName);
+ //this.serializer = cluster.environment().jsonSerializer();
}
private SimpleCouchbaseClientFactory(final Supplier cluster, final String bucketName,
- final String scopeName) {
+ final String scopeName) {
this.cluster = cluster;
this.bucket = cluster.get().bucket(bucketName);
this.scope = scopeName == null ? bucket.defaultScope() : bucket.scope(scopeName);
@@ -97,9 +111,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();
}
@@ -117,10 +131,37 @@ public PersistenceExceptionTranslator getExceptionTranslator() {
}
@Override
- public void close() {
- if (cluster instanceof OwnedSupplier) {
- cluster.get().disconnect();
+ public CoreTransactionAttemptContext getCore(TransactionOptions options, CoreTransactionAttemptContext atr) {
+ // can't we just use AttemptContextReactive everywhere? Instead of creating AttemptContext(atr), then
+ // accessing at.getACR() ?
+ if (atr == null) {
+ atr = AttemptContextReactiveAccessor
+ .newCoreTranactionAttemptContext(AttemptContextReactiveAccessor.reactive(getCluster().transactions()));
}
+
+ return atr;
+ }
+
+ // @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() {
+ // todo gp
+ // if (cluster instanceof OwnedSupplier) {
+ // cluster.get().disconnect();
+ // }
+ }
+
+ private static Duration now() {
+ return Duration.of(System.nanoTime(), ChronoUnit.NANOS);
}
}
diff --git a/src/main/java/org/springframework/data/couchbase/SimpleReactiveCouchbaseClientFactory.java b/src/main/java/org/springframework/data/couchbase/SimpleReactiveCouchbaseClientFactory.java
new file mode 100644
index 000000000..5e90fc7e7
--- /dev/null
+++ b/src/main/java/org/springframework/data/couchbase/SimpleReactiveCouchbaseClientFactory.java
@@ -0,0 +1,402 @@
+package org.springframework.data.couchbase;
+
+import static com.couchbase.client.core.io.CollectionIdentifier.DEFAULT_COLLECTION;
+import static com.couchbase.client.core.io.CollectionIdentifier.DEFAULT_SCOPE;
+
+import com.couchbase.client.core.transaction.CoreTransactionAttemptContext;
+import com.couchbase.client.java.ClusterInterface;
+import com.couchbase.client.java.codec.JsonSerializer;
+import com.couchbase.client.java.transactions.AttemptContextReactiveAccessor;
+import com.couchbase.client.java.transactions.Transactions;
+import com.couchbase.client.java.transactions.config.TransactionOptions;
+import org.springframework.data.couchbase.transaction.CouchbaseTransactionalOperator;
+import org.springframework.data.couchbase.transaction.ReactiveCouchbaseResourceHolder;
+import reactor.core.publisher.Mono;
+
+import java.io.IOException;
+
+import org.springframework.aop.framework.ProxyFactory;
+import org.springframework.dao.DataAccessException;
+import org.springframework.dao.support.PersistenceExceptionTranslator;
+import org.springframework.data.couchbase.core.CouchbaseExceptionTranslator;
+import org.springframework.data.couchbase.transaction.SessionAwareMethodInterceptor;
+import org.springframework.util.ObjectUtils;
+
+import com.couchbase.client.java.Bucket;
+import com.couchbase.client.java.Cluster;
+import com.couchbase.client.java.Collection;
+import com.couchbase.client.java.Scope;
+
+public class SimpleReactiveCouchbaseClientFactory implements ReactiveCouchbaseClientFactory {
+ final Mono cluster;
+ final ClusterInterface theCluster;
+ final String bucketName;
+ final String scopeName;
+ final PersistenceExceptionTranslator exceptionTranslator;
+ JsonSerializer serializer;
+ Transactions transactions;
+ CouchbaseTransactionalOperator transactionalOperator;
+
+ 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();
+ this.serializer = cluster.environment().jsonSerializer();
+ this.transactions = cluster.transactions();
+ this.transactionalOperator = transactionalOperator;
+ }
+
+ public SimpleReactiveCouchbaseClientFactory(Cluster cluster, String bucketName, String scopeName) {
+ this(cluster, bucketName, scopeName, null);
+ }
+
+ @Override
+ public Mono getCluster() {
+ return cluster;
+ }
+
+
+ @Override
+ public ClusterInterface getBlockingCluster() {
+ return theCluster;
+ }
+
+ @Override
+ public Mono getBucket() {
+ return cluster.map((c) -> c.bucket(bucketName));
+ }
+
+ @Override
+ public String getBucketName() {
+ return bucketName;
+ }
+
+ @Override
+ public Mono 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;
+ }
+
+ @Override
+ public Mono getCollection(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 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 getDefaultCollection() {
+ if (getScopeName() != null && DEFAULT_SCOPE.equals(getScopeName())) {
+ throw new IllegalStateException("A collectionName must be provided if a non-default scope is used.");
+ }
+ return cluster.map((c) -> c.bucket(bucketName).defaultCollection());
+ }
+
+ @Override
+ public ReactiveCouchbaseClientFactory withScope(String scopeName) {
+ return new SimpleReactiveCouchbaseClientFactory((Cluster) cluster.block(), bucketName,
+ scopeName != null ? scopeName : this.scopeName);
+ }
+
+ @Override
+ public PersistenceExceptionTranslator getExceptionTranslator() {
+ return exceptionTranslator;
+ }
+
+ @Override
+ public void close() {
+ cluster.block().disconnect();
+ }
+
+ @Override
+ public Mono getTransactionResources(TransactionOptions options) {
+ return Mono.just(new ReactiveCouchbaseResourceHolder(null));
+ }
+
+ @Override
+ public ReactiveCouchbaseResourceHolder getTransactionResources(TransactionOptions options,
+ CoreTransactionAttemptContext atr) {
+ if (atr == null) {
+ atr = AttemptContextReactiveAccessor
+ .newCoreTranactionAttemptContext(AttemptContextReactiveAccessor.reactive(transactions));
+ }
+ return new ReactiveCouchbaseResourceHolder(atr);
+ }
+
+ @Override
+ public ReactiveCouchbaseClientFactory withCore(ReactiveCouchbaseResourceHolder holder) {
+ return new CoreTransactionAttemptContextBoundCouchbaseClientFactory(holder, this, transactions);
+ }
+
+ @Override
+ public boolean isTransactionActive() {
+ return false;
+ }
+
+ @Override
+ public CouchbaseTransactionalOperator getTransactionalOperator() {
+ return transactionalOperator;
+ }
+
+ @Override
+ public ReactiveCouchbaseClientFactory with(CouchbaseTransactionalOperator txOp) {
+ return new SimpleReactiveCouchbaseClientFactory((Cluster) getCluster().block(), bucketName, scopeName, txOp);
+ }
+
+ private T createProxyInstance(ReactiveCouchbaseResourceHolder session, T target, Class targetType) {
+
+ ProxyFactory factory = new ProxyFactory();
+ factory.setTarget(target);
+ factory.setInterfaces(targetType);
+ factory.setOpaque(true);
+
+ factory.addAdvice(new SessionAwareMethodInterceptor<>(session, target, ReactiveCouchbaseResourceHolder.class,
+ ClusterInterface.class, this::proxyDatabase, Collection.class, this::proxyCollection));
+
+ return targetType.cast(factory.getProxy(target.getClass().getClassLoader()));
+ }
+
+ private Collection proxyCollection(ReactiveCouchbaseResourceHolder session, Collection c) {
+ return createProxyInstance(session, c, Collection.class);
+ }
+
+ private ClusterInterface proxyDatabase(ReactiveCouchbaseResourceHolder session, ClusterInterface cluster) {
+ return createProxyInstance(session, cluster, ClusterInterface.class);
+ }
+
+ /**
+ * {@link CoreTransactionAttemptContext} bound TODO decorating the database with a
+ * {@link SessionAwareMethodInterceptor}.
+ *
+ * @author Christoph Strobl
+ * @since 2.1
+ */
+ static final class CoreTransactionAttemptContextBoundCouchbaseClientFactory
+ implements ReactiveCouchbaseClientFactory {
+
+ private final ReactiveCouchbaseResourceHolder transactionResources;
+ private final ReactiveCouchbaseClientFactory delegate;
+ // private final Transactions transactions;
+
+ CoreTransactionAttemptContextBoundCouchbaseClientFactory(ReactiveCouchbaseResourceHolder transactionResources,
+ ReactiveCouchbaseClientFactory delegate, Transactions transactions) {
+ this.transactionResources = transactionResources;
+ this.delegate = delegate;
+ // this.transactions = transactions;
+ }
+
+ /*
+ * (non-Javadoc)
+ * @see org.springframework.data.mongodb.ReactiveMongoDatabaseFactory#getMongoDatabase()
+ */
+ @Override
+ public Mono getCluster() throws DataAccessException {
+ return delegate.getCluster().map(this::decorateDatabase);
+ }
+
+ @Override
+ public ClusterInterface getBlockingCluster() throws DataAccessException {
+ return decorateDatabase(delegate.getBlockingCluster());
+ }
+
+ @Override
+ public Mono getBucket() {
+ return delegate.getBucket();
+ }
+
+ @Override
+ public Mono getScope() {
+ return delegate.getScope();
+ }
+
+ @Override
+ public Mono 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 getDefaultCollection() {
+ return delegate.getDefaultCollection();
+ }
+
+ @Override
+ public ReactiveCouchbaseClientFactory withScope(String scopeName) {
+ return delegate.withScope(scopeName);
+ }
+
+ /*
+ * (non-Javadoc)
+ * @see org.springframework.data.mongodb.ReactiveMongoDatabaseFactory#getExceptionTranslator()
+ */
+ @Override
+ public PersistenceExceptionTranslator getExceptionTranslator() {
+ return delegate.getExceptionTranslator();
+ }
+
+ @Override
+ public String getBucketName() {
+ return delegate.getBucketName();
+ }
+
+ @Override
+ public String getScopeName() {
+ return delegate.getScopeName();
+ }
+
+ @Override
+ public void close() throws IOException {
+ delegate.close();
+ }
+
+ /*
+ * (non-Javadoc)
+ * @see org.springframework.data.mongodb.ReactiveMongoDatabaseFactory#getSession(com.mongodb.CoreTransactionAttemptContextOptions)
+ */
+
+ @Override
+ public Mono getTransactionResources(TransactionOptions options) {
+ return Mono.just(transactionResources);
+ }
+
+ @Override
+ public ReactiveCouchbaseResourceHolder getTransactionResources(TransactionOptions options,
+ CoreTransactionAttemptContext atr) {
+ ReactiveCouchbaseResourceHolder holder = delegate.getTransactionResources(options, atr);
+ return holder;
+ }
+
+ /*
+ * (non-Javadoc)
+ * @see org.springframework.data.mongodb.ReactiveMongoDatabaseFactory#withSession(com.mongodb.session.CoreTransactionAttemptContext)
+ */
+ @Override
+ public ReactiveCouchbaseClientFactory withCore(ReactiveCouchbaseResourceHolder core) {
+ return delegate.withCore(core);
+ }
+
+ /*
+ * (non-Javadoc)
+ * @see org.springframework.data.mongodb.ReactiveMongoDatabaseFactory#isTransactionActive()
+ */
+ @Override
+ public boolean isTransactionActive() {
+ return transactionResources != null && transactionResources.hasActiveTransaction();
+ }
+
+ @Override
+ public CouchbaseTransactionalOperator getTransactionalOperator() {
+ return delegate.getTransactionalOperator();
+ }
+
+ @Override
+ public ReactiveCouchbaseClientFactory with(CouchbaseTransactionalOperator txOp) {
+ return delegate.with(txOp);
+ }
+
+ private ClusterInterface decorateDatabase(ClusterInterface database) {
+ return createProxyInstance(transactionResources, database, ClusterInterface.class);
+ }
+
+ private ClusterInterface proxyDatabase(ReactiveCouchbaseResourceHolder session, ClusterInterface database) {
+ return createProxyInstance(session, database, ClusterInterface.class);
+ }
+
+ private Collection proxyCollection(ReactiveCouchbaseResourceHolder session, Collection collection) {
+ return createProxyInstance(session, collection, Collection.class);
+ }
+
+ private T createProxyInstance(ReactiveCouchbaseResourceHolder session, T target, Class targetType) {
+
+ ProxyFactory factory = new ProxyFactory();
+ factory.setTarget(target);
+ factory.setInterfaces(targetType);
+ factory.setOpaque(true);
+
+ factory.addAdvice(new SessionAwareMethodInterceptor<>(session, target, ReactiveCouchbaseResourceHolder.class,
+ ClusterInterface.class, this::proxyDatabase, Collection.class, this::proxyCollection));
+
+ return targetType.cast(factory.getProxy(target.getClass().getClassLoader()));
+ }
+
+ public ReactiveCouchbaseResourceHolder getTransactionResources() {
+ return this.transactionResources;
+ }
+
+ public ReactiveCouchbaseClientFactory getDelegate() {
+ return this.delegate;
+ }
+
+ @Override
+ public boolean equals(Object o) {
+ if (this == o)
+ return true;
+ if (o == null || getClass() != o.getClass())
+ return false;
+
+ CoreTransactionAttemptContextBoundCouchbaseClientFactory that = (CoreTransactionAttemptContextBoundCouchbaseClientFactory) o;
+
+ if (!ObjectUtils.nullSafeEquals(this.transactionResources, that.transactionResources)) {
+ return false;
+ }
+ return ObjectUtils.nullSafeEquals(this.delegate, that.delegate);
+ }
+
+ @Override
+ public int hashCode() {
+ int result = ObjectUtils.nullSafeHashCode(this.transactionResources);
+ result = 31 * result + ObjectUtils.nullSafeHashCode(this.delegate);
+ return result;
+ }
+
+ public String toString() {
+ return "SimpleReactiveCouchbaseDatabaseFactory.CoreTransactionAttemptContextBoundCouchDbFactory(session="
+ + this.getTransactionResources() + ", delegate=" + this.getDelegate() + ")";
+ }
+ }
+}
diff --git a/src/main/java/org/springframework/data/couchbase/config/AbstractCouchbaseConfiguration.java b/src/main/java/org/springframework/data/couchbase/config/AbstractCouchbaseConfiguration.java
index 6982d64f4..80f4e72b9 100644
--- a/src/main/java/org/springframework/data/couchbase/config/AbstractCouchbaseConfiguration.java
+++ b/src/main/java/org/springframework/data/couchbase/config/AbstractCouchbaseConfiguration.java
@@ -17,19 +17,31 @@
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;
+import com.couchbase.client.core.msg.kv.DurabilityLevel;
+import com.couchbase.client.core.transaction.config.CoreTransactionsConfig;
+import com.couchbase.client.java.query.QueryScanConsistency;
+import com.couchbase.client.java.transactions.config.TransactionOptions;
+import com.couchbase.client.java.transactions.config.TransactionsCleanupConfig;
+import com.couchbase.client.java.transactions.config.TransactionsConfig;
import org.springframework.beans.factory.config.BeanDefinition;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.ClassPathScanningCandidateComponentProvider;
import org.springframework.context.annotation.Configuration;
+import org.springframework.context.annotation.Role;
import org.springframework.core.type.filter.AnnotationTypeFilter;
import org.springframework.data.convert.CustomConversions;
import org.springframework.data.couchbase.CouchbaseClientFactory;
+import org.springframework.data.couchbase.ReactiveCouchbaseClientFactory;
import org.springframework.data.couchbase.SimpleCouchbaseClientFactory;
+import org.springframework.data.couchbase.SimpleReactiveCouchbaseClientFactory;
import org.springframework.data.couchbase.core.CouchbaseTemplate;
import org.springframework.data.couchbase.core.ReactiveCouchbaseTemplate;
import org.springframework.data.couchbase.core.convert.CouchbaseCustomConversions;
@@ -40,12 +52,22 @@
import org.springframework.data.couchbase.core.mapping.Document;
import org.springframework.data.couchbase.repository.config.ReactiveRepositoryOperationsMapping;
import org.springframework.data.couchbase.repository.config.RepositoryOperationsMapping;
+import org.springframework.data.couchbase.transaction.CouchbaseSimpleCallbackTransactionManager;
+import org.springframework.data.couchbase.transaction.CouchbaseTransactionManager;
+import org.springframework.data.couchbase.transaction.ReactiveCouchbaseTransactionManager;
+import org.springframework.data.couchbase.transaction.interceptor.CouchbaseTransactionInterceptor;
import org.springframework.data.mapping.model.CamelCaseAbbreviatingFieldNamingStrategy;
import org.springframework.data.mapping.model.FieldNamingStrategy;
import org.springframework.data.mapping.model.PropertyNameFieldNamingStrategy;
+import org.springframework.transaction.annotation.AnnotationTransactionAttributeSource;
+import org.springframework.transaction.config.TransactionManagementConfigUtils;
+import org.springframework.transaction.interceptor.BeanFactoryTransactionAttributeSourceAdvisor;
+import org.springframework.transaction.interceptor.TransactionAttributeSource;
+import org.springframework.transaction.interceptor.TransactionInterceptor;
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,7 +79,6 @@
import com.couchbase.client.java.env.ClusterEnvironment;
import com.couchbase.client.java.json.JacksonTransformers;
import com.couchbase.client.java.json.JsonValueModule;
-import com.couchbase.client.java.query.QueryScanConsistency;
import com.fasterxml.jackson.databind.ObjectMapper;
/**
@@ -123,15 +144,21 @@ 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());
}
+ @Bean
+ public ReactiveCouchbaseClientFactory reactiveCouchbaseClientFactory(final Cluster couchbaseCluster) {
+ return new SimpleReactiveCouchbaseClientFactory(couchbaseCluster, getBucketName(), getScopeName());
+ }
+
@Bean(destroyMethod = "disconnect")
public Cluster couchbaseCluster(ClusterEnvironment couchbaseClusterEnvironment) {
- return Cluster.connect(getConnectionString(),
+ Cluster c = Cluster.connect(getConnectionString(),
clusterOptions(authenticator()).environment(couchbaseClusterEnvironment));
+ return c;
}
@Bean(destroyMethod = "shutdown")
@@ -141,6 +168,9 @@ public ClusterEnvironment couchbaseClusterEnvironment() {
throw new CouchbaseException("non-shadowed Jackson not present");
}
builder.jsonSerializer(JacksonJsonSerializer.create(couchbaseObjectMapper()));
+ // todo gp only suitable for tests
+ TransactionsConfig.cleanupConfig(TransactionsCleanupConfig.cleanupLostAttempts(false));
+ builder.transactionsConfig(transactionsConfig());
configureEnvironment(builder);
return builder.build();
}
@@ -156,26 +186,31 @@ protected void configureEnvironment(final ClusterEnvironment.Builder builder) {
@Bean(name = BeanNames.COUCHBASE_TEMPLATE)
public CouchbaseTemplate couchbaseTemplate(CouchbaseClientFactory couchbaseClientFactory,
- MappingCouchbaseConverter mappingCouchbaseConverter, TranslationService couchbaseTranslationService) {
- return new CouchbaseTemplate(couchbaseClientFactory, mappingCouchbaseConverter, couchbaseTranslationService,
- getDefaultConsistency());
+ ReactiveCouchbaseClientFactory reactiveCouchbaseClientFactory,
+ MappingCouchbaseConverter mappingCouchbaseConverter, TranslationService couchbaseTranslationService) {
+ return new CouchbaseTemplate(couchbaseClientFactory, reactiveCouchbaseClientFactory, mappingCouchbaseConverter,
+ couchbaseTranslationService, getDefaultConsistency());
}
public CouchbaseTemplate couchbaseTemplate(CouchbaseClientFactory couchbaseClientFactory,
- MappingCouchbaseConverter mappingCouchbaseConverter) {
- return couchbaseTemplate(couchbaseClientFactory, mappingCouchbaseConverter, new JacksonTranslationService());
+ ReactiveCouchbaseClientFactory reactiveCouchbaseClientFactory,
+ MappingCouchbaseConverter mappingCouchbaseConverter) {
+ return couchbaseTemplate(couchbaseClientFactory, reactiveCouchbaseClientFactory, mappingCouchbaseConverter,
+ new JacksonTranslationService());
}
@Bean(name = BeanNames.REACTIVE_COUCHBASE_TEMPLATE)
- public ReactiveCouchbaseTemplate reactiveCouchbaseTemplate(CouchbaseClientFactory couchbaseClientFactory,
+ public ReactiveCouchbaseTemplate reactiveCouchbaseTemplate(
+ ReactiveCouchbaseClientFactory reactiveCouchbaseClientFactory,
MappingCouchbaseConverter mappingCouchbaseConverter, TranslationService couchbaseTranslationService) {
- return new ReactiveCouchbaseTemplate(couchbaseClientFactory, mappingCouchbaseConverter, couchbaseTranslationService,
- getDefaultConsistency());
+ return new ReactiveCouchbaseTemplate(reactiveCouchbaseClientFactory, mappingCouchbaseConverter,
+ couchbaseTranslationService, getDefaultConsistency());
}
- public ReactiveCouchbaseTemplate reactiveCouchbaseTemplate(CouchbaseClientFactory couchbaseClientFactory,
+ public ReactiveCouchbaseTemplate reactiveCouchbaseTemplate(
+ ReactiveCouchbaseClientFactory reactiveCouchbaseClientFactory,
MappingCouchbaseConverter mappingCouchbaseConverter) {
- return reactiveCouchbaseTemplate(couchbaseClientFactory, mappingCouchbaseConverter,
+ return reactiveCouchbaseTemplate( reactiveCouchbaseClientFactory, mappingCouchbaseConverter,
new JacksonTranslationService());
}
@@ -257,7 +292,7 @@ public String typeKey() {
*/
@Bean
public MappingCouchbaseConverter mappingCouchbaseConverter(CouchbaseMappingContext couchbaseMappingContext,
- CouchbaseCustomConversions couchbaseCustomConversions) {
+ CouchbaseCustomConversions couchbaseCustomConversions) {
MappingCouchbaseConverter converter = new MappingCouchbaseConverter(couchbaseMappingContext, typeKey());
converter.setCustomConversions(couchbaseCustomConversions);
return converter;
@@ -282,7 +317,7 @@ public TranslationService couchbaseTranslationService() {
* Creates a {@link CouchbaseMappingContext} equipped with entity classes scanned from the mapping base package.
*
*/
- @Bean
+ @Bean(COUCHBASE_MAPPING_CONTEXT)
public CouchbaseMappingContext couchbaseMappingContext(CustomConversions customConversions) throws Exception {
CouchbaseMappingContext mappingContext = new CouchbaseMappingContext();
mappingContext.setInitialEntitySet(getInitialEntitySet());
@@ -310,6 +345,55 @@ public ObjectMapper couchbaseObjectMapper() {
return mapper;
}
+ /***** ALL THIS TX SHOULD BE MOVED OUT INTO THE IMPL OF AbstractCouchbaseConfiguration *****/
+
+ @Bean(BeanNames.REACTIVE_COUCHBASE_TRANSACTION_MANAGER)
+ ReactiveCouchbaseTransactionManager reactiveTransactionManager(
+ ReactiveCouchbaseClientFactory reactiveCouchbaseClientFactory) {
+ return new ReactiveCouchbaseTransactionManager(reactiveCouchbaseClientFactory);
+ }
+
+// @Bean(BeanNames.COUCHBASE_TRANSACTION_MANAGER)
+// CouchbaseTransactionManager transactionManager(CouchbaseClientFactory couchbaseClientFactory) {
+// return new CouchbaseTransactionManager(couchbaseClientFactory);
+// }
+
+ // todo gp experimenting with making CouchbaseSimpleCallbackTransactionManager the default - but it doesn't play
+ // nice with MR's changes to insert CouchbaseTransactionInterceptor
+ // todo mr THIS DOES NOT WORK WELL with @TestTransaction / @BeforeTransaction / @AfterTransaction etc.
+ // todo mr Maybe it is only useful with @Transactional?
+ @Bean(BeanNames.COUCHBASE_SIMPLE_CALLBACK_TRANSACTION_MANAGER)
+ CouchbaseSimpleCallbackTransactionManager callbackTransactionManager(ReactiveCouchbaseClientFactory clientFactory, TransactionOptions options) {
+ return new CouchbaseSimpleCallbackTransactionManager(clientFactory, options);
+ }
+
+ @Bean(BeanNames.COUCHBASE_TRANSACTION_MANAGER)
+ CouchbaseTransactionManager transactionManager(CouchbaseClientFactory clientFactory, TransactionOptions options) {
+ return new CouchbaseTransactionManager(clientFactory, options);
+ }
+
+ // todo gpx these would be per-transactions options so it seems odd to have a global bean? Surely would want to configure everything at global level instead?
+ @Bean
+ public TransactionOptions transactionsOptions(){
+ return TransactionOptions.transactionOptions();
+ }
+
+ // todo gpx transactions config is now done in standard ClusterConfig - so I think we don't want a separate bean?
+ public TransactionsConfig.Builder transactionsConfig(){
+ return TransactionsConfig.builder().durabilityLevel(DurabilityLevel.NONE).timeout(Duration.ofMinutes(20));// for testing
+ }
+
+ /**
+ * Blocking Transaction Manager
+ *
+ * @param couchbaseTemplate
+ * @return
+ */
+// @Bean(BeanNames.COUCHBASE_CALLBACK_TRANSACTION_MANAGER)
+// CouchbaseCallbackTransactionManager callbackTransactionManager(CouchbaseTemplate couchbaseTemplate, ReactiveCouchbaseTemplate couchbaseReactiveTemplate) {
+// return new CouchbaseCallbackTransactionManager(couchbaseTemplate, couchbaseReactiveTemplate);
+// }
+
/**
* Configure whether to automatically create indices for domain types by deriving the from the entity or not.
*/
@@ -375,5 +459,33 @@ private boolean nonShadowedJacksonPresent() {
public QueryScanConsistency getDefaultConsistency() {
return null;
}
+/*
+ @Bean
+ @Role(BeanDefinition.ROLE_INFRASTRUCTURE)
+ public TransactionInterceptor transactionInterceptor(TransactionAttributeSource transactionAttributeSource) {
+ TransactionInterceptor interceptor = new CouchbaseTransactionInterceptor();
+ interceptor.setTransactionAttributeSource(transactionAttributeSource);
+ return interceptor;
+ }
+
+ @Bean
+ @Role(BeanDefinition.ROLE_INFRASTRUCTURE)
+ public TransactionAttributeSource transactionAttributeSource() {
+ return new AnnotationTransactionAttributeSource();
+ }
+ @Bean(name = TransactionManagementConfigUtils.TRANSACTION_ADVISOR_BEAN_NAME)
+ @Role(BeanDefinition.ROLE_INFRASTRUCTURE)
+ public BeanFactoryTransactionAttributeSourceAdvisor transactionAdvisor(
+ TransactionAttributeSource transactionAttributeSource, TransactionInterceptor transactionInterceptor) {
+
+ BeanFactoryTransactionAttributeSourceAdvisor advisor = new BeanFactoryTransactionAttributeSourceAdvisor();
+ advisor.setTransactionAttributeSource(transactionAttributeSource);
+ advisor.setAdvice(transactionInterceptor);
+ // if (this.enableTx != null) {
+ // advisor.setOrder(this.enableTx.getNumber("order"));
+ // }
+ return advisor;
+ }
+ */
}
diff --git a/src/main/java/org/springframework/data/couchbase/config/BeanNames.java b/src/main/java/org/springframework/data/couchbase/config/BeanNames.java
index cb9bf63ea..100c841e5 100644
--- a/src/main/java/org/springframework/data/couchbase/config/BeanNames.java
+++ b/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,12 @@ 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";
+
+ public static final String REACTIVE_COUCHBASE_TRANSACTION_MANAGER = "reactiveCouchbaseTransactionManager";
+
+ public static final String COUCHBASE_TRANSACTION_MANAGER = "couchbaseTransactionManager";
+
+ public static final String COUCHBASE_SIMPLE_CALLBACK_TRANSACTION_MANAGER = "couchbaseSimpleCallbackTransactionManager";
}
diff --git a/src/main/java/org/springframework/data/couchbase/core/AbstractTemplateSupport.java b/src/main/java/org/springframework/data/couchbase/core/AbstractTemplateSupport.java
new file mode 100644
index 000000000..8164f2cd2
--- /dev/null
+++ b/src/main/java/org/springframework/data/couchbase/core/AbstractTemplateSupport.java
@@ -0,0 +1,226 @@
+/*
+ * 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 org.springframework.data.couchbase.core;
+
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+import org.springframework.context.ApplicationContext;
+import org.springframework.data.couchbase.core.convert.CouchbaseConverter;
+import org.springframework.data.couchbase.core.convert.join.N1qlJoinResolver;
+import org.springframework.data.couchbase.core.convert.translation.TranslationService;
+import org.springframework.data.couchbase.core.mapping.CouchbaseDocument;
+import org.springframework.data.couchbase.core.mapping.CouchbasePersistentEntity;
+import org.springframework.data.couchbase.core.mapping.CouchbasePersistentProperty;
+import org.springframework.data.couchbase.core.mapping.event.AfterSaveEvent;
+import org.springframework.data.couchbase.core.mapping.event.CouchbaseMappingEvent;
+import org.springframework.data.couchbase.repository.support.MappingCouchbaseEntityInformation;
+import org.springframework.data.couchbase.repository.support.TransactionResultHolder;
+import org.springframework.data.couchbase.transaction.ReactiveCouchbaseResourceHolder;
+import org.springframework.data.mapping.PersistentPropertyAccessor;
+import org.springframework.data.mapping.context.MappingContext;
+import org.springframework.data.mapping.model.ConvertingPropertyAccessor;
+
+import com.couchbase.client.core.error.CouchbaseException;
+import org.springframework.util.ClassUtils;
+
+import java.util.Map;
+import java.util.Set;
+
+public abstract class AbstractTemplateSupport {
+
+ final ReactiveCouchbaseTemplate template;
+ final CouchbaseConverter converter;
+ final MappingContext, CouchbasePersistentProperty> mappingContext;
+ final TranslationService translationService;
+ ApplicationContext applicationContext;
+ static final Logger LOG = LoggerFactory.getLogger(AbstractTemplateSupport.class);
+
+ public AbstractTemplateSupport(ReactiveCouchbaseTemplate template, CouchbaseConverter converter, TranslationService translationService) {
+ this.template = template;
+ this.converter = converter;
+ this.mappingContext = converter.getMappingContext();
+ this.translationService = translationService;
+ }
+
+ abstract ReactiveCouchbaseTemplate getReactiveTemplate();
+
+ public T decodeEntityBase(String id, String source, long cas, Class entityClass, String scope, String collection,
+ TransactionResultHolder txResultHolder, ReactiveCouchbaseResourceHolder holder) {
+ final CouchbaseDocument converted = new CouchbaseDocument(id);
+ converted.setId(id);
+
+ // this is the entity class defined for the repository. It may not be the class of the document that was read
+ // we will reset it after reading the document
+ //
+ // This will fail for the case where:
+ // 1) The version is defined in the concrete class, but not in the abstract class; and
+ // 2) The constructor takes a "long version" argument resulting in an exception would be thrown if version in
+ // the source is null.
+ // We could expose from the MappingCouchbaseConverter determining the persistent entity from the source,
+ // but that is a lot of work to do every time just for this very rare and avoidable case.
+ // TypeInformation typeToUse = typeMapper.readType(source, type);
+
+ CouchbasePersistentEntity persistentEntity = couldBePersistentEntity(entityClass);
+
+ if (persistentEntity == null) { // method could return a Long, Boolean, String etc.
+ // QueryExecutionConverters.unwrapWrapperTypes will recursively unwrap until there is nothing left
+ // to unwrap. This results in List being unwrapped past String[] to String, so this may also be a
+ // Collection (or Array) of entityClass. We have no way of knowing - so just assume it is what we are told.
+ // if this is a Collection or array, only the first element will be returned.
+ Set> set = ((CouchbaseDocument) translationService.decode(source, converted))
+ .getContent().entrySet();
+ return (T) set.iterator().next().getValue();
+ }
+
+ // if possible, set the version property in the source so that if the constructor has a long version argument,
+ // it will have a value an not fail (as null is not a valid argument for a long argument). This possible failure
+ // can be avoid by defining the argument as Long instead of long.
+ // persistentEntity is still the (possibly abstract) class specified in the repository definition
+ // it's possible that the abstract class does not have a version property, and this won't be able to set the version
+ if (cas != 0 && persistentEntity.getVersionProperty() != null) {
+ converted.put(persistentEntity.getVersionProperty().getName(), cas);
+ }
+
+ // if the constructor has an argument that is long version, then construction will fail if the 'version'
+ // is not available as 'null' is not a legal value for a long. Changing the arg to "Long version" would solve this.
+ // (Version doesn't come from 'source', it comes from the cas argument to decodeEntity)
+ T readEntity = converter.read(entityClass, (CouchbaseDocument) translationService.decode(source, converted));
+ final ConvertingPropertyAccessor accessor = getPropertyAccessor(readEntity);
+
+ persistentEntity = couldBePersistentEntity(readEntity.getClass());
+
+ if (cas != 0 && persistentEntity.getVersionProperty() != null) {
+ accessor.setProperty(persistentEntity.getVersionProperty(), cas);
+ }
+ if (persistentEntity.transactionResultProperty() != null) {
+ accessor.setProperty(persistentEntity.transactionResultProperty(), System.identityHashCode(txResultHolder));
+ }
+ N1qlJoinResolver.handleProperties(persistentEntity, accessor, getReactiveTemplate(), id, scope, collection);
+
+ if(holder != null){
+ holder.transactionResultHolder(txResultHolder, (T)accessor.getBean());
+ }
+
+ return accessor.getBean();
+ }
+
+ CouchbasePersistentEntity couldBePersistentEntity(Class entityClass) {
+ if (ClassUtils.isPrimitiveOrWrapper(entityClass) || entityClass == String.class) {
+ return null;
+ }
+ return mappingContext.getPersistentEntity(entityClass);
+ }
+
+
+
+ public T applyResultBase(T entity, CouchbaseDocument converted, Object id, long cas,
+ TransactionResultHolder txResultHolder, ReactiveCouchbaseResourceHolder holder) {
+ ConvertingPropertyAccessor