Skip to content

Commit 282f77e

Browse files
orange-buffaloorange-buffalo
committed
#93 - Adding support for optimistic locking based on @Version column
1 parent 7ed68ee commit 282f77e

File tree

9 files changed

+204
-13
lines changed

9 files changed

+204
-13
lines changed

‎src/main/java/org/springframework/data/r2dbc/core/DatabaseClient.java

+3-2
Original file line numberDiff line numberDiff line change
@@ -46,6 +46,7 @@
4646
* to create an instance.
4747
*
4848
* @author Mark Paluch
49+
* @author Bogdan Ilchyshyn
4950
*/
5051
public interface DatabaseClient {
5152

@@ -707,9 +708,9 @@ interface TypedUpdateSpec<T> {
707708
*
708709
* @param objectToUpdate the object of which the attributes will provide the values for the update and the primary
709710
* key. Must not be {@literal null}.
710-
* @return a {@link UpdateSpec} for further configuration of the update. Guaranteed to be not {@literal null}.
711+
* @return a {@link UpdateMatchingSpec} for further configuration of the update. Guaranteed to be not {@literal null}.
711712
*/
712-
UpdateSpec using(T objectToUpdate);
713+
UpdateMatchingSpec using(T objectToUpdate);
713714

714715
/**
715716
* Use the given {@code tableName} as update target.

‎src/main/java/org/springframework/data/r2dbc/core/DefaultDatabaseClient.java

+24-7
Original file line numberDiff line numberDiff line change
@@ -69,6 +69,7 @@
6969
* Default implementation of {@link DatabaseClient}.
7070
*
7171
* @author Mark Paluch
72+
* @author Bogdan Ilchyshyn
7273
*/
7374
class DefaultDatabaseClient implements DatabaseClient, ConnectionAccessor {
7475

@@ -1188,7 +1189,7 @@ public <T> TypedUpdateSpec<T> table(Class<T> table) {
11881189

11891190
assertRegularClass(table);
11901191

1191-
return new DefaultTypedUpdateSpec<>(table, null, null);
1192+
return new DefaultTypedUpdateSpec<>(table, null, null, null);
11921193
}
11931194
}
11941195

@@ -1262,32 +1263,43 @@ private UpdatedRowsFetchSpec exchange(SqlIdentifier table) {
12621263
}
12631264
}
12641265

1265-
class DefaultTypedUpdateSpec<T> implements TypedUpdateSpec<T>, UpdateSpec {
1266+
class DefaultTypedUpdateSpec<T> implements TypedUpdateSpec<T>, UpdateMatchingSpec {
12661267

12671268
private final @Nullable Class<T> typeToUpdate;
12681269
private final @Nullable SqlIdentifier table;
12691270
private final T objectToUpdate;
1271+
private final @Nullable Criteria where;
12701272

1271-
DefaultTypedUpdateSpec(@Nullable Class<T> typeToUpdate, @Nullable SqlIdentifier table, T objectToUpdate) {
1273+
DefaultTypedUpdateSpec(@Nullable Class<T> typeToUpdate, @Nullable SqlIdentifier table, T objectToUpdate,
1274+
@Nullable Criteria where) {
12721275
this.typeToUpdate = typeToUpdate;
12731276
this.table = table;
12741277
this.objectToUpdate = objectToUpdate;
1278+
this.where = where;
12751279
}
12761280

12771281
@Override
1278-
public UpdateSpec using(T objectToUpdate) {
1282+
public UpdateMatchingSpec using(T objectToUpdate) {
12791283

12801284
Assert.notNull(objectToUpdate, "Object to update must not be null");
12811285

1282-
return new DefaultTypedUpdateSpec<>(this.typeToUpdate, this.table, objectToUpdate);
1286+
return new DefaultTypedUpdateSpec<>(this.typeToUpdate, this.table, objectToUpdate, this.where);
12831287
}
12841288

12851289
@Override
12861290
public TypedUpdateSpec<T> table(SqlIdentifier tableName) {
12871291

12881292
Assert.notNull(tableName, "Table name must not be null!");
12891293

1290-
return new DefaultTypedUpdateSpec<>(this.typeToUpdate, tableName, this.objectToUpdate);
1294+
return new DefaultTypedUpdateSpec<>(this.typeToUpdate, tableName, this.objectToUpdate, this.where);
1295+
}
1296+
1297+
@Override
1298+
public UpdateSpec matching(Criteria criteria) {
1299+
1300+
Assert.notNull(criteria, "Criteria must not be null!");
1301+
1302+
return new DefaultTypedUpdateSpec<>(this.typeToUpdate, this.table, this.objectToUpdate, criteria);
12911303
}
12921304

12931305
@Override
@@ -1330,8 +1342,13 @@ private UpdatedRowsFetchSpec exchange(SqlIdentifier table) {
13301342
}
13311343
}
13321344

1345+
Criteria updateCriteria = Criteria.where(dataAccessStrategy.toSql(ids.get(0))).is(id);
1346+
if (this.where != null) {
1347+
updateCriteria = updateCriteria.and(this.where);
1348+
}
1349+
13331350
PreparedOperation<?> operation = mapper.getMappedObject(
1334-
mapper.createUpdate(table, update).withCriteria(Criteria.where(dataAccessStrategy.toSql(ids.get(0))).is(id)));
1351+
mapper.createUpdate(table, update).withCriteria(updateCriteria));
13351352

13361353
return exchangeUpdate(operation);
13371354
}

‎src/main/java/org/springframework/data/r2dbc/core/R2dbcEntityTemplate.java

+61-3
Original file line numberDiff line numberDiff line change
@@ -30,10 +30,12 @@
3030
import org.springframework.beans.BeansException;
3131
import org.springframework.beans.factory.BeanFactory;
3232
import org.springframework.beans.factory.BeanFactoryAware;
33+
import org.springframework.core.convert.ConversionService;
3334
import org.springframework.dao.DataAccessException;
3435
import org.springframework.dao.TransientDataAccessResourceException;
3536
import org.springframework.data.mapping.IdentifierAccessor;
3637
import org.springframework.data.mapping.MappingException;
38+
import org.springframework.data.mapping.PersistentPropertyAccessor;
3739
import org.springframework.data.mapping.context.MappingContext;
3840
import org.springframework.data.projection.ProjectionInformation;
3941
import org.springframework.data.projection.SpelAwareProxyProjectionFactory;
@@ -58,6 +60,7 @@
5860
* prepared in an application context and given to services as bean reference.
5961
*
6062
* @author Mark Paluch
63+
* @author Bogdan Ilchyshyn
6164
* @since 1.1
6265
*/
6366
public class R2dbcEntityTemplate implements R2dbcEntityOperations, BeanFactoryAware {
@@ -372,6 +375,8 @@ <T> Mono<T> doInsert(T entity, SqlIdentifier tableName) {
372375

373376
RelationalPersistentEntity<T> persistentEntity = getRequiredEntity(entity);
374377

378+
setVersionIfNecessary(persistentEntity, entity);
379+
375380
return this.databaseClient.insert() //
376381
.into(persistentEntity.getType()) //
377382
.table(tableName).using(entity) //
@@ -380,6 +385,24 @@ <T> Mono<T> doInsert(T entity, SqlIdentifier tableName) {
380385
.defaultIfEmpty(entity);
381386
}
382387

388+
private <T> void setVersionIfNecessary(RelationalPersistentEntity<T> persistentEntity, T entity) {
389+
PersistentPropertyAccessor<?> propertyAccessor = persistentEntity.getPropertyAccessor(entity);
390+
RelationalPersistentProperty versionProperty = persistentEntity.getVersionProperty();
391+
392+
if (versionProperty == null) {
393+
return;
394+
}
395+
396+
Object currentVersionValue = propertyAccessor.getProperty(versionProperty);
397+
if (currentVersionValue != null) {
398+
return;
399+
}
400+
401+
Class<?> versionPropertyType = versionProperty.getType();
402+
ConversionService conversionService = this.dataAccessStrategy.getConverter().getConversionService();
403+
propertyAccessor.setProperty(versionProperty, conversionService.convert(1L, versionPropertyType));
404+
}
405+
383406
/*
384407
* (non-Javadoc)
385408
* @see org.springframework.data.r2dbc.core.R2dbcEntityOperations#update(java.lang.Object)
@@ -391,11 +414,19 @@ public <T> Mono<T> update(T entity) throws DataAccessException {
391414

392415
RelationalPersistentEntity<T> persistentEntity = getRequiredEntity(entity);
393416

394-
return this.databaseClient.update() //
417+
DatabaseClient.UpdateMatchingSpec updateMatchingSpec = this.databaseClient.update() //
395418
.table(persistentEntity.getType()) //
396-
.table(persistentEntity.getTableName()).using(entity) //
397-
.fetch().rowsUpdated().handle((rowsUpdated, sink) -> {
419+
.table(persistentEntity.getTableName()) //
420+
.using(entity);
421+
422+
DatabaseClient.UpdateSpec updateSpec = updateMatchingSpec;
423+
if (persistentEntity.hasVersionProperty()) {
424+
updateSpec = updateMatchingSpec.matching(createMatchingVersionCriteria(entity, persistentEntity));
425+
incrementVersion(entity, persistentEntity);
426+
}
398427

428+
return updateSpec.fetch() //
429+
.rowsUpdated().handle((rowsUpdated, sink) -> {
399430
if (rowsUpdated == 0) {
400431
sink.error(new TransientDataAccessResourceException(
401432
String.format("Failed to update table [%s]. Row with Id [%s] does not exist.",
@@ -406,6 +437,33 @@ public <T> Mono<T> update(T entity) throws DataAccessException {
406437
});
407438
}
408439

440+
private <T> void incrementVersion(T entity, RelationalPersistentEntity<T> persistentEntity) {
441+
PersistentPropertyAccessor<?> propertyAccessor = persistentEntity.getPropertyAccessor(entity);
442+
RelationalPersistentProperty versionProperty = persistentEntity.getVersionProperty();
443+
444+
ConversionService conversionService = this.dataAccessStrategy.getConverter().getConversionService();
445+
Object currentVersionValue = propertyAccessor.getProperty(versionProperty);
446+
long newVersionValue = 1L;
447+
if (currentVersionValue != null) {
448+
newVersionValue = conversionService.convert(currentVersionValue, Long.class) + 1;
449+
}
450+
Class<?> versionPropertyType = versionProperty.getType();
451+
propertyAccessor.setProperty(versionProperty, conversionService.convert(newVersionValue, versionPropertyType));
452+
}
453+
454+
private <T> Criteria createMatchingVersionCriteria(T entity, RelationalPersistentEntity<T> persistentEntity) {
455+
PersistentPropertyAccessor<?> propertyAccessor = persistentEntity.getPropertyAccessor(entity);
456+
RelationalPersistentProperty versionProperty = persistentEntity.getVersionProperty();
457+
458+
Object version = propertyAccessor.getProperty(versionProperty);
459+
Criteria.CriteriaStep versionColumn = Criteria.where(dataAccessStrategy.toSql(versionProperty.getColumnName()));
460+
if (version == null) {
461+
return versionColumn.isNull();
462+
} else {
463+
return versionColumn.is(version);
464+
}
465+
}
466+
409467
/*
410468
* (non-Javadoc)
411469
* @see org.springframework.data.r2dbc.core.R2dbcEntityOperations#delete(java.lang.Object)

‎src/main/java/org/springframework/data/r2dbc/query/Criteria.java

+15
Original file line numberDiff line numberDiff line change
@@ -30,6 +30,7 @@
3030
*
3131
* @author Mark Paluch
3232
* @author Oliver Drotbohm
33+
* @author Bogdan Ilchyshyn
3334
*/
3435
public class Criteria {
3536

@@ -85,6 +86,20 @@ protected Criteria createCriteria(Comparator comparator, Object value) {
8586
};
8687
}
8788

89+
/**
90+
* Create a new {@link Criteria} as a combination of current and provided {@code other} criteria
91+
* with {@code AND} operator.
92+
*
93+
* @param other Must not be {@literal null}.
94+
* @return a new {@link Criteria} object combining current and provided criteria.
95+
*/
96+
public Criteria and(Criteria other) {
97+
98+
Assert.notNull(other, "Provided criteria must not be null!");
99+
100+
return new Criteria(this, Combinator.AND, other.column, other.comparator, other.value);
101+
}
102+
88103
/**
89104
* Create a new {@link Criteria} and combine it with {@code OR} using the provided {@code column} name.
90105
*

‎src/test/java/org/springframework/data/r2dbc/repository/support/AbstractSimpleR2dbcRepositoryIntegrationTests.java

+89-1
Original file line numberDiff line numberDiff line change
@@ -33,10 +33,11 @@
3333

3434
import org.junit.Before;
3535
import org.junit.Test;
36-
3736
import org.springframework.beans.factory.annotation.Autowired;
3837
import org.springframework.dao.DataAccessException;
38+
import org.springframework.dao.OptimisticLockingFailureException;
3939
import org.springframework.data.annotation.Id;
40+
import org.springframework.data.annotation.Version;
4041
import org.springframework.data.domain.Persistable;
4142
import org.springframework.data.r2dbc.convert.MappingR2dbcConverter;
4243
import org.springframework.data.r2dbc.core.DatabaseClient;
@@ -53,6 +54,7 @@
5354
* Abstract integration tests for {@link SimpleR2dbcRepository} to be ran against various databases.
5455
*
5556
* @author Mark Paluch
57+
* @author Bogdan Ilchyshyn
5658
*/
5759
public abstract class AbstractSimpleR2dbcRepositoryIntegrationTests extends R2dbcIntegrationTestSupport {
5860

@@ -117,6 +119,42 @@ public void shouldSaveNewObject() {
117119
assertThat(map).containsEntry("name", "SCHAUFELRADBAGGER").containsEntry("manual", 12).containsKey("id");
118120
}
119121

122+
@Test
123+
public void shouldSaveNewObjectAndSetVersionIfVersionPropertyExists() {
124+
125+
LegoSetVersionable legoSet = new LegoSetVersionable(null, "SCHAUFELRADBAGGER", 12, null);
126+
127+
repository.save(legoSet) //
128+
.as(StepVerifier::create) //
129+
.consumeNextWith(actual -> assertThat(actual.getVersion()).isEqualTo(1)) //
130+
.verifyComplete();
131+
132+
Map<String, Object> map = jdbc.queryForMap("SELECT * FROM legoset");
133+
assertThat(map) //
134+
.containsEntry("name", "SCHAUFELRADBAGGER") //
135+
.containsEntry("manual", 12) //
136+
.containsEntry("version", 1) //
137+
.containsKey("id");
138+
}
139+
140+
@Test
141+
public void shouldSaveNewObjectAndUseInitialVersionIfVersionPropertyExists() {
142+
143+
LegoSetVersionable legoSet = new LegoSetVersionable(null, "SCHAUFELRADBAGGER", 12, 42);
144+
145+
repository.save(legoSet) //
146+
.as(StepVerifier::create) //
147+
.consumeNextWith(actual -> assertThat(actual.getVersion()).isEqualTo(42)) //
148+
.verifyComplete();
149+
150+
Map<String, Object> map = jdbc.queryForMap("SELECT * FROM legoset");
151+
assertThat(map) //
152+
.containsEntry("name", "SCHAUFELRADBAGGER") //
153+
.containsEntry("manual", 12) //
154+
.containsEntry("version", 42) //
155+
.containsKey("id");
156+
}
157+
120158
@Test
121159
public void shouldUpdateObject() {
122160

@@ -135,6 +173,44 @@ public void shouldUpdateObject() {
135173
assertThat(map).containsEntry("name", "SCHAUFELRADBAGGER").containsEntry("manual", 14).containsKey("id");
136174
}
137175

176+
@Test
177+
public void shouldUpdateVersionableObjectAndIncreaseVersion() {
178+
179+
jdbc.execute("INSERT INTO legoset (name, manual, version) VALUES('SCHAUFELRADBAGGER', 12, 42)");
180+
Integer id = jdbc.queryForObject("SELECT id FROM legoset", Integer.class);
181+
182+
LegoSetVersionable legoSet = new LegoSetVersionable(id, "SCHAUFELRADBAGGER", 12, 42);
183+
legoSet.setManual(14);
184+
185+
repository.save(legoSet) //
186+
.as(StepVerifier::create) //
187+
.expectNextCount(1) //
188+
.verifyComplete();
189+
190+
assertThat(legoSet.getVersion()).isEqualTo(43);
191+
192+
Map<String, Object> map = jdbc.queryForMap("SELECT * FROM legoset");
193+
assertThat(map)
194+
.containsEntry("name", "SCHAUFELRADBAGGER") //
195+
.containsEntry("manual", 14) //
196+
.containsEntry("version", 43) //
197+
.containsKey("id");
198+
}
199+
200+
@Test
201+
public void shouldFailWithOptimistickLockingWhenVersionDoesNotMatchOnUpdate() {
202+
203+
jdbc.execute("INSERT INTO legoset (name, manual, version) VALUES('SCHAUFELRADBAGGER', 12, 42)");
204+
Integer id = jdbc.queryForObject("SELECT id FROM legoset", Integer.class);
205+
206+
LegoSetVersionable legoSet = new LegoSetVersionable(id, "SCHAUFELRADBAGGER", 12, 0);
207+
208+
repository.save(legoSet) //
209+
.as(StepVerifier::create) //
210+
.expectError(OptimisticLockingFailureException.class) //
211+
.verify();
212+
}
213+
138214
@Test
139215
public void shouldSaveObjectsUsingIterable() {
140216

@@ -392,4 +468,16 @@ public boolean isNew() {
392468
return true;
393469
}
394470
}
471+
472+
@Data
473+
@Table("legoset")
474+
@NoArgsConstructor
475+
static class LegoSetVersionable extends LegoSet {
476+
@Version Integer version;
477+
478+
public LegoSetVersionable(Integer id, String name, Integer manual, Integer version) {
479+
super(id, name, manual);
480+
this.version = version;
481+
}
482+
}
395483
}

‎src/test/java/org/springframework/data/r2dbc/testing/H2TestSupport.java

+3
Original file line numberDiff line numberDiff line change
@@ -27,18 +27,21 @@
2727
* Utility class for testing against H2.
2828
*
2929
* @author Mark Paluch
30+
* @author Bogdan Ilchyshyn
3031
*/
3132
public class H2TestSupport {
3233

3334
public static String CREATE_TABLE_LEGOSET = "CREATE TABLE legoset (\n" //
3435
+ " id integer CONSTRAINT id PRIMARY KEY,\n" //
36+
+ " version integer NULL,\n" //
3537
+ " name varchar(255) NOT NULL,\n" //
3638
+ " manual integer NULL\n," //
3739
+ " cert bytea NULL\n" //
3840
+ ");";
3941

4042
public static String CREATE_TABLE_LEGOSET_WITH_ID_GENERATION = "CREATE TABLE legoset (\n" //
4143
+ " id serial CONSTRAINT id PRIMARY KEY,\n" //
44+
+ " version integer NULL,\n" //
4245
+ " name varchar(255) NOT NULL,\n" //
4346
+ " manual integer NULL\n" //
4447
+ ");";

0 commit comments

Comments
 (0)