Spring/spring-data-couchbase
1
0
Fork 0
Code Issues Packages Projects Releases Wiki Activity

DATACOUCH-287 - Add CAS operations to RxJavaCouchbaseOperations.

Browse Source 
Original pull request: #135.
This commit is contained in:
Alexander Derkach
2017-03-28 18:38:02 +03:00
committed by Subhashni Balakrishnan
parent 49045b031c
commit 6429dca5e6
6 changed files with 488 additions and 89 deletions
Download Patch File Download Diff File Expand all files Collapse all files

35
src/integration/java/org/springframework/data/couchbase/core/AsyncUtils.java
View File

@@ -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 <T> void awaitCompleted(TestSubscriber<T> testSubscriber) {
testSubscriber.awaitTerminalEvent();
testSubscriber.assertNoErrors();
testSubscriber.assertNoValues();
testSubscriber.assertCompleted();
}
public static <T> void awaitCompletedWithAnyValue(TestSubscriber<T> testSubscriber) {
testSubscriber.awaitTerminalEvent();
testSubscriber.assertNoErrors();
testSubscriber.assertCompleted();
}
public static <T> void awaitCompletedWithValueCount(TestSubscriber<T> testSubscriber, int count) {
testSubscriber.awaitTerminalEvent();
testSubscriber.assertNoErrors();
testSubscriber.assertValueCount(count);
testSubscriber.assertCompleted();
}
public static <T> void awaitError(TestSubscriber<T> testSubscriber, Class<? extends Throwable> throwableClazz) {
testSubscriber.awaitTerminalEvent();
testSubscriber.assertError(throwableClazz);
testSubscriber.assertNoValues();
}
public static <T> void awaitValue(TestSubscriber<T> testSubscriber, T value) {
testSubscriber.awaitTerminalEvent();
testSubscriber.assertNoErrors();
testSubscriber.assertValue(value);
testSubscriber.assertCompleted();
}
}

245
src/integration/java/org/springframework/data/couchbase/core/RxJavaCouchbaseTemplateTests.java
View File

@@ -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<Object> subscriber = TestSubscriber.create();
template.remove(key)
.subscribe(subscriber);
subscriber.awaitTerminalEvent();
}
private void removeCollectionIfExist(Collection<ReactiveBeer> beers) {
TestSubscriber<Object> 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<ReactiveBeer> firstSaveSubscriber = TestSubscriber.create();
TestSubscriber<ReactiveBeer> 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<String, Object> resultConv = MAPPER.readValue(result, new TypeReference<Map<String, Object>>() {
});
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<VersionedReactiveBeer> 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<VersionedReactiveBeer> 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<ReactiveBeer> beers = new ArrayList<>();
String name = "The Awesome Stout";
for (int i=0; i < 10000; i++) {
TestSubscriber<ReactiveBeer> 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<ReactiveBeer> 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<ReactiveBeer> firstInsertSubscriber = TestSubscriber.create();
TestSubscriber<ReactiveBeer> 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<ReactiveBeer> testSubscriber = TestSubscriber.create();
Collection<ReactiveBeer> 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<VersionedReactiveBeer> firstInsertSubscriber = TestSubscriber.create();
TestSubscriber<VersionedReactiveBeer> 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<VersionedReactiveBeer> 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<ReactiveBeer> firstInsertSubscriber = TestSubscriber.create();
TestSubscriber<ReactiveBeer> firstFindSubscriber = TestSubscriber.create();
TestSubscriber<ReactiveBeer> removalTestSubscriber = TestSubscriber.create();
TestSubscriber<ReactiveBeer> 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<ReactiveBeer> saveSubscriber = TestSubscriber.create();
TestSubscriber<ReactiveBeer> 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<String, Object> resultConv = MAPPER.readValue(result, new TypeReference<Map<String, Object>>() {});
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.
*/

19
src/main/java/org/springframework/data/couchbase/core/RxJavaCouchbaseOperations.java
View File

@@ -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 {
<T>Observable<T> save(Iterable<T> batchToSave, PersistTo persistTo, ReplicateTo replicateTo);
<T>Observable<T> insert(T objectToSave);
<T>Observable<T> insert(Iterable<T> batchToSave);
<T>Observable<T> insert(T objectToSave, PersistTo persistTo, ReplicateTo replicateTo);
<T>Observable<T> insert(Iterable<T> batchToSave, PersistTo persistTo, ReplicateTo replicateTo);
<T>Observable<T> update(T objectToSave);
<T>Observable<T> update(Iterable<T> batchToSave);
<T>Observable<T> update(T objectToSave, PersistTo persistTo, ReplicateTo replicateTo);
<T>Observable<T> update(Iterable<T> batchToSave, PersistTo persistTo, ReplicateTo replicateTo);
<T>Observable<T> remove(T objectToRemove);
<T>Observable<T> remove(T objectToRemove, PersistTo persistTo, ReplicateTo replicateTo);

176
src/main/java/org/springframework/data/couchbase/core/RxJavaCouchbaseTemplate.java
View File

@@ -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<? extends CouchbasePersistentEntity<?>, CouchbasePersistentProperty> mappingContext;
private Bucket syncClient;
private AsyncBucket client;
private final ClusterInfo clusterInfo;
private final CouchbaseConverter converter;
private final TranslationService translationService;
protected final MappingContext<? extends CouchbasePersistentEntity<?>, 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 <T> Observable<T> save(T objectToSave) {
return doPersist(objectToSave, PersistTo.NONE, ReplicateTo.NONE);
return save(objectToSave, PersistTo.NONE, ReplicateTo.NONE);
}
public <T> Observable<T> save(Iterable<T> batchToSave) {
public <T> Observable<T> save(Iterable<T> batchToSave) {
return Observable.from(batchToSave)
.flatMap(object -> save(object));
.flatMap(this::save);
}
public <T> Observable<T> save(T objectToSave, PersistTo persistTo, ReplicateTo replicateTo) {
return doPersist(objectToSave, persistTo, replicateTo);
return doPersist(objectToSave, PersistType.SAVE, persistTo, replicateTo);
}
public <T> Observable<T> save(Iterable<T> batchToSave, PersistTo persistTo, ReplicateTo replicateTo) {
@@ -84,13 +90,57 @@ public class RxJavaCouchbaseTemplate implements RxJavaCouchbaseOperations {
.flatMap(object -> save(object, persistTo, replicateTo));
}
@Override
public <T> Observable<T> insert(T objectToSave) {
return insert(objectToSave, PersistTo.NONE, ReplicateTo.NONE);
}
@Override
public <T> Observable<T> insert(Iterable<T> batchToSave) {
return Observable.from(batchToSave)
.flatMap(this::insert);
}
@Override
public <T> Observable<T> insert(T objectToSave, PersistTo persistTo, ReplicateTo replicateTo) {
return doPersist(objectToSave, PersistType.INSERT, persistTo, replicateTo);
}
@Override
public <T> Observable<T> insert(Iterable<T> batchToSave, PersistTo persistTo, ReplicateTo replicateTo) {
return Observable.from(batchToSave)
.flatMap(objectToSave -> insert(objectToSave, persistTo, replicateTo));
}
@Override
public <T> Observable<T> update(T objectToSave) {
return update(objectToSave, PersistTo.NONE, ReplicateTo.NONE);
}
@Override
public <T> Observable<T> update(Iterable<T> batchToSave) {
return Observable.from(batchToSave)
.flatMap(this::update);
}
@Override
public <T> Observable<T> update(T objectToSave, PersistTo persistTo, ReplicateTo replicateTo) {
return doPersist(objectToSave, PersistType.UPDATE, persistTo, replicateTo);
}
@Override
public <T> Observable<T> update(Iterable<T> batchToSave, PersistTo persistTo, ReplicateTo replicateTo) {
return Observable.from(batchToSave)
.flatMap(objectToSave -> update(objectToSave, persistTo, replicateTo));
}
public <T> Observable<T> remove(T objectToRemove) {
return doRemove(objectToRemove, PersistTo.NONE, ReplicateTo.NONE);
}
public <T> Observable<T> remove(Iterable<T> batchToRemove) {
return Observable.from(batchToRemove)
.flatMap(object -> remove(object));
.flatMap(this::remove);
}
public <T> Observable<T> 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 <T> Observable<T> doPersist(T objectToPersist, PersistType persistType, PersistTo persistTo, ReplicateTo replicateTo) {
// If version is not set - assumption that document is new, otherwise updating
Optional<Long> version = getVersion(objectToPersist);
Func3<RawJsonDocument, PersistTo, ReplicateTo, Observable<RawJsonDocument>> 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 <T> Observable<T> doPersist(T objectToPersist, final PersistTo persistTo, final ReplicateTo replicateTo) {
ensureNotIterable(objectToPersist);
final ConvertingPropertyAccessor accessor = getPropertyAccessor(objectToPersist);
final CouchbasePersistentEntity<?> persistentEntity = mappingContext.getRequiredPersistentEntity(objectToPersist.getClass());
Optional<CouchbasePersistentProperty> versionProperty = persistentEntity.getVersionProperty();
final Long version = versionProperty.flatMap(p -> accessor.getProperty(p, Long.class)).orElse(null);
private <T> 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 <T> Optional<CouchbasePersistentProperty> versionProperty(T object) {
final CouchbasePersistentEntity<?> persistentEntity = mappingContext.getRequiredPersistentEntity(object.getClass());
return persistentEntity.getVersionProperty();
}
private <T> Optional<Long> getVersion(T object) {
final ConvertingPropertyAccessor accessor = getPropertyAccessor(object);
Optional<CouchbasePersistentProperty> versionProperty = versionProperty(object);
return versionProperty.flatMap(p -> accessor.getProperty(p, Long.class));
}
private <T> void setVersion(T object, long cas) {
final ConvertingPropertyAccessor accessor = getPropertyAccessor(object);
versionProperty(object).ifPresent(p -> accessor.setProperty(p, Optional.ofNullable(cas)));
}
private <T> Observable<T> 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<CouchbasePersistentProperty> 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;
}
}
}

5
src/test/java/org/springframework/data/couchbase/core/ReactiveBeer.java
View File

@@ -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

97
src/test/java/org/springframework/data/couchbase/core/VersionedReactiveBeer.java Normal file
View File

@@ -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;
}
}