Two new options for rsync provider

- rsync_no_timeout=true/false # disable --timeout option
- rsync_timeout=n # set --timeout=n
related to issue #121

worker/common.go

@@ -1,6 +1,6 @@
 package worker
 
-// put global viables and types here
+// put global variables and types here
 
 import (
 	"gopkg.in/op/go-logging.v1"

worker/config.go

@@ -142,6 +142,8 @@ type mirrorConfig struct {
 	ExcludeFile   string   `toml:"exclude_file"`
 	Username      string   `toml:"username"`
 	Password      string   `toml:"password"`
+	RsyncNoTimeo  bool     `toml:"rsync_no_timeout"`
+	RsyncTimeout  int      `toml:"rsync_timeout"`
 	RsyncOptions  []string `toml:"rsync_options"`
 	RsyncOverride []string `toml:"rsync_override"`
 	Stage1Profile string   `toml:"stage1_profile"`

worker/provider.go

@@ -140,6 +140,8 @@ func newMirrorProvider(mirror mirrorConfig, cfg *Config) mirrorProvider {
 			password:          mirror.Password,
 			excludeFile:       mirror.ExcludeFile,
 			extraOptions:      mirror.RsyncOptions,
+			rsyncNeverTimeout: mirror.RsyncNoTimeo,
+			rsyncTimeoutValue: mirror.RsyncTimeout,
 			overriddenOptions: mirror.RsyncOverride,
 			rsyncEnv:          mirror.Env,
 			workingDir:        mirrorDir,
@@ -159,22 +161,24 @@ func newMirrorProvider(mirror mirrorConfig, cfg *Config) mirrorProvider {
 		provider = p
 	case provTwoStageRsync:
 		rc := twoStageRsyncConfig{
-			name:          mirror.Name,
-			stage1Profile: mirror.Stage1Profile,
-			upstreamURL:   mirror.Upstream,
-			rsyncCmd:      mirror.Command,
-			username:      mirror.Username,
-			password:      mirror.Password,
-			excludeFile:   mirror.ExcludeFile,
-			extraOptions:  mirror.RsyncOptions,
-			rsyncEnv:      mirror.Env,
-			workingDir:    mirrorDir,
-			logDir:        logDir,
-			logFile:       filepath.Join(logDir, "latest.log"),
-			useIPv6:       mirror.UseIPv6,
-			interval:      time.Duration(mirror.Interval) * time.Minute,
-			retry:         mirror.Retry,
-			timeout:       time.Duration(mirror.Timeout) * time.Second,
+			name:              mirror.Name,
+			stage1Profile:     mirror.Stage1Profile,
+			upstreamURL:       mirror.Upstream,
+			rsyncCmd:          mirror.Command,
+			username:          mirror.Username,
+			password:          mirror.Password,
+			excludeFile:       mirror.ExcludeFile,
+			extraOptions:      mirror.RsyncOptions,
+			rsyncNeverTimeout: mirror.RsyncNoTimeo,
+			rsyncTimeoutValue: mirror.RsyncTimeout,
+			rsyncEnv:          mirror.Env,
+			workingDir:        mirrorDir,
+			logDir:            logDir,
+			logFile:           filepath.Join(logDir, "latest.log"),
+			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

@@ -148,18 +148,19 @@ func TestRsyncProviderWithAuthentication(t *testing.T) {
 		proxyAddr := "127.0.0.1:1233"
 
 		c := rsyncConfig{
-			name:         "tuna",
-			upstreamURL:  "rsync://rsync.tuna.moe/tuna/",
-			rsyncCmd:     scriptFile,
-			username:     "tunasync",
-			password:     "tunasyncpassword",
-			workingDir:   tmpDir,
-			extraOptions: []string{"--delete-excluded"},
-			rsyncEnv:     map[string]string{"RSYNC_PROXY": proxyAddr},
-			logDir:       tmpDir,
-			logFile:      tmpFile,
-			useIPv4:      true,
-			interval:     600 * time.Second,
+			name:              "tuna",
+			upstreamURL:       "rsync://rsync.tuna.moe/tuna/",
+			rsyncCmd:          scriptFile,
+			username:          "tunasync",
+			password:          "tunasyncpassword",
+			workingDir:        tmpDir,
+			extraOptions:      []string{"--delete-excluded"},
+			rsyncTimeoutValue: 30,
+			rsyncEnv:          map[string]string{"RSYNC_PROXY": proxyAddr},
+			logDir:            tmpDir,
+			logFile:           tmpFile,
+			useIPv4:           true,
+			interval:          600 * time.Second,
 		}
 
 		provider, err := newRsyncProvider(c)
@@ -191,7 +192,7 @@ exit 0
 				fmt.Sprintf(
 					"%s %s %s -aHvh --no-o --no-g --stats --exclude .~tmp~/ "+
 						"--delete --delete-after --delay-updates --safe-links "+
-						"--timeout=120 -4 --delete-excluded %s %s",
+						"--timeout=30 -4 --delete-excluded %s %s",
 					provider.username, provider.password, proxyAddr,
 					provider.upstreamURL, provider.WorkingDir(),
 				),
@@ -221,6 +222,7 @@ func TestRsyncProviderWithOverriddenOptions(t *testing.T) {
 			upstreamURL:       "rsync://rsync.tuna.moe/tuna/",
 			rsyncCmd:          scriptFile,
 			workingDir:        tmpDir,
+			rsyncNeverTimeout: true,
 			overriddenOptions: []string{"-aHvh", "--no-o", "--no-g", "--stats"},
 			extraOptions:      []string{"--delete-excluded"},
 			logDir:            tmpDir,
@@ -490,18 +492,19 @@ func TestTwoStageRsyncProvider(t *testing.T) {
 		tmpFile := filepath.Join(tmpDir, "log_file")
 
 		c := twoStageRsyncConfig{
-			name:          "tuna-two-stage-rsync",
-			upstreamURL:   "rsync://mirrors.tuna.moe/",
-			stage1Profile: "debian",
-			rsyncCmd:      scriptFile,
-			workingDir:    tmpDir,
-			logDir:        tmpDir,
-			logFile:       tmpFile,
-			useIPv6:       true,
-			excludeFile:   tmpFile,
-			extraOptions:  []string{"--delete-excluded", "--cache"},
-			username:      "hello",
-			password:      "world",
+			name:              "tuna-two-stage-rsync",
+			upstreamURL:       "rsync://mirrors.tuna.moe/",
+			stage1Profile:     "debian",
+			rsyncCmd:          scriptFile,
+			workingDir:        tmpDir,
+			logDir:            tmpDir,
+			logFile:           tmpFile,
+			useIPv6:           true,
+			excludeFile:       tmpFile,
+			rsyncTimeoutValue: 30,
+			extraOptions:      []string{"--delete-excluded", "--cache"},
+			username:          "hello",
+			password:          "world",
 		}
 
 		provider, err := newTwoStageRsyncProvider(c)
@@ -539,7 +542,7 @@ exit 0
 				targetDir,
 				fmt.Sprintf(
 					"-aHvh --no-o --no-g --stats --exclude .~tmp~/ --safe-links "+
-						"--timeout=120 --exclude dists/ -6 "+
+						"--exclude dists/ --timeout=30 -6 "+
 						"--exclude-from %s %s %s",
 					provider.excludeFile, provider.upstreamURL, provider.WorkingDir(),
 				),
@@ -547,7 +550,7 @@ exit 0
 				fmt.Sprintf(
 					"-aHvh --no-o --no-g --stats --exclude .~tmp~/ "+
 						"--delete --delete-after --delay-updates --safe-links "+
-						"--timeout=120 --delete-excluded --cache -6 --exclude-from %s %s %s",
+						"--delete-excluded --cache --timeout=30 -6 --exclude-from %s %s %s",
 					provider.excludeFile, provider.upstreamURL, provider.WorkingDir(),
 				),
 			)
@@ -581,7 +584,7 @@ exit 0
 
 			expectedOutput := fmt.Sprintf(
 				"-aHvh --no-o --no-g --stats --exclude .~tmp~/ --safe-links "+
-					"--timeout=120 --exclude dists/ -6 "+
+					"--exclude dists/ --timeout=30 -6 "+
 					"--exclude-from %s %s %s\n",
 				provider.excludeFile, provider.upstreamURL, provider.WorkingDir(),
 			)

worker/rsync_provider.go

@@ -2,6 +2,7 @@ package worker
 
 import (
 	"errors"
+	"fmt"
 	"strings"
 	"time"
 
@@ -14,6 +15,8 @@ type rsyncConfig struct {
 	upstreamURL, username, password, excludeFile string
 	extraOptions                                 []string
 	overriddenOptions                            []string
+	rsyncNeverTimeout                            bool
+	rsyncTimeoutValue                            int
 	rsyncEnv                                     map[string]string
 	workingDir, logDir, logFile                  string
 	useIPv6, useIPv4                             bool
@@ -66,12 +69,20 @@ func newRsyncProvider(c rsyncConfig) (*rsyncProvider, error) {
 		"-aHvh", "--no-o", "--no-g", "--stats",
 		"--exclude", ".~tmp~/",
 		"--delete", "--delete-after", "--delay-updates",
-		"--safe-links", "--timeout=120",
+		"--safe-links",
 	}
 	if c.overriddenOptions != nil {
 		options = c.overriddenOptions
 	}
 
+	if !c.rsyncNeverTimeout {
+		timeo := 120
+		if c.rsyncTimeoutValue > 0 {
+			timeo = c.rsyncTimeoutValue
+		}
+		options = append(options, fmt.Sprintf("--timeout=%d", timeo))
+	}
+
 	if c.useIPv6 {
 		options = append(options, "-6")
 	} else if c.useIPv4 {

worker/two_stage_rsync_provider.go

@@ -15,6 +15,8 @@ type twoStageRsyncConfig struct {
 	stage1Profile                                string
 	upstreamURL, username, password, excludeFile string
 	extraOptions                                 []string
+	rsyncNeverTimeout                            bool
+	rsyncTimeoutValue                            int
 	rsyncEnv                                     map[string]string
 	workingDir, logDir, logFile                  string
 	useIPv6                                      bool
@@ -61,13 +63,13 @@ func newTwoStageRsyncProvider(c twoStageRsyncConfig) (*twoStageRsyncProvider, er
 		stage1Options: []string{
 			"-aHvh", "--no-o", "--no-g", "--stats",
 			"--exclude", ".~tmp~/",
-			"--safe-links", "--timeout=120",
+			"--safe-links",
 		},
 		stage2Options: []string{
 			"-aHvh", "--no-o", "--no-g", "--stats",
 			"--exclude", ".~tmp~/",
 			"--delete", "--delete-after", "--delay-updates",
-			"--safe-links", "--timeout=120",
+			"--safe-links",
 		},
 	}
 
@@ -124,6 +126,14 @@ func (p *twoStageRsyncProvider) Options(stage int) ([]string, error) {
 		return []string{}, fmt.Errorf("Invalid stage: %d", stage)
 	}
 
+	if !p.rsyncNeverTimeout {
+		timeo := 120
+		if p.rsyncTimeoutValue > 0 {
+			timeo = p.rsyncTimeoutValue
+		}
+		options = append(options, fmt.Sprintf("--timeout=%d", timeo))
+	}
+
 	if p.useIPv6 {
 		options = append(options, "-6")
 	}