Merge pull request #9 from Fanatics/consumer_topic

export topic name

pulsar/consumer.go

@@ -255,4 +255,7 @@ type Consumer interface {
 
 	// Name returns the name of consumer.
 	Name() string
+
+	// TopicName returns the topic of the single topic consumer
+	TopicName() (string, error)
 }

pulsar/consumer_impl.go

@@ -690,3 +690,7 @@ func addMessageCryptoIfMissing(client *client, options *ConsumerOptions, topics
 	}
 	return nil
 }
+
+func (c *consumer) TopicName() (string, error) {
+	return c.topic, nil
+}

pulsar/consumer_multitopic.go

@@ -229,3 +229,7 @@ func (c *multiTopicConsumer) SeekByTime(time time.Time) error {
 func (c *multiTopicConsumer) Name() string {
 	return c.consumerName
 }
+
+func (c *multiTopicConsumer) TopicName() (string, error) {
+	return "", newError(SeekFailed, "topic command not allowed for multi topic consumer")
+}

pulsar/consumer_partition.go

@@ -552,6 +552,8 @@ func (pc *partitionConsumer) internalSeekByTime(seek *seekByTimeRequest) {
 		seek.err = err
 		return
 	}
+	pc.log.Debugf("Successfully reset subscriptio to timestamp : %v",
+		uint64(seek.publishTime.UnixNano()/int64(time.Millisecond)))
 	pc.clearMessageChannels()
 }
 

pulsar/consumer_regex.go

@@ -248,6 +248,10 @@ func (c *regexConsumer) SeekByTime(time time.Time) error {
 	return newError(SeekFailed, "seek command not allowed for regex consumer")
 }
 
+func (c *regexConsumer) TopicName() (string, error) {
+	return "", newError(SeekFailed, "topic command not allowed for multi topic consumer")
+}
+
 // Name returns the name of consumer.
 func (c *regexConsumer) Name() string {
 	return c.consumerName

pulsar/internal/pulsartracing/consumer_interceptor_test.go

@@ -91,3 +91,7 @@ func (c *mockConsumer) SeekByTime(time time.Time) error {
 func (c *mockConsumer) Name() string {
 	return ""
 }
+
+func (c *mockConsumer) TopicName() (string, error) {
+	return "", nil
+}