Remove dsQueryBatch in favor of ds.Batcher.

appengine/datastorecache/manager.go

 // Copyright 2016 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 datastorecache
 
 import (
         "bytes"
         "fmt"
         "net/http"
         "strings"
         "sync/atomic"
         "time"
 
         "github.com/luci/luci-go/appengine/memlock"
         "github.com/luci/luci-go/common/clock"
         "github.com/luci/luci-go/common/errors"
         log "github.com/luci/luci-go/common/logging"
         "github.com/luci/luci-go/common/sync/parallel"
         "github.com/luci/luci-go/server/router"
 
-        "github.com/luci/gae/filter/dsQueryBatch"
         "github.com/luci/gae/service/datastore"
         "github.com/luci/gae/service/info"
 
         "github.com/julienschmidt/httprouter"
         "golang.org/x/net/context"
 )
 
-const (
-        // managerQueryBatchSize is the query batch size for cache entries that are
-        // processed by a manager shard.
-        managerQueryBatchSize = 200
-)
-
 func errHTTPHandler(fn func(c context.Context, req *http.Request, params httprouter.Params) error) router.Handler {
         return func(ctx *router.Context) {
                 err := fn(ctx.Context, ctx.Request, ctx.Params)
                 if err == nil {
                         // Handler returned no error, everything is good.
                         return
                 }
 
                 // Handler returned an error, dump it to output.
                 ctx.Writer.WriteHeader(http.StatusInternalServerError)
@@ skipping 273 matching lines @@
                                         } else {
                                                 totalPruned += int(deleteIdx)
                                         }
                                         return nil
                                 }
                         }
                 })
                 return nil
         }
 
-        err := datastore.Run(dsQueryBatch.BatchQueries(c, int32(ms.queryBatchSize), handleEntries), q, func(e *entry) error {
+        b := datastore.Batcher{
+                Size:     ms.queryBatchSize,
+                Callback: handleEntries,
+        }
+        err := b.Run(c, q, func(e *entry) error {
                 totalEntries++
                 ms.observeEntry()
                 entries = append(entries, e)
                 return nil
         })
         if err != nil {
                 return errors.Annotate(err).Reason("failed to run entry query").Err()
         }
 
         // Flush any outstanding entries (ignore error, will always be nil).
         _ = handleEntries(c)
         if totalErrors > 0 {
                 ms.observeErrors(totalErrors)
         }
 
         log.Fields{
                 "entries":   totalEntries,
                 "errors":    totalErrors,
                 "refreshed": totalRefreshed,
                 "pruned":    totalPruned,
         }.Infof(c, "Successfully updated cache entries.")
         return nil
 }