[go client] add SubscriptionInitPos option in ConsumerOptions (apache…

…#3588)

* [go client] add SubscriptionInitPos option in ConsumerOptions

Signed-off-by: xiaolong.ran <[email protected]>

* add comment for SubscriptionInitPos option

Signed-off-by: xiaolong.ran <[email protected]>

pulsar-client-go/pulsar/c_consumer.go

@@ -100,6 +100,10 @@ func subscribeAsync(client *client, options ConsumerOptions, callback func(Consu
 		C.pulsar_consumer_configuration_set_consumer_type(conf, C.pulsar_consumer_type(options.Type))
 	}
 
+	if options.SubscriptionInitPos != Latest {
+		C.pulsar_consumer_set_subscription_initial_position(conf, C.initial_position(options.SubscriptionInitPos))
+	}
+
 	// ReceiverQueueSize==0 means to use the default queue size
 	// -1 means to disable the consumer prefetching
 	if options.ReceiverQueueSize > 0 {

pulsar-client-go/pulsar/consumer.go

@@ -46,6 +46,16 @@ const (
 	Failover
 )
 
+type InitialPosition int
+
+const (
+	// Latest position which means the start consuming position will be the last message
+	Latest InitialPosition = iota
+
+	// Earliest position which means the start consuming position will be the first message
+	Earliest
+)
+
 // ConsumerBuilder is used to configure and create instances of Consumer
 type ConsumerOptions struct {
 	// Specify the topic this consumer will subscribe on.
@@ -77,6 +87,10 @@ type ConsumerOptions struct {
 	// Default is `Exclusive`
 	Type SubscriptionType
 
+	// InitialPosition at which the cursor will be set when subscribe
+	// Default is `Latest`
+	SubscriptionInitPos InitialPosition
+
 	// Sets a `MessageChannel` for the consumer
 	// When a message is received, it will be pushed to the channel for consumption
 	MessageChannel chan ConsumerMessage

pulsar-client-go/pulsar/consumer_test.go

@@ -454,7 +454,53 @@ func TestConsumer_Seek(t *testing.T) {
 	msg, err := consumer.Receive(ctx)
 	assert.Nil(t, err)
 	t.Logf("again received message:%+v", msg.ID())
-	assert.Equal(t,string(msg.Payload()),"msg-content-0")
+	assert.Equal(t, "msg-content-0", string(msg.Payload()))
 
 	consumer.Unsubscribe()
 }
+
+func TestConsumer_SubscriptionInitPos(t *testing.T) {
+	client, err := NewClient(ClientOptions{
+		URL: "pulsar://localhost:6650",
+	})
+
+	assert.Nil(t, err)
+	defer client.Close()
+
+	topicName := "persistent://public/default/testSeek"
+	subName := "test-subscription-initial-earliest-position"
+
+	// create producer
+	producer, err := client.CreateProducer(ProducerOptions{
+		Topic: topicName,
+	})
+	assert.Nil(t, err)
+	defer producer.Close()
+
+	//sent message
+	ctx := context.Background()
+
+	err = producer.Send(ctx, ProducerMessage{
+		Payload: []byte("msg-1-content-1"),
+	})
+	assert.Nil(t, err)
+
+	err = producer.Send(ctx, ProducerMessage{
+		Payload: []byte("msg-1-content-2"),
+	})
+	assert.Nil(t, err)
+
+	// create consumer
+	consumer, err := client.Subscribe(ConsumerOptions{
+		Topic:               topicName,
+		SubscriptionName:    subName,
+		SubscriptionInitPos: Earliest,
+	})
+	assert.Nil(t, err)
+	defer consumer.Close()
+
+	msg, err := consumer.Receive(ctx)
+	assert.Nil(t, err)
+
+	assert.Equal(t, "msg-1-content-1", string(msg.Payload()))
+}