feature(worker): mirrorConfig -> mirrorProvider

worker/config.go

@@ -48,18 +48,24 @@ func (p *ProviderEnum) UnmarshalText(text []byte) error {
 
 type Config struct {
 	Global  globalConfig   `toml:"global"`
+	Manager managerConfig  `toml:"manager"`
 	Mirrors []mirrorConfig `toml:"mirrors"`
 }
 
 type globalConfig struct {
 	Name       string `toml:"name"`
-	Token      string `toml:"token"`
 	LogDir     string `toml:"log_dir"`
 	MirrorDir  string `toml:"mirror_dir"`
 	Concurrent int    `toml:"concurrent"`
 	Interval   int    `toml:"interval"`
 }
 
+type managerConfig struct {
+	APIBase string `toml:"api_base"`
+	CACert  string `toml:"ca_cert"`
+	Token   string `toml:"token"`
+}
+
 type mirrorConfig struct {
 	Name      string            `toml:"name"`
 	Provider  ProviderEnum      `toml:"provider"`

worker/config_test.go

@@ -12,12 +12,15 @@ func TestConfig(t *testing.T) {
 	var cfgBlob = `
 [global]
 name = "test_worker"
-token = "some_token"
+log_dir = "/var/log/tunasync/{{.Name}}"
-log_dir = "/var/log/tunasync"
 mirror_dir = "/data/mirrors"
 concurrent = 10
 interval = 240
 
+[manager]
+api_base = "https://127.0.0.1:5000"
+token = "some_token"
+
 [[mirrors]]
 name = "AOSP"
 provider = "command"
@@ -64,6 +67,8 @@ exclude_file = "/etc/tunasync.d/fedora-exclude.txt"
 		So(cfg.Global.Interval, ShouldEqual, 240)
 		So(cfg.Global.MirrorDir, ShouldEqual, "/data/mirrors")
 
+		So(cfg.Manager.APIBase, ShouldEqual, "https://127.0.0.1:5000")
+
 		m := cfg.Mirrors[0]
 		So(m.Name, ShouldEqual, "AOSP")
 		So(m.MirrorDir, ShouldEqual, "/data/git/AOSP")

worker/main.go

@@ -1,9 +1,82 @@
 package worker
 
-import "time"
+import (
+	"bytes"
+	"errors"
+	"html/template"
+	"path/filepath"
+	"time"
+)
 
 // toplevel module for workers
 
+func initProviders(c *Config) []mirrorProvider {
+
+	formatLogDir := func(logDir string, m mirrorConfig) string {
+		tmpl, err := template.New("logDirTmpl-" + m.Name).Parse(logDir)
+		if err != nil {
+			panic(err)
+		}
+		var formatedLogDir bytes.Buffer
+		tmpl.Execute(&formatedLogDir, m)
+		return formatedLogDir.String()
+	}
+
+	providers := []mirrorProvider{}
+
+	for _, mirror := range c.Mirrors {
+		logDir := mirror.LogDir
+		mirrorDir := mirror.MirrorDir
+		if logDir == "" {
+			logDir = c.Global.LogDir
+		}
+		if mirrorDir == "" {
+			mirrorDir = c.Global.MirrorDir
+		}
+		logDir = formatLogDir(logDir, mirror)
+		switch mirror.Provider {
+		case ProvCommand:
+			pc := cmdConfig{
+				name:        mirror.Name,
+				upstreamURL: mirror.Upstream,
+				command:     mirror.Command,
+				workingDir:  filepath.Join(mirrorDir, mirror.Name),
+				logDir:      logDir,
+				logFile:     filepath.Join(logDir, "latest.log"),
+				interval:    time.Duration(mirror.Interval) * time.Minute,
+				env:         mirror.Env,
+			}
+			p, err := newCmdProvider(pc)
+			if err != nil {
+				panic(err)
+			}
+			providers = append(providers, p)
+		case ProvRsync:
+			rc := rsyncConfig{
+				name:        mirror.Name,
+				upstreamURL: mirror.Upstream,
+				password:    mirror.Password,
+				excludeFile: mirror.ExcludeFile,
+				workingDir:  filepath.Join(mirrorDir, mirror.Name),
+				logDir:      logDir,
+				logFile:     filepath.Join(logDir, "latest.log"),
+				useIPv6:     mirror.UseIPv6,
+				interval:    time.Duration(mirror.Interval) * time.Minute,
+			}
+			p, err := newRsyncProvider(rc)
+			if err != nil {
+				panic(err)
+			}
+			providers = append(providers, p)
+		default:
+			panic(errors.New("Invalid mirror provider"))
+
+		}
+
+	}
+	return providers
+}
+
 func main() {
 
 	for {

worker/rsync_provider.go

@@ -35,11 +35,16 @@ func newRsyncProvider(c rsyncConfig) (*rsyncProvider, error) {
 }
 
 // TODO: implement this
-func (p *rsyncProvider) Start() {
+func (p *rsyncProvider) Start() error {
-
+	return nil
 }
 
 // TODO: implement this
-func (p *rsyncProvider) Terminate() {
+func (p *rsyncProvider) Terminate() error {
-
+	return nil
+}
+
+// TODO: implement this
+func (p *rsyncProvider) Wait() error {
+	return nil
 }