Clean up callback interfaces.

impl/memory/datastore_data.go

 // Copyright 2015 The Chromium Authors. All rights reserved.
 // Use of this source code is governed by a BSD-style license that can be
 // found in the LICENSE file.
 
 package memory
 
 import (
         "bytes"
         "fmt"
         "sync"
@@ skipping 231 matching lines @@
         if key.Incomplete() {
                 id, err := d.allocateIDsLocked(ents, key, 1)
                 if err != nil {
                         return key, err
                 }
                 key = ds.NewKey(key.AppID(), key.Namespace(), key.Kind(), "", id, key.Parent())
         }
         return key, nil
 }
 
-func (d *dataStoreData) putMulti(keys []*ds.Key, vals []ds.PropertyMap, cb ds.PutMultiCB) {
+func (d *dataStoreData) putMulti(keys []*ds.Key, vals []ds.PropertyMap, cb ds.PutMultiCB) error {
         ns := keys[0].Namespace()
 
         for i, k := range keys {
                 pmap, _ := vals[i].Save(false)
                 dataBytes := serialize.ToBytes(pmap)
 
                 k, err := func() (ret *ds.Key, err error) {
                         d.Lock()
                         defer d.Unlock()
 
@@ skipping 12 matching lines @@
                         if old != nil {
                                 if oldPM, err = rpm(old); err != nil {
                                         return
                                 }
                         }
                         ents.Set(keyBytes(ret), dataBytes)
                         updateIndexes(d.head, ret, oldPM, pmap)
                         return
                 }()
                 if cb != nil {
-                        cb(k, err)
+                        if err := cb(k, err); err != nil {
+                                if err == ds.Stop {
+                                        return nil
+                                }
+                                return err
+                        }
                 }
         }
+        return nil
 }
 
 func getMultiInner(keys []*ds.Key, cb ds.GetMultiCB, getColl func() (*memCollection, error)) error {
         ents, err := getColl()
         if err != nil {
                 return err
         }
         if ents == nil {
                 for range keys {
                         cb(nil, ds.ErrNoSuchEntity)
@@ skipping 13 matching lines @@
 }
 
 func (d *dataStoreData) getMulti(keys []*ds.Key, cb ds.GetMultiCB) error {
         return getMultiInner(keys, cb, func() (*memCollection, error) {
                 s := d.takeSnapshot()
 
                 return s.GetCollection("ents:" + keys[0].Namespace()), nil
         })
 }
 
-func (d *dataStoreData) delMulti(keys []*ds.Key, cb ds.DeleteMultiCB) {
+func (d *dataStoreData) delMulti(keys []*ds.Key, cb ds.DeleteMultiCB) error {
         ns := keys[0].Namespace()
 
         hasEntsInNS := func() bool {
                 d.Lock()
                 defer d.Unlock()
                 return d.mutableEntsLocked(ns) != nil
         }()
 
         if hasEntsInNS {
                 for _, k := range keys {
@@ skipping 12 matching lines @@
                                         oldPM, err := rpm(old)
                                         if err != nil {
                                                 return err
                                         }
                                         ents.Delete(kb)
                                         updateIndexes(d.head, k, oldPM, nil)
                                 }
                                 return nil
                         }()
                         if cb != nil {
-                                cb(err)
+                                if err := cb(err); err != nil {
+                                        if err == ds.Stop {
+                                                return nil
+                                        }
+                                        return err
+                                }
                         }
                 }
         } else if cb != nil {
                 for range keys {
-                        cb(nil)
+                        if err := cb(nil); err != nil {
+                                if err == ds.Stop {
+                                        return nil
+                                }
+                                return err
+                        }
                 }
         }
+        return nil
 }
 
 func (d *dataStoreData) canApplyTxn(obj memContextObj) bool {
         // TODO(riannucci): implement with Flush/FlushRevert for persistance.
 
         txn := obj.(*txnDataStoreData)
         for rk, muts := range txn.muts {
                 if len(muts) == 0 { // read-only
                         continue
                 }
@@ skipping 16 matching lines @@
 }
 
 func (d *dataStoreData) applyTxn(c context.Context, obj memContextObj) {
         txn := obj.(*txnDataStoreData)
         for _, muts := range txn.muts {
                 if len(muts) == 0 { // read-only
                         continue
                 }
                 // TODO(riannucci): refactor to do just 1 putMulti, and 1 delMulti
                 for _, m := range muts {
-                        err := error(nil)
                         k := m.key
                         if m.data == nil {
-                                d.delMulti([]*ds.Key{k},
-                                        func(e error) { err = e })
+                                impossible(d.delMulti([]*ds.Key{k},
+                                        func(e error) error { return e }))
                         } else {
-                                d.putMulti([]*ds.Key{m.key}, []ds.PropertyMap{m.data},
-                                        func(_ *ds.Key, e error) { err = e })
+                                impossible(d.putMulti([]*ds.Key{m.key}, []ds.PropertyMap{m.data},
+                                        func(_ *ds.Key, e error) error { return e }))
                         }
-                        impossible(err)
                 }
         }
 }
 
 func (d *dataStoreData) mkTxn(o *ds.TransactionOptions) memContextObj {
         return &txnDataStoreData{
                 // alias to the main datastore's so that testing code can have primitive
                 // access to break features inside of transactions.
                 parent: d,
                 isXG:   o != nil && o.XG,
@@ skipping 139 matching lines @@
 }
 
 func keyBytes(key *ds.Key) []byte {
         return serialize.ToBytes(ds.MkProperty(key))
 }
 
 func rpm(data []byte) (ds.PropertyMap, error) {
         return serialize.ReadPropertyMap(bytes.NewBuffer(data),
                 serialize.WithContext, "", "")
 }