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