1 files changed, 41 insertions, 18 deletions

diff --git a/weed/command/filer_replication.go b/weed/command/filer_replication.go
index 05076143a..737f0d24a 100644
--- a/weed/command/filer_replication.go
+++ b/weed/command/filer_replication.go
@@ -1,6 +1,7 @@
 package command
 
 import (
+	"context"
 	"strings"
 
 	"github.com/chrislusf/seaweedfs/weed/glog"
@@ -11,7 +12,8 @@ import (
 	_ "github.com/chrislusf/seaweedfs/weed/replication/sink/filersink"
 	_ "github.com/chrislusf/seaweedfs/weed/replication/sink/gcssink"
 	_ "github.com/chrislusf/seaweedfs/weed/replication/sink/s3sink"
-	"github.com/chrislusf/seaweedfs/weed/server"
+	"github.com/chrislusf/seaweedfs/weed/replication/sub"
+	"github.com/chrislusf/seaweedfs/weed/util"
 	"github.com/spf13/viper"
 )
 
@@ -27,22 +29,25 @@ var cmdFilerReplicate = &Command{
 	filer.replicate listens on filer notifications. If any file is updated, it will fetch the updated content,
 	and write to the other destination.
 
-	Run "weed scaffold -config replication" to generate a replication.toml file and customize the parameters.
+	Run "weed scaffold -config=replication" to generate a replication.toml file and customize the parameters.
 
   `,
 }
 
 func runFilerReplicate(cmd *Command, args []string) bool {
 
-	weed_server.LoadConfiguration("replication", true)
-	config := viper.GetViper()
+	util.LoadConfiguration("security", false)
+	util.LoadConfiguration("replication", true)
+	util.LoadConfiguration("notification", true)
+	config := util.GetViper()
 
-	var notificationInput replication.NotificationInput
+	var notificationInput sub.NotificationInput
 
-	for _, input := range replication.NotificationInputs {
+	validateOneEnabledInput(config)
+
+	for _, input := range sub.NotificationInputs {
 		if config.GetBool("notification." + input.GetName() + ".enabled") {
-			viperSub := config.Sub("notification." + input.GetName())
-			if err := input.Initialize(viperSub); err != nil {
+			if err := input.Initialize(config, "notification."+input.GetName()+"."); err != nil {
 				glog.Fatalf("Failed to initialize notification input for %s: %+v",
 					input.GetName(), err)
 			}
@@ -53,16 +58,16 @@ func runFilerReplicate(cmd *Command, args []string) bool {
 	}
 
 	if notificationInput == nil {
-		println("Please follow 'weed scaffold -config=repliaction' to see example notification configurations.")
+		println("No notification is defined in notification.toml file.")
+		println("Please follow 'weed scaffold -config=notification' to see example notification configurations.")
 		return true
 	}
 
 	// avoid recursive replication
 	if config.GetBool("notification.source.filer.enabled") && config.GetBool("notification.sink.filer.enabled") {
-		sourceConfig, sinkConfig := config.Sub("source.filer"), config.Sub("sink.filer")
-		if sourceConfig.GetString("grpcAddress") == sinkConfig.GetString("grpcAddress") {
-			fromDir := sourceConfig.GetString("directory")
-			toDir := sinkConfig.GetString("directory")
+		if config.GetString("source.filer.grpcAddress") == config.GetString("sink.filer.grpcAddress") {
+			fromDir := config.GetString("source.filer.directory")
+			toDir := config.GetString("sink.filer.directory")
 			if strings.HasPrefix(toDir, fromDir) {
 				glog.Fatalf("recursive replication! source directory %s includes the sink directory %s", fromDir, toDir)
 			}
@@ -72,8 +77,7 @@ func runFilerReplicate(cmd *Command, args []string) bool {
 	var dataSink sink.ReplicationSink
 	for _, sk := range sink.Sinks {
 		if config.GetBool("sink." + sk.GetName() + ".enabled") {
-			viperSub := config.Sub("sink." + sk.GetName())
-			if err := sk.Initialize(viperSub); err != nil {
+			if err := sk.Initialize(config, "sink."+sk.GetName()+"."); err != nil {
 				glog.Fatalf("Failed to initialize sink for %s: %+v",
 					sk.GetName(), err)
 			}
@@ -84,14 +88,14 @@ func runFilerReplicate(cmd *Command, args []string) bool {
 	}
 
 	if dataSink == nil {
-		println("no data sink configured:")
+		println("no data sink configured in replication.toml:")
 		for _, sk := range sink.Sinks {
 			println("    " + sk.GetName())
 		}
 		return true
 	}
 
-	replicator := replication.NewReplicator(config.Sub("source.filer"), dataSink)
+	replicator := replication.NewReplicator(config, "source.filer.", dataSink)
 
 	for {
 		key, m, err := notificationInput.ReceiveMessage()
@@ -99,6 +103,10 @@ func runFilerReplicate(cmd *Command, args []string) bool {
 			glog.Errorf("receive %s: %+v", key, err)
 			continue
 		}
+		if key == "" {
+			// long poll received no messages
+			continue
+		}
 		if m.OldEntry != nil && m.NewEntry == nil {
 			glog.V(1).Infof("delete: %s", key)
 		} else if m.OldEntry == nil && m.NewEntry != nil {
@@ -106,10 +114,25 @@ func runFilerReplicate(cmd *Command, args []string) bool {
 		} else {
 			glog.V(1).Infof("modify: %s", key)
 		}
-		if err = replicator.Replicate(key, m); err != nil {
+		if err = replicator.Replicate(context.Background(), key, m); err != nil {
 			glog.Errorf("replicate %s: %+v", key, err)
+		} else {
+			glog.V(1).Infof("replicated %s", key)
 		}
 	}
 
 	return true
 }
+
+func validateOneEnabledInput(config *viper.Viper) {
+	enabledInput := ""
+	for _, input := range sub.NotificationInputs {
+		if config.GetBool("notification." + input.GetName() + ".enabled") {
+			if enabledInput == "" {
+				enabledInput = input.GetName()
+			} else {
+				glog.Fatalf("Notification input is enabled for both %s and %s", enabledInput, input.GetName())
+			}
+		}
+	}
+}