c := pulsar.ManagedConsumerConfig{
	QueueSize:           1,
}

package saga

import (
	"context"
	"crypto/tls"
	"crypto/x509"
	"github.com/leenux/pulsar-client-go"
	"github.com/stretchr/testify/suite"
	"go.uber.org/zap"
	"io/ioutil"
	"testing"
	"time"
)

var params = struct {
	pulsar        string
	tlsCert       string
	tlsKey        string
	tlsCA         string
	tlsSkipVerify bool
	name          string
	topic         string
	producer      bool
	message       string
	messageRate   time.Duration
	shared        bool
}{
	pulsar:        "localhost:6650",
	tlsCert:       "",
	tlsKey:        "",
	tlsCA:         "",
	tlsSkipVerify: false,
	name:          "demo",
	topic:         "persistent://sample/standalone/ns1/demo",
	producer:      false,
	message:       "--",
	messageRate:   time.Second,
	shared:        false,
}

var mcp *pulsar.ManagedClientPool
var mp *pulsar.ManagedProducer
var mc *pulsar.ManagedConsumer

type PulsarTestSuite struct {
	suite.Suite
}

func (suite *PulsarTestSuite) SetupTest() {

}

func (suite *PulsarTestSuite) TestReceive() {
	ctx, cancel := context.WithTimeout(context.Background(), 1*time.Second)
	defer cancel()
	_, err := mp.Send(ctx, []byte(time.Now().String()))
	if err != nil {
		zap.L().Error("Send",
			zap.Any("err", err),
		)
	}

	for i := 0; i < 1; i++ {
		message, err := mc.Receive(ctx)
		if err != nil && err.Error() != "context deadline exceeded" {
			zap.L().Error("Receive",
				zap.Any("err", err),
			)
			//break
		}
		zap.L().Debug("Receive",
			zap.Any("message", message),
			zap.Any("Payload", string(message.Payload)),
		)
	}
	zap.L().Debug("RedeliverUnacknowledged")
	err = mc.RedeliverUnacknowledged(ctx)
	if err != nil {
		zap.L().Error("RedeliverUnacknowledged",
			zap.Any("err", err),
		)
	}
	zap.L().Debug("Receive again")
	for {
		message, err := mc.Receive(ctx)
		if err != nil && err.Error() != "context deadline exceeded" {
			zap.L().Error("Receive2",
				zap.Any("err", err),
			)
			break
		}
		err = mc.Ack(ctx, message)
		if err != nil {
			zap.L().Error("Ack on Receive2",
				zap.Any("err", err),
			)
			break
		}
		zap.L().Debug("Receive2",
			zap.Any("message", message),
			zap.Any("Payload", string(message.Payload)),
		)
	}
}

func TestPulsarTestSuite(t *testing.T) {
	var logger *zap.Logger
	logger, _ = zap.NewDevelopment()
	defer logger.Sync()

	zap.ReplaceGlobals(logger)

	asyncErrs := make(chan error, 8)
	go func() {
		for err := range asyncErrs {
			zap.L().Error("asyncErrs",
				zap.Any("err", err),
			)
		}
	}()

	var tlsCfg *tls.Config
	if params.tlsCert != "" && params.tlsKey != "" {
		tlsCfg = &tls.Config{
			InsecureSkipVerify: params.tlsSkipVerify,
		}
		var err error
		cert, err := tls.LoadX509KeyPair(params.tlsCert, params.tlsKey)
		if err != nil {
			zap.L().Error("error loading certificates:",
				zap.Any("err", err),
			)
		}
		tlsCfg.Certificates = []tls.Certificate{cert}

		if params.tlsCA != "" {
			rootCA, err := ioutil.ReadFile(params.tlsCA)
			if err != nil {
				zap.L().Error("error loading certificate authority:",
					zap.Any("err", err),
				)
			}
			tlsCfg.RootCAs = x509.NewCertPool()
			tlsCfg.RootCAs.AppendCertsFromPEM(rootCA)
		}

		// Inspect certificate and print the CommonName attribute,
		// since this may be used for authorization
		if len(cert.Certificate[0]) > 0 {
			x509Cert, err := x509.ParseCertificate(cert.Certificate[0])
			if err != nil {
				zap.L().Error("error loading public certificate:",
					zap.Any("err", err),
				)
			}
			zap.L().Debug("Using certificate pair with CommonName=",
				zap.Any("x509Cert.Subject.CommonName", x509Cert.Subject.CommonName),
			)
		}
	}

	mcp = pulsar.NewManagedClientPool()
	mpCfg := pulsar.ManagedProducerConfig{
		Name:                  params.name,
		Topic:                 params.topic,
		NewProducerTimeout:    time.Second,
		InitialReconnectDelay: time.Second,
		MaxReconnectDelay:     time.Minute,
		ManagedClientConfig: pulsar.ManagedClientConfig{
			ClientConfig: pulsar.ClientConfig{
				Addr:      params.pulsar,
				TLSConfig: tlsCfg,
				Errs:      asyncErrs,
			},
		},
	}
	mcCfg := pulsar.ManagedConsumerConfig{
		Name:                  params.name,
		Topic:                 params.topic,
		Exclusive:             true,
		NewConsumerTimeout:    time.Second,
		InitialReconnectDelay: time.Second,
		MaxReconnectDelay:     time.Minute,
		ManagedClientConfig: pulsar.ManagedClientConfig{
			ClientConfig: pulsar.ClientConfig{
				Addr:      params.pulsar,
				TLSConfig: tlsCfg,
				Errs:      asyncErrs,
			},
		},
	}
	mp = pulsar.NewManagedProducer(mcp, mpCfg)
	mc = pulsar.NewManagedConsumer(mcp, mcCfg)
	// Allow consumer time to connect
	time.Sleep(time.Millisecond * 500)
	suite.Run(t, new(PulsarTestSuite))
}

=== RUN   TestPulsarTestSuite
--- PASS: TestPulsarTestSuite (1.51s)
=== RUN   TestPulsarTestSuite/TestReceive
2018-12-11T16:44:23.373+0800	DEBUG	saga/pulsar_test.go:71	Receive	{"message": {"Topic":"persistent://sample/standalone/ns1/demo","Msg":{"consumer_id":0,"message_id":{"ledgerId":1446,"entryId":17,"partition":-1}},"Meta":{"producer_name":"demo","sequence_id":0,"publish_time":1544517863000,"compression":0},"Payload":"MjAxOC0xMi0xMSAxNjo0NDoyMy4zNjE4OTkxMTkgKzA4MDAgQ1NUIG09KzAuNTEzNDc4ODU0"}, "Payload": "2018-12-11 16:44:23.361899119 +0800 CST m=+0.513478854"}
2018-12-11T16:44:23.373+0800	DEBUG	saga/pulsar_test.go:76	RedeliverUnacknowledged
2018-12-11T16:44:23.373+0800	DEBUG	saga/pulsar_test.go:83	Receive again
2018-12-11T16:44:23.378+0800	DEBUG	saga/pulsar_test.go:99	Receive2	{"message": {"Topic":"persistent://sample/standalone/ns1/demo","Msg":{"consumer_id":0,"message_id":{"ledgerId":1446,"entryId":17,"partition":-1}},"Meta":{"producer_name":"demo","sequence_id":0,"publish_time":1544517863000,"compression":0},"Payload":"MjAxOC0xMi0xMSAxNjo0NDoyMy4zNjE4OTkxMTkgKzA4MDAgQ1NUIG09KzAuNTEzNDc4ODU0"}, "Payload": "2018-12-11 16:44:23.361899119 +0800 CST m=+0.513478854"}
2018-12-11T16:44:24.363+0800	ERROR	saga/pulsar_test.go:94	Ack on Receive2	{"err": "proto: repeated field MessageId has nil element"}
...
...
Process finished with exit code 0