1 // Copyright 2014 Google Inc. All rights reserved. 2 // Use of this source code is governed by the Apache 2.0 3 // license that can be found in the LICENSE file. 4 5 package internal 6 7 // This file implements hooks for applying datastore transactions. 8 9 import ( 10 "errors" 11 "reflect" 12 13 "github.com/golang/protobuf/proto" 14 netcontext "golang.org/x/net/context" 15 16 basepb "google.golang.org/appengine/internal/base" 17 pb "google.golang.org/appengine/internal/datastore" 18 ) 19 20 var transactionSetters = make(map[reflect.Type]reflect.Value) 21 22 // RegisterTransactionSetter registers a function that sets transaction information 23 // in a protocol buffer message. f should be a function with two arguments, 24 // the first being a protocol buffer type, and the second being *datastore.Transaction. 25 func RegisterTransactionSetter(f interface{}) { 26 v := reflect.ValueOf(f) 27 transactionSetters[v.Type().In(0)] = v 28 } 29 30 // applyTransaction applies the transaction t to message pb 31 // by using the relevant setter passed to RegisterTransactionSetter. 32 func applyTransaction(pb proto.Message, t *pb.Transaction) { 33 v := reflect.ValueOf(pb) 34 if f, ok := transactionSetters[v.Type()]; ok { 35 f.Call([]reflect.Value{v, reflect.ValueOf(t)}) 36 } 37 } 38 39 var transactionKey = "used for *Transaction" 40 41 func transactionFromContext(ctx netcontext.Context) *transaction { 42 t, _ := ctx.Value(&transactionKey).(*transaction) 43 return t 44 } 45 46 func withTransaction(ctx netcontext.Context, t *transaction) netcontext.Context { 47 return netcontext.WithValue(ctx, &transactionKey, t) 48 } 49 50 type transaction struct { 51 transaction pb.Transaction 52 finished bool 53 } 54 55 var ErrConcurrentTransaction = errors.New("internal: concurrent transaction") 56 57 func RunTransactionOnce(c netcontext.Context, f func(netcontext.Context) error, xg bool) error { 58 if transactionFromContext(c) != nil { 59 return errors.New("nested transactions are not supported") 60 } 61 62 // Begin the transaction. 63 t := &transaction{} 64 req := &pb.BeginTransactionRequest{ 65 App: proto.String(FullyQualifiedAppID(c)), 66 } 67 if xg { 68 req.AllowMultipleEg = proto.Bool(true) 69 } 70 if err := Call(c, "datastore_v3", "BeginTransaction", req, &t.transaction); err != nil { 71 return err 72 } 73 74 // Call f, rolling back the transaction if f returns a non-nil error, or panics. 75 // The panic is not recovered. 76 defer func() { 77 if t.finished { 78 return 79 } 80 t.finished = true 81 // Ignore the error return value, since we are already returning a non-nil 82 // error (or we're panicking). 83 Call(c, "datastore_v3", "Rollback", &t.transaction, &basepb.VoidProto{}) 84 }() 85 if err := f(withTransaction(c, t)); err != nil { 86 return err 87 } 88 t.finished = true 89 90 // Commit the transaction. 91 res := &pb.CommitResponse{} 92 err := Call(c, "datastore_v3", "Commit", &t.transaction, res) 93 if ae, ok := err.(*APIError); ok { 94 /* TODO: restore this conditional 95 if appengine.IsDevAppServer() { 96 */ 97 // The Python Dev AppServer raises an ApplicationError with error code 2 (which is 98 // Error.CONCURRENT_TRANSACTION) and message "Concurrency exception.". 99 if ae.Code == int32(pb.Error_BAD_REQUEST) && ae.Detail == "ApplicationError: 2 Concurrency exception." { 100 return ErrConcurrentTransaction 101 } 102 if ae.Code == int32(pb.Error_CONCURRENT_TRANSACTION) { 103 return ErrConcurrentTransaction 104 } 105 } 106 return err 107 } 108
