add a timeout field to providers

worker/base_provider.go

@@ -16,6 +16,7 @@ type baseProvider struct {
 	name     string
 	interval time.Duration
 	retry    int
+	timeout  time.Duration
 	isMaster bool
 
 	cmd       *cmdJob
@@ -56,6 +57,10 @@ func (p *baseProvider) Retry() int {
 	return p.retry
 }
 
+func (p *baseProvider) Timeout() time.Duration {
+	return p.timeout
+}
+
 func (p *baseProvider) IsMaster() bool {
 	return p.isMaster
 }

worker/cmd_provider.go

@@ -16,6 +16,7 @@ type cmdConfig struct {
 	workingDir, logDir, logFile string
 	interval                    time.Duration
 	retry                       int
+	timeout                     time.Duration
 	env                         map[string]string
 	failOnMatch                 string
 	sizePattern                 string
@@ -41,6 +42,7 @@ func newCmdProvider(c cmdConfig) (*cmdProvider, error) {
 			ctx:      NewContext(),
 			interval: c.interval,
 			retry:    c.retry,
+			timeout:  c.timeout,
 		},
 		cmdConfig: c,
 	}

worker/config_test.go

@@ -6,6 +6,7 @@ import (
 	"os"
 	"path/filepath"
 	"testing"
+	"time"
 
 	. "github.com/smartystreets/goconvey/convey"
 )
@@ -140,7 +141,6 @@ use_ipv6 = true
 		So(m.Name, ShouldEqual, "debian")
 		So(m.MirrorDir, ShouldEqual, "")
 		So(m.Provider, ShouldEqual, provTwoStageRsync)
-		So(m.Timeout, ShouldEqual, 86400)
 
 		m = cfg.Mirrors[2]
 		So(m.Name, ShouldEqual, "fedora")
@@ -321,6 +321,7 @@ log_dir = "/var/log/tunasync/{{.Name}}"
 mirror_dir = "/data/mirrors"
 concurrent = 10
 interval = 240
+timeout = 86400
 retry = 3
 
 [manager]
@@ -393,5 +394,6 @@ use_ipv6 = true
 		rp, ok := p.(*rsyncProvider)
 		So(ok, ShouldBeTrue)
 		So(rp.WorkingDir(), ShouldEqual, "/data/mirrors/debian-cd")
+		So(p.Timeout(), ShouldEqual, 86400*time.Second)
 	})
 }

worker/provider.go

@@ -46,6 +46,7 @@ type mirrorProvider interface {
 
 	Interval() time.Duration
 	Retry() int
+	Timeout() time.Duration
 
 	WorkingDir() string
 	LogDir() string
@@ -91,6 +92,9 @@ func newMirrorProvider(mirror mirrorConfig, cfg *Config) mirrorProvider {
 	if mirror.Retry == 0 {
 		mirror.Retry = cfg.Global.Retry
 	}
+	if mirror.Timeout == 0 {
+		mirror.Timeout = cfg.Global.Timeout
+	}
 	logDir = formatLogDir(logDir, mirror)
 
 	// IsMaster
@@ -118,6 +122,7 @@ func newMirrorProvider(mirror mirrorConfig, cfg *Config) mirrorProvider {
 			logFile:     filepath.Join(logDir, "latest.log"),
 			interval:    time.Duration(mirror.Interval) * time.Minute,
 			retry:       mirror.Retry,
+			timeout:     time.Duration(mirror.Timeout) * time.Second,
 			env:         mirror.Env,
 		}
 		p, err := newCmdProvider(pc)
@@ -144,6 +149,7 @@ func newMirrorProvider(mirror mirrorConfig, cfg *Config) mirrorProvider {
 			useIPv4:           mirror.UseIPv4,
 			interval:          time.Duration(mirror.Interval) * time.Minute,
 			retry:             mirror.Retry,
+			timeout:           time.Duration(mirror.Timeout) * time.Second,
 		}
 		p, err := newRsyncProvider(rc)
 		if err != nil {
@@ -168,6 +174,7 @@ func newMirrorProvider(mirror mirrorConfig, cfg *Config) mirrorProvider {
 			useIPv6:       mirror.UseIPv6,
 			interval:      time.Duration(mirror.Interval) * time.Minute,
 			retry:         mirror.Retry,
+			timeout:       time.Duration(mirror.Timeout) * time.Second,
 		}
 		p, err := newTwoStageRsyncProvider(rc)
 		if err != nil {

worker/provider_test.go

@@ -28,6 +28,7 @@ func TestRsyncProvider(t *testing.T) {
 			logDir:      tmpDir,
 			logFile:     tmpFile,
 			useIPv6:     true,
+			timeout:     100 * time.Second,
 			interval:    600 * time.Second,
 		}
 
@@ -40,6 +41,7 @@ func TestRsyncProvider(t *testing.T) {
 		So(provider.LogDir(), ShouldEqual, c.logDir)
 		So(provider.LogFile(), ShouldEqual, c.logFile)
 		So(provider.Interval(), ShouldEqual, c.interval)
+		So(provider.Timeout(), ShouldEqual, c.timeout)
 
 		Convey("When entering a context (auto exit)", func() {
 			func() {

worker/rsync_provider.go

@@ -19,6 +19,7 @@ type rsyncConfig struct {
 	useIPv6, useIPv4                             bool
 	interval                                     time.Duration
 	retry                                        int
+	timeout                                      time.Duration
 }
 
 // An RsyncProvider provides the implementation to rsync-based syncing jobs
@@ -43,6 +44,7 @@ func newRsyncProvider(c rsyncConfig) (*rsyncProvider, error) {
 			ctx:      NewContext(),
 			interval: c.interval,
 			retry:    c.retry,
+			timeout:  c.timeout,
 		},
 		rsyncConfig: c,
 	}

worker/two_stage_rsync_provider.go

@@ -20,6 +20,7 @@ type twoStageRsyncConfig struct {
 	useIPv6                                      bool
 	interval                                     time.Duration
 	retry                                        int
+	timeout                                      time.Duration
 }
 
 // An RsyncProvider provides the implementation to rsync-based syncing jobs
@@ -54,6 +55,7 @@ func newTwoStageRsyncProvider(c twoStageRsyncConfig) (*twoStageRsyncProvider, er
 			ctx:      NewContext(),
 			interval: c.interval,
 			retry:    c.retry,
+			timeout:  c.timeout,
 		},
 		twoStageRsyncConfig: c,
 		stage1Options: []string{