Add Name method to Consumer interface

pulsar/consumer.go

@@ -197,4 +197,7 @@ type Consumer interface {
 	//            the message publish time where to reposition the subscription
 	//
 	SeekByTime(time time.Time) error
+
+	// Name returns the name of consumer.
+	Name() string
 }

pulsar/consumer_impl.go

@@ -74,6 +74,10 @@ func newConsumer(client *client, options ConsumerOptions) (Consumer, error) {
 		options.ReceiverQueueSize = 1000
 	}
 
+	if options.Name == "" {
+		options.Name = generateRandomName()
+	}
+
 	// did the user pass in a message channel?
 	messageCh := options.MessageChannel
 	if options.MessageChannel == nil {
@@ -136,12 +140,7 @@ func newInternalConsumer(client *client, options ConsumerOptions, topic string,
 		errorCh:                   make(chan error),
 		dlq:                       dlq,
 		log:                       log.WithField("topic", topic),
-	}
-
-	if options.Name != "" {
-		consumer.consumerName = options.Name
-	} else {
-		consumer.consumerName = generateRandomName()
+		consumerName:              options.Name,
 	}
 
 	err := consumer.internalTopicSubscribeToPartitions()
@@ -166,6 +165,11 @@ func newInternalConsumer(client *client, options ConsumerOptions, topic string,
 	return consumer, nil
 }
 
+// Name returns the name of consumer.
+func (c *consumer) Name() string {
+	return c.consumerName
+}
+
 func (c *consumer) internalTopicSubscribeToPartitions() error {
 	partitions, err := c.client.TopicPartitions(c.topic)
 	if err != nil {

pulsar/consumer_multitopic.go

@@ -32,7 +32,8 @@ import (
 type multiTopicConsumer struct {
 	options ConsumerOptions
 
-	messageCh chan ConsumerMessage
+	consumerName string
+	messageCh    chan ConsumerMessage
 
 	consumers map[string]Consumer
 
@@ -46,12 +47,13 @@ type multiTopicConsumer struct {
 func newMultiTopicConsumer(client *client, options ConsumerOptions, topics []string,
 	messageCh chan ConsumerMessage, dlq *dlqRouter) (Consumer, error) {
 	mtc := &multiTopicConsumer{
-		options:   options,
-		messageCh: messageCh,
-		consumers: make(map[string]Consumer, len(topics)),
-		closeCh:   make(chan struct{}),
-		dlq:       dlq,
-		log:       &log.Entry{},
+		options:      options,
+		messageCh:    messageCh,
+		consumers:    make(map[string]Consumer, len(topics)),
+		closeCh:      make(chan struct{}),
+		dlq:          dlq,
+		log:          &log.Entry{},
+		consumerName: options.Name,
 	}
 
 	var errs error
@@ -173,3 +175,8 @@ func (c *multiTopicConsumer) Seek(msgID MessageID) error {
 func (c *multiTopicConsumer) SeekByTime(time time.Time) error {
 	return errors.New("seek command not allowed for multi topic consumer")
 }
+
+// Name returns the name of consumer.
+func (c *multiTopicConsumer) Name() string {
+	return c.consumerName
+}

pulsar/consumer_regex.go

@@ -59,6 +59,8 @@ type regexConsumer struct {
 	ticker *time.Ticker
 
 	log *log.Entry
+
+	consumerName string
 }
 
 func newRegexConsumer(c *client, opts ConsumerOptions, tn *internal.TopicName, pattern *regexp.Regexp,
@@ -78,7 +80,8 @@ func newRegexConsumer(c *client, opts ConsumerOptions, tn *internal.TopicName, p
 
 		closeCh: make(chan struct{}),
 
-		log: log.WithField("topic", tn.Name),
+		log:          log.WithField("topic", tn.Name),
+		consumerName: opts.Name,
 	}
 
 	topics, err := rc.topics()
@@ -222,6 +225,11 @@ func (c *regexConsumer) SeekByTime(time time.Time) error {
 	return errors.New("seek command not allowed for regex consumer")
 }
 
+// Name returns the name of consumer.
+func (c *regexConsumer) Name() string {
+	return c.consumerName
+}
+
 func (c *regexConsumer) closed() bool {
 	select {
 	case <-c.closeCh:

pulsar/consumer_test.go

@@ -1342,3 +1342,28 @@ func TestProducerName(t *testing.T) {
 		consumer.Ack(msg)
 	}
 }
+
+func TestConsumerName(t *testing.T) {
+	assert := assert.New(t)
+
+	client, err := NewClient(ClientOptions{
+		URL: lookupURL,
+	})
+	assert.Nil(err)
+	defer client.Close()
+
+	topic := newTopicName()
+
+	// create consumer
+	consumerName := "test-consumer-name"
+	consumer, err := client.Subscribe(ConsumerOptions{
+		Name:             consumerName,
+		Topic:            topic,
+		SubscriptionName: "my-sub",
+	})
+
+	assert.Nil(err)
+	defer consumer.Close()
+
+	assert.Equal(consumerName, consumer.Name())
+}