Merge branch 'master' into support_ssd_volume

1 files changed, 15 insertions, 3 deletions

diff --git a/weed/command/filer_replication.go b/weed/command/filer_replication.go
index 40f2b570b..e8c06b208 100644
--- a/weed/command/filer_replication.go
+++ b/weed/command/filer_replication.go
@@ -11,10 +11,10 @@ import (
 	_ "github.com/chrislusf/seaweedfs/weed/replication/sink/b2sink"
 	_ "github.com/chrislusf/seaweedfs/weed/replication/sink/filersink"
 	_ "github.com/chrislusf/seaweedfs/weed/replication/sink/gcssink"
+	_ "github.com/chrislusf/seaweedfs/weed/replication/sink/localsink"
 	_ "github.com/chrislusf/seaweedfs/weed/replication/sink/s3sink"
 	"github.com/chrislusf/seaweedfs/weed/replication/sub"
 	"github.com/chrislusf/seaweedfs/weed/util"
-	"github.com/spf13/viper"
 )
 
 func init() {
@@ -98,13 +98,19 @@ func runFilerReplicate(cmd *Command, args []string) bool {
 	replicator := replication.NewReplicator(config, "source.filer.", dataSink)
 
 	for {
-		key, m, err := notificationInput.ReceiveMessage()
+		key, m, onSuccessFn, onFailureFn, err := notificationInput.ReceiveMessage()
 		if err != nil {
 			glog.Errorf("receive %s: %+v", key, err)
+			if onFailureFn != nil {
+				onFailureFn()
+			}
 			continue
 		}
 		if key == "" {
 			// long poll received no messages
+			if onSuccessFn != nil {
+				onSuccessFn()
+			}
 			continue
 		}
 		if m.OldEntry != nil && m.NewEntry == nil {
@@ -116,14 +122,20 @@ func runFilerReplicate(cmd *Command, args []string) bool {
 		}
 		if err = replicator.Replicate(context.Background(), key, m); err != nil {
 			glog.Errorf("replicate %s: %+v", key, err)
+			if onFailureFn != nil {
+				onFailureFn()
+			}
 		} else {
 			glog.V(1).Infof("replicated %s", key)
+			if onSuccessFn != nil {
+				onSuccessFn()
+			}
 		}
 	}
 
 }
 
-func validateOneEnabledInput(config *viper.Viper) {
+func validateOneEnabledInput(config *util.ViperProxy) {
 	enabledInput := ""
 	for _, input := range sub.NotificationInputs {
 		if config.GetBool("notification." + input.GetName() + ".enabled") {