1 files changed, 24 insertions, 10 deletions
diff --git a/weed/notification/configuration.go b/weed/notification/configuration.go
index e0ba61d58..36211692c 100644
--- a/weed/notification/configuration.go
+++ b/weed/notification/configuration.go
@@ -11,7 +11,7 @@ type MessageQueue interface {
 	// GetName gets the name to locate the configuration in filer.toml file
 	GetName() string
 	// Initialize initializes the file store
-	Initialize(configuration util.Configuration) error
+	Initialize(configuration util.Configuration, prefix string) error
 	SendMessage(key string, message proto.Message) error
 }
 
@@ -21,23 +21,37 @@ var (
 	Queue MessageQueue
 )
 
-func LoadConfiguration(config *viper.Viper) {
+func LoadConfiguration(config *viper.Viper, prefix string) {
 
 	if config == nil {
 		return
 	}
 
-	for _, store := range MessageQueues {
-		if config.GetBool(store.GetName() + ".enabled") {
-			viperSub := config.Sub(store.GetName())
-			if err := store.Initialize(viperSub); err != nil {
-				glog.Fatalf("Failed to initialize store for %s: %+v",
-					store.GetName(), err)
+	validateOneEnabledQueue(config)
+
+	for _, queue := range MessageQueues {
+		if config.GetBool(prefix + queue.GetName() + ".enabled") {
+			if err := queue.Initialize(config, prefix+queue.GetName()+"."); err != nil {
+				glog.Fatalf("Failed to initialize notification for %s: %+v",
+					queue.GetName(), err)
 			}
-			Queue = store
-			glog.V(0).Infof("Configure message queue for %s", store.GetName())
+			Queue = queue
+			glog.V(0).Infof("Configure notification message queue for %s", queue.GetName())
 			return
 		}
 	}
 
 }
+
+func validateOneEnabledQueue(config *viper.Viper) {
+	enabledQueue := ""
+	for _, queue := range MessageQueues {
+		if config.GetBool(queue.GetName() + ".enabled") {
+			if enabledQueue == "" {
+				enabledQueue = queue.GetName()
+			} else {
+				glog.Fatalf("Notification message queue is enabled for both %s and %s", enabledQueue, queue.GetName())
+			}
+		}
+	}
+}