[ISSUE #487] Implement Unsubscribe method for push consumer (#626)

* Implement Unsubscribe method for push consumer

Author: maixiaohai

consumer/push_consumer.go

@@ -227,6 +227,14 @@ func (pc *pushConsumer) Subscribe(topic string, selector MessageSelector,
 		return errors.New("cannot subscribe topic since client either failed to start or has been shutdown.")
 	}
 
+	// add retry topic subscription for resubscribe
+	retryTopic := internal.GetRetryTopic(pc.consumerGroup)
+	_, exists := pc.subscriptionDataTable.Load(retryTopic)
+	if !exists {
+		sub := buildSubscriptionData(retryTopic, MessageSelector{TAG, _SubAll})
+		pc.subscriptionDataTable.Store(retryTopic, sub)
+	}
+
 	if pc.option.Namespace != "" {
 		topic = pc.option.Namespace + "%" + topic
 	}
@@ -241,7 +249,10 @@ func (pc *pushConsumer) Subscribe(topic string, selector MessageSelector,
 	return nil
 }
 
-func (pc *pushConsumer) Unsubscribe(string) error {
+func (pc *pushConsumer) Unsubscribe(topic string) error {
+	pc.subscriptionDataTable.Delete(topic)
+	retryTopic := internal.GetRetryTopic(pc.consumerGroup)
+	pc.subscriptionDataTable.Delete(retryTopic)
 	return nil
 }
 

consumer/push_consumer_test.go

@@ -52,6 +52,23 @@ func TestStart(t *testing.T) {
 			return ConsumeSuccess, nil
 		})
 
+		_, exists := c.subscriptionDataTable.Load("TopicTest")
+		So(exists, ShouldBeTrue)
+
+		err = c.Unsubscribe("TopicTest")
+		So(err, ShouldBeNil)
+		_, exists = c.subscriptionDataTable.Load("TopicTest")
+		So(exists, ShouldBeFalse)
+
+		err = c.Subscribe("TopicTest", MessageSelector{}, func(ctx context.Context,
+			msgs ...*primitive.MessageExt) (ConsumeResult, error) {
+			fmt.Printf("subscribe callback: %v \n", msgs)
+			return ConsumeSuccess, nil
+		})
+
+		_, exists = c.subscriptionDataTable.Load("TopicTest")
+		So(exists, ShouldBeTrue)
+
 		client.EXPECT().ClientID().Return("127.0.0.1@DEFAULT")
 		client.EXPECT().Start().Return()
 		client.EXPECT().RegisterConsumer(gomock.Any(), gomock.Any()).Return(nil)