Add a task queue task router to reduce amount of boilerplate.

scheduler/appengine/engine/tq/tq.go

+// Copyright 2017 The LUCI Authors.
+//
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+//      http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+// Package tq implements simple routing layer for task queue tasks.
+package tq
+
+import (
+        "bytes"
+        "encoding/json"
+        "fmt"
+        "io/ioutil"
+        "net/http"
+        "reflect"
+        "strconv"
+        "strings"
+        "sync"
+        "time"
+
+        "github.com/golang/protobuf/jsonpb"
+        "github.com/golang/protobuf/proto"
+        "golang.org/x/net/context"
+
+        "github.com/luci/gae/service/taskqueue"
+
+        "github.com/luci/luci-go/appengine/gaemiddleware"
+        "github.com/luci/luci-go/common/data/stringset"
+        "github.com/luci/luci-go/common/logging"
+        "github.com/luci/luci-go/common/retry/transient"
+        "github.com/luci/luci-go/server/router"
+)
+
+// Dispatcher submits and handles task queue tasks.
+type Dispatcher struct {
+        BaseURL string // URL prefix for all URLs, "/internal/tasks/" by default
+
+        mu       sync.RWMutex
+        handlers map[string]handler // the key is proto message type name
+}
+
+// Task contains task body and additional parameters that influence how it is
+// routed.
+type Task struct {
+        // Payload is task's payload as well as indicator of its type.
+        //
+        // Tasks are routed based on type of the payload message, see RegisterTask.
+        Payload proto.Message
+
+        // Title is optional string that identifies the task in HTTP logs.
+        //
+        // It will show up as a suffix in task handler URL. It exists exclusively to
+        // simplify reading HTTP logs. It serves no other purpose! In particular,
+        // it is NOT a task name.
+        //
+        // Handlers won't ever see it. Pass all information through the task body.
+        Title string
+
+        // Delay specifies the duration the task queue service must wait before
+        // executing the task.
+        //
+        // Either Delay or ETA may be set, but not both.
+        Delay time.Duration
+
+        // ETA specifies the earliest time a task may be executed.
+        //
+        // Either Delay or ETA may be set, but not both.
+        ETA time.Time
+
+        // Retry options for this task.
+        //
+        // If given, overrides default options set when this task was registered.
+        RetryOptions *taskqueue.RetryOptions
+}
+
+// Handler is called to handle one enqueued task.
+//
+// The passed context is produced by a middleware chain installed with
+// InstallHandlers.
+//
+// execCount corresponds to X-AppEngine-TaskExecutionCount header value: it is
+// 1 on first execution attempt, 2 on a retry, and so on.
+//
+// May return transient errors. In this case, task queue may attempt to
+// redeliver the task (depending on RetryOptions).
+//
+// A fatal error (or success) mark the task as "done", it won't be retried.
+type Handler func(c context.Context, payload proto.Message, execCount int) error
+
+// RegisterTask tells the dispatcher that tasks of given proto type should be
+// handled by the given handler and routed through the given task queue.
+//
+// 'prototype' should be a pointer to some concrete proto message. It will be
+// used only for its type signature.
+//
+// Intended to be called during process startup. Panics if such message has
+// already been registered.
+func (d *Dispatcher) RegisterTask(prototype proto.Message, cb Handler, queue string, opts *taskqueue.RetryOptions) {
+        if queue == "" {
+                queue = "default" // default GAE task queue name, always exists
+        }
+
+        name := proto.MessageName(prototype)
+        if name == "" {
+                panic(fmt.Sprintf("unregistered proto message type %T", prototype))
+        }
+
+        d.mu.Lock()
+        defer d.mu.Unlock()
+
+        if _, ok := d.handlers[name]; ok {
+                panic(fmt.Sprintf("handler for %q has already been registered", name))
+        }
+
+        if d.handlers == nil {
+                d.handlers = make(map[string]handler)
+        }
+
+        d.handlers[name] = handler{cb: cb, queue: queue, retryOpts: opts}
+}
+
+// AddTask submits the given task to an appropriate task queue.
+//
+// It means, add some later time in some other GAE process, the callback
+// registered as a handler for corresponding proto type will be called.
+//
+// If the given context is transactional, inherits the transaction.
+//
+// May return transient errors.
+func (d *Dispatcher) AddTask(c context.Context, task Task) error {
+        handler, err := d.handler(task.Payload)
+        if err != nil {
+                return err
+        }
+
+        blob, err := serializePayload(task.Payload)
+        if err != nil {
+                return err
+        }
+
+        title := "-"
+        if task.Title != "" {
+                title = task.Title
+        }
+
+        retryOpts := handler.retryOpts
+        if task.RetryOptions != nil {
+                retryOpts = task.RetryOptions
+        }
+
+        t := taskqueue.Task{
+                Path:         fmt.Sprintf("%s%s/%s", d.baseURL(), handler.queue, title),
+                Method:       "POST",
+                Payload:      blob,
+                ETA:          task.ETA,
+                Delay:        task.Delay,
+                RetryOptions: retryOpts,
+        }
+        if err := taskqueue.Add(c, handler.queue, &t); err != nil {
+                return transient.Tag.Apply(err)
+        }
+
+        return nil
+}
+
+// InstallRoutes installs appropriate HTTP routes in the router.
+//
+// Must be called only after all task handlers are registered!
+func (d *Dispatcher) InstallRoutes(r *router.Router, mw router.MiddlewareChain) {
+        queues := stringset.New(0)
+
+        d.mu.RLock()
+        for _, h := range d.handlers {
+                queues.Add(h.queue)
+        }
+        d.mu.RUnlock()
+
+        for _, q := range queues.ToSlice() {
+                r.POST(
+                        fmt.Sprintf("%s%s/:Title", d.baseURL(), q),
+                        mw.Extend(gaemiddleware.RequireTaskQueue(q)),
+                        d.processHTTPRequest)
+        }
+}
+
+////////////////////////////////////////////////////////////////////////////////
+
+type handler struct {
+        cb        Handler                 // the actual handler
+        queue     string                  // name of the task queue
+        retryOpts *taskqueue.RetryOptions // default retry options
+}
+
+// baseURL returns a URL prefix for all HTTP routes used by Dispatcher.
+//
+// It ends with '/'.
+func (d *Dispatcher) baseURL() string {
+        switch {
+        case d.BaseURL != "" && strings.HasSuffix(d.BaseURL, "/"):
+                return d.BaseURL
+        case d.BaseURL != "":
+                return d.BaseURL + "/"
+        default:
+                return "/internal/tasks/"
+        }
+}
+
+// handler returns a handler struct registered with Register.
+func (d *Dispatcher) handler(payload proto.Message) (handler, error) {
+        name := proto.MessageName(payload)
+
+        d.mu.RLock()
+        defer d.mu.RUnlock()
+
+        handler, registered := d.handlers[name]
+        if !registered {
+                return handler, fmt.Errorf("handler for %q is not registered", name)
+        }
+        return handler, nil
+}
+
+// processHTTPRequest is invoked on each HTTP POST.
+//
+// It deserializes the task and invokes an appropriate callback.
+func (d *Dispatcher) processHTTPRequest(c *router.Context) {
+        body, err := ioutil.ReadAll(c.Request.Body)
+        if err != nil {
+                httpReply(c, false, 500, "Failed to read request body: %s", err)
+                return
+        }
+
+        payload, err := deserializePayload(body)
+        if err != nil {
+                httpReply(c, false, 400, "Bad payload, can't deserialize: %s", err)
+                return
+        }
+
+        h, err := d.handler(payload)
+        if err != nil {
+                httpReply(c, false, 404, "Bad task: %s", err)
+                return
+        }
+
+        execCount, _ := strconv.Atoi(c.Request.Header.Get("X-AppEngine-TaskExecutionCount"))
+        switch h.cb(c.Context, payload, execCount); {
+        case err == nil:
+                httpReply(c, true, 200, "OK")
+        case transient.Tag.In(err):
+                httpReply(c, false, 500, "Transient error: %s", err)
+        default:
+                httpReply(c, false, 200, "Fatal error: %s", err) // return 200 to stop retries
+        }
+}
+
+func httpReply(c *router.Context, ok bool, code int, msg string, args ...interface{}) {
+        body := fmt.Sprintf(msg, args...)
+        if !ok {
+                logging.Errorf(c.Context, "%s", body)
+        }
+        http.Error(c.Writer, body, code)
+}
+
+////////////////////////////////////////////////////////////////////////////////
+
+var marshaller = jsonpb.Marshaler{}
+
+type envelope struct {
+        Type string           `json:"type"`
+        Body *json.RawMessage `json:"body"`
+}
+
+func serializePayload(task proto.Message) ([]byte, error) {
+        var buf bytes.Buffer
+        if err := marshaller.Marshal(&buf, task); err != nil {

[Vadim Sh., 2017/07/15 01:16:29]

Using JSON to be able to see the task bodies in Cloud Console UI when debugging
stuff.

+                return nil, err
+        }
+        raw := json.RawMessage(buf.Bytes())
+        return json.Marshal(envelope{
+                Type: proto.MessageName(task),
+                Body: &raw,
+        })
+}
+
+func deserializePayload(blob []byte) (proto.Message, error) {
+        env := envelope{}
+        if err := json.Unmarshal(blob, &env); err != nil {
+                return nil, err
+        }
+
+        tp := proto.MessageType(env.Type) // this is **ConcreteStruct{}
+        if tp == nil {
+                return nil, fmt.Errorf("unregistered proto message name %q", env.Type)
+        }
+        if env.Body == nil {
+                return nil, fmt.Errorf("no task body given")
+        }
+
+        task := reflect.New(tp.Elem()).Interface().(proto.Message)
+        if err := jsonpb.Unmarshal(bytes.NewReader(*env.Body), task); err != nil {
+                return nil, err
+        }
+
+        return task, nil
+}