Add support for Pull Queues to prod implementation.

impl/prod/taskqueue.go

Index: impl/prod/taskqueue.go
diff --git a/impl/prod/taskqueue.go b/impl/prod/taskqueue.go
index 6b7c1a5fc23664a2c162059a75c43e920944da53..758fb2b45107d912068c8f9a81fe7b8cb1279a03 100644
--- a/impl/prod/taskqueue.go
+++ b/impl/prod/taskqueue.go
@@ -7,6 +7,7 @@ package prod
 import (
 	"fmt"
 	"reflect"
+	"time"
 
 	tq "github.com/luci/gae/service/taskqueue"
 	"golang.org/x/net/context"
@@ -85,6 +86,15 @@ func tqMF2R(ns []*tq.Task) []*taskqueue.Task {
 	return ret
 }
 
+// tqMR2F (TQ multi-real-to-fake) converts []*taskqueue.Task to []*tq.Task.
+func tqMR2F(ns []*taskqueue.Task) []*tq.Task {
+	ret := make([]*tq.Task, len(ns))
+	for i, t := range ns {
+		ret[i] = tqR2F(t)
+	}
+	return ret
+}
+
 func (t tqImpl) AddMulti(tasks []*tq.Task, queueName string, cb tq.RawTaskCB) error {
 	realTasks, err := taskqueue.AddMulti(t.aeCtx, tqMF2R(tasks), queueName)
 	if err != nil {
@@ -117,6 +127,26 @@ func (t tqImpl) DeleteMulti(tasks []*tq.Task, queueName string, cb tq.RawCB) err
 	return err
 }
 
+func (t tqImpl) Lease(maxTasks int, queueName string, leaseTime time.Duration) ([]*tq.Task, error) {
+	tasks, err := taskqueue.Lease(t.aeCtx, maxTasks, queueName, int(leaseTime/time.Second))
+	if err != nil {
+		return nil, err
+	}
+	return tqMR2F(tasks), nil
+}
+
+func (t tqImpl) LeaseByTag(maxTasks int, queueName string, leaseTime time.Duration, tag string) ([]*tq.Task, error) {
+	tasks, err := taskqueue.LeaseByTag(t.aeCtx, maxTasks, queueName, int(leaseTime/time.Second), tag)
+	if err != nil {
+		return nil, err
+	}
+	return tqMR2F(tasks), nil
+}
+
+func (t tqImpl) ModifyLease(task *tq.Task, queueName string, leaseTime time.Duration) error {
+	return taskqueue.ModifyLease(t.aeCtx, tqF2R(task), queueName, int(leaseTime/time.Second))
+}
+
 func (t tqImpl) Purge(queueName string) error {
 	return taskqueue.Purge(t.aeCtx, queueName)
 }