MF-1630 - Replace old subscriptions with a new one instead of throwing an error

pkg/messaging/mqtt/pubsub.go

@@ -25,7 +25,6 @@ var (
 	errSubscribeTimeout       = errors.New("failed to subscribe due to timeout reached")
 	errUnsubscribeTimeout     = errors.New("failed to unsubscribe due to timeout reached")
 	errUnsubscribeDeleteTopic = errors.New("failed to unsubscribe due to deletion of topic")
-	errAlreadySubscribed      = errors.New("already subscribed to topic")
 	errNotSubscribed          = errors.New("not subscribed")
 	errEmptyTopic             = errors.New("empty topic")
 	errEmptyID                = errors.New("empty ID")
@@ -80,7 +79,20 @@ func (ps pubsub) Subscribe(id, topic string, handler messaging.MessageHandler) e
 	case true:
 		// Check topic
 		if ok = s.contains(topic); ok {
-			return errAlreadySubscribed
+			// Unlocking, so that Unsubscribe() can access ps.subscriptions
+			ps.mu.Unlock()
+			if err := ps.Unsubscribe(id, topic); err != nil {
+				return err
+			}
+			ps.mu.Lock()
+			client, err := newClient(ps.address, id, ps.timeout)
+			if err != nil {
+				return err
+			}
+			s = subscription{
+				client: client,
+				topics: []string{topic},
+			}
 		}
 		s.topics = append(s.topics, topic)
 	default:

pkg/messaging/nats/pubsub.go

@@ -69,6 +69,7 @@ func (ps *pubsub) Subscribe(id, topic string, handler messaging.MessageHandler)
 	if topic == "" {
 		return ErrEmptyTopic
 	}
+
 	ps.mu.Lock()
 	defer ps.mu.Unlock()
 	// Check topic
@@ -77,7 +78,14 @@ func (ps *pubsub) Subscribe(id, topic string, handler messaging.MessageHandler)
 	case true:
 		// Check topic ID
 		if _, ok := s[id]; ok {
-			return ErrAlreadySubscribed
+			// Unlocking, so that Unsubscribe() can access ps.subscriptions
+			ps.mu.Unlock()
+			if err := ps.Unsubscribe(id, topic); err != nil {
+				return err
+			}
+			ps.mu.Lock()
+			s = make(map[string]subscription)
+			ps.subscriptions[topic] = s
 		}
 	default:
 		s = make(map[string]subscription)
@@ -104,6 +112,7 @@ func (ps *pubsub) Subscribe(id, topic string, handler messaging.MessageHandler)
 		Subscription: sub,
 		cancel:       handler.Cancel,
 	}
+
 	return nil
 }
 

pkg/messaging/nats/pubsub_test.go

@@ -107,7 +107,7 @@ func TestPubsub(t *testing.T) {
 			desc:         "Subscribe to an already subscribed topic with an ID",
 			topic:        fmt.Sprintf("%s.%s", chansPrefix, topic),
 			clientID:     "clientid1",
-			errorMessage: nats.ErrAlreadySubscribed,
+			errorMessage: nil,
 			pubsub:       true,
 		},
 		{
@@ -156,7 +156,7 @@ func TestPubsub(t *testing.T) {
 			desc:         "Subscribe to an already subscribed topic with a subtopic with an ID",
 			topic:        fmt.Sprintf("%s.%s.%s", chansPrefix, topic, subtopic),
 			clientID:     "clientidd1",
-			errorMessage: nats.ErrAlreadySubscribed,
+			errorMessage: nil,
 			pubsub:       true,
 		},
 		{

pkg/messaging/rabbitmq/pubsub.go

@@ -79,7 +79,15 @@ func (ps *pubsub) Subscribe(id, topic string, handler messaging.MessageHandler)
 	case true:
 		// Check topic ID
 		if _, ok := s[id]; ok {
-			return ErrAlreadySubscribed
+			// Unlocking, so Unsubscribe() can access ps.subscriptions
+			ps.mu.Unlock()
+			if err := ps.Unsubscribe(id, topic); err != nil {
+				return err
+			}
+			ps.mu.Lock()
+
+			s = make(map[string]subscription)
+			ps.subscriptions[topic] = s
 		}
 	default:
 		s = make(map[string]subscription)

pkg/messaging/rabbitmq/pubsub_test.go

@@ -104,7 +104,7 @@ func TestPubsub(t *testing.T) {
 			desc:         "Subscribe to an already subscribed topic with an ID",
 			topic:        fmt.Sprintf("%s.%s", chansPrefix, topic),
 			clientID:     "clientid1",
-			errorMessage: rabbitmq.ErrAlreadySubscribed,
+			errorMessage: nil,
 			pubsub:       true,
 		},
 		{
@@ -153,7 +153,7 @@ func TestPubsub(t *testing.T) {
 			desc:         "Subscribe to an already subscribed topic with a subtopic with an ID",
 			topic:        fmt.Sprintf("%s.%s.%s", chansPrefix, topic, subtopic),
 			clientID:     "clientidd1",
-			errorMessage: rabbitmq.ErrAlreadySubscribed,
+			errorMessage: nil,
 			pubsub:       true,
 		},
 		{