Removed unacked messages tracker (apache#90)

* Removed unacked messages tracker

* Fixed tests

pulsar/consumer.go

@@ -19,7 +19,6 @@ package pulsar
 
 import (
 	"context"
-	"time"
 )
 
 // Pair of a Consumer and Message
@@ -82,11 +81,6 @@ type ConsumerOptions struct {
 	// This properties will be visible in the topic stats
 	Properties map[string]string
 
-	// Set the timeout for unacked messages
-	// Message not acknowledged within the give time, will be replayed by the broker to the same or a different consumer
-	// Default is 0, which means message are not being replayed based on ack time
-	AckTimeout time.Duration
-
 	// Select the subscription type to be used when subscribing to the topic.
 	// Default is `Exclusive`
 	Type SubscriptionType
@@ -128,9 +122,6 @@ type ConsumerOptions struct {
 
 // Consumer is an interface that abstracts behavior of Pulsar's consumer
 type Consumer interface {
-	// Topic get the topic for the consumer
-	Topic() string
-
 	// Subscription get a subscription for the consumer
 	Subscription() string
 

pulsar/consumer_impl.go

@@ -34,8 +34,6 @@ import (
 var ErrConsumerClosed = errors.New("consumer closed")
 
 type consumer struct {
-	topic string
-
 	options ConsumerOptions
 
 	consumers []*partitionConsumer
@@ -88,7 +86,6 @@ func newConsumer(client *client, options ConsumerOptions) (Consumer, error) {
 func topicSubscribe(client *client, options ConsumerOptions, topic string,
 	messageCh chan ConsumerMessage) (Consumer, error) {
 	consumer := &consumer{
-		topic:     topic,
 		messageCh: messageCh,
 		errorCh:   make(chan error),
 		log:       log.WithField("topic", topic),
@@ -165,10 +162,6 @@ func topicSubscribe(client *client, options ConsumerOptions, topic string,
 	return consumer, nil
 }
 
-func (c *consumer) Topic() string {
-	return c.topic
-}
-
 func (c *consumer) Subscription() string {
 	return c.options.SubscriptionName
 }
@@ -177,7 +170,7 @@ func (c *consumer) Unsubscribe() error {
 	var errMsg string
 	for _, consumer := range c.consumers {
 		if err := consumer.Unsubscribe(); err != nil {
-			errMsg += fmt.Sprintf("topic %s, subscription %s: %s", c.Topic(), c.Subscription(), err)
+			errMsg += fmt.Sprintf("topic %s, subscription %s: %s", consumer.topic, c.Subscription(), err)
 		}
 	}
 	if errMsg != "" {

pulsar/consumer_test.go

@@ -149,7 +149,6 @@ func TestBatchMessageReceive(t *testing.T) {
 		SubscriptionName: subName,
 	})
 	assert.Nil(t, err)
-	assert.Equal(t, topicName, consumer.Topic())
 	count := 0
 
 	for i := 0; i < numOfMessages; i++ {
@@ -397,7 +396,6 @@ func TestConsumerReceiveTimeout(t *testing.T) {
 		Topic:            topic,
 		SubscriptionName: "my-sub1",
 		Type:             Shared,
-		AckTimeout:       5 * 1000,
 	})
 	assert.Nil(t, err)
 	defer consumer.Close()