#93 - Adding support for optimistic locking based on @Version column

Author: orange-buffalo

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

@@ -46,6 +46,7 @@
  * to create an instance.
  *
  * @author Mark Paluch
+ * @author Bogdan Ilchyshyn
  */
 public interface DatabaseClient {
 
@@ -707,9 +708,9 @@ interface TypedUpdateSpec<T> {
  *
  * @param objectToUpdate the object of which the attributes will provide the values for the update and the primary
  * key. Must not be {@literal null}.
- * @return a {@link UpdateSpec} for further configuration of the update. Guaranteed to be not {@literal null}.
+ * @return a {@link UpdateMatchingSpec} for further configuration of the update. Guaranteed to be not {@literal null}.
  */
-UpdateSpec using(T objectToUpdate);
+UpdateMatchingSpec using(T objectToUpdate);
 
 /**
  * Use the given {@code tableName} as update target.

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

@@ -69,6 +69,7 @@
  * Default implementation of {@link DatabaseClient}.
  *
  * @author Mark Paluch
+ * @author Bogdan Ilchyshyn
  */
 class DefaultDatabaseClient implements DatabaseClient, ConnectionAccessor {
 
@@ -1188,7 +1189,7 @@ public <T> TypedUpdateSpec<T> table(Class<T> table) {
 
 assertRegularClass(table);
 
-return new DefaultTypedUpdateSpec<>(table, null, null);
+return new DefaultTypedUpdateSpec<>(table, null, null, null);
 }
 }
 
@@ -1262,32 +1263,43 @@ private UpdatedRowsFetchSpec exchange(SqlIdentifier table) {
 }
 }
 
-class DefaultTypedUpdateSpec<T> implements TypedUpdateSpec<T>, UpdateSpec {
+class DefaultTypedUpdateSpec<T> implements TypedUpdateSpec<T>, UpdateMatchingSpec {
 
 private final @Nullable Class<T> typeToUpdate;
 private final @Nullable SqlIdentifier table;
 private final T objectToUpdate;
+private final @Nullable Criteria where;
 
-DefaultTypedUpdateSpec(@Nullable Class<T> typeToUpdate, @Nullable SqlIdentifier table, T objectToUpdate) {
+DefaultTypedUpdateSpec(@Nullable Class<T> typeToUpdate, @Nullable SqlIdentifier table, T objectToUpdate,
+ @Nullable Criteria where) {
 this.typeToUpdate = typeToUpdate;
 this.table = table;
 this.objectToUpdate = objectToUpdate;
+this.where = where;
 }
 
 @Override
-public UpdateSpec using(T objectToUpdate) {
+public UpdateMatchingSpec using(T objectToUpdate) {
 
 Assert.notNull(objectToUpdate, "Object to update must not be null");
 
-return new DefaultTypedUpdateSpec<>(this.typeToUpdate, this.table, objectToUpdate);
+return new DefaultTypedUpdateSpec<>(this.typeToUpdate, this.table, objectToUpdate, this.where);
 }
 
 @Override
 public TypedUpdateSpec<T> table(SqlIdentifier tableName) {
 
 Assert.notNull(tableName, "Table name must not be null!");
 
-return new DefaultTypedUpdateSpec<>(this.typeToUpdate, tableName, this.objectToUpdate);
+return new DefaultTypedUpdateSpec<>(this.typeToUpdate, tableName, this.objectToUpdate, this.where);
+}
+
+@Override
+public UpdateSpec matching(Criteria criteria) {
+
+Assert.notNull(criteria, "Criteria must not be null!");
+
+return new DefaultTypedUpdateSpec<>(this.typeToUpdate, this.table, this.objectToUpdate, criteria);
 }
 
 @Override
@@ -1330,8 +1342,13 @@ private UpdatedRowsFetchSpec exchange(SqlIdentifier table) {
 }
 }
 
+Criteria updateCriteria = Criteria.where(dataAccessStrategy.toSql(ids.get(0))).is(id);
+if (this.where != null) {
+updateCriteria = updateCriteria.and(this.where);
+}
+
 PreparedOperation<?> operation = mapper.getMappedObject(
-mapper.createUpdate(table, update).withCriteria(Criteria.where(dataAccessStrategy.toSql(ids.get(0))).is(id)));
+mapper.createUpdate(table, update).withCriteria(updateCriteria));
 
 return exchangeUpdate(operation);
 }

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

@@ -30,10 +30,12 @@
 import org.springframework.beans.BeansException;
 import org.springframework.beans.factory.BeanFactory;
 import org.springframework.beans.factory.BeanFactoryAware;
+import org.springframework.core.convert.ConversionService;
 import org.springframework.dao.DataAccessException;
 import org.springframework.dao.TransientDataAccessResourceException;
 import org.springframework.data.mapping.IdentifierAccessor;
 import org.springframework.data.mapping.MappingException;
+import org.springframework.data.mapping.PersistentPropertyAccessor;
 import org.springframework.data.mapping.context.MappingContext;
 import org.springframework.data.projection.ProjectionInformation;
 import org.springframework.data.projection.SpelAwareProxyProjectionFactory;
@@ -58,6 +60,7 @@
  * prepared in an application context and given to services as bean reference.
  *
  * @author Mark Paluch
+ * @author Bogdan Ilchyshyn
  * @since 1.1
  */
 public class R2dbcEntityTemplate implements R2dbcEntityOperations, BeanFactoryAware {
@@ -372,6 +375,8 @@ <T> Mono<T> doInsert(T entity, SqlIdentifier tableName) {
 
 RelationalPersistentEntity<T> persistentEntity = getRequiredEntity(entity);
 
+setVersionIfNecessary(persistentEntity,entity);
+
 return this.databaseClient.insert() //
 .into(persistentEntity.getType()) //
 .table(tableName).using(entity) //
@@ -380,6 +385,24 @@ <T> Mono<T> doInsert(T entity, SqlIdentifier tableName) {
 .defaultIfEmpty(entity);
 }
 
+private <T> void setVersionIfNecessary(RelationalPersistentEntity<T> persistentEntity, T entity) {
+PersistentPropertyAccessor<?> propertyAccessor = persistentEntity.getPropertyAccessor(entity);
+RelationalPersistentProperty versionProperty = persistentEntity.getVersionProperty();
+
+if (versionProperty == null) {
+return;
+}
+
+Object currentVersionValue = propertyAccessor.getProperty(versionProperty);
+if (currentVersionValue != null) {
+return;
+}
+
+Class<?> versionPropertyType = versionProperty.getType();
+ConversionService conversionService = this.dataAccessStrategy.getConverter().getConversionService();
+propertyAccessor.setProperty(versionProperty, conversionService.convert(1L, versionPropertyType));
+}
+
 /*
  * (non-Javadoc)
  * @see org.springframework.data.r2dbc.core.R2dbcEntityOperations#update(java.lang.Object)
@@ -391,11 +414,19 @@ public <T> Mono<T> update(T entity) throws DataAccessException {
 
 RelationalPersistentEntity<T> persistentEntity = getRequiredEntity(entity);
 
-return this.databaseClient.update() //
+DatabaseClient.UpdateMatchingSpec updateMatchingSpec = this.databaseClient.update() //
 .table(persistentEntity.getType()) //
-.table(persistentEntity.getTableName()).using(entity) //
-.fetch().rowsUpdated().handle((rowsUpdated, sink) -> {
+.table(persistentEntity.getTableName()) //
+.using(entity);
+
+DatabaseClient.UpdateSpec updateSpec = updateMatchingSpec;
+if (persistentEntity.hasVersionProperty()) {
+updateSpec = updateMatchingSpec.matching(createMatchingVersionCriteria(entity, persistentEntity));
+incrementVersion(entity, persistentEntity);
+}
 
+return updateSpec.fetch() //
+.rowsUpdated().handle((rowsUpdated, sink) -> {
 if (rowsUpdated == 0) {
 sink.error(new TransientDataAccessResourceException(
 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 {
 });
 }
 
+private <T> void incrementVersion(T entity, RelationalPersistentEntity<T> persistentEntity) {
+PersistentPropertyAccessor<?> propertyAccessor = persistentEntity.getPropertyAccessor(entity);
+RelationalPersistentProperty versionProperty = persistentEntity.getVersionProperty();
+
+ConversionService conversionService = this.dataAccessStrategy.getConverter().getConversionService();
+Object currentVersionValue = propertyAccessor.getProperty(versionProperty);
+long newVersionValue = 1L;
+if (currentVersionValue != null) {
+newVersionValue = conversionService.convert(currentVersionValue, Long.class) + 1;
+}
+Class<?> versionPropertyType = versionProperty.getType();
+propertyAccessor.setProperty(versionProperty, conversionService.convert(newVersionValue, versionPropertyType));
+}
+
+private <T> Criteria createMatchingVersionCriteria(T entity, RelationalPersistentEntity<T> persistentEntity) {
+PersistentPropertyAccessor<?> propertyAccessor = persistentEntity.getPropertyAccessor(entity);
+RelationalPersistentProperty versionProperty = persistentEntity.getVersionProperty();
+
+Object version = propertyAccessor.getProperty(versionProperty);
+Criteria.CriteriaStep versionColumn = Criteria.where(dataAccessStrategy.toSql(versionProperty.getColumnName()));
+if (version == null) {
+return versionColumn.isNull();
+} else {
+return versionColumn.is(version);
+}
+}
+
 /*
  * (non-Javadoc)
  * @see org.springframework.data.r2dbc.core.R2dbcEntityOperations#delete(java.lang.Object)

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

@@ -33,10 +33,11 @@
 
 import org.junit.Before;
 import org.junit.Test;
-
 import org.springframework.beans.factory.annotation.Autowired;
 import org.springframework.dao.DataAccessException;
+import org.springframework.dao.OptimisticLockingFailureException;
 import org.springframework.data.annotation.Id;
+import org.springframework.data.annotation.Version;
 import org.springframework.data.domain.Persistable;
 import org.springframework.data.r2dbc.convert.MappingR2dbcConverter;
 import org.springframework.data.r2dbc.core.DatabaseClient;
@@ -53,6 +54,7 @@
  * Abstract integration tests for {@link SimpleR2dbcRepository} to be ran against various databases.
  *
  * @author Mark Paluch
+ * @author Bogdan Ilchyshyn
  */
 public abstract class AbstractSimpleR2dbcRepositoryIntegrationTests extends R2dbcIntegrationTestSupport {
 
@@ -117,6 +119,42 @@ public void shouldSaveNewObject() {
 assertThat(map).containsEntry("name", "SCHAUFELRADBAGGER").containsEntry("manual", 12).containsKey("id");
 }
 
+@Test
+public void shouldSaveNewObjectAndSetVersionIfVersionPropertyExists() {
+
+LegoSetVersionable legoSet = new LegoSetVersionable(null, "SCHAUFELRADBAGGER", 12, null);
+
+repository.save(legoSet) //
+.as(StepVerifier::create) //
+.consumeNextWith(actual -> assertThat(actual.getVersion()).isEqualTo(1)) //
+.verifyComplete();
+
+Map<String, Object> map = jdbc.queryForMap("SELECT * FROM legoset");
+assertThat(map) //
+.containsEntry("name", "SCHAUFELRADBAGGER") //
+.containsEntry("manual", 12) //
+.containsEntry("version", 1) //
+.containsKey("id");
+}
+
+@Test
+public void shouldSaveNewObjectAndUseInitialVersionIfVersionPropertyExists() {
+
+LegoSetVersionable legoSet = new LegoSetVersionable(null, "SCHAUFELRADBAGGER", 12, 42);
+
+repository.save(legoSet) //
+.as(StepVerifier::create) //
+.consumeNextWith(actual -> assertThat(actual.getVersion()).isEqualTo(42)) //
+.verifyComplete();
+
+Map<String, Object> map = jdbc.queryForMap("SELECT * FROM legoset");
+assertThat(map) //
+.containsEntry("name", "SCHAUFELRADBAGGER") //
+.containsEntry("manual", 12) //
+.containsEntry("version", 42) //
+.containsKey("id");
+}
+
 @Test
 public void shouldUpdateObject() {
 
@@ -135,6 +173,44 @@ public void shouldUpdateObject() {
 assertThat(map).containsEntry("name", "SCHAUFELRADBAGGER").containsEntry("manual", 14).containsKey("id");
 }
 
+@Test
+public void shouldUpdateVersionableObjectAndIncreaseVersion() {
+
+jdbc.execute("INSERT INTO legoset (name, manual, version) VALUES('SCHAUFELRADBAGGER', 12, 42)");
+Integer id = jdbc.queryForObject("SELECT id FROM legoset", Integer.class);
+
+LegoSetVersionable legoSet = new LegoSetVersionable(id, "SCHAUFELRADBAGGER", 12, 42);
+legoSet.setManual(14);
+
+repository.save(legoSet) //
+.as(StepVerifier::create) //
+.expectNextCount(1) //
+.verifyComplete();
+
+assertThat(legoSet.getVersion()).isEqualTo(43);
+
+Map<String, Object> map = jdbc.queryForMap("SELECT * FROM legoset");
+assertThat(map)
+.containsEntry("name", "SCHAUFELRADBAGGER") //
+.containsEntry("manual", 14) //
+.containsEntry("version", 43) //
+.containsKey("id");
+}
+
+@Test
+public void shouldFailWithOptimistickLockingWhenVersionDoesNotMatchOnUpdate() {
+
+jdbc.execute("INSERT INTO legoset (name, manual, version) VALUES('SCHAUFELRADBAGGER', 12, 42)");
+Integer id = jdbc.queryForObject("SELECT id FROM legoset", Integer.class);
+
+LegoSetVersionable legoSet = new LegoSetVersionable(id, "SCHAUFELRADBAGGER", 12, 0);
+
+repository.save(legoSet) //
+.as(StepVerifier::create) //
+.expectError(OptimisticLockingFailureException.class) //
+.verify();
+}
+
 @Test
 public void shouldSaveObjectsUsingIterable() {
 
@@ -392,4 +468,16 @@ public boolean isNew() {
 return true;
 }
 }
+
+@Data
+@Table("legoset")
+@NoArgsConstructor
+static class LegoSetVersionable extends LegoSet {
+@Version Integer version;
+
+public LegoSetVersionable(Integer id, String name, Integer manual, Integer version) {
+super(id, name, manual);
+this.version = version;
+}
+}
 }

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

@@ -27,18 +27,21 @@
  * Utility class for testing against H2.
  *
  * @author Mark Paluch
+ * @author Bogdan Ilchyshyn
  */
 public class H2TestSupport {
 
 public static String CREATE_TABLE_LEGOSET = "CREATE TABLE legoset (\n" //
 + " id integer CONSTRAINT id PRIMARY KEY,\n" //
++ " version integer NULL,\n" //
 + " name varchar(255) NOT NULL,\n" //
 + " manual integer NULL\n," //
 + " cert bytea NULL\n" //
 + ");";
 
 public static String CREATE_TABLE_LEGOSET_WITH_ID_GENERATION = "CREATE TABLE legoset (\n" //
 + " id serial CONSTRAINT id PRIMARY KEY,\n" //
++ " version integer NULL,\n" //
 + " name varchar(255) NOT NULL,\n" //
 + " manual integer NULL\n" //
 + ");";

src/test/java/org/springframework/data/r2dbc/testing/MySqlTestSupport.java

@@ -35,20 +35,23 @@
  * Utility class for testing against MySQL.
  *
  * @author Mark Paluch
+ * @author Bogdan Ilchyshyn
  */
 public class MySqlTestSupport {
 
 private static ExternalDatabase testContainerDatabase;
 
 public static String CREATE_TABLE_LEGOSET = "CREATE TABLE legoset (\n" //
 + " id integer PRIMARY KEY,\n" //
++ " version integer NULL,\n" //
 + " name varchar(255) NOT NULL,\n" //
 + " manual integer NULL\n," //
 + " cert varbinary(255) NULL\n" //
 + ") ENGINE=InnoDB;";
 
 public static String CREATE_TABLE_LEGOSET_WITH_ID_GENERATION = "CREATE TABLE legoset (\n" //
 + " id integer AUTO_INCREMENT PRIMARY KEY,\n" //
++ " version integer NULL,\n" //
 + " name varchar(255) NOT NULL,\n" //
 + " manual integer NULL\n" //
 + ") ENGINE=InnoDB;";