Merge pull request #132 from SpineEventEngine/thread-safe-transactions

Thread-safe transactions

Author: dmdashenkov

.gitignore

@@ -29,8 +29,6 @@ gcs-auth-key.json
 # They are stored in the encrypted file and are decrypted on the CI server only.
 
 # This file extracted form credentials.tar
-spine-dev*.json
-/datastore/src/test/resources/spine-dev.json
-/stackdriver-trace/src/test/resources/spine-dev.json
+**/spine-dev*.json
 
 package-lock.json

config

@@ -432,6 +432,17 @@ private void deleteEntities(Key[] keys) {
         }
     }
 
+    /**
+     * Starts a new database transaction.
+     *
+     * @return the new transaction
+     * @see TransactionWrapper
+     */
+    public final TransactionWrapper newTransaction() {
+        Transaction tx = datastore.newTransaction();
+        return new TransactionWrapper(tx);
+    }
+
     /**
      * Starts a transaction.
      *
@@ -443,7 +454,9 @@ private void deleteEntities(Key[] keys) {
      *         if a transaction is already started on this instance of
      *         {@code DatastoreWrapper}
      * @see #isTransactionActive()
+     * @deprecated Use {@link #newTransaction()} instead.
      */
+    @Deprecated
     public void startTransaction() throws IllegalStateException {
         checkState(!isTransactionActive(), NOT_ACTIVE_TRANSACTION_CONDITION_MESSAGE);
         activeTransaction = datastore.newTransaction();
@@ -461,7 +474,9 @@ public void startTransaction() throws IllegalStateException {
      *         if no transaction is started on this instance of
      *         {@code DatastoreWrapper}
      * @see #isTransactionActive()
+     * @deprecated Use {@link #newTransaction()} instead.
      */
+    @Deprecated
     public void commitTransaction() throws IllegalStateException {
         checkState(isTransactionActive(), ACTIVE_TRANSACTION_CONDITION_MESSAGE);
         activeTransaction.commit();
@@ -481,7 +496,9 @@ public void commitTransaction() throws IllegalStateException {
      *         if no transaction is active for the current
      *         instance of {@code DatastoreWrapper}
      * @see #isTransactionActive()
+     * @deprecated Use {@link #newTransaction()} instead.
      */
+    @Deprecated
     public void rollbackTransaction() throws IllegalStateException {
         checkState(isTransactionActive(), ACTIVE_TRANSACTION_CONDITION_MESSAGE);
         activeTransaction.rollback();
@@ -492,7 +509,9 @@ public void rollbackTransaction() throws IllegalStateException {
      * Checks whether there is an active transaction on this instance of {@code DatastoreWrapper}.
      *
      * @return {@code true} if there is an active transaction, {@code false} otherwise
+     * @deprecated Use {@link #newTransaction()} instead.
      */
+    @Deprecated
     public boolean isTransactionActive() {
         return activeTransaction != null && activeTransaction.isActive();
     }

datastore/src/main/java/io/spine/server/storage/datastore/DatastoreWrapper.java

@@ -141,31 +141,20 @@ public final void write(I id, M message) {
      * Behaves similarly to {@link #write(Message)}, but performs the operation
      * in scope of a Datastore transaction.
      *
-     * <p>If there is no active transaction at the moment, the new transaction is started
-     * and committed. In case there is already an active transaction, the write operation
-     * is performed in its scope.
-     *
      * @param message
      *         the message to write
      */
-    @SuppressWarnings("OverlyBroadCatchBlock")  // handling all possible transaction-related issues.
     final void writeTransactionally(M message) {
         checkNotNull(message);
 
-        boolean txRequired = !datastore.isTransactionActive();
-        if (txRequired) {
-            datastore.startTransaction();
-        }
-        try {
-            write(message);
-            if (txRequired) {
-                datastore.commitTransaction();
-            }
-        } catch (Exception e) {
-            if (txRequired && datastore.isTransactionActive()) {
-                datastore.rollbackTransaction();
-            }
-            throw newIllegalStateException("Error committing the transaction.", e);
+        try (TransactionWrapper tx = datastore.newTransaction()) {
+            Entity entity = toEntity(message);
+            tx.createOrUpdate(entity);
+            tx.commit();
+        } catch (RuntimeException e) {
+            throw newIllegalStateException(e,
+                                           "Error writing a `%s` in transaction.",
+                                           message.getClass().getName());
         }
     }
 

datastore/src/main/java/io/spine/server/storage/datastore/DsMessageStorage.java

@@ -22,6 +22,7 @@
 
 import com.google.cloud.datastore.Entity;
 import com.google.cloud.datastore.Value;
+import com.google.errorprone.annotations.CanIgnoreReturnValue;
 import com.google.errorprone.annotations.Immutable;
 import com.google.protobuf.Message;
 
@@ -55,6 +56,7 @@ interface MessageColumn<M extends Message> {
      *         the message which field value is going to be set
      * @return the same instance of {@code builder} with the property filled
      */
+    @CanIgnoreReturnValue
     default Entity.Builder fill(Entity.Builder builder, M message) {
         Value<?> value = getter().apply(message);
         builder.set(columnName(), value);

datastore/src/main/java/io/spine/server/storage/datastore/MessageColumn.java

@@ -20,6 +20,7 @@
 
 package io.spine.server.storage.datastore;
 
+import com.google.errorprone.annotations.CanIgnoreReturnValue;
 import com.google.protobuf.Descriptors.Descriptor;
 import io.spine.server.entity.Entity;
 import io.spine.server.entity.model.EntityClass;
@@ -58,6 +59,7 @@ abstract class RecordStorageBuilder<I,
     /**
      * Sets the type URL of the entity state, which is stored in the resulting storage.
      */
+    @CanIgnoreReturnValue
     public B setStateType(TypeUrl stateTypeUrl) {
         checkNotNull(stateTypeUrl);
         Descriptor descriptor = stateTypeUrl.toTypeName()
@@ -69,6 +71,7 @@ public B setStateType(TypeUrl stateTypeUrl) {
     /**
      * Assignts the ID class of the stored entities.
      */
+    @CanIgnoreReturnValue
     public B setIdClass(Class<I> idClass) {
         this.idClass = checkNotNull(idClass);
         return self();
@@ -77,6 +80,7 @@ public B setIdClass(Class<I> idClass) {
     /**
      * Assigns the class of the stored entity.
      */
+    @CanIgnoreReturnValue
     public B setEntityClass(Class<? extends Entity<?, ?>> entityClass) {
         this.entityClass = checkNotNull(entityClass);
         return self();
@@ -90,6 +94,7 @@ public B setEntityClass(Class<? extends Entity<?, ?>> entityClass) {
      * {@linkplain #setEntityClass(Class) entity class} separately.
      */
     @SuppressWarnings("unchecked") // The ID class is ensured by the parameter type.
+    @CanIgnoreReturnValue
     public B setModelClass(EntityClass<? extends Entity<I, ?>> modelClass) {
         TypeUrl stateType = modelClass.stateType();
         Class<I> idClass = (Class<I>) modelClass.idClass();
@@ -104,6 +109,7 @@ public B setModelClass(EntityClass<? extends Entity<I, ?>> modelClass) {
     /**
      * Sets the {@link io.spine.server.storage.datastore.DatastoreWrapper} to use in this storage.
      */
+    @CanIgnoreReturnValue
     public B setDatastore(DatastoreWrapper datastore) {
         this.datastore = checkNotNull(datastore);
         return self();
@@ -117,6 +123,7 @@ public B setDatastore(DatastoreWrapper datastore) {
      *         {@link io.spine.server.storage.Storage#isMultitenant multitenant},
      *         {@code false} otherwise
      */
+    @CanIgnoreReturnValue
     public B setMultitenant(boolean multitenant) {
         this.multitenant = multitenant;
         return self();
@@ -126,6 +133,7 @@ public B setMultitenant(boolean multitenant) {
      * Assigns the type registry of
      * the {@linkplain io.spine.server.entity.storage.EntityColumn entity columns}.
      */
+    @CanIgnoreReturnValue
     public B setColumnTypeRegistry(
             ColumnTypeRegistry<? extends DatastoreColumnType<?, ?>> columnTypeRegistry) {
         this.columnTypeRegistry = checkNotNull(columnTypeRegistry);

datastore/src/main/java/io/spine/server/storage/datastore/RecordStorageBuilder.java

@@ -0,0 +1,88 @@
+/*
+ * Copyright 2019, TeamDev. All rights reserved.
+ *
+ * Redistribution and use in source and/or binary forms, with or without
+ * modification, must retain the above copyright notice and the following
+ * disclaimer.
+ *
+ * THIS SOFTWARE IS PROVIDED BY THE COPYRIGHT HOLDERS AND CONTRIBUTORS
+ * "AS IS" AND ANY EXPRESS OR IMPLIED WARRANTIES, INCLUDING, BUT NOT
+ * LIMITED TO, THE IMPLIED WARRANTIES OF MERCHANTABILITY AND FITNESS FOR
+ * A PARTICULAR PURPOSE ARE DISCLAIMED. IN NO EVENT SHALL THE COPYRIGHT
+ * OWNER OR CONTRIBUTORS BE LIABLE FOR ANY DIRECT, INDIRECT, INCIDENTAL,
+ * SPECIAL, EXEMPLARY, OR CONSEQUENTIAL DAMAGES (INCLUDING, BUT NOT
+ * LIMITED TO, PROCUREMENT OF SUBSTITUTE GOODS OR SERVICES; LOSS OF USE,
+ * DATA, OR PROFITS; OR BUSINESS INTERRUPTION) HOWEVER CAUSED AND ON ANY
+ * THEORY OF LIABILITY, WHETHER IN CONTRACT, STRICT LIABILITY, OR TORT
+ * (INCLUDING NEGLIGENCE OR OTHERWISE) ARISING IN ANY WAY OUT OF THE USE
+ * OF THIS SOFTWARE, EVEN IF ADVISED OF THE POSSIBILITY OF SUCH DAMAGE.
+ */
+
+package io.spine.server.storage.datastore;
+
+import com.google.cloud.datastore.DatastoreException;
+import com.google.cloud.datastore.Entity;
+import com.google.cloud.datastore.Key;
+import com.google.cloud.datastore.Transaction;
+
+import java.util.Optional;
+
+import static com.google.common.base.Preconditions.checkNotNull;
+
+/**
+ * A Cloud Datastore transaction wrapper.
+ */
+public final class TransactionWrapper implements AutoCloseable {
+
+    private final Transaction tx;
+
+    TransactionWrapper(Transaction tx) {
+        this.tx = checkNotNull(tx);
+    }
+
+    /**
+     * Puts the given entity into the Datastore in the transaction.
+     */
+    public void createOrUpdate(Entity entity) {
+        tx.put(entity);
+    }
+
+    /**
+     * Reads an entity from the Datastore in the transaction.
+     *
+     * @return the entity with the given key or {@code Optional.empty()} if such an entity does not
+     *         exist
+     */
+    public Optional<Entity> read(Key key) {
+        Entity entity = tx.get(key);
+        return Optional.ofNullable(entity);
+    }
+
+    /**
+     * Commits this transaction.
+     *
+     * @throws DatastoreException if the transaction is no longer active
+     */
+    public void commit() {
+        tx.commit();
+    }
+
+    /**
+     * Rolls back this transaction.
+     *
+     * @throws DatastoreException if the transaction is no longer active
+     */
+    public void rollback() {
+        tx.rollback();
+    }
+
+    /**
+     * Rolls back this transaction if it's still active.
+     */
+    @Override
+    public void close() {
+        if (tx.isActive()) {
+            rollback();
+        }
+    }
+}

datastore/src/main/java/io/spine/server/storage/datastore/TransactionWrapper.java

@@ -17,12 +17,16 @@
  * (INCLUDING NEGLIGENCE OR OTHERWISE) ARISING IN ANY WAY OUT OF THE USE
  * OF THIS SOFTWARE, EVEN IF ADVISED OF THE POSSIBILITY OF SUCH DAMAGE.
  */
+
 /**
  * This package contains Google Cloud Datastore implementation of the storages.
  *
  * @see io.spine.server.storage.AbstractStorage
  */
+@CheckReturnValue
 @ParametersAreNonnullByDefault
 package io.spine.server.storage.datastore;
 
+import com.google.errorprone.annotations.CheckReturnValue;
+
 import javax.annotation.ParametersAreNonnullByDefault;

datastore/src/main/java/io/spine/server/storage/datastore/package-info.java

@@ -77,6 +77,7 @@ static void tearDown() {
         wrapper.dropTable(NAMESPACE_HOLDER_KIND);
     }
 
+    @SuppressWarnings("deprecation") // To be deleted alongside with the tested API.
     @Nested
     class SingleTenant {