Add cloud datastore implementation.

impl/prod/raw_datastore.go

 // Copyright 2015 The LUCI Authors. All rights reserved.
 // Use of this source code is governed under the Apache License, Version 2.0
 // that can be found in the LICENSE file.
 
 package prod
 
 import (
         ds "github.com/luci/gae/service/datastore"
         "github.com/luci/luci-go/common/errors"
         "golang.org/x/net/context"
@@ skipping 21 matching lines @@
 
 type rdsImpl struct {
         // userCtx is the context that has the luci/gae services and user objects in
         // it.
         userCtx context.Context
 
         // aeCtx is the context with the appengine connection information in it.
         aeCtx context.Context
 }
 
-func idxCallbacker(err error, amt int, cb func(idx int, err error)) error {
+func idxCallbacker(err error, amt int, cb func(idx int, err error) error) error {
         if err == nil {
                 for i := 0; i < amt; i++ {
-                        cb(i, nil)
+                        if err := cb(i, nil); err != nil {
+                                return err
+                        }
                 }
                 return nil
         }
         err = errors.Fix(err)
         me, ok := err.(errors.MultiError)
         if ok {
                 for i, err := range me {
-                        cb(i, err)
+                        if err := cb(i, err); err != nil {
+                                return err
+                        }
                 }
                 return nil
         }
         return err
 }
 
 func (d rdsImpl) AllocateIDs(keys []*ds.Key, cb ds.NewKeyCB) error {
         // Map keys by entity type.
         entityMap := make(map[string][]int)
         for i, key := range keys {
@@ skipping 36 matching lines @@
                 }
         }
         return nil
 }
 
 func (d rdsImpl) DeleteMulti(ks []*ds.Key, cb ds.DeleteMultiCB) error {
         keys, err := dsMF2R(d.aeCtx, ks)
         if err == nil {
                 err = datastore.DeleteMulti(d.aeCtx, keys)
         }
-        return idxCallbacker(err, len(ks), func(_ int, err error) {
-                cb(err)
+        return idxCallbacker(err, len(ks), func(_ int, err error) error {
+                return cb(err)
         })
 }
 
 func (d rdsImpl) GetMulti(keys []*ds.Key, _meta ds.MultiMetaGetter, cb ds.GetMultiCB) error {
         vals := make([]datastore.PropertyLoadSaver, len(keys))
         rkeys, err := dsMF2R(d.aeCtx, keys)
         if err == nil {
                 for i := range keys {
                         vals[i] = &typeFilter{d.aeCtx, ds.PropertyMap{}}
                 }
                 err = datastore.GetMulti(d.aeCtx, rkeys, vals)
         }
-        return idxCallbacker(err, len(keys), func(idx int, err error) {
+        return idxCallbacker(err, len(keys), func(idx int, err error) error {
                 if pls := vals[idx]; pls != nil {
-                        cb(pls.(*typeFilter).pm, err)
-                } else {
-                        cb(nil, err)
+                        return cb(pls.(*typeFilter).pm, err)
                 }
+                return cb(nil, err)
         })
 }
 
 func (d rdsImpl) PutMulti(keys []*ds.Key, vals []ds.PropertyMap, cb ds.NewKeyCB) error {
         rkeys, err := dsMF2R(d.aeCtx, keys)
         if err == nil {
                 rvals := make([]datastore.PropertyLoadSaver, len(vals))
                 for i, val := range vals {
                         rvals[i] = &typeFilter{d.aeCtx, val}
                 }
                 rkeys, err = datastore.PutMulti(d.aeCtx, rkeys, rvals)
         }
-        return idxCallbacker(err, len(keys), func(idx int, err error) {
+        return idxCallbacker(err, len(keys), func(idx int, err error) error {
                 k := (*ds.Key)(nil)
                 if err == nil {
                         k = dsR2F(rkeys[idx])
                 }
-                cb(k, err)
+                return cb(k, err)
         })
 }
 
 func (d rdsImpl) fixQuery(fq *ds.FinalizedQuery) (*datastore.Query, error) {
         ret := datastore.NewQuery(fq.Kind())
 
         start, end := fq.Bounds()
         if start != nil {
                 ret = ret.Start(start.(datastore.Cursor))
         }
@@ skipping 107 matching lines @@
 func (d rdsImpl) RunInTransaction(f func(c context.Context) error, opts *ds.TransactionOptions) error {
         ropts := (*datastore.TransactionOptions)(opts)
         return datastore.RunInTransaction(d.aeCtx, func(c context.Context) error {
                 return f(context.WithValue(d.userCtx, prodContextKey, c))
         }, ropts)
 }
 
 func (d rdsImpl) Testable() ds.Testable {
         return nil
 }