Merge pull request #15 from haimshalev/dev

[rxrepo-core][rxrepo-mem][rxrepo-sql-core] Handled concurrent delete …
slim-gears · Nov 25, 2019 · 47ece78 · 47ece78
2 parents 4f9bb74 + 3cf8642
commit 47ece78
Show file tree

Hide file tree

Showing 6 changed files with 187 additions and 113 deletions.
diff --git a/rxrepo-core/src/main/java/com/slimgears/rxrepo/query/DefaultEntitySet.java b/rxrepo-core/src/main/java/com/slimgears/rxrepo/query/DefaultEntitySet.java
@@ -411,6 +411,7 @@ public Maybe<S> update(K key, Function<Maybe<S>, Maybe<S>> updater) {
     private static boolean isConcurrencyException(Throwable exception) {
         log.debug("Checking exception: {}", exception.getMessage(), exception);
         return exception instanceof ConcurrentModificationException ||
+                exception instanceof NoSuchElementException ||
                 (exception instanceof CompositeException && ((CompositeException)exception)
                         .getExceptions()
                         .stream()

diff --git a/rxrepo-mem/src/main/java/com/slimgears/rxrepo/mem/MemoryEntityQueryProvider.java b/rxrepo-mem/src/main/java/com/slimgears/rxrepo/mem/MemoryEntityQueryProvider.java
@@ -27,6 +27,7 @@
 import java.util.concurrent.ExecutorService;
 import java.util.concurrent.Executors;
 import java.util.concurrent.atomic.AtomicReference;
+import java.util.function.Supplier;
 import java.util.stream.Collectors;
 
 public class MemoryEntityQueryProvider<K, S> implements EntityQueryProvider<K, S>, AutoCloseable {
@@ -63,12 +64,12 @@ public MetaClassWithKey<K, S> metaClass() {
     @Override
     public Maybe<S> insertOrUpdate(K key, Function<Maybe<S>, Maybe<S>> entityUpdater) {
         return Maybe.defer(() -> {
-            AtomicReference<S> reference = objects.computeIfAbsent(key, k -> new AtomicReference<>());
-            AtomicReference<S> oldValue = new AtomicReference<>(reference.get());
+            Supplier<AtomicReference<S>> referenceResolver = () -> objects.computeIfAbsent(key, k -> new AtomicReference<>());
+            AtomicReference<S> oldValue = new AtomicReference<>(referenceResolver.get().get());
             return entityUpdater
-                    .apply(Optional.ofNullable(reference.get()).map(Maybe::just).orElseGet(Maybe::empty))
-                    .flatMap(e -> reference.compareAndSet(oldValue.get(), e)
-                            ? Maybe.just(e)
+                    .apply(Optional.ofNullable(referenceResolver.get().get()).map(Maybe::just).orElseGet(Maybe::empty))
+                    .flatMap(e -> referenceResolver.get().compareAndSet(oldValue.get(), e)
+                            ? (e != null ? Maybe.just(e): Maybe.empty())
                             : Maybe.error(new ConcurrentModificationException("Concurrent modification of " + metaClass.simpleName() + " detected")))
                     .doOnSuccess(e -> {
                         if (!Objects.equals(oldValue.get(), e)) {

diff --git a/rxrepo-sql-core/src/main/java/com/slimgears/rxrepo/sql/DefaultSqlStatementProvider.java b/rxrepo-sql-core/src/main/java/com/slimgears/rxrepo/sql/DefaultSqlStatementProvider.java
@@ -79,23 +79,32 @@ public <K, S> SqlStatement forDelete(DeleteInfo<K, S> deleteInfo) {
                 limitClause(deleteInfo)));
     }
 
+    @Override
+    public <K, S> SqlStatement forUpdate(MetaClassWithKey<K, S> metaClass, PropertyResolver propertyResolver, ReferenceResolver resolver) {
+        return forInsertOrUpdate(metaClass, propertyResolver, resolver, false);
+    }
+
     @Override
     public <K, S> SqlStatement forInsertOrUpdate(MetaClassWithKey<K, S> metaClass, PropertyResolver propertyResolver, ReferenceResolver resolver) {
+        return forInsertOrUpdate(metaClass, propertyResolver, resolver, true);
+    }
+
+    private <K, S> SqlStatement forInsertOrUpdate(MetaClassWithKey<K, S> metaClass, PropertyResolver propertyResolver, ReferenceResolver resolver, boolean forced) {
         PropertyMeta<S, K> keyProperty = metaClass.keyProperty();
 
         return statement(() -> of(
-                        "update",
-                        schemaProvider.tableName(metaClass),
-                        "set",
-                        Streams
-                                .fromIterable(propertyResolver.propertyNames())
-                                .flatMap(sqlAssignmentGenerator.toAssignment(metaClass, propertyResolver, resolver))
-                                .collect(Collectors.joining(", ")),
-                        "upsert",
-                        "return after",
-                        "where",
-                        toConditionClause(PropertyExpression.ofObject(keyProperty).eq(propertyResolver.getProperty(keyProperty)))
-                ));
+                "update",
+                schemaProvider.tableName(metaClass),
+                "set",
+                Streams
+                        .fromIterable(propertyResolver.propertyNames())
+                        .flatMap(sqlAssignmentGenerator.toAssignment(metaClass, propertyResolver, resolver))
+                        .collect(Collectors.joining(", ")),
+                forced ? "upsert" : "",
+                "return after",
+                "where",
+                toConditionClause(PropertyExpression.ofObject(keyProperty).eq(propertyResolver.getProperty(keyProperty)))
+        ));
     }
 
     @Override

diff --git a/rxrepo-sql-core/src/main/java/com/slimgears/rxrepo/sql/SqlQueryProvider.java b/rxrepo-sql-core/src/main/java/com/slimgears/rxrepo/sql/SqlQueryProvider.java
@@ -83,13 +83,17 @@ public <K, S> Maybe<S> insertOrUpdate(MetaClassWithKey<K, S> metaClass, K key, F
                             .apply(Maybe.just(oldObj))
                             .map(newObj -> pr.mergeWith(PropertyResolver.fromObject(metaClass, newObj)))
                             .filter(newPr -> !pr.equals(newPr))
-                            .flatMap(newPr -> insertOrUpdate(metaClass, newPr).toMaybe());
+                            .flatMap(newPr -> update(metaClass, newPr).toMaybe());
                 })
                 .switchIfEmpty(Maybe.defer(() -> entityUpdater
                         .apply(Maybe.empty())
                         .flatMap(e -> insert(metaClass, e).toMaybe()))));
     }
 
+    private <K, S> Single<S> update(MetaClassWithKey<K, S> metaClass, PropertyResolver propertyResolver) {
+        SqlStatement statement = statementProvider.forUpdate(metaClass, propertyResolver, referenceResolver);
+        return insertOrUpdate(metaClass, statement);
+    }
 
     private <K, S> Single<S> insertOrUpdate(MetaClassWithKey<K, S> metaClass, PropertyResolver propertyResolver) {
         SqlStatement statement = statementProvider.forInsertOrUpdate(metaClass, propertyResolver, referenceResolver);

diff --git a/rxrepo-sql-core/src/main/java/com/slimgears/rxrepo/sql/SqlStatementProvider.java b/rxrepo-sql-core/src/main/java/com/slimgears/rxrepo/sql/SqlStatementProvider.java
@@ -9,19 +9,29 @@
 
 public interface SqlStatementProvider {
     <K, S, T> SqlStatement forQuery(QueryInfo<K, S, T> queryInfo);
+
     <K, S, T, R> SqlStatement forAggregation(QueryInfo<K, S, T> queryInfo,
-                                                                               ObjectExpression<T, R> aggregation,
-                                                                               String projectedName);
+                                             ObjectExpression<T, R> aggregation,
+                                             String projectedName);
+
     <K, S> SqlStatement forUpdate(UpdateInfo<K, S> updateInfo);
+
     <K, S> SqlStatement forDelete(DeleteInfo<K, S> deleteInfo);
+
     <K, S> SqlStatement forInsert(MetaClassWithKey<K, S> metaClass,
-                                                                    PropertyResolver propertyResolver,
-                                                                    ReferenceResolver referenceResolver);
+                                  PropertyResolver propertyResolver,
+                                  ReferenceResolver referenceResolver);
+
     <K, S> SqlStatement forInsertOrUpdate(MetaClassWithKey<K, S> metaClass,
-                                                                            PropertyResolver propertyResolver,
-                                                                            ReferenceResolver referenceResolver);
+                                          PropertyResolver propertyResolver,
+                                          ReferenceResolver referenceResolver);
+
+    <K, S> SqlStatement forUpdate(MetaClassWithKey<K, S> metaClass,
+                                  PropertyResolver propertyResolver,
+                                  ReferenceResolver referenceResolver);
 
     <K, S> SqlStatement forDrop(MetaClassWithKey<K, S> metaClass);
+
     SqlStatement forDrop();
 
     default <K, S> SqlStatement forInsertOrUpdate(MetaClassWithKey<K, S> metaClass, S entity, ReferenceResolver referenceResolver) {