support AWS SQS as file change notification message queue

weed/command/filer_replication.go

@@ -13,6 +13,7 @@ import (
 	_ "github.com/chrislusf/seaweedfs/weed/replication/sink/s3sink"
 	"github.com/chrislusf/seaweedfs/weed/server"
 	"github.com/spf13/viper"
+	"github.com/chrislusf/seaweedfs/weed/replication/sub"
 )
 
 func init() {
@@ -37,9 +38,9 @@ func runFilerReplicate(cmd *Command, args []string) bool {
 	weed_server.LoadConfiguration("replication", true)
 	config := viper.GetViper()
 
-	var notificationInput replication.NotificationInput
+	var notificationInput sub.NotificationInput
 
-	for _, input := range replication.NotificationInputs {
+	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 {
@@ -99,6 +100,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 {

weed/command/scaffold.go

@@ -146,6 +146,14 @@ hosts = [
 ]
 topic = "seaweedfs_filer"
 
+[notification.aws_sqs]
+# experimental, let me know if it works
+enabled = false
+aws_access_key_id     = ""        # if empty, loads from the shared credentials file (~/.aws/credentials).
+aws_secret_access_key = ""        # if empty, loads from the shared credentials file (~/.aws/credentials).
+region = "us-east-2"
+sqs_queue_name = "my_filer_queue" # an existing queue name
+
 `
 	REPLICATION_TOML_EXAMPLE = `
 # A sample TOML config file for replicating SeaweedFS filer
@@ -169,6 +177,13 @@ topic = "seaweedfs_filer1_to_filer2"
 offsetFile = "./last.offset"
 offsetSaveIntervalSeconds = 10
 
+[notification.aws_sqs]
+enabled = false
+aws_access_key_id     = ""        # if empty, loads from the shared credentials file (~/.aws/credentials).
+aws_secret_access_key = ""        # if empty, loads from the shared credentials file (~/.aws/credentials).
+region = "us-east-2"
+sqs_queue_name = "my_filer_queue" # an existing queue name
+
 [sink.filer]
 enabled = false
 grpcAddress = "localhost:18888"