Skip to content

Datacouch 1145 transaction support #1423

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Merged
merged 4 commits into from
May 3, 2022
Merged

Datacouch 1145 transaction support #1423

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
4 commits
Select commit Hold shift + click to select a range
02a48f8
Porting to SDK-integrated version of transactions
programmatix Apr 26, 2022
a778c6d
Continuing work to get the ExtSDKIntegration port working
programmatix Apr 28, 2022
5012c4e
Added CouchbaseSimpleCallbackTransactionManager, the simplest
programmatix Apr 29, 2022
98b695e
Adding WIP get-and-replace @Transactional support
programmatix May 3, 2022
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
27 changes: 1 addition & 26 deletions pom.xml
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -21,7 +21,6 @@
<couchbase>3.2.5</couchbase>
<couchbase.osgi>3.2.5</couchbase.osgi>
<springdata.commons>2.7.0-SNAPSHOT</springdata.commons>
<couchbase-transactions>1.2.2</couchbase-transactions>
<java-module-name>spring.data.couchbase</java-module-name>
<!--
<spring.boot>2.6.0-RC1</spring.boot>
Expand Down Expand Up @@ -56,18 +55,6 @@
-->


<dependency>
<groupId>com.couchbase.client</groupId>
<artifactId>couchbase-transactions</artifactId>
<version>${couchbase-transactions}</version>
<exclusions>
<exclusion>
<groupId>com.couchbase.client</groupId>
<artifactId>java-client</artifactId>
</exclusion>
</exclusions>
</dependency>

<dependency>
<groupId>org.springframework</groupId>
<artifactId>spring-context-support</artifactId>
Expand Down Expand Up @@ -244,18 +231,6 @@
<version>4.0.3</version>
<scope>test</scope>
</dependency>
<dependency>
<groupId>com.couchbase.client</groupId>
<artifactId>couchbase-transactions</artifactId>
<version>${couchbase-transactions}</version>
<scope>compile</scope>
<exclusions>
<exclusion>
<groupId>com.couchbase.client</groupId>
<artifactId>java-client</artifactId>
</exclusion>
</exclusions>
</dependency>
<dependency>
<groupId>org.testcontainers</groupId>
<artifactId>testcontainers</artifactId>
Expand All @@ -264,7 +239,7 @@
<dependency>
<groupId>com.couchbase.client</groupId>
<artifactId>java-client</artifactId>
<version>3.2.5</version>
<version>3.3.0-SNAPSHOT</version>
</dependency>

<dependency>
Expand Down
11 changes: 11 additions & 0 deletions src/main/java/com/couchbase/client/java/Cluster.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -44,6 +44,7 @@
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;
Expand Down Expand Up @@ -102,6 +103,7 @@
* 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 gp is this required?
public class Cluster implements ClusterInterface {

/**
Expand Down Expand Up @@ -574,5 +576,14 @@ public void waitUntilReady(final Duration timeout, final WaitUntilReadyOptions o
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());
}
}

101 changes: 54 additions & 47 deletions src/main/java/com/couchbase/transactions/AttemptContextReactiveAccessor.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -17,74 +17,81 @@
package com.couchbase.transactions;

import com.couchbase.client.core.annotation.Stability;
import com.couchbase.transactions.config.MergedTransactionConfig;
import com.couchbase.transactions.config.PerTransactionConfig;
import com.couchbase.transactions.config.PerTransactionConfigBuilder;
import com.couchbase.transactions.config.TransactionConfig;
import com.couchbase.transactions.forwards.Supported;
import com.couchbase.transactions.log.TransactionLogger;
import com.couchbase.client.core.transaction.log.CoreTransactionLogger;
import com.couchbase.client.java.transactions.ReactiveTransactionAttemptContext;
import com.couchbase.client.java.transactions.TransactionAttemptContext;
import org.springframework.transaction.reactive.TransactionContext;

import java.time.Duration;
import java.time.temporal.ChronoUnit;
import java.util.Optional;
import java.util.UUID;

/**
* To access the AttemptContextReactive held by AttemptContext
* To access the ReactiveTransactionAttemptContext held by TransactionAttemptContext
*
* @author Michael Reiche
*/
public class AttemptContextReactiveAccessor {

public static AttemptContextReactive getACR(AttemptContext attemptContext) {
return attemptContext.ctx();
public static ReactiveTransactionAttemptContext getACR(TransactionAttemptContext attemptContext) {
// return attemptContext.ctx();
// todo gp is this access needed. Could hold the raw CoreTransactionAttemptContext instead.
return null;
}

public static AttemptContext from(AttemptContextReactive attemptContextReactive) {
return new AttemptContext(attemptContextReactive);
public static TransactionAttemptContext from(ReactiveTransactionAttemptContext attemptContextReactive) {
// todo gp needed?
return null;
// return new TransactionAttemptContext(attemptContextReactive);
}

public static TransactionLogger getLogger(AttemptContextReactive attemptContextReactive){
return attemptContextReactive.LOGGER;
}
@Stability.Internal
public static AttemptContextReactive newAttemptContextReactive(TransactionsReactive transactions){
PerTransactionConfig perConfig = PerTransactionConfigBuilder.create().build();
MergedTransactionConfig merged = new MergedTransactionConfig(transactions.config(), Optional.of(perConfig));

TransactionContext overall = new TransactionContext(
transactions.cleanup().clusterData().cluster().environment().requestTracer(),
transactions.cleanup().clusterData().cluster().environment().eventBus(),
UUID.randomUUID().toString(), now(), Duration.ZERO, merged);

String txnId = UUID.randomUUID().toString();
overall.LOGGER.info(configDebug(transactions.config(), perConfig));
return transactions.createAttemptContext(overall, merged, txnId);
public static CoreTransactionLogger getLogger(ReactiveTransactionAttemptContext attemptContextReactive){
// todo gp needed?
return null;
//return attemptContextReactive;
}
// todo gp needed?
// @Stability.Internal
// public static ReactiveTransactionAttemptContext newAttemptContextReactive(TransactionsReactive transactions){
// return null;
// PerTransactionConfig perConfig = PerTransactionConfigBuilder.create().build();
// MergedTransactionConfig merged = new MergedTransactionConfig(transactions.config(), Optional.of(perConfig));
//
// TransactionContext overall = new TransactionContext(
// transactions.cleanup().clusterData().cluster().environment().requestTracer(),
// transactions.cleanup().clusterData().cluster().environment().eventBus(),
// UUID.randomUUID().toString(), now(), Duration.ZERO, merged);
//
// String txnId = UUID.randomUUID().toString();
// overall.LOGGER.info(configDebug(transactions.config(), perConfig));
// return transactions.createAttemptContext(overall, merged, txnId);
// }

private static Duration now() {
return Duration.of(System.nanoTime(), ChronoUnit.NANOS);
}

static private 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();
}
// todo gp if needed let's expose in the SDK
// static private 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();
// }

}
Loading