Initial distributor implementation

appengine/cmd/dm/distributor/tq_handler.go

Index: appengine/cmd/dm/distributor/tq_handler.go
diff --git a/appengine/cmd/dm/distributor/tq_handler.go b/appengine/cmd/dm/distributor/tq_handler.go
new file mode 100644
index 0000000000000000000000000000000000000000..8293c542621cc9b06476f4cf6c66bf083f09458b
--- /dev/null
+++ b/appengine/cmd/dm/distributor/tq_handler.go
@@ -0,0 +1,49 @@
+// 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 distributor
+
+import (
+	"net/http"
+	"net/url"
+	"strings"
+
+	"github.com/julienschmidt/httprouter"
+	"github.com/luci/luci-go/appengine/gaemiddleware"
+	"github.com/luci/luci-go/server/middleware"
+	"golang.org/x/net/context"
+)
+
+const handlerPattern = "/tq/distributor/:cfgName"
+
+func handlerPath(cfgName string) string {
+	return strings.Replace(handlerPattern, ":cfgName", url.QueryEscape(cfgName), 1)
+}
+
+// InstallHandlers installs the taskqueue callback handler.
+func InstallHandlers(r *httprouter.Router, base middleware.Base) {
+	r.POST(handlerPattern, base(
+		gaemiddleware.RequireTaskQueue("", taskqueHandler)))
+}
+
+func taskqueHandler(c context.Context, rw http.ResponseWriter, r *http.Request, p httprouter.Params) {
+	defer r.Body.Close()
+
+	cfg, err := LoadConfiguration(c, p.ByName("cfgName"))
+	if err != nil {
+		http.Error(rw, "bad configuration name", http.StatusBadRequest)
+		return
+	}
+	dist, err := MakeDistributor(c, cfg)
+	if err != nil {
+		http.Error(rw, "bad distributor", http.StatusBadRequest)
+		return
+	}
+	err = dist.HandleTaskQueueTask(c, r)
+	if err != nil {
+		http.Error(rw, "failure to execute handler", http.StatusInternalServerError)
+		return
+	}
+	rw.WriteHeader(http.StatusOK)
+}