Fix missing/broken ds.Stop handling in raw DS.

filter/txnBuf/state.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 txnBuf
 
 import (
         "bytes"
         "sync"
 
@@ skipping 298 matching lines @@
                         }
                 }
                 return nil
         }()
         if err != nil {
                 return err
         }
 
         for i, itm := range data {
                 err := lme.GetOne(i)
+                var cbErr error
                 if err != nil {
-                        cb(nil, err)
+                        cbErr = cb(nil, err)
                 } else if itm.data == nil {
-                        cb(nil, datastore.ErrNoSuchEntity)
+                        cbErr = cb(nil, datastore.ErrNoSuchEntity)
                 } else {
-                        cb(itm.data, nil)
+                        cbErr = cb(itm.data, nil)
+                }
+                if cbErr != nil {
+                        return cbErr
                 }
         }
         return nil
 }
 
 func (t *txnBufState) deleteMulti(keys []*datastore.Key, cb datastore.DeleteMultiCB, haveLock bool) error {
         encKeys, roots := toEncoded(keys)
 
         err := func() error {
                 if !haveLock {
@@ skipping 13 matching lines @@
                         return nil
                 })
                 impossible(err)
                 return nil
         }()
         if err != nil {
                 return err
         }
 
         for range keys {
-                cb(nil)
+                if err := cb(nil); err != nil {
+                        return err
+                }
         }
 
         return nil
 }
 
 func (t *txnBufState) fixKeys(keys []*datastore.Key) ([]*datastore.Key, error) {
         lme := errors.NewLazyMultiError(len(keys))
         realKeys := []*datastore.Key(nil)
         for i, key := range keys {
                 if key.Incomplete() {
@@ skipping 16 matching lines @@
         if realKeys != nil {
                 return realKeys, err
         }
         return keys, err
 }
 
 func (t *txnBufState) putMulti(keys []*datastore.Key, vals []datastore.PropertyMap, cb datastore.PutMultiCB, haveLock bool) error {
         keys, err := t.fixKeys(keys)
         if err != nil {
                 for _, e := range err.(errors.MultiError) {
-                        cb(nil, e)
+                        if err := cb(nil, e); err != nil {
+                                return err
+                        }
                 }
                 return nil
         }
 
         encKeys, roots := toEncoded(keys)
 
         err = func() error {
                 if !haveLock {
                         t.Lock()
                         defer t.Unlock()
@@ skipping 11 matching lines @@
                         return nil
                 })
                 impossible(err)
                 return nil
         }()
         if err != nil {
                 return err
         }
 
         for _, k := range keys {
-                cb(k, nil)
+                if err := cb(k, nil); err != nil {
+                        return err
+                }
         }
         return nil
 }
 
 func commitToReal(s *txnBufState) error {
         toPut, toPutKeys, toDel := s.effect()
 
         return parallel.FanOutIn(func(ch chan<- func() error) {
                 if len(toPut) > 0 {
                         ch <- func() error {
@@ skipping 99 matching lines @@
 // plus a stringset of all the encoded root keys that `keys` represents.
 func toEncoded(keys []*datastore.Key) (full []string, roots stringset.Set) {
         roots = stringset.New(len(keys))
         full = make([]string, len(keys))
         for i, k := range keys {
                 roots.Add(string(serialize.ToBytes(k.Root())))
                 full[i] = string(serialize.ToBytes(k))
         }
         return
 }