Make the rest of the services have a similar raw/user interface structure.

impl/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"
         "net/http"
@@ skipping 88 matching lines @@
         }
 }
 
 func (t *taskQueueData) ResetTasks() {
         t.Lock()
         defer t.Unlock()
 
         t.resetTasksWithLock()
 }
 
-func (t *taskQueueData) getQueueName(queueName string) (string, error) {
+func (t *taskQueueData) getQueueNameLocked(queueName string) (string, error) {
         if queueName == "" {
                 queueName = "default"
         }
         if _, ok := t.named[queueName]; !ok {
                 return "", errors.New("UNKNOWN_QUEUE")
         }
         return queueName, nil
 }
 
+func (t *taskQueueData) purgeLocked(queueName string) error {
+        queueName, err := t.getQueueNameLocked(queueName)
+        if err != nil {
+                return err
+        }
+
+        t.named[queueName] = map[string]*tq.Task{}
+        t.archived[queueName] = map[string]*tq.Task{}
+        return nil
+}
+
 var tqOkMethods = map[string]struct{}{
         "GET":    {},
         "POST":   {},
         "HEAD":   {},
         "PUT":    {},
         "DELETE": {},
 }
 
-func (t *taskQueueData) prepTask(c context.Context, ns string, task *tq.Task, queueName string) (*tq.Task, string, error) {
-        queueName, err := t.getQueueName(queueName)
-        if err != nil {
-                return nil, "", err
-        }
-
-        toSched := dupTask(task)
+func (t *taskQueueData) prepTask(c context.Context, ns string, task *tq.Task, queueName string) (*tq.Task, error) {
+        toSched := task.Duplicate()
 
         if toSched.Path == "" {
                 toSched.Path = "/_ah/queue/" + queueName
         }
 
         if toSched.ETA.IsZero() {
                 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 := tqOkMethods[toSched.Method]; !ok {
-                return nil, "", fmt.Errorf("taskqueue: bad method %q", toSched.Method)
+                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(c, "", t.named[queueName])
         } else {
                 if !validTaskName.MatchString(toSched.Name) {
-                        return nil, "", errors.New("INVALID_TASK_NAME")
+                        return nil, errors.New("INVALID_TASK_NAME")
                 }
         }
 
-        return toSched, queueName, nil
+        return toSched, nil
 }
 
 /////////////////////////////// txnTaskQueueData ///////////////////////////////
 
 type txnTaskQueueData struct {
         lock sync.Mutex
 
         // boolean 0 or 1, use atomic.*Int32 to access.
         closed int32
         anony  tq.AnonymousQueueData
         parent *taskQueueData
 }
 
 var (
         _ = memContextObj((*txnTaskQueueData)(nil))
         _ = tq.Testable((*txnTaskQueueData)(nil))
 )
 
 func (t *txnTaskQueueData) canApplyTxn(obj memContextObj) bool         { return false }
 func (t *txnTaskQueueData) applyTxn(context.Context, memContextObj)    { panic("impossible") }
 func (t *txnTaskQueueData) mkTxn(*ds.TransactionOptions) memContextObj { panic("impossible") }
 
 func (t *txnTaskQueueData) endTxn() {
         if atomic.LoadInt32(&t.closed) == 1 {
                 panic("cannot end transaction twice")
         }
         atomic.StoreInt32(&t.closed, 1)
 }
 
-func (t *txnTaskQueueData) run(f func() error) error {
-        // Slightly different from the SDK... datastore and taskqueue each implement
-        // this here, where in the SDK only datastore.transaction.Call does.
-        if atomic.LoadInt32(&t.closed) == 1 {
-                return fmt.Errorf("taskqueue: transaction context has expired")
-        }
-        return f()
-}
-
 func (t *txnTaskQueueData) ResetTasks() {
         t.Lock()
         defer t.Unlock()
 
         for queuename := range t.anony {
                 t.anony[queuename] = nil
         }
         t.parent.resetTasksWithLock()
 }
 
 func (t *txnTaskQueueData) Lock() {
         t.lock.Lock()
         t.parent.Lock()
 }
 func (t *txnTaskQueueData) Unlock() {
         t.parent.Unlock()
         t.lock.Unlock()
 }
 
 func (t *txnTaskQueueData) GetTransactionTasks() tq.AnonymousQueueData {
         t.Lock()
         defer t.Unlock()
 
         ret := make(tq.AnonymousQueueData, len(t.anony))
         for k, vs := range t.anony {
                 ret[k] = make([]*tq.Task, len(vs))
                 for i, v := range vs {
-                        tsk := dupTask(v)
+                        tsk := v.Duplicate()
                         tsk.Name = ""
                         ret[k][i] = tsk
                 }
         }
 
         return ret
 }
 
 func (t *txnTaskQueueData) GetTombstonedTasks() tq.QueueData {
         return t.parent.GetTombstonedTasks()
 }
 
 func (t *txnTaskQueueData) GetScheduledTasks() tq.QueueData {
         return t.parent.GetScheduledTasks()
 }
 
 func (t *txnTaskQueueData) CreateQueue(queueName string) {
         t.parent.CreateQueue(queueName)
 }