Add support for Pull Queues to prod implementation.

impl/memory/taskqueue.go

 // Copyright 2015 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 memory
 
 import (
         "regexp"
+        "time"
 
         "golang.org/x/net/context"
 
         tq "github.com/luci/gae/service/taskqueue"
 
         "github.com/luci/luci-go/common/data/rand/mathrand"
         "github.com/luci/luci-go/common/errors"
 )
 
 /////////////////////////////// public functions ///////////////////////////////
@@ skipping 79 matching lines @@
         if err != nil {
                 return err
         }
 
         for _, task := range tasks {
                 cb(t.deleteLocked(task, queueName))
         }
         return nil
 }
 
+func (t *taskqueueImpl) Lease(maxTasks int, queueName string, leaseTime time.Duration) ([]*tq.Task, error) {
+        panic("not implemented yet")
+}
+
+func (t *taskqueueImpl) LeaseByTag(maxTasks int, queueName string, leaseTime time.Duration, tag string) ([]*tq.Task, error) {
+        panic("not implemented yet")
+}
+
+func (t *taskqueueImpl) ModifyLease(task *tq.Task, queueName string, leaseTime time.Duration) error {
+        panic("not implemented yet")
+}
+
 func (t *taskqueueImpl) Purge(queueName string) error {
         t.Lock()
         defer t.Unlock()
 
         return t.purgeLocked(queueName)
 }
 
 func (t *taskqueueImpl) Stats(queueNames []string, cb tq.RawStatsCB) error {
         t.Lock()
         defer t.Unlock()
@@ skipping 83 matching lines @@
         for _, task := range tasks {
                 cb(t.addLocked(task, queueName))
         }
         return nil
 }
 
 func (t *taskqueueTxnImpl) DeleteMulti([]*tq.Task, string, tq.RawCB) error {
         return errors.New("taskqueue: cannot DeleteMulti from a transaction")
 }
 
+func (t *taskqueueTxnImpl) Lease(maxTasks int, queueName string, leaseTime time.Duration) ([]*tq.Task, error) {
+        return nil, errors.New("taskqueue: cannot Lease from a transaction")
+}
+
+func (t *taskqueueTxnImpl) LeaseByTag(maxTasks int, queueName string, leaseTime time.Duration, tag string) ([]*tq.Task, error) {
+        return nil, errors.New("taskqueue: cannot LeaseByTag from a transaction")
+}
+
+func (t *taskqueueTxnImpl) ModifyLease(task *tq.Task, queueName string, leaseTime time.Duration) error {
+        return errors.New("taskqueue: cannot ModifyLease from a transaction")
+}
+
 func (t *taskqueueTxnImpl) Purge(string) error {
         return errors.New("taskqueue: cannot Purge from a transaction")
 }
 
 func (t *taskqueueTxnImpl) Stats([]string, tq.RawStatsCB) error {
         return errors.New("taskqueue: cannot Stats from a transaction")
 }
 
 func (t *taskqueueTxnImpl) GetTestable() tq.Testable { return t }
 
@@ skipping 19 matching lines @@
 func dupQueue(q tq.QueueData) tq.QueueData {
         r := make(tq.QueueData, len(q))
         for k, q := range q {
                 r[k] = make(map[string]*tq.Task, len(q))
                 for tn, t := range q {
                         r[k][tn] = t.Duplicate()
                 }
         }
         return r
 }