Spring/spring-data-couchbase
1
0
Fork 0
Code Issues Packages Projects Releases Wiki Activity

Update transactions documentation for native transaction support.

Browse Source 
Closes #1454,#1512.
This commit is contained in:
Michael Reiche
2022-07-21 10:28:26 -07:00
parent 258dc55496
commit 28d9f8f27b
19 changed files with 290 additions and 208 deletions
Download Patch File Download Diff File Expand all files Collapse all files

23
README.adoc
View File

@@ -179,6 +179,29 @@ Building the documentation builds also the project without running tests.
The generated documentation is available from `target/site/reference/html/index.html`.
=== Building and staging reference documentation for review
[source,bash]
----
export MY_GIT_USER=<github-user>
mvn generate-resources
docs=`pwd`/target/site/reference/html
pushd /tmp
mkdir $$
cd $$
# see https://docs.github.com/en/pages/getting-started-with-github-pages/creating-a-github-pages-site
# this examples uses a repository named "staged"
git clone git@github.com:${MY_GIT_USER}/staged.git -b gh-pages
cd staged
cp -R $docs/* .
git add .
git commit --message "stage for review"
git push origin gh-pages
popd
----
The generated documentation is available from `target/site/reference/html/index.html`.
== Examples
* https://github.com/spring-projects/spring-data-examples/[Spring Data Examples] contains example projects that explain specific features in more detail.

12
pom.xml
View File

@@ -301,8 +301,20 @@
<artifactId>maven-assembly-plugin</artifactId>
</plugin>
<plugin>
<!-- generate asciidoc to stage for review -->
<groupId>org.asciidoctor</groupId>
<artifactId>asciidoctor-maven-plugin</artifactId>
<executions>
<execution>
<phase>generate-resources</phase>
<goals>
<goal>process-asciidoc</goal>
</goals>
<configuration>
<outputDirectory>target/site/reference/html</outputDirectory>
</configuration>
</execution>
</executions>
</plugin>
<plugin>
<groupId>com.mysema.maven</groupId>

198
src/main/asciidoc/transactions.adoc
View File

@@ -1,114 +1,178 @@
[[couchbase.transactions]]
= Transaction Support
= Couchbase Transactions
Couchbase supports https://docs.couchbase.com/server/6.5/learn/data/transactions.html[Distributed Transactions]. This section documents on how to use it with Spring Data Couchbase.
Couchbase supports https://docs.couchbase.com/server/current/learn/data/transactions.html[Distributed Transactions]. This section documents how to use it with Spring Data Couchbase.
== Requirements
- Couchbase Server 6.5 or above.
- Couchbase Java client 3.0.0 or above. It is recommended to follow the transitive dependency for the transactions library from maven.
- Couchbase Server 6.6.1 or aabove.
- Spring Data Couchbase 5.0.0-M5 or above.
- NTP should be configured so nodes of the Couchbase cluster are in sync with time. The time being out of sync will not cause incorrect behavior, but can impact metadata cleanup.
- Set spring.main.allow-bean-definition-overriding=true either in application.properties or as a SpringApplicationBuilder property.
== Overview
The Spring Data Couchbase template operations insert, find, replace and delete and repository methods that use those calls can participate in a Couchbase Transaction. They can be executed in a transaction by using the @Transactional annotation, the CouchbaseTransactionalOperator, or in the lambda of a Couchbase Transaction.
== Getting Started & Configuration
The `couchbase-transactions` artifact needs to be included into your `pom.xml` if maven is being used (or equivalent).
Couchbase Transactions are normally leveraged with a method annotated with @Transactional.
The @Transactional operator is implemented with the CouchbaseTransactionManager which is supplied as a bean in the AbstractCouchbaseConfiguration.
Couchbase Transactions can be used without defining a service class by using CouchbaseTransactionOperator which is also supplied as a bean in AbtractCouchbaseConfiguration.
Couchbase Transactions can also be used directly using Spring Data Couchbase operations within a lambda https://docs.couchbase.com/server/current/learn/data/transactions.html#using-transactions[Using Transactions]
- Group: `com.couchbase.client`
- Artifact: `couchbase-transactions`
- Version: latest one, i.e. `1.0.0`
== Transactions with @Transactional
Once it is included in your project, you need to create a single `Transactions` object. Conveniently, it can be part of
your spring data couchbase `AbstractCouchbaseConfiguration` implementation:
.Transaction Configuration
@Transactional defines as transactional a method or all methods on a class.
When this annotation is declared at the class level, it applies as a default
to all methods of the declaring class and its subclasses.
=== Attribute Semantics
In this release, the Couchbase Transactions ignores the rollback attributes.
The transaction isolation level is read-committed;
.Transaction Configuration and Use by @Transactional
====
.The Configuration
[source,java]
----
@Configuration
@EnableCouchbaseRepositories("<parent-dir-of-repository-interfaces>")
@EnableReactiveCouchbaseRepositories("<parent-dir-of-repository-interfaces>")
@EnableTransactionManagement // <1>
static class Config extends AbstractCouchbaseConfiguration {
// Usual Setup
@Override public String getConnectionString() { /* ... */ }
@Override public String getUserName() { /* ... */ }
@Override public String getPassword() { /* ... */ }
@Override public String getBucketName() { /* ... */ }
// Usual Setup
@Override public String getConnectionString() { /* ... */ }
@Override public String getUserName() { /* ... */ }
@Override public String getPassword() { /* ... */ }
@Override public String getBucketName() { /* ... */ }
@Bean
public Transactions transactions(final Cluster couchbaseCluster) {
return Transactions.create(couchbaseCluster, TransactionConfigBuilder.create()
// The configuration can be altered here, but in most cases the defaults are fine.
.build());
}
// Customization of transaction behavior is via the configureEnvironment() method
@Override protected void configureEnvironment(final Builder builder) {
builder.transactionsConfig(
TransactionsConfig.builder().timeout(Duration.ofSeconds(30)));
}
}
----
.The Transactional Service Class
Note that the body of @Transactional methods can be re-executed if the transaction fails.
It is imperative that everthing in the method body be idempotent.
[source,java]
----
import reactor.core.publisher.Mono;
import reactor.core.publisher.Flux;
import org.springframework.stereotype.Service;
import org.springframework.transaction.annotation.Transactional;
final CouchbaseOperations personOperations;
final ReactiveCouchbaseOperations reactivePersonOperations;
@Service // <2>
public class PersonService {
final CouchbaseOperations operations;
final ReactiveCouchbaseOperations reactiveOperations;
public PersonService(CouchbaseOperations ops, ReactiveCouchbaseOperations reactiveOps) {
operations = ops;
reactiveOperations = reactiveOps;
}
// no annotation results in this method being executed not in a transaction
public Person save(Person p) {
return operations.save(p);
}
@Transactional
public Person changeFirstName(String id, String newFirstName) {
Person p = operations.findById(Person.class).one(id); // <3>
return operations.replaceById(Person.class).one(p.withFirstName(newFirstName);
}
@Transactional
public Mono<Person> reactiveChangeFirstName(String id, String newFirstName) {
return personOperationsRx.findById(Person.class).one(person.id())
.flatMap(p -> personOperationsRx.replaceById(Person.class).one(p.withFirstName(newFirstName)));
}
}
----
[source,java]
.Using the @Transactional Service.
----
@Autowired PersonService personService; // <4>
Person walterWhite = new Person( "Walter", "White");
Person p = personService.save(walterWhite); // this is not a transactional method
...
Person renamedPerson = personService.changeFirstName(walterWhite.getId(), "Ricky"); // <5>
----
Functioning of the @Transactional method annotation requires
[start=1]
. the configuration class to be annotated with @EnableTransactionManagement;
. the service object with the annotated methods must be annotated with @Service;
. the body of the method is executed in a transaction.
. the service object with the annotated methods must be obtained via @Autowired.
. the call to the method must be made from a different class than service because calling an annotated
method from the same class will not invoke the Method Interceptor that does the transaction processing.
====
Once the `@Bean` is configured, you can autowire it from your service (or any other class) to make use of it. Please
see the https://docs.couchbase.com/java-sdk/3.0/howtos/distributed-acid-transactions-from-the-sdk.html[Reference Documentation]
on how to use the `Transactions` class. Since you need access to the current `Collection` as well, we recommend you to also
autowire the `CouchbaseClientFactory` and access it from there:
== Transactions with CouchbaseTransactionalOperator
.Transaction Access
CouchbaseTransactionalOperator can be used to construct a transaction in-line without creating a service class that uses @Transactional.
CouchbaseTransactionalOperator is available as a bean and can be instantiated with @Autowired.
If creating one explicitly, it must be created with CouchbaseTransactionalOperator.create(manager) (NOT TransactionalOperator.create(manager)).
.Transaction Access Using TransactionalOperator.execute()
====
[source,java]
----
@Autowired
Transactions transactions;
@Autowired TransactionalOperator txOperator;
@Autowired ReactiveCouchbaseTemplate reactiveCouchbaseTemplate;
@Autowired
CouchbaseClientFactory couchbaseClientFactory;
public void doSomething() {
transactions.run(ctx -> {
ctx.insert(couchbaseClientFactory.getDefaultCollection(), "id", "content");
ctx.commit();
});
}
Flux<Person> result = txOperator.execute((ctx) ->
reactiveCouchbaseTemplate.findById(Person.class).one(person.id())
.flatMap(p -> reactiveCouchbaseTemplate.replaceById(Person.class).one(p.withFirstName("Walt")))
);
----
====
== Object Conversions
== Transactions Directly with the SDK
Since the transactions library itself has no knowledge of your spring data entity types, you need to convert it back and
forth when reading/writing to interact properly. Fortunately, all you need to do is autowire the `MappingCouchbaseConverter` and
utilize it:
Spring Data Couchbase works seamlessly with the Couchbase Java SDK for transaction processing. Spring Data Couchbase operations that
can be executed in a transaction will work directly within the lambda of a transactions().run() without involving any of the Spring
Transactions mechanisms. This is the most straight-forward way to leverage Couchbase Transactions in Spring Data Couchbase.
.Transaction Conversion on Write
Please see the https://docs.couchbase.com/java-sdk/current/howtos/distributed-acid-transactions-from-the-sdk.html[Reference Documentation]
.Transaction Access - Blocking
====
[source,java]
----
@Autowired
MappingCouchbaseConverter mappingCouchbaseConverter;
@Autowired CouchbaseTemplate couchbaseTemplate;
public void doSomething() {
transactions.run(ctx -> {
Airline airline = new Airline("demo-airline", "at");
CouchbaseDocument target = new CouchbaseDocument();
mappingCouchbaseConverter.write(airline, target);
ctx.insert(couchbaseClientFactory.getDefaultCollection(), target.getId(), target.getContent());
ctx.commit();
});
}
TransactionResult result = couchbaseTemplate.getCouchbaseClientFactory().getCluster().transactions().run(ctx -> {
Person p = couchbaseTemplate.findById(Person.class).one(personId);
couchbaseTemplate.replaceById(Person.class).one(p.withFirstName("Walt"));
});
----
====
The same approach can be used on read:
.Transaction Conversion on Read
.Transaction Access - Reactive
====
[source,java]
----
TransactionGetResult getResult = ctx.get(couchbaseClientFactory.getDefaultCollection(), "doc-id");
@Autowired ReactiveCouchbaseTemplate reactiveCouchbaseTemplate;
CouchbaseDocument source = new CouchbaseDocument(getResult.id());
source.setContent(getResult.contentAsObject());
Airline read = mappingCouchbaseConverter.read(Airline.class, source);
Mono<TransactionResult> result = reactiveCouchbaseTemplate.getCouchbaseClientFactory().getCluster().reactive().transactions()
.run(ctx ->
reactiveCouchbaseTemplate.findById(Person.class).one(personId)
.flatMap(p -> reactiveCouchbaseTemplate.replaceById(Person.class).one(p.withFirstName("Walt")))
);
----
====
We are also looking into tighter integration of the transaction library into the spring data library
ecosystem.

11
src/main/java/org/springframework/data/couchbase/config/AbstractCouchbaseConfiguration.java
View File

@@ -51,6 +51,7 @@ import org.springframework.transaction.TransactionManager;
import org.springframework.transaction.annotation.AnnotationTransactionAttributeSource;
import org.springframework.transaction.interceptor.TransactionAttributeSource;
import org.springframework.transaction.interceptor.TransactionInterceptor;
import org.springframework.transaction.support.TransactionTemplate;
import org.springframework.util.ClassUtils;
import org.springframework.util.StringUtils;
@@ -330,6 +331,16 @@ public abstract class AbstractCouchbaseConfiguration {
return new CouchbaseCallbackTransactionManager(clientFactory);
}
/**
* The default transaction template manager.
*
* @param couchbaseTransactionManager
* @return
*/
@Bean(BeanNames.COUCHBASE_TRANSACTION_TEMPLATE)
TransactionTemplate couchbaseTransactionTemplate(CouchbaseCallbackTransactionManager couchbaseTransactionManager) {
return new TransactionTemplate(couchbaseTransactionManager);
}
/**
* The default TransactionalOperator.
*

2
src/main/java/org/springframework/data/couchbase/config/BeanNames.java
View File

@@ -64,5 +64,7 @@ public class BeanNames {
public static final String COUCHBASE_TRANSACTION_MANAGER = "couchbaseTransactionManager";
public static final String COUCHBASE_TRANSACTION_TEMPLATE = "couchbaseTransactionTemplate";
public static final String COUCHBASE_TRANSACTIONAL_OPERATOR = "couchbaseTransactionalOperator";
}

2
src/test/java/org/springframework/data/couchbase/transactions/CouchbasePersonTransactionReactiveIntegrationTests.java
View File

@@ -108,7 +108,7 @@ public class CouchbasePersonTransactionReactiveIntegrationTests extends JavaInte
@Test
public void shouldRollbackAfterExceptionOfTxAnnotatedMethod() {
assertThrowsWithCause(() -> personService.declarativeSavePersonErrors(WalterWhite).blockLast(),
assertThrowsWithCause(() -> personService.declarativeSavePersonErrors(WalterWhite).block(),
TransactionSystemUnambiguousException.class, SimulateFailureException.class);
}

5
src/test/java/org/springframework/data/couchbase/transactions/CouchbaseTransactionalNonAllowableOperationsIntegrationTests.java
View File

@@ -36,7 +36,6 @@ import org.springframework.data.couchbase.util.Capabilities;
import org.springframework.data.couchbase.util.ClusterType;
import org.springframework.data.couchbase.util.IgnoreWhen;
import org.springframework.data.couchbase.util.JavaIntegrationTests;
import org.springframework.stereotype.Component;
import org.springframework.stereotype.Service;
import org.springframework.test.context.junit.jupiter.SpringJUnitConfig;
import org.springframework.transaction.annotation.EnableTransactionManagement;
@@ -114,9 +113,7 @@ public class CouchbaseTransactionalNonAllowableOperationsIntegrationTests extend
});
}
@Service
@Component
@EnableTransactionManagement
@Service // this will work in the unit tests even without @Service because of explicit loading by @SpringJUnitConfig
static class PersonService {
final CouchbaseOperations personOperations;

5
src/test/java/org/springframework/data/couchbase/transactions/CouchbaseTransactionalOptionsIntegrationTests.java
View File

@@ -36,7 +36,6 @@ import org.springframework.data.couchbase.transactions.util.TransactionTestUtil;
import org.springframework.data.couchbase.util.ClusterType;
import org.springframework.data.couchbase.util.IgnoreWhen;
import org.springframework.data.couchbase.util.JavaIntegrationTests;
import org.springframework.stereotype.Component;
import org.springframework.stereotype.Service;
import org.springframework.test.context.junit.jupiter.SpringJUnitConfig;
import org.springframework.transaction.annotation.EnableTransactionManagement;
@@ -100,9 +99,7 @@ public class CouchbaseTransactionalOptionsIntegrationTests extends JavaIntegrati
personService.supportedIsolation();
}
@Service
@Component
@EnableTransactionManagement
@Service // this will work in the unit tests even without @Service because of explicit loading by @SpringJUnitConfig
static class PersonService {
final CouchbaseOperations ops;

5
src/test/java/org/springframework/data/couchbase/transactions/CouchbaseTransactionalPropagationIntegrationTests.java
View File

@@ -43,7 +43,6 @@ import org.springframework.data.couchbase.util.ClusterType;
import org.springframework.data.couchbase.util.IgnoreWhen;
import org.springframework.data.couchbase.util.JavaIntegrationTests;
import org.springframework.lang.Nullable;
import org.springframework.stereotype.Component;
import org.springframework.stereotype.Service;
import org.springframework.test.context.junit.jupiter.SpringJUnitConfig;
import org.springframework.transaction.IllegalTransactionStateException;
@@ -285,9 +284,7 @@ public class CouchbaseTransactionalPropagationIntegrationTests extends JavaInteg
assertEquals(3, attempts.get());
}
@Service
@Component
@EnableTransactionManagement
@Service // this will work in the unit tests even without @Service because of explicit loading by @SpringJUnitConfig
static class PersonService {
final CouchbaseOperations ops;

6
src/test/java/org/springframework/data/couchbase/transactions/CouchbaseTransactionalRepositoryIntegrationTests.java
View File

@@ -40,7 +40,6 @@ import org.springframework.data.couchbase.transaction.error.TransactionSystemUna
import org.springframework.data.couchbase.util.ClusterType;
import org.springframework.data.couchbase.util.IgnoreWhen;
import org.springframework.data.couchbase.util.JavaIntegrationTests;
import org.springframework.stereotype.Component;
import org.springframework.stereotype.Service;
import org.springframework.test.context.junit.jupiter.SpringJUnitConfig;
import org.springframework.transaction.annotation.EnableTransactionManagement;
@@ -114,7 +113,6 @@ public class CouchbaseTransactionalRepositoryIntegrationTests extends JavaIntegr
String id = UUID.randomUUID().toString();
assertThrowsWithCause(() -> {
;
userService.run(repo -> {
User user = repo.save(new User(id, "Ada", "Lovelace"));
SimulateFailureException.throwEx("fail");
@@ -125,9 +123,7 @@ public class CouchbaseTransactionalRepositoryIntegrationTests extends JavaIntegr
assertNull(user);
}
@Service
@Component
@EnableTransactionManagement
@Service // this will work in the unit tests even without @Service because of explicit loading by @SpringJUnitConfig
static class UserService {
@Autowired UserRepository userRepo;

5
src/test/java/org/springframework/data/couchbase/transactions/CouchbaseTransactionalTemplateIntegrationTests.java
View File

@@ -52,7 +52,6 @@ import org.springframework.data.couchbase.util.Capabilities;
import org.springframework.data.couchbase.util.ClusterType;
import org.springframework.data.couchbase.util.IgnoreWhen;
import org.springframework.data.couchbase.util.JavaIntegrationTests;
import org.springframework.stereotype.Component;
import org.springframework.stereotype.Service;
import org.springframework.test.context.junit.jupiter.SpringJUnitConfig;
import org.springframework.transaction.annotation.EnableTransactionManagement;
@@ -366,9 +365,7 @@ public class CouchbaseTransactionalTemplateIntegrationTests extends JavaIntegrat
}, TransactionSystemUnambiguousException.class, IllegalArgumentException.class);
}
@Service
@Component
@EnableTransactionManagement
@Service // this will work in the unit tests even without @Service because of explicit loading by @SpringJUnitConfig
static class PersonService {
final CouchbaseOperations personOperations;
final ReactiveCouchbaseOperations personOperationsRx;

6
src/test/java/org/springframework/data/couchbase/transactions/CouchbaseTransactionalUnsettableParametersIntegrationTests.java
View File

@@ -36,10 +36,8 @@ import org.springframework.data.couchbase.util.Capabilities;
import org.springframework.data.couchbase.util.ClusterType;
import org.springframework.data.couchbase.util.IgnoreWhen;
import org.springframework.data.couchbase.util.JavaIntegrationTests;
import org.springframework.stereotype.Component;
import org.springframework.stereotype.Service;
import org.springframework.test.context.junit.jupiter.SpringJUnitConfig;
import org.springframework.transaction.annotation.EnableTransactionManagement;
import org.springframework.transaction.annotation.Transactional;
import com.couchbase.client.core.msg.kv.DurabilityLevel;
@@ -202,9 +200,7 @@ public class CouchbaseTransactionalUnsettableParametersIntegrationTests extends
});
}
@Service
@Component
@EnableTransactionManagement
@Service // this will work in the unit tests even without @Service because of explicit loading by @SpringJUnitConfig
static class PersonService {
final CouchbaseOperations personOperations;

128
src/test/java/org/springframework/data/couchbase/transactions/PersonService.java
View File

@@ -20,7 +20,6 @@ import static com.couchbase.client.java.query.QueryScanConsistency.REQUEST_PLUS;
import static org.springframework.data.couchbase.util.JavaIntegrationTests.throwSimulateFailureException;
import static org.springframework.data.couchbase.util.Util.assertInAnnotationTransaction;
import reactor.core.publisher.Flux;
import reactor.core.publisher.Mono;
import java.util.List;
@@ -32,93 +31,76 @@ import org.springframework.data.couchbase.core.ReactiveCouchbaseOperations;
import org.springframework.data.couchbase.domain.Person;
import org.springframework.stereotype.Component;
import org.springframework.stereotype.Service;
import org.springframework.transaction.annotation.EnableTransactionManagement;
import org.springframework.transaction.annotation.Transactional;
import org.springframework.transaction.reactive.TransactionalOperator;
/**
* PersonService for tests
*
* @author Michael Reiche
*/
@Service
@Component
@EnableTransactionManagement
@Service // this will work in the unit tests even without @Service because of explicit loading by @SpringJUnitConfig
class PersonService {
final CouchbaseOperations personOperations;
final ReactiveCouchbaseOperations personOperationsRx;
final TransactionalOperator transactionalOperator;
final ReactiveCouchbaseOperations reactivePersonOperations;
public PersonService(CouchbaseOperations ops, ReactiveCouchbaseOperations opsRx,
TransactionalOperator transactionalOperator) {
public PersonService(CouchbaseOperations ops, ReactiveCouchbaseOperations reactiveOps) {
personOperations = ops;
personOperationsRx = opsRx;
this.transactionalOperator = transactionalOperator;
reactivePersonOperations = reactiveOps;
}
@Transactional
public Person savePersonErrors(Person person) {
assertInAnnotationTransaction(false);
return personOperationsRx.insertById(Person.class).one(person)//
.<Person> flatMap(it -> Mono.error(new SimulateFailureException()))//
.as(transactionalOperator::transactional).block();
Person p = personOperations.insertById(Person.class).one(person);
SimulateFailureException.throwEx("savePersonErrors");
return p;
}
@Transactional
public Person savePerson(Person person) {
assertInAnnotationTransaction(false);
return personOperationsRx.insertById(Person.class).one(person)//
.as(transactionalOperator::transactional).block();
assertInAnnotationTransaction(true);
return personOperations.insertById(Person.class).one(person);
}
@Transactional
public Long countDuringTx(Person person) {
assertInAnnotationTransaction(false);
return personOperationsRx.insertById(Person.class).one(person)//
.then(personOperationsRx.findByQuery(Person.class).withConsistency(REQUEST_PLUS).count())
.as(transactionalOperator::transactional).block();
assertInAnnotationTransaction(true);
Person p = personOperations.insertById(Person.class).one(person);
return personOperations.findByQuery(Person.class).withConsistency(REQUEST_PLUS).count();
}
@Transactional
public List<CouchbasePersonTransactionIntegrationTests.EventLog> saveWithLogs(Person person) {
assertInAnnotationTransaction(false);
return Flux
.merge(
personOperationsRx.insertById(CouchbasePersonTransactionIntegrationTests.EventLog.class)
.one(new CouchbasePersonTransactionIntegrationTests.EventLog(new ObjectId(), "beforeConvert")),
personOperationsRx.insertById(CouchbasePersonTransactionIntegrationTests.EventLog.class)
.one(new CouchbasePersonTransactionIntegrationTests.EventLog(new ObjectId(), "afterConvert")),
personOperationsRx.insertById(CouchbasePersonTransactionIntegrationTests.EventLog.class)
.one(new CouchbasePersonTransactionIntegrationTests.EventLog(new ObjectId(), "beforeInsert")),
personOperationsRx.insertById(Person.class).one(person),
personOperationsRx.insertById(CouchbasePersonTransactionIntegrationTests.EventLog.class)
.one(new CouchbasePersonTransactionIntegrationTests.EventLog(new ObjectId(), "afterInsert"))) //
.thenMany(personOperationsRx.findByQuery(CouchbasePersonTransactionIntegrationTests.EventLog.class)
.withConsistency(REQUEST_PLUS).all()) //
.as(transactionalOperator::transactional).collectList().block();
assertInAnnotationTransaction(true);
personOperations.insertById(CouchbasePersonTransactionIntegrationTests.EventLog.class)
.one(new CouchbasePersonTransactionIntegrationTests.EventLog(new ObjectId(), "beforeConvert"));
personOperations.insertById(CouchbasePersonTransactionIntegrationTests.EventLog.class)
.one(new CouchbasePersonTransactionIntegrationTests.EventLog(new ObjectId(), "afterConvert"));
personOperations.insertById(CouchbasePersonTransactionIntegrationTests.EventLog.class)
.one(new CouchbasePersonTransactionIntegrationTests.EventLog(new ObjectId(), "beforeInsert"));
personOperations.insertById(Person.class).one(person);
personOperations.insertById(CouchbasePersonTransactionIntegrationTests.EventLog.class)
.one(new CouchbasePersonTransactionIntegrationTests.EventLog(new ObjectId(), "afterInsert"));
return personOperations.findByQuery(CouchbasePersonTransactionIntegrationTests.EventLog.class)
.withConsistency(REQUEST_PLUS).all();
}
@Transactional
public List<CouchbasePersonTransactionIntegrationTests.EventLog> saveWithErrorLogs(Person person) {
assertInAnnotationTransaction(false);
return Flux
.merge(
personOperationsRx.insertById(CouchbasePersonTransactionIntegrationTests.EventLog.class)
.one(new CouchbasePersonTransactionIntegrationTests.EventLog(new ObjectId(), "beforeConvert")),
//
personOperationsRx.insertById(CouchbasePersonTransactionIntegrationTests.EventLog.class)
.one(new CouchbasePersonTransactionIntegrationTests.EventLog(new ObjectId(), "afterConvert")),
//
personOperationsRx.insertById(CouchbasePersonTransactionIntegrationTests.EventLog.class)
.one(new CouchbasePersonTransactionIntegrationTests.EventLog(new ObjectId(), "beforeInsert")),
//
personOperationsRx.insertById(Person.class).one(person),
//
personOperationsRx.insertById(CouchbasePersonTransactionIntegrationTests.EventLog.class)
.one(new CouchbasePersonTransactionIntegrationTests.EventLog(new ObjectId(), "afterInsert"))) //
.thenMany(personOperationsRx.findByQuery(CouchbasePersonTransactionIntegrationTests.EventLog.class)
.withConsistency(REQUEST_PLUS).all()) //
.<CouchbasePersonTransactionIntegrationTests.EventLog> flatMap(it -> Mono.error(new SimulateFailureException()))
.as(transactionalOperator::transactional).collectList().block();
assertInAnnotationTransaction(true);
personOperations.insertById(CouchbasePersonTransactionIntegrationTests.EventLog.class)
.one(new CouchbasePersonTransactionIntegrationTests.EventLog(new ObjectId(), "beforeConvert"));
personOperations.insertById(CouchbasePersonTransactionIntegrationTests.EventLog.class)
.one(new CouchbasePersonTransactionIntegrationTests.EventLog(new ObjectId(), "afterConvert"));
personOperations.insertById(CouchbasePersonTransactionIntegrationTests.EventLog.class)
.one(new CouchbasePersonTransactionIntegrationTests.EventLog(new ObjectId(), "beforeInsert"));
personOperations.insertById(Person.class).one(person);
personOperations.insertById(CouchbasePersonTransactionIntegrationTests.EventLog.class)
.one(new CouchbasePersonTransactionIntegrationTests.EventLog(new ObjectId(), "afterInsert"));
SimulateFailureException.throwEx("saveEventError");
return personOperations.findByQuery(CouchbasePersonTransactionIntegrationTests.EventLog.class)
.withConsistency(REQUEST_PLUS).all();
}
// org.springframework.beans.factory.NoUniqueBeanDefinitionException:
@@ -160,14 +142,12 @@ class PersonService {
* @param person
* @return
*/
// @Transactional(transactionManager = BeanNames.REACTIVE_COUCHBASE_TRANSACTION_MANAGER)
// must use transactionalOperator
@Transactional
public Mono<Person> declarativeFindReplacePersonReactive(Person person, AtomicInteger tryCount) {
// assertInAnnotationTransaction(true);
return personOperationsRx.findById(Person.class).one(person.id())
assertInAnnotationTransaction(true);
return reactivePersonOperations.findById(Person.class).one(person.id())
.map((p) -> ReplaceLoopThread.updateOutOfTransaction(personOperations, p, tryCount.incrementAndGet()))
.flatMap(p -> personOperationsRx.replaceById(Person.class).one(p.withFirstName(person.getFirstname())))
.as(transactionalOperator::transactional);
.flatMap(p -> reactivePersonOperations.replaceById(Person.class).one(p.withFirstName(person.getFirstname())));
}
/**
@@ -183,19 +163,17 @@ class PersonService {
return personOperations.replaceById(Person.class).one(p.withFirstName(person.getFirstname()));
}
// @Transactional(transactionManager = BeanNames.REACTIVE_COUCHBASE_TRANSACTION_MANAGER)
// must use transactionalOperator
@Transactional
public Mono<Person> declarativeSavePersonReactive(Person person) {
// assertInAnnotationTransaction(true);
return personOperationsRx.insertById(Person.class).one(person).as(transactionalOperator::transactional);
assertInAnnotationTransaction(true);
return reactivePersonOperations.insertById(Person.class).one(person);
}
// @Transactional(transactionManager = BeanNames.REACTIVE_COUCHBASE_TRANSACTION_MANAGER)
// must use transactionalOperator
@Transactional
public Mono<Person> declarativeSavePersonErrorsReactive(Person person) {
// assertInAnnotationTransaction(true);
return personOperationsRx.insertById(Person.class).one(person).map((pp) -> throwSimulateFailureException(pp))
.as(transactionalOperator::transactional); //
assertInAnnotationTransaction(true);
return reactivePersonOperations.insertById(Person.class).one(person).map((pp) -> throwSimulateFailureException(pp));
}
}

47
src/test/java/org/springframework/data/couchbase/transactions/PersonServiceReactive.java
View File

@@ -17,11 +17,13 @@
package org.springframework.data.couchbase.transactions;
import static com.couchbase.client.java.query.QueryScanConsistency.REQUEST_PLUS;
import static org.junit.jupiter.api.Assertions.assertTrue;
import org.springframework.data.couchbase.core.TransactionalSupport;
import org.springframework.stereotype.Service;
import reactor.core.publisher.Flux;
import reactor.core.publisher.Mono;
import org.springframework.data.couchbase.config.BeanNames;
import org.springframework.data.couchbase.core.CouchbaseOperations;
import org.springframework.data.couchbase.core.ReactiveCouchbaseOperations;
import org.springframework.data.couchbase.domain.Person;
@@ -33,6 +35,7 @@ import org.springframework.transaction.reactive.TransactionalOperator;
*
* @author Michael Reiche
*/
@Service // this will work in the unit tests even without @Service because of explicit loading by @SpringJUnitConfig
class PersonServiceReactive {
final ReactiveCouchbaseOperations personOperationsRx;
@@ -47,24 +50,28 @@ class PersonServiceReactive {
return;
}
@Transactional
public Mono<Person> savePersonErrors(Person person) {
return personOperationsRx.insertById(Person.class).one(person) //
.<Person> flatMap(it -> Mono.error(new SimulateFailureException())) //
.as(transactionalOperator::transactional);
.<Person> flatMap(it -> Mono.error(new SimulateFailureException()));
}
@Transactional
public Mono<Person> savePerson(Person person) {
return personOperationsRx.insertById(Person.class).one(person) //
.flatMap(Mono::just) //
.as(transactionalOperator::transactional);
return TransactionalSupport.checkForTransactionInThreadLocalStorage().map(stat -> {
assertTrue(stat.isPresent(), "Not in transaction");
System.err.println("In a transaction!!");
return stat;
}).flatMap(ignored -> personOperationsRx.insertById(Person.class).one(person));
}
@Transactional
public Mono<Long> countDuringTx(Person person) {
return personOperationsRx.save(person) //
.then(personOperationsRx.findByQuery(Person.class).withConsistency(REQUEST_PLUS).count()) //
.as(transactionalOperator::transactional);
.then(personOperationsRx.findByQuery(Person.class).withConsistency(REQUEST_PLUS).count());
}
@Transactional
public Flux<CouchbasePersonTransactionReactiveIntegrationTests.EventLog> saveWithLogs(Person person) {
return Flux
.merge(
@@ -78,10 +85,10 @@ class PersonServiceReactive {
personOperationsRx
.save(new CouchbasePersonTransactionReactiveIntegrationTests.EventLog(new ObjectId(), "afterInsert"))) //
.thenMany(personOperationsRx.findByQuery(CouchbasePersonTransactionReactiveIntegrationTests.EventLog.class)
.withConsistency(REQUEST_PLUS).all()) //
.as(transactionalOperator::transactional);
.withConsistency(REQUEST_PLUS).all());
}
@Transactional
public Flux<Void> saveWithErrorLogs(Person person) {
return Flux
.merge(
@@ -94,20 +101,18 @@ class PersonServiceReactive {
personOperationsRx.save(person),
personOperationsRx
.save(new CouchbasePersonTransactionReactiveIntegrationTests.EventLog(new ObjectId(), "afterInsert"))) //
.<Void> flatMap(it -> Mono.error(new SimulateFailureException())) //
.as(transactionalOperator::transactional);
.<Void> flatMap(it -> Mono.error(new SimulateFailureException()));
}
// @Transactional(transactionManager = BeanNames.COUCHBASE_TRANSACTION_MANAGER)
public Flux<Person> declarativeSavePerson(Person person) {
return transactionalOperator.execute(reactiveTransaction -> personOperationsRx.save(person));
@Transactional
public Mono<Person> declarativeSavePerson(Person person) {
return personOperationsRx.save(person);
}
@Transactional(transactionManager = BeanNames.COUCHBASE_TRANSACTION_MANAGER)
public Flux<Person> declarativeSavePersonErrors(Person person) {
Person p = personOperations.insertById(Person.class).one(person);
Person pp = personOperations.findByQuery(Person.class).withConsistency(REQUEST_PLUS).all().get(0);
SimulateFailureException.throwEx(); // so the following lines is not flagged as unreachable
return Flux.just(p);
@Transactional
public Mono<Person> declarativeSavePersonErrors(Person person) {
return personOperationsRx.insertById(Person.class).one(person)
.flatMap(pp -> personOperationsRx.findById(Person.class).one(pp.id()))
.flatMap(ppp -> Mono.error(new SimulateFailureException()));
}
}

5
src/test/java/org/springframework/data/couchbase/transactions/ReactiveTransactionalTemplateIntegrationTests.java
View File

@@ -46,7 +46,6 @@ import org.springframework.data.couchbase.transaction.error.TransactionSystemUna
import org.springframework.data.couchbase.util.ClusterType;
import org.springframework.data.couchbase.util.IgnoreWhen;
import org.springframework.data.couchbase.util.JavaIntegrationTests;
import org.springframework.stereotype.Component;
import org.springframework.stereotype.Service;
import org.springframework.test.context.junit.jupiter.SpringJUnitConfig;
import org.springframework.transaction.annotation.EnableTransactionManagement;
@@ -166,9 +165,7 @@ public class ReactiveTransactionalTemplateIntegrationTests extends JavaIntegrati
assertEquals(3, fromLambda.size());
}
@Service
@Component
@EnableTransactionManagement
@Service // this will work in the unit tests even without @Service because of explicit loading by @SpringJUnitConfig
static class PersonService {
final ReactiveCouchbaseOperations ops;

15
src/test/java/org/springframework/data/couchbase/transactions/TransactionTemplateIntegrationTests.java
View File

@@ -64,7 +64,7 @@ import org.springframework.transaction.support.TransactionTemplate;
@IgnoreWhen(missesCapabilities = Capabilities.QUERY, clusterTypes = ClusterType.MOCKED)
@SpringJUnitConfig(TransactionsConfig.class)
public class TransactionTemplateIntegrationTests extends JavaIntegrationTests {
TransactionTemplate template;
@Autowired TransactionTemplate transactionTemplate;
@Autowired CouchbaseCallbackTransactionManager transactionManager;
@Autowired CouchbaseClientFactory couchbaseClientFactory;
@Autowired CouchbaseTemplate ops;
@@ -86,8 +86,6 @@ public class TransactionTemplateIntegrationTests extends JavaIntegrationTests {
assertNotInTransaction();
List<RemoveResult> rp0 = ops.removeByQuery(Person.class).withConsistency(REQUEST_PLUS).all();
List<RemoveResult> rp1 = ops.removeByQuery(PersonWithoutVersion.class).withConsistency(REQUEST_PLUS).all();
template = new TransactionTemplate(transactionManager);
}
@AfterEach
@@ -106,7 +104,7 @@ public class TransactionTemplateIntegrationTests extends JavaIntegrationTests {
private RunResult doInTransaction(Consumer<TransactionStatus> lambda) {
AtomicInteger tryCount = new AtomicInteger();
template.executeWithoutResult(status -> {
transactionTemplate.executeWithoutResult(status -> {
TransactionTestUtil.assertInTransaction();
assertFalse(status.hasSavepoint());
assertFalse(status.isRollbackOnly());
@@ -346,7 +344,7 @@ public class TransactionTemplateIntegrationTests extends JavaIntegrationTests {
@DisplayName("Setting an unsupported isolation level should fail")
@Test
public void unsupportedIsolationLevel() {
template.setIsolationLevel(TransactionDefinition.ISOLATION_SERIALIZABLE);
transactionTemplate.setIsolationLevel(TransactionDefinition.ISOLATION_SERIALIZABLE);
assertThrowsWithCause(() -> doInTransaction(status -> {}), IllegalArgumentException.class);
}
@@ -354,9 +352,10 @@ public class TransactionTemplateIntegrationTests extends JavaIntegrationTests {
@DisplayName("Setting PROPAGATION_MANDATORY should fail, as not in a transaction")
@Test
public void propagationMandatoryOutsideTransaction() {
template.setPropagationBehavior(TransactionDefinition.PROPAGATION_MANDATORY);
int propagation = transactionTemplate.getPropagationBehavior();
transactionTemplate.setPropagationBehavior(TransactionDefinition.PROPAGATION_MANDATORY);
assertThrowsWithCause(() -> doInTransaction(status -> {}), IllegalTransactionStateException.class);
transactionTemplate.setPropagationBehavior(propagation);
}
@Test
@@ -364,7 +363,7 @@ public class TransactionTemplateIntegrationTests extends JavaIntegrationTests {
TransactionTemplate template2 = new TransactionTemplate(transactionManager);
template2.setPropagationBehavior(TransactionDefinition.PROPAGATION_MANDATORY);
template.executeWithoutResult(status -> {
transactionTemplate.executeWithoutResult(status -> {
template2.executeWithoutResult(status2 -> {
Person person = ops.insertById(Person.class).one(WalterWhite);
});

4
src/test/java/org/springframework/data/couchbase/transactions/sdk/SDKReactiveTransactionsNonAllowableOperationsIntegrationTests.java
View File

@@ -37,7 +37,6 @@ import org.springframework.data.couchbase.util.Capabilities;
import org.springframework.data.couchbase.util.ClusterType;
import org.springframework.data.couchbase.util.IgnoreWhen;
import org.springframework.data.couchbase.util.JavaIntegrationTests;
import org.springframework.stereotype.Component;
import org.springframework.stereotype.Service;
import org.springframework.test.context.junit.jupiter.SpringJUnitConfig;
@@ -117,8 +116,7 @@ public class SDKReactiveTransactionsNonAllowableOperationsIntegrationTests exten
}
// This is intentionally not a @Transactional service
@Service
@Component
@Service // this will work in the unit tests even without @Service because of explicit loading by @SpringJUnitConfig
static class PersonService {
final ReactiveCouchbaseOperations personOperations;

15
src/test/java/org/springframework/data/couchbase/transactions/sdk/SDKReactiveTransactionsTemplateIntegrationTests.java
View File

@@ -126,6 +126,21 @@ public class SDKReactiveTransactionsTemplateIntegrationTests extends JavaIntegra
assertEquals(1, rr.attempts);
}
private RunResult doInTransaction2(Function<ReactiveTransactionAttemptContext, Mono<?>> lambda,
@Nullable TransactionOptions options) {
AtomicInteger attempts = new AtomicInteger();
TransactionResult result = couchbaseClientFactory.getCluster().reactive().transactions().run(ctx -> {
return TransactionalSupport.checkForTransactionInThreadLocalStorage().then(Mono.defer(() -> {
return lambda.apply(ctx);
}));
}, options).block();
assertNotInTransaction();
return new RunResult(result, attempts.get());
}
@DisplayName("A basic golden path replace should succeed")
@Test
public void committedReplace() {

4
src/test/java/org/springframework/data/couchbase/transactions/sdk/SDKTransactionsNonAllowableOperationsIntegrationTests.java
View File

@@ -36,7 +36,6 @@ import org.springframework.data.couchbase.util.Capabilities;
import org.springframework.data.couchbase.util.ClusterType;
import org.springframework.data.couchbase.util.IgnoreWhen;
import org.springframework.data.couchbase.util.JavaIntegrationTests;
import org.springframework.stereotype.Component;
import org.springframework.stereotype.Service;
import org.springframework.test.context.junit.jupiter.SpringJUnitConfig;
@@ -117,8 +116,7 @@ public class SDKTransactionsNonAllowableOperationsIntegrationTests extends JavaI
}
// This is intentionally not a @Transactional service
@Service
@Component
@Service // this will work in the unit tests even without @Service because of explicit loading by @SpringJUnitConfig
static class PersonService {
final CouchbaseOperations personOperations;