DATACOUCH-955 - 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 <[email protected]>

Author: jorgerod

src/main/java/org/springframework/data/couchbase/config/BeanNames.java

@@ -1,5 +1,5 @@
 /*
- * Copyright 2012-2020 the original author or authors
+ * 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.
@@ -24,6 +24,7 @@
  * @author Michael Nitschinger
  * @author Simon Baslé
  * @author Michael Reiche
+ * @author Jorge Rodríguez Martín
  */
 public class BeanNames {
 
@@ -53,4 +54,9 @@ public class BeanNames {
 	 * The name for the bean that will handle audit trail marking of entities.
 	 */
 	public static final String COUCHBASE_AUDITING_HANDLER = "couchbaseAuditingHandler";
+
+	/**
+	 * The name for the bean that will handle reactive audit trail marking of entities.
+	 */
+	public static final String REACTIVE_COUCHBASE_AUDITING_HANDLER = "reactiveCouchbaseAuditingHandler";
 }

src/main/java/org/springframework/data/couchbase/core/CouchbaseTemplate.java

@@ -1,5 +1,5 @@
 /*
- * Copyright 2012-2020 the original author or authors
+ * 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.
@@ -180,4 +180,8 @@ private void prepareIndexCreator(final ApplicationContext context) {
 			}
 		}
 	}
+
+	TemplateSupport support() {
+		return templateSupport;
+	}
 }

src/main/java/org/springframework/data/couchbase/core/CouchbaseTemplateSupport.java

@@ -1,5 +1,6 @@
 /*
- * Copyright 2012-2020 the original author or authors
+/*
+ * 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.
@@ -45,9 +46,10 @@
  * @author Michael Nitschinger
  * @author Michael Reiche
  * @author Jorge Rodriguez Martin
+ * @author Carlos Espinaco
  * @since 3.0
  */
-class CouchbaseTemplateSupport implements ApplicationContextAware {
+class CouchbaseTemplateSupport implements ApplicationContextAware, TemplateSupport {
 
 	private static final Logger LOG = LoggerFactory.getLogger(CouchbaseTemplateSupport.class);
 
@@ -63,6 +65,7 @@ public CouchbaseTemplateSupport(final CouchbaseConverter converter, final Transl
 		this.translationService = translationService;
 	}
 
+	@Override
 	public CouchbaseDocument encodeEntity(final Object entityToEncode) {
 		maybeEmitEvent(new BeforeConvertEvent<>(entityToEncode));
 		Object maybeNewEntity = maybeCallBeforeConvert(entityToEncode, "");
@@ -73,6 +76,7 @@ public CouchbaseDocument encodeEntity(final Object entityToEncode) {
 		return converted;
 	}
 
+	@Override
 	public <T> T decodeEntity(String id, String source, long cas, Class<T> entityClass) {
 		final CouchbaseDocument converted = new CouchbaseDocument(id);
 		converted.setId(id);
@@ -90,6 +94,7 @@ public <T> T decodeEntity(String id, String source, long cas, Class<T> entityCla
 		return accessor.getBean();
 	}
 
+	@Override
 	public Object applyUpdatedCas(final Object entity, final long cas) {
 		final ConvertingPropertyAccessor<Object> accessor = getPropertyAccessor(entity);
 		final CouchbasePersistentEntity<?> persistentEntity = mappingContext.getRequiredPersistentEntity(entity.getClass());
@@ -102,6 +107,7 @@ public Object applyUpdatedCas(final Object entity, final long cas) {
 		return entity;
 	}
 
+	@Override
 	public Object applyUpdatedId(final Object entity, Object id) {
 		final ConvertingPropertyAccessor<Object> accessor = getPropertyAccessor(entity);
 		final CouchbasePersistentEntity<?> persistentEntity = mappingContext.getRequiredPersistentEntity(entity.getClass());
@@ -114,6 +120,7 @@ public Object applyUpdatedId(final Object entity, Object id) {
 		return entity;
 	}
 
+	@Override
 	public long getCas(final Object entity) {
 		final ConvertingPropertyAccessor<Object> accessor = getPropertyAccessor(entity);
 		final CouchbasePersistentEntity<?> persistentEntity = mappingContext.getRequiredPersistentEntity(entity.getClass());
@@ -129,6 +136,7 @@ public long getCas(final Object entity) {
 		return cas;
 	}
 
+	@Override
 	public String getJavaNameForEntity(final Class<?> clazz) {
 		final CouchbasePersistentEntity<?> persistentEntity = mappingContext.getRequiredPersistentEntity(clazz);
 		MappingCouchbaseEntityInformation<?, Object> info = new MappingCouchbaseEntityInformation<>(persistentEntity);

src/main/java/org/springframework/data/couchbase/core/ExecutableFindByAnalyticsOperationSupport.java

@@ -1,5 +1,5 @@
 /*
- * Copyright 2012-2020 the original author or authors
+ * 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.
@@ -53,7 +53,7 @@ static class ExecutableFindByAnalyticsSupport<T> implements ExecutableFindByAnal
 			this.domainType = domainType;
 			this.query = query;
 			this.reactiveSupport = new ReactiveFindByAnalyticsSupport<>(template.reactive(), domainType, query,
-					scanConsistency);
+					scanConsistency, new NonReactiveSupportWrapper(template.support()));
 			this.scanConsistency = scanConsistency;
 		}
 

src/main/java/org/springframework/data/couchbase/core/ExecutableFindByIdOperationSupport.java

@@ -1,5 +1,5 @@
 /*
- * Copyright 2012-2020 the original author or authors
+ * 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.
@@ -48,7 +48,7 @@ static class ExecutableFindByIdSupport<T> implements ExecutableFindById<T> {
 			this.domainType = domainType;
 			this.collection = collection;
 			this.fields = fields;
-			this.reactiveSupport = new ReactiveFindByIdSupport<>(template.reactive(), domainType, collection, fields);
+			this.reactiveSupport = new ReactiveFindByIdSupport<>(template.reactive(), domainType, collection, fields, new NonReactiveSupportWrapper(template.support()));
 		}
 
 		@Override

src/main/java/org/springframework/data/couchbase/core/ExecutableFindByQueryOperationSupport.java

@@ -1,5 +1,5 @@
 /*
- * Copyright 2012-2020 the original author or authors
+ * 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.
@@ -65,7 +65,7 @@ static class ExecutableFindByQuerySupport<T> implements ExecutableFindByQuery<T>
 			this.returnType = returnType;
 			this.query = query;
 			this.reactiveSupport = new ReactiveFindByQuerySupport<T>(template.reactive(), domainType, returnType, query,
-					scanConsistency, collection, distinctFields);
+					scanConsistency, collection, distinctFields, new NonReactiveSupportWrapper(template.support()));
 			this.scanConsistency = scanConsistency;
 			this.collection = collection;
 			this.distinctFields = distinctFields;

src/main/java/org/springframework/data/couchbase/core/ExecutableFindFromReplicasByIdOperationSupport.java

@@ -1,5 +1,5 @@
 /*
- * Copyright 2012-2020 the original author or authors
+ * 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.
@@ -48,7 +48,7 @@ static class ExecutableFindFromReplicasByIdSupport<T> implements ExecutableFindF
 			this.collection = collection;
 			this.returnType = returnType;
 			this.reactiveSupport = new ReactiveFindFromReplicasByIdSupport<>(template.reactive(), domainType, returnType,
-					collection);
+					collection, new NonReactiveSupportWrapper(template.support()));
 		}
 
 		@Override

src/main/java/org/springframework/data/couchbase/core/ExecutableInsertByIdOperationSupport.java

@@ -1,5 +1,5 @@
 /*
- * Copyright 2012-2020 the original author or authors
+ * 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.
@@ -62,7 +62,7 @@ static class ExecutableInsertByIdSupport<T> implements ExecutableInsertById<T> {
 			this.durabilityLevel = durabilityLevel;
 			this.expiry = expiry;
 			this.reactiveSupport = new ReactiveInsertByIdSupport<>(template.reactive(), domainType, collection, persistTo,
-					replicateTo, durabilityLevel, expiry);
+					replicateTo, durabilityLevel, expiry, new NonReactiveSupportWrapper(template.support()));
 		}
 
 		@Override

src/main/java/org/springframework/data/couchbase/core/ExecutableReplaceByIdOperationSupport.java

@@ -1,5 +1,5 @@
 /*
- * Copyright 2012-2020 the original author or authors
+ * 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.
@@ -62,7 +62,7 @@ static class ExecutableReplaceByIdSupport<T> implements ExecutableReplaceById<T>
 			this.durabilityLevel = durabilityLevel;
 			this.expiry = expiry;
 			this.reactiveSupport = new ReactiveReplaceByIdSupport<>(template.reactive(),
-					domainType, collection, persistTo, replicateTo, durabilityLevel, expiry);
+					domainType, collection, persistTo, replicateTo, durabilityLevel, expiry, new NonReactiveSupportWrapper(template.support()));
 		}
 
 		@Override

src/main/java/org/springframework/data/couchbase/core/ExecutableUpsertByIdOperationSupport.java

@@ -1,5 +1,5 @@
 /*
- * Copyright 2012-2020 the original author or authors
+ * 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.
@@ -62,7 +62,7 @@ static class ExecutableUpsertByIdSupport<T> implements ExecutableUpsertById<T> {
 			this.durabilityLevel = durabilityLevel;
 			this.expiry = expiry;
 			this.reactiveSupport = new ReactiveUpsertByIdSupport<>(template.reactive(),
-					domainType, collection, persistTo, replicateTo, durabilityLevel, expiry);
+					domainType, collection, persistTo, replicateTo, durabilityLevel, expiry, new NonReactiveSupportWrapper(template.support()));
 		}
 
 		@Override

src/main/java/org/springframework/data/couchbase/core/NonReactiveSupportWrapper.java

@@ -0,0 +1,65 @@
+/*
+ * 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.springframework.data.couchbase.core.mapping.CouchbaseDocument;
+
+import reactor.core.publisher.Mono;
+
+/**
+ * Wrapper of {@link TemplateSupport} methods to adapt them to {@link ReactiveTemplateSupport}.
+ *
+ * @author Carlos Espinaco
+ * @since 4.2
+ */
+public class NonReactiveSupportWrapper implements ReactiveTemplateSupport {
+
+	private final TemplateSupport support;
+
+	public NonReactiveSupportWrapper(TemplateSupport support) {
+		this.support = support;
+	}
+
+	@Override
+	public Mono<CouchbaseDocument> encodeEntity(Object entityToEncode) {
+		return Mono.fromSupplier(() -> support.encodeEntity(entityToEncode));
+	}
+
+	@Override
+	public <T> Mono<T> decodeEntity(String id, String source, long cas, Class<T> entityClass) {
+		return Mono.fromSupplier(() -> support.decodeEntity(id, source, cas, entityClass));
+	}
+
+	@Override
+	public Mono<Object> applyUpdatedCas(Object entity, long cas) {
+		return Mono.fromSupplier(() -> support.applyUpdatedCas(entity, cas));
+	}
+
+	@Override
+	public Mono<Object> applyUpdatedId(Object entity, Object id) {
+		return Mono.fromSupplier(() -> support.applyUpdatedId(entity, id));
+	}
+
+	@Override
+	public Long getCas(Object entity) {
+		return support.getCas(entity);
+	}
+
+	@Override
+	public String getJavaNameForEntity(Class<?> clazz) {
+		return support.getJavaNameForEntity(clazz);
+	}
+}

src/main/java/org/springframework/data/couchbase/core/ReactiveCouchbaseTemplate.java

@@ -1,5 +1,5 @@
 /*
- * Copyright 2012-2020 the original author or authors
+ * 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.
@@ -34,13 +34,14 @@
  * @author Michael Nitschinger
  * @author Michael Reiche
  * @author Jorge Rodriguez Martin
+ * @author Carlos Espinaco
  */
 public class ReactiveCouchbaseTemplate implements ReactiveCouchbaseOperations, ApplicationContextAware {
 
 	private final CouchbaseClientFactory clientFactory;
 	private final CouchbaseConverter converter;
 	private final PersistenceExceptionTranslator exceptionTranslator;
-	private final CouchbaseTemplateSupport templateSupport;
+	private final ReactiveCouchbaseTemplateSupport templateSupport;
 
 	public ReactiveCouchbaseTemplate(final CouchbaseClientFactory clientFactory, final CouchbaseConverter converter) {
 		this(clientFactory, converter, new JacksonTranslationService());
@@ -51,7 +52,7 @@ public ReactiveCouchbaseTemplate(final CouchbaseClientFactory clientFactory, fin
 		this.clientFactory = clientFactory;
 		this.converter = converter;
 		this.exceptionTranslator = clientFactory.getExceptionTranslator();
-		this.templateSupport = new CouchbaseTemplateSupport(converter, translationService);
+		this.templateSupport = new ReactiveCouchbaseTemplateSupport(converter, translationService);
 	}
 
 	@Override
@@ -134,7 +135,7 @@ public CouchbaseConverter getConverter() {
 		return converter;
 	}
 
-	CouchbaseTemplateSupport support() {
+	ReactiveTemplateSupport support() {
 		return templateSupport;
 	}