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

* Implement Unsubscribe method for push consumer
apache · Mar 16, 2021 · 8019e59 · 8019e59
1 parent c97d492
commit 8019e59
Show file tree

Hide file tree

Showing 2 changed files with 29 additions and 1 deletion.
diff --git a/consumer/push_consumer.go b/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
 }
 

diff --git a/consumer/push_consumer_test.go b/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)