#208 Fixed creating subscriptions on startup to exsiting queues

Admiral-Piett · Oct 12, 2019 · b6cccf8 · b6cccf8
1 parent 44f52f0
commit b6cccf8
Show file tree

Hide file tree

Showing 2 changed files with 35 additions and 32 deletions.
diff --git a/app/conf/config.go b/app/conf/config.go
@@ -104,27 +104,24 @@ func LoadYamlConfig(filename string, env string) []string {
 		newTopic.Subscriptions = make([]*app.Subscription, 0, 0)
 
 		for _, subs := range topic.Subscriptions {
-			if _, ok := app.SyncQueues.Queues[subs.QueueName]; !ok {
-				var newSub *app.Subscription
-				if strings.Contains(subs.Protocol, "http") {
-					newSub = createHttpSubscription(subs)
-				} else {
-					//Queue does not exist yet, create it.
-					newSub = createSqsSubscription(subs, topicArn)
-				}
-				if subs.FilterPolicy != "" {
-					filterPolicy := &app.FilterPolicy{}
-					err = json.Unmarshal([]byte(subs.FilterPolicy), filterPolicy)
-					if err != nil {
-						log.Errorf("err: %s", err)
-						return ports
-					}
-					newSub.FilterPolicy = filterPolicy
+			var newSub *app.Subscription
+			if strings.Contains(subs.Protocol, "http") {
+				newSub = createHttpSubscription(subs)
+			} else {
+				//Queue does not exist yet, create it.
+				newSub = createSqsSubscription(subs, topicArn)
+			}
+			if subs.FilterPolicy != "" {
+				filterPolicy := &app.FilterPolicy{}
+				err = json.Unmarshal([]byte(subs.FilterPolicy), filterPolicy)
+				if err != nil {
+					log.Errorf("err: %s", err)
+					return ports
 				}
-
-				newTopic.Subscriptions = append(newTopic.Subscriptions, newSub)
+				newSub.FilterPolicy = filterPolicy
 			}
 
+			newTopic.Subscriptions = append(newTopic.Subscriptions, newSub)
 		}
 		app.SyncTopics.Topics[topic.Name] = newTopic
 	}
@@ -144,20 +141,22 @@ func createHttpSubscription(configSubscription app.EnvSubsciption) *app.Subscrip
 }
 
 func createSqsSubscription(configSubscription app.EnvSubsciption, topicArn string) *app.Subscription {
-	queueUrl := "http://" + app.CurrentEnvironment.Host + ":" + app.CurrentEnvironment.Port +
-		"/" + app.CurrentEnvironment.AccountID + "/" + configSubscription.QueueName
-	if app.CurrentEnvironment.Region != "" {
-		queueUrl = "http://" + app.CurrentEnvironment.Region + "." + app.CurrentEnvironment.Host + ":" +
-			app.CurrentEnvironment.Port + "/" + app.CurrentEnvironment.AccountID + "/" + configSubscription.QueueName
-	}
-	queueArn := "arn:aws:sqs:" + app.CurrentEnvironment.Region + ":" + app.CurrentEnvironment.AccountID + ":" + configSubscription.QueueName
-	app.SyncQueues.Queues[configSubscription.QueueName] = &app.Queue{
-		Name:                configSubscription.QueueName,
-		TimeoutSecs:         app.CurrentEnvironment.QueueAttributeDefaults.VisibilityTimeout,
-		Arn:                 queueArn,
-		URL:                 queueUrl,
-		ReceiveWaitTimeSecs: app.CurrentEnvironment.QueueAttributeDefaults.ReceiveMessageWaitTimeSeconds,
-		IsFIFO:              app.HasFIFOQueueName(configSubscription.QueueName),
+	if _, ok := app.SyncQueues.Queues[configSubscription.QueueName]; !ok {
+		queueUrl := "http://" + app.CurrentEnvironment.Host + ":" + app.CurrentEnvironment.Port +
+			"/" + app.CurrentEnvironment.AccountID + "/" + configSubscription.QueueName
+		if app.CurrentEnvironment.Region != "" {
+			queueUrl = "http://" + app.CurrentEnvironment.Region + "." + app.CurrentEnvironment.Host + ":" +
+				app.CurrentEnvironment.Port + "/" + app.CurrentEnvironment.AccountID + "/" + configSubscription.QueueName
+		}
+		queueArn := "arn:aws:sqs:" + app.CurrentEnvironment.Region + ":" + app.CurrentEnvironment.AccountID + ":" + configSubscription.QueueName
+		app.SyncQueues.Queues[configSubscription.QueueName] = &app.Queue{
+			Name:                configSubscription.QueueName,
+			TimeoutSecs:         app.CurrentEnvironment.QueueAttributeDefaults.VisibilityTimeout,
+			Arn:                 queueArn,
+			URL:                 queueUrl,
+			ReceiveWaitTimeSecs: app.CurrentEnvironment.QueueAttributeDefaults.ReceiveMessageWaitTimeSeconds,
+			IsFIFO:              app.HasFIFOQueueName(configSubscription.QueueName),
+		}
 	}
 	qArn := app.SyncQueues.Queues[configSubscription.QueueName].Arn
 	newSub := &app.Subscription{EndPoint: qArn, Protocol: "sqs", TopicArn: topicArn, Raw: configSubscription.Raw}

diff --git a/app/conf/goaws.yaml b/app/conf/goaws.yaml
@@ -25,6 +25,10 @@ Local:                              # Environment name that can be passed on the
         - QueueName: local-queue4   # Queue name
           Raw: true                 # Raw message delivery (true/false)
           #FilterPolicy: '{"foo": ["bar"]}' # Subscription's FilterPolicy, json object as a string
+    - Name: sub-topic            # Topic name - with some Subscriptions
+      Subscriptions:                # List of Subscriptions to create for this topic (queues will be created as required)
+          - QueueName: local-queue1   # Queue name
+            Raw: true                # Raw message delivery (true/false)
     - Name: local-topic2            # Topic name - no Subscriptions
     - Name: my_topic                # Topic name - http subscription
       Subscriptions: