java/src/main/com/google/appengine/api/datastore/TransactionImpl.java

   1 // Copyright 2008 Google Inc. All Rights Reserved.
   2 package com.google.appengine.api.datastore;
   3
   4 import com.google.appengine.api.datastore.EntityCachingStrategy.PreMutationCachingResult;
   5 import com.google.appengine.api.utils.FutureWrapper;
   6 import com.google.apphosting.api.ApiBasePb;
   7 import com.google.apphosting.api.ApiProxy;
   8 import com.google.apphosting.api.ApiProxy.ApiConfig;
   9 import com.google.apphosting.datastore.DatastoreV3Pb;
  10 import com.google.apphosting.datastore.DatastoreV3Pb.CommitResponse;
  11 import com.google.io.protocol.ProtocolMessage;
  12
  13 import java.util.List;
  14 import java.util.concurrent.Future;
  15
  16 /**
  17  * Implementation of the {@link Transaction} interface that routes all calls
  18  * through the {@link ApiProxy}.  Our implementation is implicitly async.
  19  * BeginTransaction RPCs always return instantly, and this class maintains a
  20  * reference to the {@link Future} associated with the RPC.  We service as
  21  * much of the {@link Transaction} interface as we can without retrieving
  22  * the result of the future.
  23  *
  24  * There is no synchronization in this code because transactions are associated
  25  * with a single thread and are documented as such.
  26  *
  27  */
  28 class TransactionImpl implements Transaction, CurrentTransactionProvider {
  29
  30   enum TransactionState {
  31     BEGUN,
  32     COMPLETION_IN_PROGRESS,
  33     COMMITTED,
  34     ROLLED_BACK,
  35     ERROR
  36   }
  37
  38   private final ApiConfig apiConfig;
  39
  40   private final String app;
  41
  42   /**
  43    * The {@link Future} associated with the BeginTransaction RPC we sent to the
  44    * datastore server.
  45    */
  46   private final Future<DatastoreV3Pb.Transaction> future;
  47
  48   private final TransactionStack txnStack;
  49
  50   private final DatastoreCallbacks callbacks;
  51
  52   private final EntityCachingStrategy entityCachingStrategy;
  53
  54   private final boolean isExplicit;
  55
  56   TransactionState state = TransactionState.BEGUN;
  57
  58   /**
  59    * A {@link PostOpFuture} implementation that runs both post put and post
  60    * delete callbacks.
  61    */
  62   private class PostCommitFuture extends PostOpFuture<Void> {
  63     private final List<Entity> putEntities;
  64     private final List<Key> deletedKeys;
  65
  66     private PostCommitFuture(
  67         List<Entity> putEntities, List<Key> deletedKeys, Future<Void> delegate) {
  68       super(delegate, callbacks);
  69       this.putEntities = putEntities;
  70       this.deletedKeys = deletedKeys;
  71     }
  72
  73     @Override
  74     void executeCallbacks(Void ignoreMe) {
  75       PutContext putContext = new PutContext(TransactionImpl.this, putEntities);
  76       callbacks.executePostPutCallbacks(putContext);
  77       DeleteContext deleteContext = new DeleteContext(TransactionImpl.this, deletedKeys);
  78       callbacks.executePostDeleteCallbacks(deleteContext);
  79     }
  80   }
  81
  82   TransactionImpl(ApiConfig apiConfig, String app, Future<DatastoreV3Pb.Transaction> future,
  83       TransactionStack txnStack, DatastoreCallbacks callbacks,
  84       EntityCachingStrategy entityCachingStrategy, boolean isExplicit) {
  85     this.apiConfig = apiConfig;
  86     this.app = app;
  87     this.future = future;
  88     this.txnStack = txnStack;
  89     this.callbacks = callbacks;
  90     this.entityCachingStrategy = entityCachingStrategy;
  91     this.isExplicit = isExplicit;
  92   }
  93
  94   /**
  95    * Provides the unique identifier for the txn.
  96    * Blocks on the future since the handle comes back from the datastore
  97    * server.
  98    */
  99   private long getHandle() {
 100     return FutureHelper.quietGet(future).getHandle();
 101   }
 102
 103   <T extends ProtocolMessage<T>> Future<T> makeAsyncCall(
 104       String methodName, ProtocolMessage<?> request, T response) {
 105     return DatastoreApiHelper.makeAsyncCall(apiConfig, methodName, request, response);
 106   }
 107
 108   private <T extends ProtocolMessage<T>> Future<T> makeAsyncCall(String methodName, T response) {
 109     DatastoreV3Pb.Transaction txn = new DatastoreV3Pb.Transaction();
 110     txn.setApp(app);
 111     txn.setHandle(getHandle());
 112
 113     return makeAsyncCall(methodName, txn, response);
 114   }
 115
 116   @Override
 117   public void commit() {
 118     FutureHelper.quietGet(commitAsync());
 119   }
 120
 121   @Override
 122   public Future<Void> commitAsync() {
 123     ensureTxnStarted();
 124     try {
 125       for (Future<?> f : txnStack.getFutures(this)) {
 126         FutureHelper.quietGet(f);
 127       }
 128       PreMutationCachingResult preMutationCachingResult =
 129           entityCachingStrategy.preCommit(txnStack.getPutEntities(this),
 130               txnStack.getDeletedKeys(this));
 131       Future<CommitResponse> commitResponse = makeAsyncCall("Commit", new CommitResponse());
 132       state = TransactionState.COMPLETION_IN_PROGRESS;
 133       Future<Void> result = new FutureWrapper<CommitResponse, Void>(commitResponse) {
 134         @Override
 135         protected Void wrap(CommitResponse ignore) throws Exception {
 136           state = TransactionState.COMMITTED;
 137           return null;
 138         }
 139
 140         @Override
 141         protected Throwable convertException(Throwable cause) {
 142           state = TransactionState.ERROR;
 143           return cause;
 144         }
 145       };
 146       result = entityCachingStrategy.createPostMutationFuture(result, preMutationCachingResult);
 147       return new PostCommitFuture(txnStack.getPutEntities(this), txnStack.getDeletedKeys(this),
 148           result);
 149     } finally {
 150       if (isExplicit) {
 151         txnStack.remove(this);
 152       }
 153     }
 154   }
 155
 156   @Override
 157   public void rollback() {
 158     FutureHelper.quietGet(rollbackAsync());
 159   }
 160
 161   @Override
 162   public Future<Void> rollbackAsync() {
 163     ensureTxnStarted();
 164     try {
 165       for (Future<?> f : txnStack.getFutures(this)) {
 166         FutureHelper.quietGet(f);
 167       }
 168       Future<ApiBasePb.VoidProto> future = makeAsyncCall("Rollback", new ApiBasePb.VoidProto());
 169       state = TransactionState.COMPLETION_IN_PROGRESS;
 170       return new FutureWrapper<ApiBasePb.VoidProto, Void>(future) {
 171         @Override
 172         protected Void wrap(ApiBasePb.VoidProto ignore) throws Exception {
 173           state = TransactionState.ROLLED_BACK;
 174           return null;
 175         }
 176
 177         @Override
 178         protected Throwable convertException(Throwable cause) {
 179           state = TransactionState.ERROR;
 180           return cause;
 181         }
 182       };
 183     } finally {
 184       if (isExplicit) {
 185         txnStack.remove(this);
 186       }
 187     }
 188   }
 189
 190   @Override
 191   public String getApp() {
 192     return app;
 193   }
 194
 195   @Override
 196   public String getId() {
 197     return Long.toString(getHandle());
 198   }
 199
 200   @Override
 201   public boolean equals(Object o) {
 202     if (this == o) {
 203       return true;
 204     }
 205     if (o == null || getClass() != o.getClass()) {
 206       return false;
 207     }
 208
 209     TransactionImpl that = (TransactionImpl) o;
 210
 211     return getHandle() == that.getHandle();
 212   }
 213
 214   @Override
 215   public int hashCode() {
 216     return (int) (getHandle() ^ (getHandle() >>> 32));
 217   }
 218
 219   @Override
 220   public String toString() {
 221     return "Txn [" + app + "." + getHandle() + ", " + state + "]";
 222   }
 223
 224   @Override
 225   public boolean isActive() {
 226     return state == TransactionState.BEGUN || state == TransactionState.COMPLETION_IN_PROGRESS;
 227   }
 228
 229   @Override
 230   public Transaction getCurrentTransaction(Transaction defaultValue) {
 231     return this;
 232   }
 233
 234   /**
 235    * If {@code txn} is not null and not active, throw
 236    * {@link IllegalStateException}.
 237    */
 238   static void ensureTxnActive(Transaction txn) {
 239     if (txn != null && !txn.isActive()) {
 240       throw new IllegalStateException("Transaction with which this operation is "
 241           + "associated is not active.");
 242     }
 243   }
 244
 245   private void ensureTxnStarted() {
 246     if (state != TransactionState.BEGUN) {
 247       throw new IllegalStateException("Transaction is in state " + state + ".  There is no legal "
 248           + "transition out of this state.");
 249     }
 250   }
 251 }