feat: support nameserver resolver (#457)

use nameserver resolver instead of nameserver address, we can pass
passThrough for direct address, httpResolver for a domain, an env resolver
for env param.

.travis.yml

@@ -3,9 +3,8 @@ language: go
 go:
   - "1.11.x"
   - "1.12.x"
-
+  - "1.13.x"
 go_import_path: github.com/apache/rocketmq-client-go/v2
-
 env:
   global:
     - NAME_SERVER_ADDRESS=127.0.0.1:9876
@@ -27,7 +26,10 @@ before_script:
   - nohup sh bin/mqbroker -n localhost:9876 &
 
 script:
+  - cd ${TRAVIS_HOME}
+  - ls -al
   - cd ${GOPATH}/src/github.com/apache/rocketmq-client-go/v2
+  - ls -al
   - go fmt ./... && [[ -z `git status -s` ]]
   - go mod vendor && go test ./... -coverprofile=coverage.txt -covermode=atomic
 

api.go

@@ -19,6 +19,7 @@ package rocketmq
 
 import (
 	"context"
+
 	"github.com/pkg/errors"
 
 	"github.com/apache/rocketmq-client-go/v2/consumer"

consumer/consumer.go

@@ -27,7 +27,8 @@ import (
 	"sync/atomic"
 	"time"
 
-	"github.com/json-iterator/go"
+	jsoniter "github.com/json-iterator/go"
+
 	"github.com/pkg/errors"
 	"github.com/tidwall/gjson"
 
@@ -273,9 +274,6 @@ type defaultConsumer struct {
 }
 
 func (dc *defaultConsumer) start() error {
-	if len(dc.option.NameServerAddrs) == 0 {
-		dc.namesrv.UpdateNameServerAddress(dc.option.NameServerDomain, dc.option.InstanceName)
-	}
 	if dc.model == Clustering {
 		// set retry topic
 		retryTopic := internal.GetRetryTopic(dc.consumerGroup)

consumer/offset_store.go

@@ -26,7 +26,7 @@ import (
 	"sync"
 	"time"
 
-	"github.com/json-iterator/go"
+	jsoniter "github.com/json-iterator/go"
 
 	"github.com/apache/rocketmq-client-go/v2/internal"
 	"github.com/apache/rocketmq-client-go/v2/internal/remote"
@@ -255,12 +255,6 @@ func (r *remoteBrokerOffsetStore) persist(mqs []*primitive.MessageQueue) {
 				rlog.LogKeyUnderlayError: err.Error(),
 				"offset":                 off,
 			})
-		} else {
-			rlog.Info("update offset to broker success", map[string]interface{}{
-				rlog.LogKeyConsumerGroup: r.group,
-				rlog.LogKeyMessageQueue:  mq.String(),
-				"offset":                 off,
-			})
 		}
 	}
 }

consumer/option.go

@@ -104,6 +104,8 @@ type consumerOptions struct {
 	//
 	AutoCommit            bool
 	RebalanceLockInterval time.Duration
+
+	Resolver primitive.NsResolver
 }
 
 func defaultPushConsumerOptions() consumerOptions {
@@ -115,6 +117,7 @@ func defaultPushConsumerOptions() consumerOptions {
 		MaxReconsumeTimes:          -1,
 		ConsumerModel:              Clustering,
 		AutoCommit:                 true,
+		Resolver:                   primitive.NewHttpResolver("DEFAULT"),
 	}
 	opts.ClientOptions.GroupName = "DEFAULT_CONSUMER"
 	return opts
@@ -125,6 +128,7 @@ type Option func(*consumerOptions)
 func defaultPullConsumerOptions() consumerOptions {
 	opts := consumerOptions{
 		ClientOptions: internal.DefaultClientOptions(),
+		Resolver:      primitive.NewHttpResolver("DEFAULT"),
 	}
 	opts.ClientOptions.GroupName = "DEFAULT_CONSUMER"
 	return opts
@@ -179,20 +183,6 @@ func WithInstance(name string) Option {
 	}
 }
 
-// WithNameServer set NameServer address, only support one NameServer cluster in alpha2
-func WithNameServer(nameServers primitive.NamesrvAddr) Option {
-	return func(opts *consumerOptions) {
-		opts.NameServerAddrs = nameServers
-	}
-}
-
-// WithNameServerDomain set NameServer domain
-func WithNameServerDomain(nameServerUrl string) Option {
-	return func(opts *consumerOptions) {
-		opts.NameServerDomain = nameServerUrl
-	}
-}
-
 // WithNamespace set the namespace of consumer
 func WithNamespace(namespace string) Option {
 	return func(opts *consumerOptions) {
@@ -263,3 +253,24 @@ func WithPullInterval(interval time.Duration) Option {
 		options.PullInterval = interval
 	}
 }
+
+// WithNsResovler set nameserver resolver to fetch nameserver addr
+func WithNsResovler(resolver primitive.NsResolver) Option {
+	return func(options *consumerOptions) {
+		options.Resolver = resolver
+	}
+}
+
+// WithNameServer set NameServer address, only support one NameServer cluster in alpha2
+func WithNameServer(nameServers primitive.NamesrvAddr) Option {
+	return func(options *consumerOptions) {
+		options.Resolver = primitive.NewPassthroughResolver(nameServers)
+	}
+}
+
+// WithNameServerDomain set NameServer domain
+func WithNameServerDomain(nameServerUrl string) Option {
+	return func(opts *consumerOptions) {
+		opts.Resolver = primitive.NewHttpResolver("DEFAULT", nameServerUrl)
+	}
+}

consumer/pull_consumer.go

@@ -76,7 +76,7 @@ func NewPullConsumer(options ...Option) (*defaultPullConsumer, error) {
 		apply(&defaultOpts)
 	}
 
-	srvs, err := internal.NewNamesrv(defaultOpts.NameServerAddrs)
+	srvs, err := internal.NewNamesrv(defaultOpts.Resolver)
 	if err != nil {
 		return nil, errors.Wrap(err, "new Namesrv failed.")
 	}

consumer/push_consumer.go

@@ -76,7 +76,7 @@ func NewPushConsumer(opts ...Option) (*pushConsumer, error) {
 	for _, apply := range opts {
 		apply(&defaultOpts)
 	}
-	srvs, err := internal.NewNamesrv(defaultOpts.NameServerAddrs)
+	srvs, err := internal.NewNamesrv(defaultOpts.Resolver)
 	if err != nil {
 		return nil, errors.Wrap(err, "new Namesrv failed.")
 	}

consumer/push_consumer_test.go

@@ -20,11 +20,12 @@ package consumer
 import (
 	"context"
 	"fmt"
+	"testing"
+
 	"github.com/apache/rocketmq-client-go/v2/internal"
 	"github.com/apache/rocketmq-client-go/v2/primitive"
 	"github.com/golang/mock/gomock"
 	. "github.com/smartystreets/goconvey/convey"
-	"testing"
 )
 
 func mockB4Start(c *pushConsumer) {
@@ -35,7 +36,7 @@ func TestStart(t *testing.T) {
 	Convey("test Start method", t, func() {
 		c, _ := NewPushConsumer(
 			WithGroupName("testGroup"),
-			WithNameServer([]string{"127.0.0.1:9876"}),
+			WithNsResovler(primitive.NewPassthroughResolver([]string{"127.0.0.1:9876"})),
 			WithConsumerModel(BroadCasting),
 		)
 

consumer/strategy_test.go

@@ -19,9 +19,10 @@ package consumer
 
 import (
 	"fmt"
+	"testing"
+
 	"github.com/apache/rocketmq-client-go/v2/primitive"
 	. "github.com/smartystreets/goconvey/convey"
-	"testing"
 )
 
 func TestAllocateByAveragely(t *testing.T) {

examples/consumer/acl/main.go

@@ -31,7 +31,7 @@ import (
 func main() {
 	c, err := rocketmq.NewPushConsumer(
 		consumer.WithGroupName("testGroup"),
-		consumer.WithNameServer([]string{"127.0.0.1:9876"}),
+		consumer.WithNsResovler(primitive.NewPassthroughResolver([]string{"127.0.0.1:9876"})),
 		consumer.WithCredentials(primitive.Credentials{
 			AccessKey: "RocketMQ",
 			SecretKey: "12345678",

examples/consumer/broadcast/main.go

@@ -31,7 +31,7 @@ import (
 func main() {
 	c, _ := rocketmq.NewPushConsumer(
 		consumer.WithGroupName("testGroup"),
-		consumer.WithNameServer([]string{"127.0.0.1:9876"}),
+		consumer.WithNsResovler(primitive.NewPassthroughResolver([]string{"127.0.0.1:9876"})),
 		consumer.WithConsumeFromWhere(consumer.ConsumeFromFirstOffset),
 		consumer.WithConsumerModel(consumer.BroadCasting),
 	)

examples/consumer/delay/main.go

@@ -31,7 +31,7 @@ import (
 func main() {
 	c, _ := rocketmq.NewPushConsumer(
 		consumer.WithGroupName("testGroup"),
-		consumer.WithNameServer([]string{"127.0.0.1:9876"}),
+		consumer.WithNsResovler(primitive.NewPassthroughResolver([]string{"127.0.0.1:9876"})),
 	)
 	err := c.Subscribe("TopicTest", consumer.MessageSelector{}, func(ctx context.Context,
 		msgs ...*primitive.MessageExt) (consumer.ConsumeResult, error) {

examples/consumer/interceptor/main.go

@@ -31,7 +31,7 @@ import (
 func main() {
 	c, _ := rocketmq.NewPushConsumer(
 		consumer.WithGroupName("testGroup"),
-		consumer.WithNameServer([]string{"127.0.0.1:9876"}),
+		consumer.WithNsResovler(primitive.NewPassthroughResolver([]string{"127.0.0.1:9876"})),
 		consumer.WithConsumerModel(consumer.Clustering),
 		consumer.WithConsumeFromWhere(consumer.ConsumeFromFirstOffset),
 		consumer.WithInterceptor(UserFistInterceptor(), UserSecondInterceptor()))

examples/consumer/namespace/main.go

@@ -31,7 +31,7 @@ import (
 func main() {
 	c, err := rocketmq.NewPushConsumer(
 		consumer.WithGroupName("testGroup"),
-		consumer.WithNameServer([]string{"127.0.0.1:9876"}),
+		consumer.WithNsResovler(primitive.NewPassthroughResolver([]string{"127.0.0.1:9876"})),
 		consumer.WithCredentials(primitive.Credentials{
 			AccessKey: "RocketMQ",
 			SecretKey: "12345678",

examples/consumer/orderly/main.go

@@ -31,7 +31,7 @@ import (
 func main() {
 	c, _ := rocketmq.NewPushConsumer(
 		consumer.WithGroupName("testGroup"),
-		consumer.WithNameServer([]string{"127.0.0.1:9876"}),
+		consumer.WithNsResovler(primitive.NewPassthroughResolver([]string{"127.0.0.1:9876"})),
 		consumer.WithConsumerModel(consumer.Clustering),
 		consumer.WithConsumeFromWhere(consumer.ConsumeFromFirstOffset),
 		consumer.WithConsumerOrder(true),

examples/consumer/pull/main.go

@@ -31,7 +31,7 @@ import (
 func main() {
 	c, err := rocketmq.NewPullConsumer(
 		consumer.WithGroupName("testGroup"),
-		consumer.WithNameServer([]string{"127.0.0.1:9876"}),
+		consumer.WithNsResovler(primitive.NewPassthroughResolver([]string{"127.0.0.1:9876"})),
 	)
 	if err != nil {
 		rlog.Fatal(fmt.Sprintf("fail to new pullConsumer: %s", err), nil)

examples/consumer/retry/concurrent/main.go

@@ -37,7 +37,7 @@ import (
 func main() {
 	c, _ := rocketmq.NewPushConsumer(
 		consumer.WithGroupName("testGroup"),
-		consumer.WithNameServer([]string{"127.0.0.1:9876"}),
+		consumer.WithNsResovler(primitive.NewPassthroughResolver([]string{"127.0.0.1:9876"})),
 		consumer.WithConsumerModel(consumer.Clustering),
 	)
 

examples/consumer/retry/order/main.go

@@ -36,7 +36,7 @@ import (
 func main() {
 	c, _ := rocketmq.NewPushConsumer(
 		consumer.WithGroupName("testGroup"),
-		consumer.WithNameServer([]string{"127.0.0.1:9876"}),
+		consumer.WithNsResovler(primitive.NewPassthroughResolver([]string{"127.0.0.1:9876"})),
 		consumer.WithConsumerModel(consumer.Clustering),
 		consumer.WithConsumeFromWhere(consumer.ConsumeFromFirstOffset),
 		consumer.WithConsumerOrder(true),

examples/consumer/simple/main.go

@@ -31,7 +31,7 @@ import (
 func main() {
 	c, _ := rocketmq.NewPushConsumer(
 		consumer.WithGroupName("testGroup"),
-		consumer.WithNameServer([]string{"127.0.0.1:9876"}),
+		consumer.WithNsResovler(primitive.NewPassthroughResolver([]string{"127.0.0.1:9876"})),
 	)
 	err := c.Subscribe("test", consumer.MessageSelector{}, func(ctx context.Context,
 		msgs ...*primitive.MessageExt) (consumer.ConsumeResult, error) {

examples/consumer/strategy/main.go

@@ -31,7 +31,7 @@ import (
 func main() {
 	c, _ := rocketmq.NewPushConsumer(
 		consumer.WithGroupName("testGroup"),
-		consumer.WithNameServer([]string{"127.0.0.1:9876"}),
+		consumer.WithNsResovler(primitive.NewPassthroughResolver([]string{"127.0.0.1:9876"})),
 		consumer.WithStrategy(consumer.AllocateByAveragely),
 	)
 	err := c.Subscribe("TopicTest", consumer.MessageSelector{}, func(ctx context.Context,

examples/consumer/tag/main.go

@@ -31,7 +31,7 @@ import (
 func main() {
 	c, _ := rocketmq.NewPushConsumer(
 		consumer.WithGroupName("testGroup"),
-		consumer.WithNameServer([]string{"127.0.0.1:9876"}),
+		consumer.WithNsResovler(primitive.NewPassthroughResolver([]string{"127.0.0.1:9876"})),
 	)
 	selector := consumer.MessageSelector{
 		Type:       consumer.TAG,

examples/consumer/trace/main.go

@@ -37,7 +37,7 @@ func main() {
 
 	c, _ := rocketmq.NewPushConsumer(
 		consumer.WithGroupName("testGroup"),
-		consumer.WithNameServer(namesrvs),
+		consumer.WithNsResovler(primitive.NewPassthroughResolver([]string{"127.0.0.1:9876"})),
 		consumer.WithTrace(traceCfg),
 	)
 	err := c.Subscribe("TopicTest", consumer.MessageSelector{}, func(ctx context.Context,

examples/producer/acl/main.go

@@ -30,7 +30,7 @@ import (
 
 func main() {
 	p, err := rocketmq.NewProducer(
-		producer.WithNameServer([]string{"127.0.0.1:9876"}),
+		producer.WithNsResovler(primitive.NewPassthroughResolver([]string{"127.0.0.1:9876"})),
 		producer.WithRetry(2),
 		producer.WithCredentials(primitive.Credentials{
 			AccessKey: "RocketMQ",

examples/producer/async/main.go

@@ -31,7 +31,7 @@ import (
 // Package main implements a async producer to send message.
 func main() {
 	p, _ := rocketmq.NewProducer(
-		producer.WithNameServer([]string{"127.0.0.1:9876"}),
+		producer.WithNsResovler(primitive.NewPassthroughResolver([]string{"127.0.0.1:9876"})),
 		producer.WithRetry(2),
 		producer.WithQueueSelector(producer.NewManualQueueSelector()))
 

examples/producer/batch/main.go

@@ -30,7 +30,7 @@ import (
 
 func main() {
 	p, _ := rocketmq.NewProducer(
-		producer.WithNameServer([]string{"127.0.0.1:9876"}),
+		producer.WithNsResovler(primitive.NewPassthroughResolver([]string{"127.0.0.1:9876"})),
 		producer.WithRetry(2),
 	)
 	err := p.Start()

examples/producer/delay/main.go

@@ -29,7 +29,7 @@ import (
 
 func main() {
 	p, _ := rocketmq.NewProducer(
-		producer.WithNameServer([]string{"127.0.0.1:9876"}),
+		producer.WithNsResovler(primitive.NewPassthroughResolver([]string{"127.0.0.1:9876"})),
 		producer.WithRetry(2),
 	)
 	err := p.Start()

examples/producer/interceptor/main.go

@@ -30,7 +30,7 @@ import (
 
 func main() {
 	p, _ := rocketmq.NewProducer(
-		producer.WithNameServer([]string{"127.0.0.1:9876"}),
+		producer.WithNsResovler(primitive.NewPassthroughResolver([]string{"127.0.0.1:9876"})),
 		producer.WithRetry(2),
 		producer.WithInterceptor(UserFirstInterceptor(), UserSecondInterceptor()),
 	)

examples/producer/namespace/main.go

@@ -30,7 +30,7 @@ import (
 
 func main() {
 	p, err := rocketmq.NewProducer(
-		producer.WithNameServer([]string{"127.0.0.1:9876"}),
+		producer.WithNsResovler(primitive.NewPassthroughResolver([]string{"127.0.0.1:9876"})),
 		producer.WithRetry(2),
 		producer.WithCredentials(primitive.Credentials{
 			AccessKey: "RocketMQ",

examples/producer/simple/main.go

@@ -31,7 +31,7 @@ import (
 // Package main implements a simple producer to send message.
 func main() {
 	p, _ := rocketmq.NewProducer(
-		producer.WithNameServer([]string{"127.0.0.1:9876"}),
+		producer.WithNsResovler(primitive.NewPassthroughResolver([]string{"127.0.0.1:9876"})),
 		producer.WithRetry(2),
 	)
 	err := p.Start()

examples/producer/tag/main.go

@@ -29,7 +29,7 @@ import (
 
 func main() {
 	p, _ := rocketmq.NewProducer(
-		producer.WithNameServer([]string{"127.0.0.1:9876"}),
+		producer.WithNsResovler(primitive.NewPassthroughResolver([]string{"127.0.0.1:9876"})),
 		producer.WithRetry(2),
 	)
 	err := p.Start()