Firestore declarative transactions tests and docs (spring-attic#2039)

* declarative transactions tests and docs; fixes spring-attic#2004

Author: dmitry-s

docs/src/main/asciidoc/firestore.adoc

@@ -180,6 +180,44 @@ Read documents are locked until the transaction finishes with a commit or a roll
 If an `Exception` is thrown within a transaction, the rollback operation is executed.
 Otherwise, the commit operation is executed.
 
+===== Declarative Transactions with @Transactional Annotation
+
+This feature requires a bean of `SpannerTransactionManager`, which is provided when using `spring-cloud-gcp-starter-data-firestore`.
+
+`FirestoreTemplate` and `FirestoreReactiveRepository` support running methods with the `@Transactional` https://docs.spring.io/spring/docs/current/spring-framework-reference/data-access.html#transaction-declarative[annotation] as transactions.
+If a method annotated with `@Transactional` calls another method also annotated, then both methods will work within the same transaction.
+
+One way to use this feature is illustrated here. You would need to do the following:
+
+. Annotate your configuration class with the `@EnableTransactionManagement` annotation.
+
+. Create a service class that has methods annotated with `@Transactional`:
+
+[source, java, indent=0]
+----
+include::{project-root}/spring-cloud-gcp-data-firestore/src/test/java/org/springframework/cloud/gcp/data/firestore/it/UserService.java[tag=user_service]
+----
+
+[start=3]
+. Make a Spring Bean provider that creates an instance of that class:
+
+[source, java, indent=0]
+----
+include::{project-root}/spring-cloud-gcp-data-firestore/src/test/java/org/springframework/cloud/gcp/data/firestore/it/FirestoreIntegrationTestsConfiguration.java[tag=user_service_bean]
+----
+
+After that, you can autowire your service like so:
+[source, java]
+----
+public class MyApplication {
+include::{project-root}/spring-cloud-gcp-data-firestore/src/test/java/org/springframework/cloud/gcp/data/firestore/it/FirestoreRepositoryIntegrationTests.java[tag=autowire_user_service]
+}
+----
+
+Now when you call the methods annotated with `@Transactional` on your service object, a transaction will be automatically started. 
+If an error occurs during the execution of a method annotated with `@Transactional`, the transaction will be rolled back. 
+If no error occurs, the transaction will be committed.
+
 === Reactive Repository Sample
 
 A https://github.com/spring-cloud/spring-cloud-gcp/tree/master/spring-cloud-gcp-samples/spring-cloud-gcp-data-firestore-sample[sample application] is available.

spring-cloud-gcp-data-firestore/src/main/java/org/springframework/cloud/gcp/data/firestore/transaction/ReactiveFirestoreTransactionManager.java

@@ -114,6 +114,7 @@ protected Mono<Void> doRollback(TransactionSynchronizationManager transactionSyn
 						obs -> this.firestore.rollback(RollbackRequest.newBuilder()
 								.setTransaction(
 										extractFirestoreTransaction(genericReactiveTransaction).getTransactionId())
+								.setDatabase(this.databasePath)
 								.build(), obs))
 				.then();
 	}

spring-cloud-gcp-data-firestore/src/test/java/org/springframework/cloud/gcp/data/firestore/it/FirestoreIntegrationTests.java

@@ -41,6 +41,10 @@
 import static org.assertj.core.api.Assertions.assertThat;
 import static org.hamcrest.Matchers.is;
 import static org.junit.Assume.assumeThat;
+import static org.mockito.ArgumentMatchers.any;
+import static org.mockito.Mockito.reset;
+import static org.mockito.Mockito.times;
+import static org.mockito.Mockito.verify;
 
 /**
  * @author Dmitry Solomakha
@@ -79,23 +83,38 @@ public void transactionTest() {
 		User alice = new User("Alice", 29);
 		User bob = new User("Bob", 60);
 
+
 		DefaultTransactionDefinition transactionDefinition = new DefaultTransactionDefinition();
 		transactionDefinition.setReadOnly(false);
 		TransactionalOperator operator = TransactionalOperator.create(this.txManager, transactionDefinition);
 
-		this.firestoreTemplate.save(alice)
-				.then(this.firestoreTemplate.save(bob))
-				.as(operator::transactional).block();
+		reset(this.txManager);
+
+		this.firestoreTemplate.save(alice).then(this.firestoreTemplate.save(bob))
+				.as(operator::transactional)
+				.block();
 
 		assertThat(this.firestoreTemplate.findAll(User.class).collectList().block())
 				.containsExactlyInAnyOrder(bob, alice);
 
+		verify(this.txManager, times(1)).commit(any());
+		verify(this.txManager, times(0)).rollback(any());
+		verify(this.txManager, times(1)).getReactiveTransaction(any());
+
+		reset(this.txManager);
 
 		// test rollback
 		this.firestoreTemplate.saveAll(Mono.defer(() -> {
 			throw new FirestoreDataException("BOOM!");
 		}))
-				.then(this.firestoreTemplate.deleteAll(User.class)).onErrorReturn(0L).block();
+				.then(this.firestoreTemplate.deleteAll(User.class))
+				.as(operator::transactional)
+				.onErrorResume(throwable -> Mono.empty())
+				.block();
+
+		verify(this.txManager, times(0)).commit(any());
+		verify(this.txManager, times(1)).rollback(any());
+		verify(this.txManager, times(1)).getReactiveTransaction(any());
 
 		assertThat(this.firestoreTemplate.count(User.class).block()).isEqualTo(2);
 
@@ -148,7 +167,7 @@ public void writeReadDeleteTest() {
 
 
 	@Test
-	public void saveTest() throws InterruptedException {
+	public void saveTest() {
 		assertThat(this.firestoreTemplate.count(User.class).block()).isEqualTo(0);
 
 		User u1 = new User("Cloud", 22);
@@ -159,7 +178,7 @@ public void saveTest() throws InterruptedException {
 	}
 
 	@Test
-	public void saveAllTest() throws InterruptedException {
+	public void saveAllTest() {
 		User u1 = new User("Cloud", 22);
 		User u2 = new User("Squall", 17);
 		Flux<User> users = Flux.fromArray(new User[]{u1, u2});
@@ -173,7 +192,7 @@ public void saveAllTest() throws InterruptedException {
 	}
 
 	@Test
-	public void saveAllBulkTest() throws InterruptedException {
+	public void saveAllBulkTest() {
 		Flux<User> users = Flux.create(sink -> {
 			for (int i = 0; i < 1000; i++) {
 				sink.next(new User("testUser " + i, i));
@@ -189,7 +208,7 @@ public void saveAllBulkTest() throws InterruptedException {
 	}
 
 	@Test
-	public void deleteTest() throws InterruptedException {
+	public void deleteTest() {
 		this.firestoreTemplate.save(new User("alpha", 45)).block();
 		this.firestoreTemplate.save(new User("beta", 23)).block();
 		this.firestoreTemplate.save(new User("gamma", 44)).block();

spring-cloud-gcp-data-firestore/src/test/java/org/springframework/cloud/gcp/data/firestore/it/FirestoreIntegrationTestsConfiguration.java

@@ -24,6 +24,7 @@
 import io.grpc.ManagedChannel;
 import io.grpc.ManagedChannelBuilder;
 import io.grpc.auth.MoreCallCredentials;
+import org.mockito.Mockito;
 
 import org.springframework.beans.factory.annotation.Value;
 import org.springframework.boot.autoconfigure.condition.ConditionalOnMissingBean;
@@ -34,13 +35,15 @@
 import org.springframework.context.annotation.Bean;
 import org.springframework.context.annotation.Configuration;
 import org.springframework.context.annotation.PropertySource;
+import org.springframework.transaction.annotation.EnableTransactionManagement;
 
 /**
  * Spring config for the integration tests.
  */
 @Configuration
 @PropertySource("application-test.properties")
 @EnableReactiveFirestoreRepositories
+@EnableTransactionManagement
 public class FirestoreIntegrationTestsConfiguration {
 	@Value("projects/${test.integration.firestore.project-id}/databases/(default)/documents")
 	String defaultParent;
@@ -72,7 +75,14 @@ public FirestoreMappingContext firestoreMappingContext() {
 	@ConditionalOnMissingBean
 	public ReactiveFirestoreTransactionManager firestoreTransactionManager(
 			FirestoreGrpc.FirestoreStub firestoreStub) {
-		return new ReactiveFirestoreTransactionManager(firestoreStub, this.defaultParent);
+		return Mockito.spy(new ReactiveFirestoreTransactionManager(firestoreStub, this.defaultParent));
 	}
 
+	//tag::user_service_bean[]
+	@Bean
+	public UserService userService() {
+		return new UserService();
+	}
+	//end::user_service_bean[]
+
 }

spring-cloud-gcp-data-firestore/src/test/java/org/springframework/cloud/gcp/data/firestore/it/FirestoreRepositoryIntegrationTests.java

@@ -25,6 +25,7 @@
 import org.junit.Test;
 import org.junit.runner.RunWith;
 import reactor.core.publisher.Flux;
+import reactor.core.publisher.Mono;
 
 import org.springframework.beans.factory.annotation.Autowired;
 import org.springframework.cloud.gcp.data.firestore.User;
@@ -38,6 +39,10 @@
 import static org.assertj.core.api.Assertions.assertThat;
 import static org.hamcrest.Matchers.is;
 import static org.junit.Assume.assumeThat;
+import static org.mockito.ArgumentMatchers.any;
+import static org.mockito.Mockito.reset;
+import static org.mockito.Mockito.times;
+import static org.mockito.Mockito.verify;
 
 @RunWith(SpringRunner.class)
 @ContextConfiguration(classes = FirestoreIntegrationTestsConfiguration.class)
@@ -52,6 +57,14 @@ public class FirestoreRepositoryIntegrationTests {
 	ReactiveFirestoreTransactionManager txManager;
 	//end::autowire_tx_manager[]
 
+	//tag::autowire_user_service[]
+	@Autowired
+	UserService userService;
+	//end::autowire_user_service[]
+
+	@Autowired
+	ReactiveFirestoreTransactionManager transactionManager;
+
 	@BeforeClass
 	public static void checkToRun() throws IOException {
 		assumeThat("Firestore-sample tests are disabled. "
@@ -62,6 +75,7 @@ public static void checkToRun() throws IOException {
 	@Before
 	public void cleanTestEnvironment() {
 		this.userRepository.deleteAll().block();
+		reset(this.transactionManager);
 	}
 
 	@Test
@@ -124,7 +138,7 @@ public void transactionalOperatorTest() {
 	public void partTreeRepositoryMethodTest() {
 		User u1 = new User("Cloud", 22);
 		User u2 = new User("Squall", 17);
-		Flux<User> users = Flux.fromArray(new User[] { u1, u2 });
+		Flux<User> users = Flux.fromArray(new User[] {u1, u2});
 
 		this.userRepository.saveAll(users).blockLast();
 
@@ -151,4 +165,47 @@ public void pageableQueryTest() {
 
 		assertThat(pagedUsers).containsExactlyInAnyOrder("blah-person5", "blah-person6");
 	}
+
+	@Test
+	public void declarativeTransactionRollbackTest() {
+		this.userService.deleteUsers().onErrorResume(throwable -> Mono.empty()).block();
+
+		verify(this.transactionManager, times(0)).commit(any());
+		verify(this.transactionManager, times(1)).rollback(any());
+		verify(this.transactionManager, times(1)).getReactiveTransaction(any());
+	}
+
+	@Test
+	public void declarativeTransactionCommitTest() {
+		User alice = new User("Alice", 29);
+		User bob = new User("Bob", 60);
+
+		this.userRepository.save(alice).then(this.userRepository.save(bob)).block();
+
+		this.userService.updateUsers().block();
+
+		verify(this.transactionManager, times(1)).commit(any());
+		verify(this.transactionManager, times(0)).rollback(any());
+		verify(this.transactionManager, times(1)).getReactiveTransaction(any());
+
+		assertThat(this.userRepository.findAll().map(User::getAge).collectList().block())
+				.containsExactlyInAnyOrder(28, 59);
+	}
+
+	@Test
+	public void transactionPropagationTest() {
+		User alice = new User("Alice", 29);
+		User bob = new User("Bob", 60);
+
+		this.userRepository.save(alice).then(this.userRepository.save(bob)).block();
+
+		this.userService.updateUsersTransactionPropagation().block();
+
+		verify(this.transactionManager, times(1)).commit(any());
+		verify(this.transactionManager, times(0)).rollback(any());
+		verify(this.transactionManager, times(1)).getReactiveTransaction(any());
+
+		assertThat(this.userRepository.findAll().map(User::getAge).collectList().block())
+				.containsExactlyInAnyOrder(28, 59);
+	}
 }

spring-cloud-gcp-data-firestore/src/test/java/org/springframework/cloud/gcp/data/firestore/it/UserService.java

@@ -0,0 +1,70 @@
+/*
+ * Copyright 2017-2019 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.cloud.gcp.data.firestore.it;
+
+import reactor.core.publisher.Flux;
+import reactor.core.publisher.Mono;
+
+import org.springframework.beans.factory.annotation.Autowired;
+import org.springframework.cloud.gcp.data.firestore.FirestoreDataException;
+import org.springframework.cloud.gcp.data.firestore.User;
+import org.springframework.transaction.annotation.Transactional;
+
+/**
+ * @author Dmitry Solomakha
+ */
+
+//tag::user_service[]
+class UserService {
+	@Autowired
+	private UserRepository userRepository;
+
+	//end::user_service[]
+	@Transactional
+	public Mono<Void> updateUsersTransactionPropagation() {
+		return findAll()
+				.flatMap(a -> {
+					a.setAge(a.getAge() - 1);
+					return this.userRepository.save(a);
+				})
+				.then();
+	}
+
+	@Transactional
+	private Flux<User> findAll() {
+		return this.userRepository.findAll();
+	}
+
+	@Transactional
+	public Mono<Void> deleteUsers() {
+		return this.userRepository.saveAll(Mono.defer(() -> {
+			throw new FirestoreDataException("BOOM!");
+		})).then(this.userRepository.deleteAll());
+	}
+
+	// tag::user_service[]
+	@Transactional
+	public Mono<Void> updateUsers() {
+		return this.userRepository.findAll()
+				.flatMap(a -> {
+					a.setAge(a.getAge() - 1);
+					return this.userRepository.save(a);
+				})
+				.then();
+	}
+}
+//end::user_service[]