Clean up callback interfaces.

impl/memory/datastore_query_execution.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"
 
         ds "github.com/luci/gae/service/datastore"
         "github.com/luci/gae/service/datastore/serialize"
         "github.com/luci/luci-go/common/cmpbin"
         "github.com/luci/luci-go/common/stringset"
 )
 
 type queryStrategy interface {
         // handle applies the strategy to the embedded user callback.
         //   - rawData is the slice of encoded Properties from the index row
         //     (correctly de-inverted).
         //   - decodedProps is the slice of decoded Properties from the index row
         //   - key is the decoded Key from the index row (the last item in rawData and
         //     decodedProps)
         //   - gc is the getCursor function to be passed to the user's callback
-        handle(rawData [][]byte, decodedProps []ds.Property, key *ds.Key, gc func() (ds.Cursor, error)) bool
+        handle(rawData [][]byte, decodedProps []ds.Property, key *ds.Key, gc func() (ds.Cursor, error)) error
 }
 
 type projectionLookup struct {
         suffixIndex  int
         propertyName string
 }
 
 type projectionStrategy struct {
         cb ds.RawRunCB
 
@@ skipping 17 matching lines @@
                 }
                 impossible(lookupErr)
         }
         ret := &projectionStrategy{cb: cb, project: projectionLookups}
         if fq.Distinct() {
                 ret.distinct = stringset.New(0)
         }
         return ret
 }
 
-func (s *projectionStrategy) handle(rawData [][]byte, decodedProps []ds.Property, key *ds.Key, gc func() (ds.Cursor, error)) bool {
+func (s *projectionStrategy) handle(rawData [][]byte, decodedProps []ds.Property, key *ds.Key, gc func() (ds.Cursor, error)) error {
         projectedRaw := [][]byte(nil)
         if s.distinct != nil {
                 projectedRaw = make([][]byte, len(decodedProps))
         }
         pmap := make(ds.PropertyMap, len(s.project))
         for i, p := range s.project {
                 if s.distinct != nil {
                         projectedRaw[i] = rawData[p.suffixIndex]
                 }
                 pmap[p.propertyName] = []ds.Property{decodedProps[p.suffixIndex]}
         }
         if s.distinct != nil {
                 if !s.distinct.Add(string(serialize.Join(projectedRaw...))) {
-                        return true
+                        return nil
                 }
         }
         return s.cb(key, pmap, gc)
 }
 
 type keysOnlyStrategy struct {
         cb ds.RawRunCB
 
         dedup stringset.Set
 }
 
-func (s *keysOnlyStrategy) handle(rawData [][]byte, _ []ds.Property, key *ds.Key, gc func() (ds.Cursor, error)) bool {
+func (s *keysOnlyStrategy) handle(rawData [][]byte, _ []ds.Property, key *ds.Key, gc func() (ds.Cursor, error)) error {
         if !s.dedup.Add(string(rawData[len(rawData)-1])) {
-                return true
+                return nil
         }
         return s.cb(key, nil, gc)
 }
 
 type normalStrategy struct {
         cb ds.RawRunCB
 
         aid   string
         ns    string
         head  *memCollection
         dedup stringset.Set
 }
 
 func newNormalStrategy(aid, ns string, cb ds.RawRunCB, head *memStore) queryStrategy {
         coll := head.GetCollection("ents:" + ns)
         if coll == nil {
                 return nil
         }
         return &normalStrategy{cb, aid, ns, coll, stringset.New(0)}
 }
 
-func (s *normalStrategy) handle(rawData [][]byte, _ []ds.Property, key *ds.Key, gc func() (ds.Cursor, error)) bool {
+func (s *normalStrategy) handle(rawData [][]byte, _ []ds.Property, key *ds.Key, gc func() (ds.Cursor, error)) error {
         rawKey := rawData[len(rawData)-1]
         if !s.dedup.Add(string(rawKey)) {
-                return true
+                return nil
         }
 
         rawEnt := s.head.Get(rawKey)
         if rawEnt == nil {
                 // entity doesn't exist at head
-                return true
+                return nil
         }
         pm, err := serialize.ReadPropertyMap(bytes.NewBuffer(rawEnt), serialize.WithoutContext, s.aid, s.ns)
         memoryCorruption(err)
 
         return s.cb(key, pm, gc)
 }
 
 func pickQueryStrategy(fq *ds.FinalizedQuery, rq *reducedQuery, cb ds.RawRunCB, head *memStore) queryStrategy {
         if fq.KeysOnly() {
                 return &keysOnlyStrategy{cb, stringset.New(0)}
@@ skipping 31 matching lines @@
         return
 }
 
 func countQuery(fq *ds.FinalizedQuery, aid, ns string, isTxn bool, idx, head *memStore) (ret int64, err error) {
         if len(fq.Project()) == 0 && !fq.KeysOnly() {
                 fq, err = fq.Original().KeysOnly(true).Finalize()
                 if err != nil {
                         return
                 }
         }
-        err = executeQuery(fq, aid, ns, isTxn, idx, head, func(_ *ds.Key, _ ds.PropertyMap, _ ds.CursorCB) bool {
+        err = executeQuery(fq, aid, ns, isTxn, idx, head, func(_ *ds.Key, _ ds.PropertyMap, _ ds.CursorCB) error {
                 ret++
-                return true
+                return nil
         })
         return
 }
 
 func executeQuery(fq *ds.FinalizedQuery, aid, ns string, isTxn bool, idx, head *memStore, cb ds.RawRunCB) error {
         rq, err := reduce(fq, aid, ns, isTxn)
         if err == ds.ErrNullQuery {
                 return nil
         }
         if err != nil {
@@ skipping 31 matching lines @@
                                         memoryCorruption(err)
                                 }
                                 cursorPrefix = buf.Bytes()
                         }
                         // TODO(riannucci): Do we need to decrement suffix instead of increment
                         // if we're sorting by __key__ DESCENDING?
                         return queryCursor(serialize.Join(cursorPrefix, increment(suffix))), nil
                 }
         }
 
-        multiIterate(idxs, func(suffix []byte) bool {
+        return multiIterate(idxs, func(suffix []byte) error {
                 if offset > 0 {
                         offset--
-                        return true
+                        return nil
                 }
                 if hasLimit {
                         if limit <= 0 {
-                                return false
+                                return ds.Stop
                         }
                         limit--
                 }
 
                 rawData, decodedProps := parseSuffix(aid, ns, rq.suffixFormat, suffix, -1)
 
                 keyProp := decodedProps[len(decodedProps)-1]
                 if keyProp.Type() != ds.PTKey {
                         impossible(fmt.Errorf("decoded index row doesn't end with a Key: %#v", keyProp))
                 }
 
                 return strategy.handle(
                         rawData, decodedProps, keyProp.Value().(*ds.Key),
                         getCursorFn(suffix))
         })
-
-        return nil
 }