Add context-aware "time" library wrapper.

go/src/infra/gae/libs/wrapper/memory/taskqueue_data.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 (
         "errors"
         "fmt"
         "infra/gae/libs/wrapper"
-        "math/rand"
         "net/http"
         "sync"
         "sync/atomic"
-        "time"
 
         "appengine/datastore"
         "appengine/taskqueue"
         pb "appengine_internal/taskqueue"
+        "golang.org/x/net/context"
+        "infra/libs/clock"
 )
 
 var (
         currentNamespace = http.CanonicalHeaderKey("X-AppEngine-Current-Namespace")
         defaultNamespace = http.CanonicalHeaderKey("X-AppEngine-Default-Namespace")
 )
 
 //////////////////////////////// taskQueueData /////////////////////////////////
 
 type taskQueueData struct {
@@ skipping 13 matching lines @@
         return &taskQueueData{
                 BrokenFeatures: wrapper.BrokenFeatures{
                         DefaultError: newTQError(pb.TaskQueueServiceError_TRANSIENT_ERROR)},
                 named:    wrapper.QueueData{"default": {}},
                 archived: wrapper.QueueData{"default": {}},
         }
 }
 
 func (t *taskQueueData) canApplyTxn(obj memContextObj) bool { return true }
 func (t *taskQueueData) endTxn()                            {}
-func (t *taskQueueData) applyTxn(rnd *rand.Rand, obj memContextObj) {
+func (t *taskQueueData) applyTxn(c context.Context, obj memContextObj) {
         txn := obj.(*txnTaskQueueData)
         for qn, tasks := range txn.anony {
                 for _, tsk := range tasks {
-                        tsk.Name = mkName(rnd, tsk.Name, t.named[qn])
+                        tsk.Name = mkName(c, tsk.Name, t.named[qn])
                         t.named[qn][tsk.Name] = tsk
                 }
         }
         txn.anony = nil
 }
 func (t *taskQueueData) mkTxn(*datastore.TransactionOptions) (memContextObj, error) {
         return &txnTaskQueueData{
                 BrokenFeatures: &t.BrokenFeatures,
                 parent:         t,
                 anony:          wrapper.AnonymousQueueData{},
@@ skipping 46 matching lines @@
 func (t *taskQueueData) getQueueName(queueName string) (string, error) {
         if queueName == "" {
                 queueName = "default"
         }
         if _, ok := t.named[queueName]; !ok {
                 return "", newTQError(pb.TaskQueueServiceError_UNKNOWN_QUEUE)
         }
         return queueName, nil
 }
 
-func (t *taskQueueData) prepTask(ns string, task *taskqueue.Task, queueName string, now time.Time, rnd *rand.Rand) (*taskqueue.Task, string, error) {
+func (t *taskQueueData) prepTask(c context.Context, ns string, task *taskqueue.Task, queueName string) (
+        *taskqueue.Task, string, error) {
         queueName, err := t.getQueueName(queueName)
         if err != nil {
                 return nil, "", err
         }
 
         toSched := dupTask(task)
 
         if toSched.Path == "" {
                 return nil, "", newTQError(pb.TaskQueueServiceError_INVALID_URL)
         }
 
         if toSched.ETA.IsZero() {
-                toSched.ETA = now.Add(toSched.Delay)
+                toSched.ETA = clock.Now(c).Add(toSched.Delay)
         } else if toSched.Delay != 0 {
                 panic("taskqueue: both Delay and ETA are set")
         }
         toSched.Delay = 0
 
         if toSched.Method == "" {
                 toSched.Method = "POST"
         }
         if _, ok := pb.TaskQueueAddRequest_RequestMethod_value[toSched.Method]; !ok {
                 return nil, "", fmt.Errorf("taskqueue: bad method %q", toSched.Method)
         }
         if toSched.Method != "POST" && toSched.Method != "PUT" {
                 toSched.Payload = nil
         }
 
         if _, ok := toSched.Header[currentNamespace]; !ok {
                 if ns != "" {
                         if toSched.Header == nil {
                                 toSched.Header = http.Header{}
                         }
                         toSched.Header[currentNamespace] = []string{ns}
                 }
         }
         // TODO(riannucci): implement DefaultNamespace
 
         if toSched.Name == "" {
-                toSched.Name = mkName(rnd, "", t.named[queueName])
+                toSched.Name = mkName(c, "", t.named[queueName])
         } else {
                 if !validTaskName.MatchString(toSched.Name) {
                         return nil, "", newTQError(pb.TaskQueueServiceError_INVALID_TASK_NAME)
                 }
         }
 
         return toSched, queueName, nil
 }
 
 /////////////////////////////// txnTaskQueueData ///////////////////////////////
 
 type txnTaskQueueData struct {
         *wrapper.BrokenFeatures
 
         lock sync.Mutex
 
         // boolean 0 or 1, use atomic.*Int32 to access.
         closed int32
         anony  wrapper.AnonymousQueueData
         parent *taskQueueData
 }
 
 var (
         _ = memContextObj((*txnTaskQueueData)(nil))
         _ = wrapper.TQTestable((*txnTaskQueueData)(nil))
 )
 
 func (t *txnTaskQueueData) canApplyTxn(obj memContextObj) bool { return false }
 
-func (t *txnTaskQueueData) applyTxn(*rand.Rand, memContextObj) {
+func (t *txnTaskQueueData) applyTxn(context.Context, memContextObj) {
         panic(errors.New("txnTaskQueueData.applyTxn is not implemented"))
 }
 
 func (t *txnTaskQueueData) mkTxn(*datastore.TransactionOptions) (memContextObj, error) {
         return nil, errors.New("txnTaskQueueData.mkTxn is not implemented")
 }
 
 func (t *txnTaskQueueData) endTxn() {
         if atomic.LoadInt32(&t.closed) == 1 {
                 panic("cannot end transaction twice")
@@ skipping 50 matching lines @@
         return t.parent.GetTombstonedTasks()
 }
 
 func (t *txnTaskQueueData) GetScheduledTasks() wrapper.QueueData {
         return t.parent.GetScheduledTasks()
 }
 
 func (t *txnTaskQueueData) CreateQueue(queueName string) {
         t.parent.CreateQueue(queueName)
 }