Initial distributor implementation

appengine/cmd/dm/distributor/registry.go

Index: appengine/cmd/dm/distributor/registry.go
diff --git a/appengine/cmd/dm/distributor/registry.go b/appengine/cmd/dm/distributor/registry.go
new file mode 100644
index 0000000000000000000000000000000000000000..b441308ca6cb12b69fefad6cf3587e8f402bba47
--- /dev/null
+++ b/appengine/cmd/dm/distributor/registry.go
@@ -0,0 +1,107 @@
+// 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 (
+	"fmt"
+	"reflect"
+	"sync"
+
+	"github.com/golang/protobuf/proto"
+	"github.com/luci/gae/service/info"
+	protos "github.com/luci/luci-go/appengine/cmd/dm/distributor/protos"
+	"golang.org/x/net/context"
+)
+
+var registry = struct {
+	sync.RWMutex
+	data map[reflect.Type]Factory
+}{data: map[reflect.Type]Factory{}}
+
+// Register enrolls a new distributor type in the registry. Calling
+// this multiple times for the same proto Message will panic.
+func Register(p proto.Message, factory Factory) {
+	if factory == nil {
+		panic("factory is nil")
+	}
+	if p == nil {
+		panic("proto.Message is nil")
+	}
+
+	typ := reflect.TypeOf(p)
+
+	registry.Lock()
+	defer registry.Unlock()
+	if _, ok := registry.data[typ]; ok {
+		panic(fmt.Errorf("trying to register %q twice", typ))
+	}
+	registry.data[typ] = factory
+}
+
+// MakeDistributor builds a distributor instance that's configured with the
+// provided config.
+//
+// The configuration for this distributor are obtained from luci-config at the
+// time an Execution is started.
+func MakeDistributor(c context.Context, cfg *Config) (D, error) {
+	typ := reflect.TypeOf(cfg.ImplConfig)
+
+	registry.RLock()
+	fn, ok := registry.data[typ]
+	registry.RUnlock()
+
+	if !ok {
+		return nil, fmt.Errorf("unknown distributor type %T", cfg.ImplConfig)
+	}
+
+	return fn(c, cfg)
+}
+
+// LoadConfiguration loads the named distributor configurtaion from luci-config,
+// possibly using the in-memory or memcache version.
+func LoadConfiguration(c context.Context, cfgName string) (ret *Config, err error) {
+	cfgVersion := ""
+	fullDMConfig := proto.Message(nil)
+	// TODO(riannucci): get config and version from luci-config in a fast cached way.
+
+	configs := fullDMConfig.(*protos.Config)
+	cfg, ok := configs.DistributorConfigs[cfgName]
+	if !ok {
+		err = fmt.Errorf("unknown distributor configuration: %q", cfgName)
+		return
+	}
+	if alias := cfg.GetAlias(); alias != nil {
+		cfg, ok = configs.DistributorConfigs[alias.OtherConfig]
+		if !ok {
+			err = fmt.Errorf("unknown distributor configuration: %q (via alias %q)", cfgName, alias.OtherConfig)
+			return
+		}
+		if cfg.GetAlias() != nil {
+			err = fmt.Errorf("too many levels of indirection for alias %q (points to alias %q)", cfgName, alias.OtherConfig)
+			return
+		}
+	}
+
+	dt := cfg.GetDistributorType()
+	if dt == nil {
+		err = fmt.Errorf("blank or unrecognized distributor_type")
+		return
+	}
+	dVal := reflect.ValueOf(dt)
+
+	// All non-nil DistributorType's have a single field which is the actual oneof
+	// value.
+	implConfig := dVal.Field(0).Interface().(proto.Message)
+
+	baseURL := fmt.Sprintf("https://%s/", info.Get(c).DefaultVersionHostname())
+	ret = &Config{
+		baseURL,
+		baseURL + handlerPath(cfgName),
+		cfgName,
+		cfgVersion,
+		implConfig,
+	}
+	return
+}