Skip to content

Commit 39233d9

Browse files
mikereichemikereiche
committed
Merge branch 'main' into datacouch_963_scopes_and_collections_for_repositories
2 parents 161d70c + 831c25c commit 39233d9

File tree

44 files changed

+548
-177
lines changed

Some content is hidden

Large Commits have some content hidden by default. Use the searchbox below for content that may be hidden.

44 files changed

+548
-177
lines changed

‎pom.xml

Lines changed: 1 addition & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -81,7 +81,7 @@
8181
<dependency>
8282
<groupId>org.apache.httpcomponents</groupId>
8383
<artifactId>httpclient</artifactId>
84-
<version>4.3.6</version>
84+
<version>4.5.13</version>
8585
<scope>test</scope>
8686
</dependency>
8787

‎src/main/java/org/springframework/data/couchbase/config/AbstractCouchbaseConfiguration.java

Lines changed: 2 additions & 6 deletions
Original file line numberDiff line numberDiff line change
@@ -1,5 +1,5 @@
11
/*
2-
* Copyright 2012-2020 the original author or authors
2+
* Copyright 2012-2021 the original author or authors
33
*
44
* Licensed under the Apache License, Version 2.0 (the "License");
55
* you may not use this file except in compliance with the License.
@@ -49,7 +49,6 @@
4949
import org.springframework.util.StringUtils;
5050

5151
import com.couchbase.client.core.deps.com.fasterxml.jackson.databind.DeserializationFeature;
52-
import com.couchbase.client.core.deps.com.fasterxml.jackson.databind.Module;
5352
import com.couchbase.client.core.encryption.CryptoManager;
5453
import com.couchbase.client.core.env.Authenticator;
5554
import com.couchbase.client.core.env.PasswordAuthenticator;
@@ -60,7 +59,6 @@
6059
import com.couchbase.client.java.env.ClusterEnvironment;
6160
import com.couchbase.client.java.json.JacksonTransformers;
6261
import com.couchbase.client.java.json.JsonValueModule;
63-
import com.couchbase.client.java.json.RepackagedJsonValueModule;
6462
import com.fasterxml.jackson.databind.ObjectMapper;
6563

6664
/**
@@ -75,8 +73,6 @@
7573
@Configuration
7674
public abstract class AbstractCouchbaseConfiguration {
7775

78-
@Autowired ObjectMapper couchbaseObjectMapper;
79-
8076
/**
8177
* The connection string which allows the SDK to connect to the cluster.
8278
* <p>
@@ -144,7 +140,7 @@ public ClusterEnvironment couchbaseClusterEnvironment() {
144140
if (!nonShadowedJacksonPresent()) {
145141
throw new CouchbaseException("non-shadowed Jackson not present");
146142
}
147-
builder.jsonSerializer(JacksonJsonSerializer.create(couchbaseObjectMapper));
143+
builder.jsonSerializer(JacksonJsonSerializer.create(couchbaseObjectMapper()));
148144
configureEnvironment(builder);
149145
return builder.build();
150146
}

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

Lines changed: 2 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -106,6 +106,7 @@ public <T> ExecutableFindByAnalytics<T> findByAnalytics(Class<T> domainType) {
106106
}
107107

108108
@Override
109+
@Deprecated
109110
public ExecutableRemoveById removeById() {
110111
return removeById(null);
111112
}
@@ -116,6 +117,7 @@ public ExecutableRemoveById removeById(Class<?> domainType) {
116117
}
117118

118119
@Override
120+
@Deprecated
119121
public ExecutableExistsById existsById() {
120122
return existsById(null);
121123
}

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

Lines changed: 10 additions & 4 deletions
Original file line numberDiff line numberDiff line change
@@ -26,6 +26,7 @@
2626
import org.springframework.data.couchbase.core.mapping.CouchbasePersistentEntity;
2727
import org.springframework.data.couchbase.core.mapping.CouchbasePersistentProperty;
2828
import org.springframework.data.couchbase.core.mapping.event.AfterConvertCallback;
29+
import org.springframework.data.couchbase.core.mapping.event.AfterSaveEvent;
2930
import org.springframework.data.couchbase.core.mapping.event.BeforeConvertCallback;
3031
import org.springframework.data.couchbase.core.mapping.event.BeforeConvertEvent;
3132
import org.springframework.data.couchbase.core.mapping.event.BeforeSaveEvent;
@@ -95,16 +96,21 @@ public <T> T decodeEntity(String id, String source, long cas, Class<T> entityCla
9596
}
9697

9798
@Override
98-
public Object applyUpdatedCas(final Object entity, final long cas) {
99+
public Object applyUpdatedCas(final Object entity, CouchbaseDocument converted, final long cas) {
100+
Object returnValue;
99101
final ConvertingPropertyAccessor<Object> accessor = getPropertyAccessor(entity);
100102
final CouchbasePersistentEntity<?> persistentEntity = mappingContext.getRequiredPersistentEntity(entity.getClass());
101103
final CouchbasePersistentProperty versionProperty = persistentEntity.getVersionProperty();
102104

103105
if (versionProperty != null) {
104106
accessor.setProperty(versionProperty, cas);
105-
return accessor.getBean();
107+
returnValue = accessor.getBean();
108+
} else {
109+
returnValue = entity;
106110
}
107-
return entity;
111+
maybeEmitEvent(new AfterSaveEvent(returnValue, converted));
112+
113+
return returnValue;
108114
}
109115

110116
@Override
@@ -172,7 +178,7 @@ public void setEntityCallbacks(EntityCallbacks entityCallbacks) {
172178
this.entityCallbacks = entityCallbacks;
173179
}
174180

175-
void maybeEmitEvent(CouchbaseMappingEvent<?> event) {
181+
public void maybeEmitEvent(CouchbaseMappingEvent<?> event) {
176182
if (canPublishEvent()) {
177183
try {
178184
this.applicationContext.publishEvent(event);

‎src/main/java/org/springframework/data/couchbase/core/ExecutableExistsByIdOperationSupport.java

Lines changed: 1 addition & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -32,6 +32,7 @@ public class ExecutableExistsByIdOperationSupport implements ExecutableExistsByI
3232
}
3333

3434
@Override
35+
@Deprecated
3536
public ExecutableExistsById existsById() {
3637
return existsById(null);
3738
}

‎src/main/java/org/springframework/data/couchbase/core/ExecutableRemoveByIdOperationSupport.java

Lines changed: 1 addition & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -35,6 +35,7 @@ public ExecutableRemoveByIdOperationSupport(final CouchbaseTemplate template) {
3535
}
3636

3737
@Override
38+
@Deprecated
3839
public ExecutableRemoveById removeById() {
3940
return removeById(null);
4041
}

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

Lines changed: 8 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -17,6 +17,7 @@
1717

1818
import org.springframework.data.couchbase.core.mapping.CouchbaseDocument;
1919

20+
import org.springframework.data.couchbase.core.mapping.event.CouchbaseMappingEvent;
2021
import reactor.core.publisher.Mono;
2122

2223
/**
@@ -44,8 +45,8 @@ public <T> Mono<T> decodeEntity(String id, String source, long cas, Class<T> ent
4445
}
4546

4647
@Override
47-
public Mono<Object> applyUpdatedCas(Object entity, long cas) {
48-
return Mono.fromSupplier(() -> support.applyUpdatedCas(entity, cas));
48+
public Mono<Object> applyUpdatedCas(Object entity, CouchbaseDocument converted, long cas) {
49+
return Mono.fromSupplier(() -> support.applyUpdatedCas(entity, converted, cas));
4950
}
5051

5152
@Override
@@ -62,4 +63,9 @@ public Long getCas(Object entity) {
6263
public String getJavaNameForEntity(Class<?> clazz) {
6364
return support.getJavaNameForEntity(clazz);
6465
}
66+
67+
@Override
68+
public void maybeEmitEvent(CouchbaseMappingEvent<?> event) {
69+
support.maybeEmitEvent(event);
70+
}
6571
}

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

Lines changed: 1 addition & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -94,7 +94,7 @@ public <T> ReactiveInsertById<T> insertById(Class<T> domainType) {
9494

9595
@Override
9696
public ReactiveRemoveById removeById() {
97-
return new ReactiveRemoveByIdOperationSupport(this).removeById(null);
97+
return removeById(null);
9898
}
9999

100100
@Override

‎src/main/java/org/springframework/data/couchbase/core/ReactiveCouchbaseTemplateSupport.java

Lines changed: 22 additions & 18 deletions
Original file line numberDiff line numberDiff line change
@@ -16,6 +16,8 @@
1616

1717
package org.springframework.data.couchbase.core;
1818

19+
import reactor.core.publisher.Mono;
20+
1921
import org.slf4j.Logger;
2022
import org.slf4j.LoggerFactory;
2123
import org.springframework.beans.BeansException;
@@ -26,6 +28,7 @@
2628
import org.springframework.data.couchbase.core.mapping.CouchbaseDocument;
2729
import org.springframework.data.couchbase.core.mapping.CouchbasePersistentEntity;
2830
import org.springframework.data.couchbase.core.mapping.CouchbasePersistentProperty;
31+
import org.springframework.data.couchbase.core.mapping.event.AfterSaveEvent;
2932
import org.springframework.data.couchbase.core.mapping.event.BeforeConvertEvent;
3033
import org.springframework.data.couchbase.core.mapping.event.BeforeSaveEvent;
3134
import org.springframework.data.couchbase.core.mapping.event.CouchbaseMappingEvent;
@@ -38,7 +41,6 @@
3841
import org.springframework.data.mapping.context.MappingContext;
3942
import org.springframework.data.mapping.model.ConvertingPropertyAccessor;
4043
import org.springframework.util.Assert;
41-
import reactor.core.publisher.Mono;
4244

4345
/**
4446
* Internal encode/decode support for {@link ReactiveCouchbaseTemplate}.
@@ -65,15 +67,12 @@ public ReactiveCouchbaseTemplateSupport(final CouchbaseConverter converter,
6567

6668
@Override
6769
public Mono<CouchbaseDocument> encodeEntity(final Object entityToEncode) {
68-
return Mono.just(entityToEncode)
69-
.doOnNext(entity -> maybeEmitEvent(new BeforeConvertEvent<>(entity)))
70-
.flatMap(entity -> maybeCallBeforeConvert(entity, ""))
71-
.map(maybeNewEntity -> {
70+
return Mono.just(entityToEncode).doOnNext(entity -> maybeEmitEvent(new BeforeConvertEvent<>(entity)))
71+
.flatMap(entity -> maybeCallBeforeConvert(entity, "")).map(maybeNewEntity -> {
7272
final CouchbaseDocument converted = new CouchbaseDocument();
7373
converter.write(maybeNewEntity, converted);
7474
return converted;
75-
})
76-
.flatMap(converted -> maybeCallAfterConvert(entityToEncode, converted, "").thenReturn(converted))
75+
}).flatMap(converted -> maybeCallAfterConvert(entityToEncode, converted, "").thenReturn(converted))
7776
.doOnNext(converted -> maybeEmitEvent(new BeforeSaveEvent<>(entityToEncode, converted)));
7877
}
7978

@@ -98,25 +97,31 @@ public <T> Mono<T> decodeEntity(String id, String source, long cas, Class<T> ent
9897
}
9998

10099
@Override
101-
public Mono<Object> applyUpdatedCas(final Object entity, final long cas) {
100+
public Mono<Object> applyUpdatedCas(final Object entity, CouchbaseDocument converted, final long cas) {
102101
return Mono.fromSupplier(() -> {
102+
Object returnValue;
103103
final ConvertingPropertyAccessor<Object> accessor = getPropertyAccessor(entity);
104-
final CouchbasePersistentEntity<?> persistentEntity = mappingContext.getRequiredPersistentEntity(entity.getClass());
104+
final CouchbasePersistentEntity<?> persistentEntity = mappingContext
105+
.getRequiredPersistentEntity(entity.getClass());
105106
final CouchbasePersistentProperty versionProperty = persistentEntity.getVersionProperty();
106107

107108
if (versionProperty != null) {
108109
accessor.setProperty(versionProperty, cas);
109-
return accessor.getBean();
110+
returnValue = accessor.getBean();
111+
} else {
112+
returnValue = entity;
110113
}
111-
return entity;
114+
maybeEmitEvent(new AfterSaveEvent(returnValue, converted));
115+
return returnValue;
112116
});
113117
}
114118

115119
@Override
116120
public Mono<Object> applyUpdatedId(final Object entity, Object id) {
117121
return Mono.fromSupplier(() -> {
118122
final ConvertingPropertyAccessor<Object> accessor = getPropertyAccessor(entity);
119-
final CouchbasePersistentEntity<?> persistentEntity = mappingContext.getRequiredPersistentEntity(entity.getClass());
123+
final CouchbasePersistentEntity<?> persistentEntity = mappingContext
124+
.getRequiredPersistentEntity(entity.getClass());
120125
final CouchbasePersistentProperty idProperty = persistentEntity.getIdProperty();
121126

122127
if (idProperty != null) {
@@ -130,8 +135,7 @@ public Mono<Object> applyUpdatedId(final Object entity, Object id) {
130135
@Override
131136
public Long getCas(final Object entity) {
132137
final ConvertingPropertyAccessor<Object> accessor = getPropertyAccessor(entity);
133-
final CouchbasePersistentEntity<?> persistentEntity = mappingContext
134-
.getRequiredPersistentEntity(entity.getClass());
138+
final CouchbasePersistentEntity<?> persistentEntity = mappingContext.getRequiredPersistentEntity(entity.getClass());
135139
final CouchbasePersistentProperty versionProperty = persistentEntity.getVersionProperty();
136140

137141
long cas = 0;
@@ -166,9 +170,9 @@ public void setApplicationContext(ApplicationContext applicationContext) throws
166170
}
167171

168172
/**
169-
* Set the {@link ReactiveEntityCallbacks} instance to use when invoking {@link
170-
* org.springframework.data.mapping.callback.ReactiveEntityCallbacks callbacks} like the {@link
171-
* ReactiveBeforeConvertCallback}.
173+
* Set the {@link ReactiveEntityCallbacks} instance to use when invoking
174+
* {@link org.springframework.data.mapping.callback.ReactiveEntityCallbacks callbacks} like the
175+
* {@link ReactiveBeforeConvertCallback}.
172176
* <p/>
173177
* Overrides potentially existing {@link EntityCallbacks}.
174178
*
@@ -180,7 +184,7 @@ public void setReactiveEntityCallbacks(ReactiveEntityCallbacks reactiveEntityCal
180184
this.reactiveEntityCallbacks = reactiveEntityCallbacks;
181185
}
182186

183-
void maybeEmitEvent(CouchbaseMappingEvent<?> event) {
187+
public void maybeEmitEvent(CouchbaseMappingEvent<?> event) {
184188
if (canPublishEvent()) {
185189
try {
186190
this.applicationContext.publishEvent(event);

‎src/main/java/org/springframework/data/couchbase/core/ReactiveExistsByIdOperationSupport.java

Lines changed: 2 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -41,6 +41,7 @@ public class ReactiveExistsByIdOperationSupport implements ReactiveExistsByIdOpe
4141
}
4242

4343
@Override
44+
@Deprecated
4445
public ReactiveExistsById existsById() {
4546
return existsById(null);
4647
}
@@ -70,7 +71,7 @@ static class ReactiveExistsByIdSupport implements ReactiveExistsById {
7071
@Override
7172
public Mono<Boolean> one(final String id) {
7273
PseudoArgs<ExistsOptions> pArgs = new PseudoArgs<>(template, scope, collection,
73-
options != null ? options : ExistsOptions.existsOptions(), null);
74+
options != null ? options : ExistsOptions.existsOptions(), domainType);
7475
LOG.trace("statement: {} scope: {} collection: {}", "exitsById", pArgs.getScope(), pArgs.getCollection());
7576
return Mono.just(id)
7677
.flatMap(docId -> template.getCouchbaseClientFactory().withScope(pArgs.getScope())

‎src/main/java/org/springframework/data/couchbase/core/ReactiveFindByIdOperationSupport.java

Lines changed: 1 addition & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -79,7 +79,7 @@ public Mono<T> one(final String id) {
7979
if (fields != null && !fields.isEmpty()) {
8080
gOptions.project(fields);
8181
}
82-
PseudoArgs<GetOptions> pArgs = new PseudoArgs(template, scope, collection, gOptions, null);
82+
PseudoArgs<GetOptions> pArgs = new PseudoArgs(template, scope, collection, gOptions, domainType);
8383
LOG.trace("statement: {} scope: {} collection: {}", "findById", pArgs.getScope(), pArgs.getCollection());
8484
return template.getCouchbaseClientFactory().withScope(pArgs.getScope()).getCollection(pArgs.getCollection())
8585
.reactive().get(docId, pArgs.getOptions());

‎src/main/java/org/springframework/data/couchbase/core/ReactiveFindByQueryOperationSupport.java

Lines changed: 2 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -157,7 +157,8 @@ public Mono<T> first() {
157157
@Override
158158
public Flux<T> all() {
159159
return Flux.defer(() -> {
160-
PseudoArgs<QueryOptions> pArgs = new PseudoArgs(template, scope, collection, options, domainType);
160+
PseudoArgs<QueryOptions> pArgs = new PseudoArgs(template, scope, collection,
161+
options != null ? options : QueryOptions.queryOptions(), domainType);
161162
String statement = assembleEntityQuery(false, distinctFields, pArgs.getCollection());
162163
LOG.trace("statement: {} {}", "findByQuery", statement);
163164
Mono<ReactiveQueryResult> allResult = pArgs.getScope() == null

‎src/main/java/org/springframework/data/couchbase/core/ReactiveFindFromReplicasByIdOperationSupport.java

Lines changed: 1 addition & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -73,7 +73,7 @@ public Mono<T> any(final String id) {
7373
if (garOptions.build().transcoder() == null) {
7474
garOptions.transcoder(RawJsonTranscoder.INSTANCE);
7575
}
76-
PseudoArgs<GetAnyReplicaOptions> pArgs = new PseudoArgs<>(template, scope, collection, garOptions, null);
76+
PseudoArgs<GetAnyReplicaOptions> pArgs = new PseudoArgs<>(template, scope, collection, garOptions, domainType);
7777
LOG.trace("statement: {} scope: {} collection: {}", "getAnyReplica", pArgs.getScope(), pArgs.getCollection());
7878
return template.getCouchbaseClientFactory().withScope(pArgs.getScope()).getCollection(pArgs.getCollection())
7979
.reactive().getAnyReplica(docId, pArgs.getOptions());

‎src/main/java/org/springframework/data/couchbase/core/ReactiveInsertByIdOperationSupport.java

Lines changed: 1 addition & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -86,7 +86,7 @@ public Mono<T> one(T object) {
8686
.getCollection(pArgs.getCollection()).reactive()
8787
.insert(converted.getId(), converted.export(), buildOptions(pArgs.getOptions(), converted))
8888
.flatMap(result -> support.applyUpdatedId(object, converted.getId())
89-
.flatMap(insertedObject -> (Mono<T>) support.applyUpdatedCas(insertedObject, result.cas()))))
89+
.flatMap(updatedObject -> support.applyUpdatedCas(updatedObject, converted, result.cas()))))
9090
.onErrorMap(throwable -> {
9191
if (throwable instanceof RuntimeException) {
9292
return template.potentiallyConvertRuntimeException((RuntimeException) throwable);

‎src/main/java/org/springframework/data/couchbase/core/ReactiveRemoveByQueryOperationSupport.java

Lines changed: 2 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -72,7 +72,8 @@ static class ReactiveRemoveByQuerySupport<T> implements ReactiveRemoveByQuery<T>
7272
@Override
7373
public Flux<RemoveResult> all() {
7474
return Flux.defer(() -> {
75-
PseudoArgs<QueryOptions> pArgs = new PseudoArgs<>(template, scope, collection, options, domainType);
75+
PseudoArgs<QueryOptions> pArgs = new PseudoArgs<>(template, scope, collection,
76+
options != null ? options : QueryOptions.queryOptions(), domainType);
7677
String statement = assembleDeleteQuery(pArgs.getCollection());
7778
LOG.trace("statement: {}", statement);
7879
Mono<ReactiveQueryResult> allResult = pArgs.getScope() == null

‎src/main/java/org/springframework/data/couchbase/core/ReactiveReplaceByIdOperationSupport.java

Lines changed: 6 additions & 5 deletions
Original file line numberDiff line numberDiff line change
@@ -80,11 +80,12 @@ static class ReactiveReplaceByIdSupport<T> implements ReactiveReplaceById<T> {
8080
public Mono<T> one(T object) {
8181
PseudoArgs<ReplaceOptions> pArgs = new PseudoArgs<>(template, scope, collection, options, domainType);
8282
LOG.trace("statement: {} pArgs: {}", "replaceById", pArgs);
83-
return Mono.just(object).flatMap(support::encodeEntity).flatMap(converted -> template.getCouchbaseClientFactory()
84-
.withScope(pArgs.getScope()).getCollection(pArgs.getCollection()).reactive()
85-
.replace(converted.getId(), converted.export(), buildReplaceOptions(pArgs.getOptions(), object, converted))
86-
.flatMap(result -> support.applyUpdatedId(object, converted.getId())
87-
.flatMap(replacedObject -> (Mono<T>) support.applyUpdatedCas(replacedObject, result.cas()))))
83+
return Mono.just(object).flatMap(support::encodeEntity)
84+
.flatMap(converted -> template.getCouchbaseClientFactory().withScope(pArgs.getScope())
85+
.getCollection(pArgs.getCollection()).reactive()
86+
.replace(converted.getId(), converted.export(),
87+
buildReplaceOptions(pArgs.getOptions(), object, converted))
88+
.flatMap(result -> support.applyUpdatedCas(object, converted, result.cas())))
8889
.onErrorMap(throwable -> {
8990
if (throwable instanceof RuntimeException) {
9091
return template.potentiallyConvertRuntimeException((RuntimeException) throwable);

‎src/main/java/org/springframework/data/couchbase/core/ReactiveTemplateSupport.java

Lines changed: 5 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -17,6 +17,7 @@
1717

1818
import org.springframework.data.couchbase.core.mapping.CouchbaseDocument;
1919

20+
import org.springframework.data.couchbase.core.mapping.event.CouchbaseMappingEvent;
2021
import reactor.core.publisher.Mono;
2122

2223
public interface ReactiveTemplateSupport {
@@ -25,11 +26,13 @@ public interface ReactiveTemplateSupport {
2526

2627
<T> Mono<T> decodeEntity(String id, String source, long cas, Class<T> entityClass);
2728

28-
Mono<Object> applyUpdatedCas(Object entity, long cas);
29+
<T> Mono<T> applyUpdatedCas(T entity, CouchbaseDocument converted, long cas);
2930

30-
Mono<Object> applyUpdatedId(Object entity, Object id);
31+
<T> Mono<T> applyUpdatedId(T entity, Object id);
3132

3233
Long getCas(Object entity);
3334

3435
String getJavaNameForEntity(Class<?> clazz);
36+
37+
void maybeEmitEvent(CouchbaseMappingEvent<?> event);
3538
}

‎src/main/java/org/springframework/data/couchbase/core/ReactiveUpsertByIdOperationSupport.java

Lines changed: 4 additions & 4 deletions
Original file line numberDiff line numberDiff line change
@@ -77,11 +77,11 @@ static class ReactiveUpsertByIdSupport<T> implements ReactiveUpsertById<T> {
7777
public Mono<T> one(T object) {
7878
PseudoArgs<UpsertOptions> pArgs = new PseudoArgs<>(template, scope, collection, options, domainType);
7979
return Mono.just(object).flatMap(support::encodeEntity)
80-
.flatMap(converted -> template.getCouchbaseClientFactory().withScope(pArgs.getScope())
81-
.getCollection(pArgs.getCollection()).reactive()
82-
.upsert(converted.getId(), converted.export(), buildUpsertOptions(pArgs.getOptions(), converted))
80+
.flatMap(converted -> template.getCouchbaseClientFactory().withScope(pArgs.getScope())
81+
.getCollection(pArgs.getCollection()).reactive()
82+
.upsert(converted.getId(), converted.export(), buildUpsertOptions(pArgs.getOptions(), converted))
8383
.flatMap(result -> support.applyUpdatedId(object, converted.getId())
84-
.flatMap(updatedObject -> (Mono<T>) support.applyUpdatedCas(updatedObject, result.cas()))))
84+
.flatMap(updatedObject -> support.applyUpdatedCas(updatedObject, converted, result.cas()))))
8585
.onErrorMap(throwable -> {
8686
if (throwable instanceof RuntimeException) {
8787
return template.potentiallyConvertRuntimeException((RuntimeException) throwable);

0 commit comments

Comments
 (0)