Add pulsar reader implementation (#11990)

Signed-off-by: godchen <qingxiang.chen@zilliz.com>

internal/util/mqclient/pulsar_client.go

@@ -55,6 +55,23 @@ func (pc *pulsarClient) CreateProducer(options ProducerOptions) (Producer, error
 	return producer, nil
 }
 
+func (pc *pulsarClient) CreateReader(options ReaderOptions) (Reader, error) {
+	opts := pulsar.ReaderOptions{
+		Topic:                   options.Topic,
+		StartMessageID:          options.StartMessageID,
+		StartMessageIDInclusive: options.StartMessageIDInclusive,
+	}
+	pr, err := pc.client.CreateReader(opts)
+	if err != nil {
+		return nil, err
+	}
+	if pr == nil {
+		return nil, errors.New("pulsar is not ready, producer is nil")
+	}
+	reader := &pulsarReader{r: pr}
+	return reader, nil
+}
+
 func (pc *pulsarClient) Subscribe(options ConsumerOptions) (Consumer, error) {
 	receiveChannel := make(chan pulsar.ConsumerMessage, options.BufSize)
 	consumer, err := pc.client.Subscribe(pulsar.ConsumerOptions{

internal/util/mqclient/pulsar_reader.go

@@ -0,0 +1,41 @@
+package mqclient
+
+import (
+	"context"
+
+	"github.com/apache/pulsar-client-go/pulsar"
+)
+
+type pulsarReader struct {
+	r pulsar.Reader
+}
+
+func (pr *pulsarReader) Topic() string {
+	return pr.r.Topic()
+}
+
+func (pr *pulsarReader) Next(ctx context.Context) (Message, error) {
+	pm, err := pr.r.Next(ctx)
+	if err != nil {
+		return nil, err
+	}
+
+	return &pulsarMessage{msg: pm}, nil
+}
+
+func (pr *pulsarReader) HasNext() bool {
+	return pr.r.HasNext()
+}
+
+func (pr *pulsarReader) Close() {
+	pr.r.Close()
+}
+
+func (pr *pulsarReader) Seek(id MessageID) error {
+	messageID := id.(*pulsarID).messageID
+	err := pr.r.Seek(messageID)
+	if err != nil {
+		return err
+	}
+	return nil
+}

internal/util/mqclient/pulsar_reader_test.go

@@ -0,0 +1,105 @@
+// Copyright (C) 2019-2020 Zilliz. All rights reserved.
+//
+// Licensed under the Apache License, Version 2.0 (the "License"); you may not use this file except in compliance
+// with the License. You may obtain a copy of the License at
+//
+// http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software distributed under the License
+// is distributed on an "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express
+// or implied. See the License for the specific language governing permissions and limitations under the License.
+
+package mqclient
+
+import (
+	"context"
+	"fmt"
+	"math/rand"
+	"testing"
+	"time"
+
+	"github.com/apache/pulsar-client-go/pulsar"
+	"github.com/stretchr/testify/assert"
+)
+
+func TestPulsarReader(t *testing.T) {
+	ctx := context.Background()
+	pulsarAddress, _ := Params.Load("_PulsarAddress")
+	pc, err := GetPulsarClientInstance(pulsar.ClientOptions{URL: pulsarAddress})
+	assert.Nil(t, err)
+	defer pc.Close()
+
+	rand.Seed(time.Now().UnixNano())
+	topic := fmt.Sprintf("test-%d", rand.Int())
+
+	producer, err := pc.CreateProducer(ProducerOptions{Topic: topic})
+	assert.Nil(t, err)
+	assert.NotNil(t, producer)
+
+	defer producer.Close()
+
+	const N = 10
+	var seekID MessageID
+	for i := 0; i < N; i++ {
+		msg := &ProducerMessage{
+			Payload:    []byte(fmt.Sprintf("helloworld-%d", i)),
+			Properties: map[string]string{},
+		}
+
+		id, err := producer.Send(ctx, msg)
+		assert.Nil(t, err)
+		if i == 4 {
+			seekID = &pulsarID{messageID: id.(*pulsarID).messageID}
+		}
+	}
+
+	reader, err := pc.CreateReader(ReaderOptions{
+		Topic:          topic,
+		StartMessageID: pulsar.EarliestMessageID(),
+	})
+	assert.Nil(t, err)
+	assert.NotNil(t, reader)
+	defer reader.Close()
+
+	str := reader.Topic()
+	assert.NotNil(t, str)
+
+	for i := 0; i < N; i++ {
+		revMsg, err := reader.Next(ctx)
+		assert.Nil(t, err)
+		assert.NotNil(t, revMsg)
+	}
+
+	readerOfStartMessageID, err := pc.CreateReader(ReaderOptions{
+		Topic:                   topic,
+		StartMessageID:          seekID,
+		StartMessageIDInclusive: true,
+	})
+	assert.Nil(t, err)
+	defer readerOfStartMessageID.Close()
+
+	for i := 4; i < N; i++ {
+		assert.True(t, readerOfStartMessageID.HasNext())
+		revMsg, err := readerOfStartMessageID.Next(ctx)
+		assert.Nil(t, err)
+		assert.NotNil(t, revMsg)
+	}
+
+	readerOfSeek, err := pc.CreateReader(ReaderOptions{
+		Topic:          topic,
+		StartMessageID: pulsar.EarliestMessageID(),
+	})
+	assert.Nil(t, err)
+	defer readerOfSeek.Close()
+
+	err = reader.Seek(seekID)
+	assert.Nil(t, err)
+
+	for i := 4; i < N; i++ {
+		assert.True(t, readerOfSeek.HasNext())
+		revMsg, err := readerOfSeek.Next(ctx)
+		assert.Nil(t, err)
+		assert.NotNil(t, revMsg)
+	}
+
+}