DATACOUCH-623 - Add replace() method to CouchbaseRepository for CAS u…

src/main/java/org/springframework/data/couchbase/repository/support/SimpleCouchbaseRepository.java

@@ -26,6 +26,7 @@
 import org.springframework.data.couchbase.core.query.Query;
 import org.springframework.data.couchbase.repository.CouchbaseRepository;
 import org.springframework.data.couchbase.repository.query.CouchbaseEntityInformation;
+import static org.springframework.data.couchbase.repository.support.Util.hasNonZeroVersionProperty;
 import org.springframework.data.domain.Page;
 import org.springframework.data.domain.PageImpl;
 import org.springframework.data.domain.Pageable;
@@ -75,15 +76,19 @@ public SimpleCouchbaseRepository(final CouchbaseEntityInformation<T, String> ent
 	@SuppressWarnings("unchecked")
 	public <S extends T> S save(final S entity) {
 		Assert.notNull(entity, "Entity must not be null!");
-		return (S) couchbaseOperations.upsertById(entityInformation.getJavaType()).one(entity);
+		// if entity has non-null, non-zero version property, then replace()
+		if (hasNonZeroVersionProperty(entity, couchbaseOperations.getConverter())) {
+			return (S) couchbaseOperations.replaceById(entityInformation.getJavaType()).one(entity);
+		} else {
+			return (S) couchbaseOperations.upsertById(entityInformation.getJavaType()).one(entity);
+		}
 	}
 
 	@Override
 	@SuppressWarnings("unchecked")
 	public <S extends T> Iterable<S> saveAll(final Iterable<S> entities) {
 		Assert.notNull(entities, "The given Iterable of entities must not be null!");
-		return (Iterable<S>) couchbaseOperations.upsertById(entityInformation.getJavaType())
-				.all(Streamable.of(entities).toList());
+		return Streamable.of(entities).stream().map((e) -> save(e)).collect(StreamUtils.toUnmodifiableList());
 	}
 
 	@Override

src/main/java/org/springframework/data/couchbase/repository/support/SimpleReactiveCouchbaseRepository.java

@@ -29,6 +29,7 @@
 import org.springframework.data.couchbase.core.query.Query;
 import org.springframework.data.couchbase.repository.ReactiveCouchbaseRepository;
 import org.springframework.data.couchbase.repository.query.CouchbaseEntityInformation;
+import static org.springframework.data.couchbase.repository.support.Util.hasNonZeroVersionProperty;
 import org.springframework.data.domain.Sort;
 import org.springframework.data.util.Streamable;
 import org.springframework.util.Assert;
@@ -75,9 +76,15 @@ public SimpleReactiveCouchbaseRepository(final CouchbaseEntityInformation<T, Str
 	}
 
 	@SuppressWarnings("unchecked")
+	@Override
 	public <S extends T> Mono<S> save(final S entity) {
 		Assert.notNull(entity, "Entity must not be null!");
-		return (Mono<S>) operations.upsertById(entityInformation.getJavaType()).one(entity);
+		// if entity has non-null version property, then replace()
+		if (hasNonZeroVersionProperty(entity, operations.getConverter())) {
+			return (Mono<S>) operations.replaceById(entityInformation.getJavaType()).one(entity);
+		} else {
+			return (Mono<S>) operations.upsertById(entityInformation.getJavaType()).one(entity);
+		}
 	}
 
 	@Override
@@ -89,7 +96,7 @@ public Flux<T> findAll(final Sort sort) {
 	@Override
 	public <S extends T> Flux<S> saveAll(final Iterable<S> entities) {
 		Assert.notNull(entities, "The given Iterable of entities must not be null!");
-		return (Flux<S>) operations.upsertById(entityInformation.getJavaType()).all(Streamable.of(entities).toList());
+		return Flux.fromIterable(entities).flatMap(this::save);
 	}
 
 	@SuppressWarnings("unchecked")

src/main/java/org/springframework/data/couchbase/repository/support/Util.java

@@ -0,0 +1,21 @@
+package org.springframework.data.couchbase.repository.support;
+
+import org.springframework.data.couchbase.core.convert.CouchbaseConverter;
+import org.springframework.data.couchbase.core.mapping.CouchbasePersistentEntity;
+import org.springframework.data.couchbase.core.mapping.CouchbasePersistentProperty;
+
+public class Util {
+
+	public static boolean hasNonZeroVersionProperty(Object entity, CouchbaseConverter converter) {
+		CouchbasePersistentEntity<?> mapperEntity = converter.getMappingContext().getPersistentEntity(entity.getClass());
+		final CouchbasePersistentProperty versionProperty = mapperEntity.getVersionProperty();
+		boolean hasVersionProperty = false;
+		try {
+			if (versionProperty != null && versionProperty.getField() != null) {
+				Object versionValue = versionProperty.getField().get(entity);
+				hasVersionProperty = versionValue != null && !versionValue.equals(Long.valueOf(0));
+			}
+		} catch (IllegalAccessException iae) {}
+		return hasVersionProperty;
+	}
+}

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

@@ -18,6 +18,7 @@
 
 import java.util.List;
 
+import org.springframework.data.couchbase.repository.CouchbaseRepository;
 import com.couchbase.client.java.json.JsonArray;
 import org.springframework.data.couchbase.repository.Query;
 import org.springframework.data.repository.PagingAndSortingRepository;
@@ -31,7 +32,7 @@
  * @author Michael Reiche
  */
 @Repository
-public interface UserRepository extends PagingAndSortingRepository<User, String> {
+public interface UserRepository extends CouchbaseRepository<User, String> {
 
 	List<User> findByFirstname(String firstname);
 

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

@@ -31,11 +31,15 @@
 import org.junit.jupiter.api.Test;
 import org.springframework.beans.factory.annotation.Autowired;
 import org.springframework.context.annotation.Configuration;
+import org.springframework.dao.DataIntegrityViolationException;
 import org.springframework.data.couchbase.CouchbaseClientFactory;
 import org.springframework.data.couchbase.config.AbstractCouchbaseConfiguration;
 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.ReactiveUserRepository;
+import org.springframework.data.couchbase.domain.User;
+import org.springframework.data.couchbase.domain.UserRepository;
 import org.springframework.data.couchbase.domain.Person;
 import org.springframework.data.couchbase.domain.PersonRepository;
 import org.springframework.data.couchbase.repository.config.EnableCouchbaseRepositories;
@@ -61,6 +65,8 @@ public class CouchbaseRepositoryQueryIntegrationTests extends ClusterAwareIntegr
 
 	@Autowired AirportRepository airportRepository;
 
+	@Autowired UserRepository userRepository;
+
 	@BeforeEach
 	void beforeEach() {
 		try {
@@ -139,6 +145,17 @@ void findBySimpleProperty() {
 
 	}
 
+	@Test
+	public void testCas() {
+		User user = new User("1", "Dave", "Wilson");
+		userRepository.save(user);
+		user.setVersion(user.getVersion() - 1);
+		assertThrows(DataIntegrityViolationException.class, () -> userRepository.save(user));
+		user.setVersion(0);
+		userRepository.save(user);
+		userRepository.delete(user);
+	}
+
 	@Test
 	void count() {
 		String[] iatas = { "JFK", "IAD", "SFO", "SJC", "SEA", "LAX", "PHX" };
@@ -153,8 +170,8 @@ void count() {
 				airportRepository.save(airport);
 			}
 
-			Long count = airportRepository.countFancyExpression( Arrays.asList("JFK"), Arrays.asList("jfk"), false);
-			assertEquals( 1, count);
+			Long count = airportRepository.countFancyExpression(Arrays.asList("JFK"), Arrays.asList("jfk"), false);
+			assertEquals(1, count);
 
 			long airportCount = airportRepository.count();
 			assertEquals(7, airportCount);

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

@@ -29,11 +29,14 @@
 import org.junit.jupiter.api.Test;
 import org.springframework.beans.factory.annotation.Autowired;
 import org.springframework.context.annotation.Configuration;
+import org.springframework.dao.DataIntegrityViolationException;
 import org.springframework.dao.DataRetrievalFailureException;
 import org.springframework.data.couchbase.CouchbaseClientFactory;
 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.ReactiveUserRepository;
+import org.springframework.data.couchbase.domain.User;
 import org.springframework.data.couchbase.repository.config.EnableReactiveCouchbaseRepositories;
 import org.springframework.data.couchbase.util.Capabilities;
 import org.springframework.data.couchbase.util.ClusterAwareIntegrationTests;
@@ -56,6 +59,7 @@ public class ReactiveCouchbaseRepositoryQueryIntegrationTests extends ClusterAwa
 	@Autowired CouchbaseClientFactory clientFactory;
 
 	@Autowired ReactiveAirportRepository airportRepository; // intellij flags "Could not Autowire", but it runs ok.
+	@Autowired ReactiveUserRepository userRepository; // intellij flags "Could not Autowire", but it runs ok.
 
 	@BeforeEach
 	void beforeEach() {
@@ -97,6 +101,17 @@ void findBySimpleProperty() {
 		}
 	}
 
+	@Test
+	public void testCas() {
+		User user = new User("1", "Dave", "Wilson");
+		userRepository.save(user).block();
+		user.setVersion(user.getVersion() - 1);
+		assertThrows(DataIntegrityViolationException.class, () -> userRepository.save(user).block());
+		user.setVersion(0);
+		userRepository.save(user).block();
+		userRepository.delete(user).block();
+	}
+
 	@Test
 	void count() {
 		String[] iatas = { "JFK", "IAD", "SFO", "SJC", "SEA", "LAX", "PHX" };