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")
@@ -47,6 +46,7 @@ type pubsub struct {
 	subscriptions map[string]subscription
 }
 
+// NewPubSub returns MQTT message publisher/subscriber.
 func NewPubSub(url, queue string, timeout time.Duration, logger log.Logger) (messaging.PubSub, error) {
 	client, err := newClient(url, "mqtt-publisher", timeout)
 	if err != nil {
@@ -80,7 +80,23 @@ 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()
+			err := ps.Unsubscribe(id, topic)
+			ps.mu.Lock() // Lock so that deferred unlock handle it
+			if err != nil {
+				return err
+			}
+			if len(ps.subscriptions) == 0 {
+				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:
@@ -93,6 +109,7 @@ func (ps pubsub) Subscribe(id, topic string, handler messaging.MessageHandler) e
 			topics: []string{topic},
 		}
 	}
+
 	token := s.client.Subscribe(topic, qos, ps.mqttHandler(handler))
 	if token.Error() != nil {
 		return token.Error()

pkg/messaging/nats/pubsub.go

@@ -22,6 +22,7 @@ var (
 	ErrNotSubscribed     = errors.New("not subscribed")
 	ErrEmptyTopic        = errors.New("empty topic")
 	ErrEmptyID           = errors.New("empty id")
+	ErrFailed            = errors.New("failed")
 )
 
 var _ messaging.PubSub = (*pubsub)(nil)
@@ -69,20 +70,30 @@ func (ps *pubsub) Subscribe(id, topic string, handler messaging.MessageHandler)
 	if topic == "" {
 		return ErrEmptyTopic
 	}
+
 	ps.mu.Lock()
-	defer ps.mu.Unlock()
 	// Check topic
 	s, ok := ps.subscriptions[topic]
-	switch ok {
-	case true:
-		// Check topic ID
+	if ok {
+		// Check client 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()
+			// value of s can be changed while ps.mu is unlocked
+			s = ps.subscriptions[topic]
 		}
-	default:
+	}
+	defer ps.mu.Unlock()
+	if s == nil {
 		s = make(map[string]subscription)
 		ps.subscriptions[topic] = s
 	}
+
 	nh := ps.natsHandler(handler)
 
 	if ps.queue != "" {
@@ -104,6 +115,7 @@ func (ps *pubsub) Subscribe(id, topic string, handler messaging.MessageHandler)
 		Subscription: sub,
 		cancel:       handler.Cancel,
 	}
+
 	return nil
 }
 

pkg/messaging/nats/pubsub_test.go

@@ -88,124 +88,173 @@ func TestPubsub(t *testing.T) {
 		clientID     string
 		errorMessage error
 		pubsub       bool //true for subscribe and false for unsubscribe
+		handler      messaging.MessageHandler
 	}{
 		{
 			desc:         "Subscribe to a topic with an ID",
 			topic:        fmt.Sprintf("%s.%s", chansPrefix, topic),
 			clientID:     "clientid1",
 			errorMessage: nil,
 			pubsub:       true,
+			handler:      handler{false},
 		},
 		{
 			desc:         "Subscribe to the same topic with a different ID",
 			topic:        fmt.Sprintf("%s.%s", chansPrefix, topic),
 			clientID:     "clientid2",
 			errorMessage: nil,
 			pubsub:       true,
+			handler:      handler{false},
 		},
 		{
 			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,
+			handler:      handler{false},
 		},
 		{
 			desc:         "Unsubscribe from a topic with an ID",
 			topic:        fmt.Sprintf("%s.%s", chansPrefix, topic),
 			clientID:     "clientid1",
 			errorMessage: nil,
 			pubsub:       false,
+			handler:      handler{false},
 		},
 		{
 			desc:         "Unsubscribe from a non-existent topic with an ID",
 			topic:        "h",
 			clientID:     "clientid1",
 			errorMessage: nats.ErrNotSubscribed,
 			pubsub:       false,
+			handler:      handler{false},
 		},
 		{
 			desc:         "Unsubscribe from the same topic with a different ID",
 			topic:        fmt.Sprintf("%s.%s", chansPrefix, topic),
 			clientID:     "clientidd2",
 			errorMessage: nats.ErrNotSubscribed,
 			pubsub:       false,
+			handler:      handler{false},
 		},
 		{
 			desc:         "Unsubscribe from the same topic with a different ID not subscribed",
 			topic:        fmt.Sprintf("%s.%s", chansPrefix, topic),
 			clientID:     "clientidd3",
 			errorMessage: nats.ErrNotSubscribed,
 			pubsub:       false,
+			handler:      handler{false},
 		},
 		{
 			desc:         "Unsubscribe from an already unsubscribed topic with an ID",
 			topic:        fmt.Sprintf("%s.%s", chansPrefix, topic),
 			clientID:     "clientid1",
 			errorMessage: nats.ErrNotSubscribed,
 			pubsub:       false,
+			handler:      handler{false},
 		},
 		{
 			desc:         "Subscribe to a topic with a subtopic with an ID",
 			topic:        fmt.Sprintf("%s.%s.%s", chansPrefix, topic, subtopic),
 			clientID:     "clientidd1",
 			errorMessage: nil,
 			pubsub:       true,
+			handler:      handler{false},
 		},
 		{
 			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,
+			handler:      handler{false},
 		},
 		{
 			desc:         "Unsubscribe from a topic with a subtopic with an ID",
 			topic:        fmt.Sprintf("%s.%s.%s", chansPrefix, topic, subtopic),
 			clientID:     "clientidd1",
 			errorMessage: nil,
 			pubsub:       false,
+			handler:      handler{false},
 		},
 		{
 			desc:         "Unsubscribe from an already unsubscribed topic with a subtopic with an ID",
 			topic:        fmt.Sprintf("%s.%s.%s", chansPrefix, topic, subtopic),
 			clientID:     "clientid1",
 			errorMessage: nats.ErrNotSubscribed,
 			pubsub:       false,
+			handler:      handler{false},
 		},
 		{
 			desc:         "Subscribe to an empty topic with an ID",
 			topic:        "",
 			clientID:     "clientid1",
 			errorMessage: nats.ErrEmptyTopic,
 			pubsub:       true,
+			handler:      handler{false},
 		},
 		{
 			desc:         "Unsubscribe from an empty topic with an ID",
 			topic:        "",
 			clientID:     "clientid1",
 			errorMessage: nats.ErrEmptyTopic,
 			pubsub:       false,
+			handler:      handler{false},
 		},
 		{
 			desc:         "Subscribe to a topic with empty id",
 			topic:        fmt.Sprintf("%s.%s", chansPrefix, topic),
 			clientID:     "",
 			errorMessage: nats.ErrEmptyID,
 			pubsub:       true,
+			handler:      handler{false},
 		},
 		{
 			desc:         "Unsubscribe from a topic with empty id",
 			topic:        fmt.Sprintf("%s.%s", chansPrefix, topic),
 			clientID:     "",
 			errorMessage: nats.ErrEmptyID,
 			pubsub:       false,
+			handler:      handler{false},
+		},
+		{
+			desc:         "Subscribe to another topic with an ID",
+			topic:        fmt.Sprintf("%s.%s", chansPrefix, topic+"1"),
+			clientID:     "clientid3",
+			errorMessage: nil,
+			pubsub:       true,
+			handler:      handler{true},
+		},
+		{
+			desc:         "Subscribe to another already subscribed topic with an ID with Unsubscribe failing",
+			topic:        fmt.Sprintf("%s.%s", chansPrefix, topic+"1"),
+			clientID:     "clientid3",
+			errorMessage: nats.ErrFailed,
+			pubsub:       true,
+			handler:      handler{true},
+		},
+		{
+			desc:         "Subscribe to a new topic with an ID",
+			topic:        fmt.Sprintf("%s.%s", chansPrefix, topic+"2"),
+			clientID:     "clientid4",
+			errorMessage: nil,
+			pubsub:       true,
+			handler:      handler{true},
+		},
+		{
+			desc:         "Unsubscribe from a topic with an ID with failing handler",
+			topic:        fmt.Sprintf("%s.%s", chansPrefix, topic+"2"),
+			clientID:     "clientid4",
+			errorMessage: nats.ErrFailed,
+			pubsub:       false,
+			handler:      handler{true},
 		},
 	}
 
 	for _, pc := range subcases {
 		if pc.pubsub == true {
-			err := pubsub.Subscribe(pc.clientID, pc.topic, handler{})
+			err := pubsub.Subscribe(pc.clientID, pc.topic, pc.handler)
 			if pc.errorMessage == nil {
 				require.Nil(t, err, fmt.Sprintf("%s got unexpected error: %s", pc.desc, err))
 			} else {
@@ -222,13 +271,18 @@ func TestPubsub(t *testing.T) {
 	}
 }
 
-type handler struct{}
+type handler struct {
+	fail bool
+}
 
 func (h handler) Handle(msg messaging.Message) error {
 	msgChan <- msg
 	return nil
 }
 
 func (h handler) Cancel() error {
+	if h.fail {
+		return nats.ErrFailed
+	}
 	return nil
 }

pkg/messaging/rabbitmq/pubsub.go

@@ -26,6 +26,7 @@ var (
 	ErrNotSubscribed     = errors.New("not subscribed")
 	ErrEmptyTopic        = errors.New("empty topic")
 	ErrEmptyID           = errors.New("empty id")
+	ErrFailed            = errors.New("failed")
 )
 
 var _ messaging.PubSub = (*pubsub)(nil)
@@ -72,16 +73,24 @@ func (ps *pubsub) Subscribe(id, topic string, handler messaging.MessageHandler)
 		return ErrEmptyTopic
 	}
 	ps.mu.Lock()
-	defer ps.mu.Unlock()
 	// Check topic
 	s, ok := ps.subscriptions[topic]
-	switch ok {
-	case true:
-		// Check topic ID
+	if ok {
+		// Check client 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()
+			// value of s can be changed while ps.mu is unlocked
+			s = ps.subscriptions[topic]
 		}
-	default:
+	}
+	defer ps.mu.Unlock()
+	if s == nil {
 		s = make(map[string]subscription)
 		ps.subscriptions[topic] = s
 	}