Clean up callback interfaces.

filter/txnBuf/query_merger.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"
         "sort"
 
         "github.com/luci/gae/impl/memory"
         ds "github.com/luci/gae/service/datastore"
         "github.com/luci/gae/service/datastore/serialize"
         "github.com/luci/luci-go/common/stringset"
 )
 
 // queryToIter takes a FinalizedQuery and returns an iterator function which
 // will produce either *items or errors.
 //
 //  - d is the raw datastore to run this query on
 //  - filter is a function which will return true if the given key should be
 //    excluded from the result set.
 func queryToIter(stopChan chan struct{}, fq *ds.FinalizedQuery, d ds.RawInterface) func() (*item, error) {
         c := make(chan *item)
 
         go func() {
                 defer close(c)
 
-                err := d.Run(fq, func(k *ds.Key, pm ds.PropertyMap, _ ds.CursorCB) (keepGoing bool) {
+                err := d.Run(fq, func(k *ds.Key, pm ds.PropertyMap, _ ds.CursorCB) error {
                         i := &item{key: k, data: pm}
                         select {
                         case c <- i:
-                                return true
+                                return nil
                         case <-stopChan:
-                                return false
+                                return ds.Stop
                         }
                 })
                 if err != nil {
                         c <- &item{err: err}
                 }
         }()
 
         return func() (*item, error) {
                 itm := <-c
                 if itm == nil {
@@ skipping 62 matching lines @@
 // runMergedQueries executes a user query `fq` against the parent datastore as
 // well as the in-memory datastore, calling `cb` with the merged result set.
 //
 // It's expected that the caller of this function will apply limit and offset
 // if the query contains those restrictions. This may convert the query to
 // an expanded projection query with more data than the user asked for. It's the
 // caller's responsibility to prune away the extra data.
 //
 // See also `dsTxnBuf.Run()`.
 func runMergedQueries(fq *ds.FinalizedQuery, sizes *sizeTracker,
-        memDS, parentDS ds.RawInterface, cb func(k *ds.Key, data ds.PropertyMap) bool) error {
+        memDS, parentDS ds.RawInterface, cb func(k *ds.Key, data ds.PropertyMap) error) error {
 
         toRun, err := adjustQuery(fq)
         if err != nil {
                 return err
         }
 
         cmpLower, cmpUpper := memory.GetBinaryBounds(fq)
         cmpOrder := fq.Orders()
         cmpFn := func(i *item) string {
                 return i.getCmpRow(cmpLower, cmpUpper, cmpOrder)
@@ skipping 98 matching lines @@
                 }
                 if distinct != nil {
                         // NOTE: We know that toUse will not be used after this point for
                         // comparison purposes, so re-use its cmpRow property for our distinct
                         // filter here.
                         toUse.cmpRow = ""
                         if !distinct.Add(toUse.getCmpRow(nil, nil, distinctOrder)) {
                                 continue
                         }
                 }
-                if !cb(toUse.key, toUse.data) {
-                        break
+                if err := cb(toUse.key, toUse.data); err != nil {
+                        if err == ds.Stop {
+                                return nil
+                        }
+                        return err
                 }
         }
 
         return nil
 }
 
 // toComparableString computes the byte-sortable 'order' string for the given
 // key/PropertyMap.
 //
 //   * start/end are byte sequences which are the inequality bounds of the
@@ skipping 40 matching lines @@
                 }
                 if !foundOne {
                         return nil, nil
                 }
         }
         if end != nil && bytes.Compare(soFar, end) >= 0 {
                 return nil, nil
         }
         return soFar, ps["__key__"][0]
 }