diff --git a/src/integration/java/org/springframework/data/couchbase/core/AsyncUtils.java b/src/integration/java/org/springframework/data/couchbase/core/AsyncUtils.java index 8df3f22e..7d4a25d4 100644 --- a/src/integration/java/org/springframework/data/couchbase/core/AsyncUtils.java +++ b/src/integration/java/org/springframework/data/couchbase/core/AsyncUtils.java @@ -1,5 +1,7 @@ package org.springframework.data.couchbase.core; +import rx.observers.TestSubscriber; + import java.util.Collection; import java.util.Collections; import java.util.List; @@ -17,4 +19,37 @@ public class AsyncUtils { future.get(numThreads, TimeUnit.SECONDS); } } + + public static void awaitCompleted(TestSubscriber testSubscriber) { + testSubscriber.awaitTerminalEvent(); + testSubscriber.assertNoErrors(); + testSubscriber.assertNoValues(); + testSubscriber.assertCompleted(); + } + + public static void awaitCompletedWithAnyValue(TestSubscriber testSubscriber) { + testSubscriber.awaitTerminalEvent(); + testSubscriber.assertNoErrors(); + testSubscriber.assertCompleted(); + } + + public static void awaitCompletedWithValueCount(TestSubscriber testSubscriber, int count) { + testSubscriber.awaitTerminalEvent(); + testSubscriber.assertNoErrors(); + testSubscriber.assertValueCount(count); + testSubscriber.assertCompleted(); + } + + public static void awaitError(TestSubscriber testSubscriber, Class throwableClazz) { + testSubscriber.awaitTerminalEvent(); + testSubscriber.assertError(throwableClazz); + testSubscriber.assertNoValues(); + } + + public static void awaitValue(TestSubscriber testSubscriber, T value) { + testSubscriber.awaitTerminalEvent(); + testSubscriber.assertNoErrors(); + testSubscriber.assertValue(value); + testSubscriber.assertCompleted(); + } } diff --git a/src/integration/java/org/springframework/data/couchbase/core/RxJavaCouchbaseTemplateTests.java b/src/integration/java/org/springframework/data/couchbase/core/RxJavaCouchbaseTemplateTests.java index 41a5d95f..c6902bce 100644 --- a/src/integration/java/org/springframework/data/couchbase/core/RxJavaCouchbaseTemplateTests.java +++ b/src/integration/java/org/springframework/data/couchbase/core/RxJavaCouchbaseTemplateTests.java @@ -22,6 +22,7 @@ import static org.hamcrest.MatcherAssert.assertThat; import static org.hamcrest.core.IsEqual.equalTo; import static org.junit.Assert.*; +import java.io.IOException; import java.util.*; import com.couchbase.client.java.Bucket; @@ -37,12 +38,15 @@ import com.couchbase.client.java.view.Stale; import com.couchbase.client.java.view.ViewQuery; import com.fasterxml.jackson.core.type.TypeReference; import com.fasterxml.jackson.databind.ObjectMapper; +import org.junit.Before; import org.junit.Rule; import org.junit.Test; import org.junit.rules.TestName; import org.junit.runner.RunWith; import org.springframework.beans.factory.annotation.Autowired; +import org.springframework.dao.DataRetrievalFailureException; +import org.springframework.dao.OptimisticLockingFailureException; import org.springframework.data.annotation.Id; import org.springframework.data.annotation.Version; import org.springframework.data.couchbase.ReactiveIntegrationTestApplicationConfig; @@ -51,9 +55,11 @@ import org.springframework.data.couchbase.core.mapping.Document; import org.springframework.test.context.ContextConfiguration; import org.springframework.test.context.TestExecutionListeners; import org.springframework.test.context.junit4.SpringJUnit4ClassRunner; +import rx.observers.TestSubscriber; /** * @author Subhashni Balakrishnan + * @author Alex Derkach **/ @RunWith(SpringJUnit4ClassRunner.class) @ContextConfiguration(classes = ReactiveIntegrationTestApplicationConfig.class) @@ -70,77 +76,199 @@ public class RxJavaCouchbaseTemplateTests { private RxJavaCouchbaseOperations template; private static final ObjectMapper MAPPER = new ObjectMapper(); + private static final String DEFAULT_ID = "reactivebeers:awesome-stout"; + private static final String DEFAULT_NAME = "The Awesome Stout"; + private static final boolean DEFAULT_ACTIVE = false; + private static final String DEFAULT_DESCRIPTION = ""; + @Before + public void setUp() throws Exception { + removeIfExist(DEFAULT_ID); + } private void removeIfExist(String key) { - template.remove(key).subscribe( - v -> {}, - err -> {} - ); + TestSubscriber subscriber = TestSubscriber.create(); + template.remove(key) + .subscribe(subscriber); + subscriber.awaitTerminalEvent(); } private void removeCollectionIfExist(Collection beers) { + TestSubscriber subscriber = TestSubscriber.create(); template.remove(beers, PersistTo.MASTER, ReplicateTo.NONE) - .subscribe( - v -> {}, - err -> {} - ); + .subscribe(subscriber); + subscriber.awaitTerminalEvent(); } @Test - public void saveSimpleEntityCorrectly() throws Exception { - String id = "reactivebeers:awesome-stout"; - removeIfExist(id); + public void upsertNonVersionedEntityCorrectlyWhenSaveIsCalled() throws Exception { + String newName = DEFAULT_NAME + "Second"; + ReactiveBeer firstBeer = new ReactiveBeer(DEFAULT_ID, DEFAULT_NAME, DEFAULT_ACTIVE, DEFAULT_DESCRIPTION); + ReactiveBeer secondBeer = new ReactiveBeer(DEFAULT_ID, newName, DEFAULT_ACTIVE, DEFAULT_DESCRIPTION); + TestSubscriber firstSaveSubscriber = TestSubscriber.create(); + TestSubscriber secondSaveSubscriber = TestSubscriber.create(); - String name = "The Awesome Stout"; - boolean active = false; - ReactiveBeer beer = new ReactiveBeer(id, name, active, ""); + template.save(firstBeer).subscribe(firstSaveSubscriber); + template.save(secondBeer).subscribe(secondSaveSubscriber); - template.save(beer) - .subscribe(); - RawJsonDocument resultDoc = client.get(id, RawJsonDocument.class); - assertNotNull(resultDoc); - String result = resultDoc.content(); - assertNotNull(result); - Map resultConv = MAPPER.readValue(result, new TypeReference>() { - }); + AsyncUtils.awaitCompletedWithAnyValue(firstSaveSubscriber); + AsyncUtils.awaitCompletedWithAnyValue(secondSaveSubscriber); + validateBeer(DEFAULT_ID, newName, DEFAULT_ACTIVE, DEFAULT_DESCRIPTION, ReactiveBeer.class); + } - assertNotNull(resultConv.get(MappingCouchbaseConverter.TYPEKEY_DEFAULT)); - assertNull(resultConv.get("javaClass")); - assertEquals("org.springframework.data.couchbase.core.ReactiveBeer", resultConv.get(MappingCouchbaseConverter.TYPEKEY_DEFAULT)); - assertEquals(false, resultConv.get("is_active")); - assertEquals("The Awesome Stout", resultConv.get("name")); - removeIfExist(id); + @Test + public void replaceVersionedEntityCorrectlyWhenSaveIsCalledAndCasIsNotZero() throws Exception { + String newName = DEFAULT_NAME + "Second"; + VersionedReactiveBeer firstBeer = new VersionedReactiveBeer(DEFAULT_ID, DEFAULT_NAME, DEFAULT_ACTIVE, DEFAULT_DESCRIPTION); + VersionedReactiveBeer secondBeer = new VersionedReactiveBeer(DEFAULT_ID, newName, DEFAULT_ACTIVE, DEFAULT_DESCRIPTION); + + long version = template.save(firstBeer).toBlocking().single().getVersion(); + assertTrue(version > 0); + secondBeer.setVersion(version); + long newVersion = template.save(secondBeer).toBlocking().single().getVersion(); + assertTrue(newVersion > 0); + assertNotEquals(version, newVersion); + + validateBeer(DEFAULT_ID, newName, DEFAULT_ACTIVE, DEFAULT_DESCRIPTION, VersionedReactiveBeer.class); + } + + @Test + public void throwExceptionWhenSaveIsCalledAndCasIsMissmatched() throws Exception { + String newName = DEFAULT_NAME + "Second"; + VersionedReactiveBeer firstBeer = new VersionedReactiveBeer(DEFAULT_ID, DEFAULT_NAME, DEFAULT_ACTIVE, DEFAULT_DESCRIPTION); + VersionedReactiveBeer secondBeer = new VersionedReactiveBeer(DEFAULT_ID, newName, DEFAULT_ACTIVE, DEFAULT_DESCRIPTION); + TestSubscriber secondSaveSubscriber = TestSubscriber.create(); + + long version = template.save(firstBeer).toBlocking().single().getVersion(); + assertTrue(version > 0); + secondBeer.setVersion(version + 1234); + template.save(secondBeer).subscribe(secondSaveSubscriber); + AsyncUtils.awaitError(secondSaveSubscriber, OptimisticLockingFailureException.class); + + validateBeer(DEFAULT_ID, DEFAULT_NAME, DEFAULT_ACTIVE, DEFAULT_DESCRIPTION, VersionedReactiveBeer.class); + } + + @Test + public void throwExceptionWhenSaveIsCalledAndCasIsZeroAndEntityAlreadyExists() throws Exception { + String newName = DEFAULT_NAME + "Second"; + VersionedReactiveBeer firstBeer = new VersionedReactiveBeer(DEFAULT_ID, DEFAULT_NAME, DEFAULT_ACTIVE, DEFAULT_DESCRIPTION); + VersionedReactiveBeer secondBeer = new VersionedReactiveBeer(DEFAULT_ID, newName, DEFAULT_ACTIVE, DEFAULT_DESCRIPTION); + TestSubscriber secondSaveSubscriber = TestSubscriber.create(); + + long version = template.save(firstBeer).toBlocking().single().getVersion(); + assertTrue(version > 0); + template.save(secondBeer).subscribe(secondSaveSubscriber); + AsyncUtils.awaitError(secondSaveSubscriber, OptimisticLockingFailureException.class); + + validateBeer(DEFAULT_ID, DEFAULT_NAME, DEFAULT_ACTIVE, DEFAULT_DESCRIPTION, VersionedReactiveBeer.class); } @Test public void saveCollectionCorrectly() throws Exception { Collection beers = new ArrayList<>(); - String name = "The Awesome Stout"; - - for (int i=0; i < 10000; i++) { + TestSubscriber testSubscriber = TestSubscriber.create(); + String name = DEFAULT_NAME; + int collectionSize = 10000; + for (int i = 0; i < collectionSize; i++) { beers.add(new ReactiveBeer("beerCollItem" + i, name + i, false, "")); } removeCollectionIfExist(beers); - template.save(beers).subscribe(); + + template.save(beers).subscribe(testSubscriber); + + AsyncUtils.awaitCompletedWithValueCount(testSubscriber, collectionSize); + } + + @Test + public void insertSimpleEntityCorrectly() throws Exception { + TestSubscriber testSubscriber = TestSubscriber.create(); + ReactiveBeer beer = new ReactiveBeer(DEFAULT_ID, DEFAULT_NAME, DEFAULT_ACTIVE, DEFAULT_DESCRIPTION); + + template.insert(beer).subscribe(testSubscriber); + + AsyncUtils.awaitCompletedWithAnyValue(testSubscriber); + validateBeer(DEFAULT_ID, DEFAULT_NAME, DEFAULT_ACTIVE, DEFAULT_DESCRIPTION, ReactiveBeer.class); + } + + @Test + public void expectErrorWhenDocumentExistsAndInsertIsCalled() throws Exception { + TestSubscriber firstInsertSubscriber = TestSubscriber.create(); + TestSubscriber secondInsertSubscriber = TestSubscriber.create(); + ReactiveBeer beer = new ReactiveBeer(DEFAULT_ID, DEFAULT_NAME, DEFAULT_ACTIVE, DEFAULT_DESCRIPTION); + + template.insert(beer).subscribe(firstInsertSubscriber); + AsyncUtils.awaitCompletedWithAnyValue(firstInsertSubscriber); + + template.insert(beer).subscribe(secondInsertSubscriber); + AsyncUtils.awaitError(secondInsertSubscriber, OptimisticLockingFailureException.class); + } + + @Test + public void insertCollectionCorrectly() throws Exception { + TestSubscriber testSubscriber = TestSubscriber.create(); + Collection beers = new ArrayList<>(); + String name = DEFAULT_NAME; + int collectionSize = 10000; + + for (int i = 0; i < collectionSize; i++) { + beers.add(new ReactiveBeer("beerCollItem" + i, name + i, false, "")); + } + removeCollectionIfExist(beers); + + template.insert(beers).subscribe(testSubscriber); + + AsyncUtils.awaitCompletedWithValueCount(testSubscriber, collectionSize); + } + + @Test + public void replaceSimpleEntityCorrectly() throws Exception { + TestSubscriber firstInsertSubscriber = TestSubscriber.create(); + TestSubscriber replaceTestSubscriber = TestSubscriber.create(); + String newName = DEFAULT_NAME + " New"; + VersionedReactiveBeer beer = new VersionedReactiveBeer(DEFAULT_ID, DEFAULT_NAME, DEFAULT_ACTIVE, DEFAULT_DESCRIPTION); + + template.insert(beer).subscribe(firstInsertSubscriber); + AsyncUtils.awaitCompletedWithAnyValue(firstInsertSubscriber); + + template.findById(DEFAULT_ID, VersionedReactiveBeer.class) + .doOnNext(v -> v.setName(newName)) + .flatMap(v -> template.update(v)) + .subscribe(replaceTestSubscriber); + + AsyncUtils.awaitCompletedWithAnyValue(replaceTestSubscriber); + validateBeer(DEFAULT_ID, newName, DEFAULT_ACTIVE, DEFAULT_DESCRIPTION, VersionedReactiveBeer.class); + } + + @Test + public void expectErrorWhenReplacingSimpleEntityWhichDoesNotExist() throws Exception { + TestSubscriber testSubscriber = TestSubscriber.create(); + VersionedReactiveBeer beer = new VersionedReactiveBeer(DEFAULT_ID, DEFAULT_NAME, DEFAULT_ACTIVE, DEFAULT_DESCRIPTION); + + template.update(beer).subscribe(testSubscriber); + AsyncUtils.awaitError(testSubscriber, DataRetrievalFailureException.class); } @Test public void removeDocument() { - String id = "beers:to-delete-stout"; - ReactiveBeer beer = new ReactiveBeer(id, "", false, ""); - removeIfExist(id); + TestSubscriber firstInsertSubscriber = TestSubscriber.create(); + TestSubscriber firstFindSubscriber = TestSubscriber.create(); + TestSubscriber removalTestSubscriber = TestSubscriber.create(); + TestSubscriber secondFindSubscriber = TestSubscriber.create(); + ReactiveBeer beer = new ReactiveBeer(DEFAULT_ID, DEFAULT_NAME, DEFAULT_ACTIVE, DEFAULT_DESCRIPTION); - template.save(beer).subscribe(); - Object result = template.findById(id, ReactiveBeer.class).toBlocking().single(); - assertNotNull(result); + template.save(beer).subscribe(firstInsertSubscriber); + AsyncUtils.awaitCompletedWithAnyValue(firstInsertSubscriber); - template.remove(beer).subscribe(); - result = client.get(id); - assertNull(result); + template.findById(DEFAULT_ID, ReactiveBeer.class).subscribe(firstFindSubscriber); + AsyncUtils.awaitCompletedWithValueCount(firstFindSubscriber, 1); + + template.remove(beer).subscribe(removalTestSubscriber); + AsyncUtils.awaitCompletedWithValueCount(removalTestSubscriber, 1); + + template.findById(DEFAULT_ID, ReactiveBeer.class).subscribe(secondFindSubscriber); + AsyncUtils.awaitValue(secondFindSubscriber, null); } - @Test public void storeListsAndMaps() { String id = "persons:lots-of-names"; @@ -171,18 +299,16 @@ public class RxJavaCouchbaseTemplateTests { @Test public void validFindById() { - String id = "reactive beers:findme-stout"; - String name = "Findme Stout"; - boolean active = true; - ReactiveBeer beer = new ReactiveBeer(id, name, active, ""); - template.save(beer).subscribe(); + TestSubscriber saveSubscriber = TestSubscriber.create(); + TestSubscriber findSubscriber = TestSubscriber.create(); - ReactiveBeer found = template.findById(id, ReactiveBeer.class).toBlocking().single(); + ReactiveBeer beer = new ReactiveBeer(DEFAULT_ID, DEFAULT_NAME, DEFAULT_ACTIVE, DEFAULT_DESCRIPTION); - assertNotNull(found); - assertEquals(id, found.getId()); - assertEquals(name, found.getName()); - assertEquals(active, found.getActive()); + template.save(beer).subscribe(saveSubscriber); + AsyncUtils.awaitCompletedWithAnyValue(saveSubscriber); + + template.findById(DEFAULT_ID, ReactiveBeer.class).subscribe(findSubscriber); + AsyncUtils.awaitValue(findSubscriber, beer); } @Test @@ -297,6 +423,21 @@ public class RxJavaCouchbaseTemplateTests { } } + private void validateBeer(String id, String name, boolean active, String description, Class clazz) throws IOException { + RawJsonDocument resultDoc = client.get(id, RawJsonDocument.class); + assertNotNull(resultDoc); + String result = resultDoc.content(); + assertNotNull(result); + Map resultConv = MAPPER.readValue(result, new TypeReference>() {}); + + assertNotNull(resultConv.get(MappingCouchbaseConverter.TYPEKEY_DEFAULT)); + assertNull(resultConv.get("javaClass")); + assertEquals(clazz.getCanonicalName(), resultConv.get(MappingCouchbaseConverter.TYPEKEY_DEFAULT)); + assertEquals(active, resultConv.get("is_active")); + assertEquals(name, resultConv.get("name")); + assertEquals(description, resultConv.get("desc")); + } + /** * A sample document with just an id and property. */ diff --git a/src/main/java/org/springframework/data/couchbase/core/RxJavaCouchbaseOperations.java b/src/main/java/org/springframework/data/couchbase/core/RxJavaCouchbaseOperations.java index 26545bd0..c950de7d 100644 --- a/src/main/java/org/springframework/data/couchbase/core/RxJavaCouchbaseOperations.java +++ b/src/main/java/org/springframework/data/couchbase/core/RxJavaCouchbaseOperations.java @@ -15,8 +15,6 @@ */ package org.springframework.data.couchbase.core; -import java.util.Collection; - import com.couchbase.client.java.Bucket; import com.couchbase.client.java.PersistTo; import com.couchbase.client.java.ReplicateTo; @@ -33,6 +31,7 @@ import rx.Observable; /** * @author Subhashni Balakrishnan + * @author Alex Derkach * @since 3.0 */ public interface RxJavaCouchbaseOperations { @@ -45,6 +44,22 @@ public interface RxJavaCouchbaseOperations { Observable save(Iterable batchToSave, PersistTo persistTo, ReplicateTo replicateTo); + Observable insert(T objectToSave); + + Observable insert(Iterable batchToSave); + + Observable insert(T objectToSave, PersistTo persistTo, ReplicateTo replicateTo); + + Observable insert(Iterable batchToSave, PersistTo persistTo, ReplicateTo replicateTo); + + Observable update(T objectToSave); + + Observable update(Iterable batchToSave); + + Observable update(T objectToSave, PersistTo persistTo, ReplicateTo replicateTo); + + Observable update(Iterable batchToSave, PersistTo persistTo, ReplicateTo replicateTo); + Observable remove(T objectToRemove); Observable remove(T objectToRemove, PersistTo persistTo, ReplicateTo replicateTo); diff --git a/src/main/java/org/springframework/data/couchbase/core/RxJavaCouchbaseTemplate.java b/src/main/java/org/springframework/data/couchbase/core/RxJavaCouchbaseTemplate.java index ab84b0c7..51f4eddc 100644 --- a/src/main/java/org/springframework/data/couchbase/core/RxJavaCouchbaseTemplate.java +++ b/src/main/java/org/springframework/data/couchbase/core/RxJavaCouchbaseTemplate.java @@ -18,7 +18,6 @@ package org.springframework.data.couchbase.core; import static org.springframework.data.couchbase.core.CouchbaseTemplate.ensureNotIterable; -import java.util.Collection; import java.util.Optional; import com.couchbase.client.java.AsyncBucket; @@ -29,10 +28,13 @@ import com.couchbase.client.java.cluster.ClusterInfo; import com.couchbase.client.java.document.Document; import com.couchbase.client.java.document.RawJsonDocument; import com.couchbase.client.java.document.json.JsonObject; +import com.couchbase.client.java.error.CASMismatchException; +import com.couchbase.client.java.error.DocumentAlreadyExistsException; import com.couchbase.client.java.query.*; import com.couchbase.client.java.view.*; import org.slf4j.Logger; import org.slf4j.LoggerFactory; +import org.springframework.dao.OptimisticLockingFailureException; import org.springframework.data.couchbase.core.convert.CouchbaseConverter; import org.springframework.data.couchbase.core.convert.MappingCouchbaseConverter; import org.springframework.data.couchbase.core.convert.translation.JacksonTranslationService; @@ -44,39 +46,43 @@ import org.springframework.data.mapping.PersistentPropertyAccessor; import org.springframework.data.mapping.context.MappingContext; import org.springframework.data.mapping.model.ConvertingPropertyAccessor; import rx.Observable; +import rx.functions.Action4; +import rx.functions.Func3; +import rx.functions.Func4; /** * RxJavaCouchbaseTemplate implements operations using rxjava1 observables * @author Subhashni Balakrishnan * @author Mark Paluch + * @author Alex Derkach * @since 3.0 */ public class RxJavaCouchbaseTemplate implements RxJavaCouchbaseOperations { private static final Logger LOGGER = LoggerFactory.getLogger(RxJavaCouchbaseTemplate.class); + private static final WriteResultChecking DEFAULT_WRITE_RESULT_CHECKING = WriteResultChecking.NONE; + + protected final MappingContext, CouchbasePersistentProperty> mappingContext; + private Bucket syncClient; private AsyncBucket client; private final ClusterInfo clusterInfo; private final CouchbaseConverter converter; private final TranslationService translationService; - protected final MappingContext, CouchbasePersistentProperty> mappingContext; private Consistency configuredConsistency = Consistency.DEFAULT_CONSISTENCY; - - private static final WriteResultChecking DEFAULT_WRITE_RESULT_CHECKING = WriteResultChecking.NONE; private WriteResultChecking writeResultChecking = DEFAULT_WRITE_RESULT_CHECKING; - public Observable save(T objectToSave) { - return doPersist(objectToSave, PersistTo.NONE, ReplicateTo.NONE); + return save(objectToSave, PersistTo.NONE, ReplicateTo.NONE); } - public Observable save(Iterable batchToSave) { + public Observable save(Iterable batchToSave) { return Observable.from(batchToSave) - .flatMap(object -> save(object)); + .flatMap(this::save); } public Observable save(T objectToSave, PersistTo persistTo, ReplicateTo replicateTo) { - return doPersist(objectToSave, persistTo, replicateTo); + return doPersist(objectToSave, PersistType.SAVE, persistTo, replicateTo); } public Observable save(Iterable batchToSave, PersistTo persistTo, ReplicateTo replicateTo) { @@ -84,13 +90,57 @@ public class RxJavaCouchbaseTemplate implements RxJavaCouchbaseOperations { .flatMap(object -> save(object, persistTo, replicateTo)); } + @Override + public Observable insert(T objectToSave) { + return insert(objectToSave, PersistTo.NONE, ReplicateTo.NONE); + } + + @Override + public Observable insert(Iterable batchToSave) { + return Observable.from(batchToSave) + .flatMap(this::insert); + } + + @Override + public Observable insert(T objectToSave, PersistTo persistTo, ReplicateTo replicateTo) { + return doPersist(objectToSave, PersistType.INSERT, persistTo, replicateTo); + } + + @Override + public Observable insert(Iterable batchToSave, PersistTo persistTo, ReplicateTo replicateTo) { + return Observable.from(batchToSave) + .flatMap(objectToSave -> insert(objectToSave, persistTo, replicateTo)); + } + + @Override + public Observable update(T objectToSave) { + return update(objectToSave, PersistTo.NONE, ReplicateTo.NONE); + } + + @Override + public Observable update(Iterable batchToSave) { + return Observable.from(batchToSave) + .flatMap(this::update); + } + + @Override + public Observable update(T objectToSave, PersistTo persistTo, ReplicateTo replicateTo) { + return doPersist(objectToSave, PersistType.UPDATE, persistTo, replicateTo); + } + + @Override + public Observable update(Iterable batchToSave, PersistTo persistTo, ReplicateTo replicateTo) { + return Observable.from(batchToSave) + .flatMap(objectToSave -> update(objectToSave, persistTo, replicateTo)); + } + public Observable remove(T objectToRemove) { return doRemove(objectToRemove, PersistTo.NONE, ReplicateTo.NONE); } public Observable remove(Iterable batchToRemove) { return Observable.from(batchToRemove) - .flatMap(object -> remove(object)); + .flatMap(this::remove); } public Observable remove(T objectToRemove, PersistTo persistTo, ReplicateTo replicateTo) { @@ -102,7 +152,6 @@ public class RxJavaCouchbaseTemplate implements RxJavaCouchbaseOperations { .flatMap(object -> remove(object, persistTo, replicateTo)); } - public RxJavaCouchbaseTemplate(final ClusterInfo clusterInfo, final Bucket client) { this(clusterInfo, client, null, null); } @@ -117,8 +166,8 @@ public class RxJavaCouchbaseTemplate implements RxJavaCouchbaseOperations { } public RxJavaCouchbaseTemplate(final ClusterInfo clusterInfo, final Bucket client, - final CouchbaseConverter converter, - final TranslationService translationService) { + final CouchbaseConverter converter, + final TranslationService translationService) { this.syncClient = client; this.clusterInfo = clusterInfo; this.client = client.async(); @@ -157,38 +206,82 @@ public class RxJavaCouchbaseTemplate implements RxJavaCouchbaseOperations { return new ConvertingPropertyAccessor(accessor, converter.getConversionService()); } + private Observable doPersist(T objectToPersist, PersistType persistType, PersistTo persistTo, ReplicateTo replicateTo) { + // If version is not set - assumption that document is new, otherwise updating + Optional version = getVersion(objectToPersist); + Func3> persistFunction; + switch (persistType) { + case SAVE: + if (!version.isPresent()) { + //No version field - no cas + persistFunction = client::upsert; + } else if (version.get() > 0) { + //Updating existing document with cas + persistFunction = client::replace; + } else { + //Creating new document + persistFunction = client::insert; + } + break; + case UPDATE: + persistFunction = client::replace; + break; + case INSERT: + default: + persistFunction = client::insert; + break; + } + return persistFunction.call(toJsonDocument(objectToPersist), persistTo, replicateTo) + .flatMap(storedDoc -> { + if (storedDoc != null && storedDoc.cas() != 0) { + setVersion(objectToPersist, storedDoc.cas()); + } + return Observable.just(objectToPersist); + }) + .onErrorResumeNext(e -> { + if (e instanceof DocumentAlreadyExistsException) { + throw new OptimisticLockingFailureException(persistType.springDataOperationName + + " document with version value failed: " + version.orElse(null), e); + } + if (e instanceof CASMismatchException) { + throw new OptimisticLockingFailureException(persistType.springDataOperationName + + " document with version value failed: " + version.orElse(null), e); + } + return TemplateUtils.translateError(e); + }); + } - private Observable doPersist(T objectToPersist, final PersistTo persistTo, final ReplicateTo replicateTo) { - ensureNotIterable(objectToPersist); - - final ConvertingPropertyAccessor accessor = getPropertyAccessor(objectToPersist); - final CouchbasePersistentEntity persistentEntity = mappingContext.getRequiredPersistentEntity(objectToPersist.getClass()); - Optional versionProperty = persistentEntity.getVersionProperty(); - final Long version = versionProperty.flatMap(p -> accessor.getProperty(p, Long.class)).orElse(null); + private RawJsonDocument toJsonDocument(T object) { + ensureNotIterable(object); final CouchbaseDocument converted = new CouchbaseDocument(); - converter.write(objectToPersist, converted); - RawJsonDocument doc = encodeAndWrap(converted, version); - return client.upsert(doc, persistTo, replicateTo) - .flatMap(rawJsonDocument -> Observable.just(objectToPersist)) - .doOnError(e -> TemplateUtils.translateError(e)); + converter.write(object, converted); + return encodeAndWrap(converted, getVersion(object).orElse(null)); + } + + private Optional versionProperty(T object) { + final CouchbasePersistentEntity persistentEntity = mappingContext.getRequiredPersistentEntity(object.getClass()); + return persistentEntity.getVersionProperty(); + } + + private Optional getVersion(T object) { + final ConvertingPropertyAccessor accessor = getPropertyAccessor(object); + Optional versionProperty = versionProperty(object); + return versionProperty.flatMap(p -> accessor.getProperty(p, Long.class)); + } + + private void setVersion(T object, long cas) { + final ConvertingPropertyAccessor accessor = getPropertyAccessor(object); + versionProperty(object).ifPresent(p -> accessor.setProperty(p, Optional.ofNullable(cas))); } private Observable doRemove(T objectToRemove, final PersistTo persistTo, final ReplicateTo replicateTo) { - ensureNotIterable(objectToRemove); if(objectToRemove instanceof String) { return client.remove((String) objectToRemove, persistTo, replicateTo) .flatMap(rawJsonDocument -> Observable.just(objectToRemove)) .doOnError(e -> TemplateUtils.translateError(e)); } else { - final ConvertingPropertyAccessor accessor = getPropertyAccessor(objectToRemove); - final CouchbasePersistentEntity persistentEntity = mappingContext.getRequiredPersistentEntity(objectToRemove.getClass()); - final Optional versionProperty = persistentEntity.getVersionProperty(); - final Long version = versionProperty.flatMap(p -> accessor.getProperty(p, Long.class)).orElse(null); - - final CouchbaseDocument converted = new CouchbaseDocument(); - converter.write(objectToRemove, converted); - RawJsonDocument doc = encodeAndWrap(converted, version); + RawJsonDocument doc = toJsonDocument(objectToRemove); return client.remove(doc, persistTo, replicateTo) .flatMap(rawJsonDocument -> Observable.just(objectToRemove)) .doOnError(e -> TemplateUtils.translateError(e)); @@ -363,4 +456,19 @@ public class RxJavaCouchbaseTemplate implements RxJavaCouchbaseOperations { return this.clusterInfo; } + private enum PersistType { + SAVE("Save", "Upsert"), + INSERT("Insert", "Insert"), + UPDATE("Update", "Replace"); + + private final String sdkOperationName; + private final String springDataOperationName; + + PersistType(String sdkOperationName, String springDataOperationName) { + this.sdkOperationName = sdkOperationName; + this.springDataOperationName = springDataOperationName; + } + + } + } diff --git a/src/test/java/org/springframework/data/couchbase/core/ReactiveBeer.java b/src/test/java/org/springframework/data/couchbase/core/ReactiveBeer.java index 1b0c98ba..5c793782 100644 --- a/src/test/java/org/springframework/data/couchbase/core/ReactiveBeer.java +++ b/src/test/java/org/springframework/data/couchbase/core/ReactiveBeer.java @@ -16,16 +16,19 @@ package org.springframework.data.couchbase.core; +import lombok.EqualsAndHashCode; import org.springframework.data.annotation.Id; import com.couchbase.client.java.repository.annotation.Field; /** - * Test class for persisting and loading from {@link ReactiveCouchbaseTemplate}. + * Test class for persisting and loading from {@link RxJavaCouchbaseTemplate}. * * @author Subhashni Balakrishnan + * @author Alex Derkach */ +@EqualsAndHashCode public class ReactiveBeer { @Id diff --git a/src/test/java/org/springframework/data/couchbase/core/VersionedReactiveBeer.java b/src/test/java/org/springframework/data/couchbase/core/VersionedReactiveBeer.java new file mode 100644 index 00000000..d89f1dfd --- /dev/null +++ b/src/test/java/org/springframework/data/couchbase/core/VersionedReactiveBeer.java @@ -0,0 +1,97 @@ +/* + * Copyright 2017 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 + * + * http://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 com.couchbase.client.java.repository.annotation.Field; +import lombok.EqualsAndHashCode; +import org.springframework.data.annotation.Id; +import org.springframework.data.annotation.Version; + + +/** + * Test class for persisting and loading from {@link RxJavaCouchbaseTemplate}. + * + * @author Alex Derkach + */ +@EqualsAndHashCode +public class VersionedReactiveBeer { + + @Id + private final String id; + + private String name; + + @Field("is_active") + private boolean active = true; + + @Field("desc") + private String description; + + @Version + private long version; + + public VersionedReactiveBeer(String id, String name, Boolean active, String description) { + this.id = id; + this.name = name; + this.active = active; + this.description = description; + } + + @Override + public String toString() { + return "Beer [id=" + id + ", name=" + name + ", active=" + active + ", description=" + description + "]"; + } + + public VersionedReactiveBeer setName(String name) { + this.name = name; + return this; + } + + public String getName() { + return name; + } + + public VersionedReactiveBeer setActive(boolean active) { + this.active = active; + return this; + } + + public boolean getActive() { + return active; + } + + public VersionedReactiveBeer setDescription(String description) { + this.description = description; + return this; + } + + public String getDescription() { + return description; + } + + public String getId() { + return id; + } + + public void setVersion(long version) { + this.version = version; + } + + public long getVersion() { + return version; + } +}