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..f980d67df1cfe633886856cd34df8445112bec3c 100644
--- a/impl/prod/taskqueue.go
+++ b/impl/prod/taskqueue.go
@@ -85,6 +85,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 +126,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 int) ([]*tq.Task, error) {
+	tasks, err := taskqueue.Lease(t.aeCtx, maxTasks, queueName, leaseTime)
+	if err != nil {
+		return nil, err
+	}
+	return tqMR2F(tasks), nil
+}
+
+func (t tqImpl) LeaseByTag(maxTasks int, queueName string, leaseTime int, tag string) ([]*tq.Task, error) {
+	tasks, err := taskqueue.LeaseByTag(t.aeCtx, maxTasks, queueName, leaseTime, tag)
+	if err != nil {
+		return nil, err
+	}
+	return tqMR2F(tasks), nil
+}
+
+func (t tqImpl) ModifyLease(task *tq.Task, queueName string, leaseTime int) error {
+	return taskqueue.ModifyLease(t.aeCtx, tqF2R(task), queueName, leaseTime)
+}
+
 func (t tqImpl) Purge(queueName string) error {
 	return taskqueue.Purge(t.aeCtx, queueName)
 }