added subscriptionID as prefix for setup multiple consumers for same …

…topic

pkg/manager/flows/inputs/umb/input.go

@@ -6,15 +6,15 @@ import (
 	"github.com/adrianriobo/qe-eventmanager/pkg/configuration/flows"
 	"github.com/adrianriobo/qe-eventmanager/pkg/manager/flows/actions"
 	"github.com/adrianriobo/qe-eventmanager/pkg/manager/flows/inputs/ack"
-	inputACK "github.com/adrianriobo/qe-eventmanager/pkg/manager/flows/inputs/ack"
 	"github.com/adrianriobo/qe-eventmanager/pkg/services/messaging/umb"
 	"github.com/adrianriobo/qe-eventmanager/pkg/services/messaging/umb/api"
 	"github.com/adrianriobo/qe-eventmanager/pkg/util/json"
 	"github.com/adrianriobo/qe-eventmanager/pkg/util/logging"
 )
 
-func Add(flowName string, input flows.UMBInput, ack inputACK.ACK, action actions.Runnable) error {
+func Add(flowName string, input flows.UMBInput, ack ack.ACK, action actions.Runnable) error {
 	if err := umb.Subscribe(
+		flowName,
 		input.Topic,
 		[]api.MessageHandler{new(flowName, input.Filters, ack, action)}); err != nil {
 		return err

pkg/services/messaging/umb/umb.go

@@ -63,11 +63,11 @@ func SendBytes(destination string, message []byte) error {
 	return _umb.client.Send(destination, message)
 }
 
-func Subscribe(topic string, handlers []api.MessageHandler) error {
+func Subscribe(subscriptionID, topic string, handlers []api.MessageHandler) error {
 	_umb.subscribe.Lock()
 	defer _umb.subscribe.Unlock()
-	logging.Infof("Adding a subscription to %s", topic)
-	internalSubscription, err := _umb.client.Subscribe(umbTopic(topic), handlers)
+	logging.Infof("Adding a subscription %s on topic %s", subscriptionID, topic)
+	internalSubscription, err := _umb.client.Subscribe(umbTopic(subscriptionID, topic), handlers)
 	if err != nil {
 		return err
 	}
@@ -137,10 +137,11 @@ func handle(msg []byte, handler api.MessageHandler) {
 // Umb uses identified consumer queues acting as (virtual) topics
 // for subscriptions, the full queue name is based on the pattern:
 // "Consumer.$SERVICE_ACCOUNT_NAME.$SUBSCRIPTION_ID.VirtualTopic.>"
-func umbTopic(topic string) string {
-	subscriptionId := strings.Split(topic, ".")
-	return fmt.Sprintf("Consumer.%s.%s.%s",
+func umbTopic(subscriptionID, topic string) string {
+	topicCrumbs := strings.Split(topic, ".")
+	return fmt.Sprintf("Consumer.%s.%s-%s.%s",
 		_umb.consumerID,
-		subscriptionId[len(subscriptionId)-1],
+		subscriptionID,
+		topicCrumbs[len(topicCrumbs)-1],
 		topic)
 }