Merge branch 'upstreamMaster' into iamapipr

1 files changed, 17 insertions, 12 deletions

diff --git a/weed/command/filer_replication.go b/weed/command/filer_replication.go
index e8c06b208..885c95540 100644
--- a/weed/command/filer_replication.go
+++ b/weed/command/filer_replication.go
@@ -74,18 +74,7 @@ func runFilerReplicate(cmd *Command, args []string) bool {
 		}
 	}
 
-	var dataSink sink.ReplicationSink
-	for _, sk := range sink.Sinks {
-		if config.GetBool("sink." + sk.GetName() + ".enabled") {
-			if err := sk.Initialize(config, "sink."+sk.GetName()+"."); err != nil {
-				glog.Fatalf("Failed to initialize sink for %s: %+v",
-					sk.GetName(), err)
-			}
-			glog.V(0).Infof("Configure sink to %s", sk.GetName())
-			dataSink = sk
-			break
-		}
-	}
+	dataSink := findSink(config)
 
 	if dataSink == nil {
 		println("no data sink configured in replication.toml:")
@@ -135,6 +124,22 @@ func runFilerReplicate(cmd *Command, args []string) bool {
 
 }
 
+func findSink(config *util.ViperProxy) sink.ReplicationSink {
+	var dataSink sink.ReplicationSink
+	for _, sk := range sink.Sinks {
+		if config.GetBool("sink." + sk.GetName() + ".enabled") {
+			if err := sk.Initialize(config, "sink."+sk.GetName()+"."); err != nil {
+				glog.Fatalf("Failed to initialize sink for %s: %+v",
+					sk.GetName(), err)
+			}
+			glog.V(0).Infof("Configure sink to %s", sk.GetName())
+			dataSink = sk
+			break
+		}
+	}
+	return dataSink
+}
+
 func validateOneEnabledInput(config *util.ViperProxy) {
 	enabledInput := ""
 	for _, input := range sub.NotificationInputs {