Add support for reactive auditing and ReactiveEntityCallbacks.

Add support for reactive auditing and ReactiveEntityCallbacks.
Also, adapt CouchbaseAuditingRegistrar for support AuditingEntityCallback.

Closes #955.
Original pull request: #1102.

Co-authored-by: Carlos Espinado <carlosemart>
Co-authored-by: mikereiche <michael.reiche@couchbase.com>

src/test/java/org/springframework/data/couchbase/domain/Airport.java

@@ -16,8 +16,10 @@
 
 package org.springframework.data.couchbase.domain;
 
+import org.springframework.data.annotation.CreatedBy;
 import org.springframework.data.annotation.Id;
 import org.springframework.data.annotation.PersistenceConstructor;
+import org.springframework.data.annotation.Version;
 import org.springframework.data.annotation.TypeAlias;
 import org.springframework.data.couchbase.core.mapping.Document;
 
@@ -36,6 +38,11 @@ public class Airport extends ComparableEntity {
 
 	String icao;
 
+	@CreatedBy private String createdBy;
+
+	@Version long version;
+
+
 	@PersistenceConstructor
 	public Airport(String id, String iata, String icao) {
 		this.id = id;
@@ -55,4 +62,7 @@ public class Airport extends ComparableEntity {
 		return icao;
 	}
 
+	public String getCreatedBy() {
+		return createdBy;
+	}
 }

src/test/java/org/springframework/data/couchbase/domain/ReactiveNaiveAuditorAware.java

@@ -0,0 +1,35 @@
+/*
+ * Copyright 2012-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.domain;
+
+import org.springframework.data.domain.ReactiveAuditorAware;
+import reactor.core.publisher.Mono;
+
+
+/**
+ * This class returns a string that represents the current user
+ *
+ * @author Jorge Rodríguez Martín
+ * @since 4.2
+ */
+public class ReactiveNaiveAuditorAware implements ReactiveAuditorAware<String> {
+
+	@Override
+	public Mono<String> getCurrentAuditor() {
+		return Mono.just("auditor");
+	}
+
+}

src/test/java/org/springframework/data/couchbase/repository/CouchbaseRepositoryQueryIntegrationTests.java

@@ -38,8 +38,10 @@ import java.util.stream.Collectors;
 import org.junit.jupiter.api.BeforeEach;
 import org.junit.jupiter.api.Test;
 import org.springframework.beans.factory.annotation.Autowired;
+import org.springframework.context.annotation.Bean;
 import org.springframework.context.annotation.Configuration;
 import org.springframework.dao.DataIntegrityViolationException;
+import org.springframework.data.auditing.DateTimeProvider;
 import org.springframework.data.couchbase.CouchbaseClientFactory;
 import org.springframework.data.couchbase.config.AbstractCouchbaseConfiguration;
 import org.springframework.data.couchbase.core.CouchbaseTemplate;
@@ -50,10 +52,13 @@ import org.springframework.data.couchbase.domain.Address;
 import org.springframework.data.couchbase.domain.Airport;
 import org.springframework.data.couchbase.domain.AirportRepository;
 import org.springframework.data.couchbase.domain.Iata;
+import org.springframework.data.couchbase.domain.NaiveAuditorAware;
 import org.springframework.data.couchbase.domain.Person;
 import org.springframework.data.couchbase.domain.PersonRepository;
 import org.springframework.data.couchbase.domain.User;
 import org.springframework.data.couchbase.domain.UserRepository;
+import org.springframework.data.couchbase.domain.time.AuditingDateTimeProvider;
+import org.springframework.data.couchbase.repository.auditing.EnableCouchbaseAuditing;
 import org.springframework.data.couchbase.repository.config.EnableCouchbaseRepositories;
 import org.springframework.data.couchbase.repository.query.CouchbaseQueryMethod;
 import org.springframework.data.couchbase.repository.query.CouchbaseRepositoryQuery;
@@ -378,6 +383,21 @@ public class CouchbaseRepositoryQueryIntegrationTests extends ClusterAwareIntegr
 		assertEquals(user, users.get(0));
 	}
 
+	@Test
+	void findBySimplePropertyAudited() {
+		Airport vie = null;
+		try {
+			vie = new Airport("airports::vie", "vie", "low2");
+			Airport saved = airportRepository.save(vie);
+			List<Airport> airports1 = airportRepository.findAllByIata("vie");
+			assertEquals(saved, airports1.get(0));
+			assertEquals(saved.getCreatedBy(), "auditor"); // NaiveAuditorAware will provide this
+		} finally {
+			airportRepository.delete(vie);
+		}
+	}
+
+
 	private void sleep(int millis) {
 		try {
 			Thread.sleep(millis); // so they are executed out-of-order
@@ -386,6 +406,7 @@ public class CouchbaseRepositoryQueryIntegrationTests extends ClusterAwareIntegr
 
 	@Configuration
 	@EnableCouchbaseRepositories("org.springframework.data.couchbase")
+	@EnableCouchbaseAuditing(auditorAwareRef = "auditorAwareRef", dateTimeProviderRef = "dateTimeProviderRef")
 	static class Config extends AbstractCouchbaseConfiguration {
 
 		@Override
@@ -408,6 +429,15 @@ public class CouchbaseRepositoryQueryIntegrationTests extends ClusterAwareIntegr
 			return bucketName();
 		}
 
+		@Bean(name = "auditorAwareRef")
+		public NaiveAuditorAware testAuditorAware() {
+			return new NaiveAuditorAware();
+		}
+
+		@Bean(name = "dateTimeProviderRef")
+		public DateTimeProvider testDateTimeProvider() {
+			return new AuditingDateTimeProvider();
+		}
 	}
 
 }

src/test/java/org/springframework/data/couchbase/repository/ReactiveCouchbaseRepositoryKeyValueIntegrationTests.java

@@ -18,15 +18,22 @@ package org.springframework.data.couchbase.repository;
 
 import static org.junit.jupiter.api.Assertions.*;
 
+import java.util.List;
 import java.util.Optional;
 import java.util.UUID;
-
 import org.junit.jupiter.api.Test;
 import org.springframework.beans.factory.annotation.Autowired;
+import org.springframework.context.annotation.Bean;
 import org.springframework.context.annotation.Configuration;
+import org.springframework.data.auditing.DateTimeProvider;
 import org.springframework.data.couchbase.config.AbstractCouchbaseConfiguration;
+import org.springframework.data.couchbase.domain.Airport;
+import org.springframework.data.couchbase.domain.ReactiveAirportRepository;
+import org.springframework.data.couchbase.domain.ReactiveNaiveAuditorAware;
 import org.springframework.data.couchbase.domain.ReactiveUserRepository;
 import org.springframework.data.couchbase.domain.User;
+import org.springframework.data.couchbase.domain.time.AuditingDateTimeProvider;
+import org.springframework.data.couchbase.repository.auditing.EnableReactiveCouchbaseAuditing;
 import org.springframework.data.couchbase.repository.config.EnableReactiveCouchbaseRepositories;
 import org.springframework.data.couchbase.util.ClusterAwareIntegrationTests;
 import org.springframework.data.couchbase.util.ClusterType;
@@ -39,23 +46,40 @@ public class ReactiveCouchbaseRepositoryKeyValueIntegrationTests extends Cluster
 
 	@Autowired ReactiveUserRepository userRepository;
 
+	@Autowired ReactiveAirportRepository airportRepository;
+
 	@Test
 	void saveAndFindById() {
 		User user = new User(UUID.randomUUID().toString(), "f", "l");
 
 		assertFalse(userRepository.existsById(user.getId()).block());
 
-		userRepository.save(user).block();
+		final User save = userRepository.save(user).block();
 
 		Optional<User> found = userRepository.findById(user.getId()).blockOptional();
 		assertTrue(found.isPresent());
-		found.ifPresent(u -> assertEquals(user, u));
+		found.ifPresent(u -> assertEquals(save, u));
 
 		assertTrue(userRepository.existsById(user.getId()).block());
 	}
 
+	@Test
+	void findBySimplePropertyAudited() {
+		Airport vie = null;
+		try {
+			vie = new Airport("airports::vie", "vie", "low2");
+			Airport saved = airportRepository.save(vie).block();
+			List<Airport> airports1 = airportRepository.findAllByIata("vie").collectList().block();
+			assertEquals(saved, airports1.get(0));
+			assertEquals(saved.getCreatedBy(), "auditor"); // NaiveAuditorAware will provide this
+		} finally {
+			airportRepository.delete(vie).block();
+		}
+	}
+
 	@Configuration
 	@EnableReactiveCouchbaseRepositories("org.springframework.data.couchbase")
+  @EnableReactiveCouchbaseAuditing
 	static class Config extends AbstractCouchbaseConfiguration {
 
 		@Override
@@ -78,6 +102,16 @@ public class ReactiveCouchbaseRepositoryKeyValueIntegrationTests extends Cluster
 			return bucketName();
 		}
 
+		@Bean(name = "auditorAwareRef")
+		public ReactiveNaiveAuditorAware testAuditorAware() {
+			return new ReactiveNaiveAuditorAware();
+		}
+
+		@Bean(name = "dateTimeProviderRef")
+		public DateTimeProvider testDateTimeProvider() {
+			return new AuditingDateTimeProvider();
+		}
+
 	}
 
 }