fixes and improvements (#55)

* util/router: sync from github
* config: add watcher interface

Signed-off-by: Vasiliy Tolstov <v.tolstov@unistack.org>

broker/broker.go

@@ -18,14 +18,6 @@ var (
 	ErrDisconnected = errors.New("broker disconnected")
 )
 
-type BatchBroker interface {
-	Broker
-	// BatchPublish messages to broker with multiple topics
-	BatchPublish(ctx context.Context, msgs []*Message, opts ...PublishOption) error
-	// BatchSubscribe subscribes to topic messages via handler
-	BatchSubscribe(ctx context.Context, topic string, h BatchHandler, opts ...SubscribeOption) (Subscriber, error)
-}
-
 // Broker is an interface used for asynchronous messaging.
 type Broker interface {
 	// Name returns broker instance name
@@ -44,6 +36,10 @@ type Broker interface {
 	Publish(ctx context.Context, topic string, msg *Message, opts ...PublishOption) error
 	// Subscribe subscribes to topic message via handler
 	Subscribe(ctx context.Context, topic string, h Handler, opts ...SubscribeOption) (Subscriber, error)
+	// BatchPublish messages to broker with multiple topics
+	BatchPublish(ctx context.Context, msgs []*Message, opts ...PublishOption) error
+	// BatchSubscribe subscribes to topic messages via handler
+	BatchSubscribe(ctx context.Context, topic string, h BatchHandler, opts ...SubscribeOption) (Subscriber, error)
 	// String type of broker
 	String() string
 }

broker/memory.go

@@ -13,10 +13,9 @@ import (
 )
 
 type memoryBroker struct {
-	subscribers      map[string][]*memorySubscriber
-	batchsubscribers map[string][]*memoryBatchSubscriber
-	addr             string
-	opts             Options
+	subscribers map[string][]*memorySubscriber
+	addr        string
+	opts        Options
 	sync.RWMutex
 	connected bool
 }
@@ -29,21 +28,13 @@ type memoryEvent struct {
 }
 
 type memorySubscriber struct {
-	ctx     context.Context
-	exit    chan bool
-	handler Handler
-	id      string
-	topic   string
-	opts    SubscribeOptions
-}
-
-type memoryBatchSubscriber struct {
-	ctx     context.Context
-	exit    chan bool
-	handler BatchHandler
-	id      string
-	topic   string
-	opts    SubscribeOptions
+	ctx          context.Context
+	exit         chan bool
+	handler      Handler
+	batchhandler BatchHandler
+	id           string
+	topic        string
+	opts         SubscribeOptions
 }
 
 func (m *memoryBroker) Options() Options {
@@ -97,6 +88,36 @@ func (m *memoryBroker) Init(opts ...Option) error {
 	return nil
 }
 
+func (m *memoryBroker) Publish(ctx context.Context, topic string, msg *Message, opts ...PublishOption) error {
+	m.RLock()
+	if !m.connected {
+		m.RUnlock()
+		return ErrNotConnected
+	}
+	m.RUnlock()
+
+	options := NewPublishOptions(opts...)
+	vs := make([]msgWrapper, 0, 1)
+	if m.opts.Codec == nil || options.BodyOnly {
+		topic, _ := msg.Header.Get(metadata.HeaderTopic)
+		vs = append(vs, msgWrapper{topic: topic, body: msg})
+	} else {
+		topic, _ := msg.Header.Get(metadata.HeaderTopic)
+		buf, err := m.opts.Codec.Marshal(msg)
+		if err != nil {
+			return err
+		}
+		vs = append(vs, msgWrapper{topic: topic, body: buf})
+	}
+
+	return m.publish(ctx, vs, opts...)
+}
+
+type msgWrapper struct {
+	topic string
+	body  interface{}
+}
+
 func (m *memoryBroker) BatchPublish(ctx context.Context, msgs []*Message, opts ...PublishOption) error {
 	m.RLock()
 	if !m.connected {
@@ -105,155 +126,90 @@ func (m *memoryBroker) BatchPublish(ctx context.Context, msgs []*Message, opts .
 	}
 	m.RUnlock()
 
-	type msgWrapper struct {
-		topic string
-		body  interface{}
-	}
-
+	options := NewPublishOptions(opts...)
 	vs := make([]msgWrapper, 0, len(msgs))
-	if m.opts.Codec == nil {
-		m.RLock()
+	if m.opts.Codec == nil || options.BodyOnly {
 		for _, msg := range msgs {
 			topic, _ := msg.Header.Get(metadata.HeaderTopic)
-			vs = append(vs, msgWrapper{topic: topic, body: m})
+			vs = append(vs, msgWrapper{topic: topic, body: msg})
 		}
-		m.RUnlock()
 	} else {
-		m.RLock()
 		for _, msg := range msgs {
 			topic, _ := msg.Header.Get(metadata.HeaderTopic)
 			buf, err := m.opts.Codec.Marshal(msg)
 			if err != nil {
-				m.RUnlock()
 				return err
 			}
 			vs = append(vs, msgWrapper{topic: topic, body: buf})
 		}
-		m.RUnlock()
 	}
 
-	if len(m.batchsubscribers) > 0 {
-		eh := m.opts.BatchErrorHandler
+	return m.publish(ctx, vs, opts...)
+}
 
-		msgTopicMap := make(map[string]Events)
-		for _, v := range vs {
-			p := &memoryEvent{
-				topic:   v.topic,
-				message: v.body,
-				opts:    m.opts,
-			}
-			msgTopicMap[p.topic] = append(msgTopicMap[p.topic], p)
-		}
-
-		for t, ms := range msgTopicMap {
-			m.RLock()
-			subs, ok := m.batchsubscribers[t]
-			m.RUnlock()
-			if !ok {
-				continue
-			}
-			for _, sub := range subs {
-				if err := sub.handler(ms); err != nil {
-					ms.SetError(err)
-					if sub.opts.BatchErrorHandler != nil {
-						eh = sub.opts.BatchErrorHandler
-					}
-					if eh != nil {
-						eh(ms)
-					} else if m.opts.Logger.V(logger.ErrorLevel) {
-						m.opts.Logger.Error(m.opts.Context, err.Error())
-					}
-				} else if sub.opts.AutoAck {
-					if err := ms.Ack(); err != nil {
-						m.opts.Logger.Errorf(m.opts.Context, "ack failed: %v", err)
-					}
-				}
-			}
-		}
-
-	}
-
-	eh := m.opts.ErrorHandler
+func (m *memoryBroker) publish(ctx context.Context, vs []msgWrapper, opts ...PublishOption) error {
+	var err error
 
+	msgTopicMap := make(map[string]Events)
 	for _, v := range vs {
 		p := &memoryEvent{
 			topic:   v.topic,
 			message: v.body,
 			opts:    m.opts,
 		}
+		msgTopicMap[p.topic] = append(msgTopicMap[p.topic], p)
+	}
 
+	beh := m.opts.BatchErrorHandler
+	eh := m.opts.ErrorHandler
+
+	for t, ms := range msgTopicMap {
 		m.RLock()
-		subs, ok := m.subscribers[p.topic]
+		subs, ok := m.subscribers[t]
 		m.RUnlock()
 		if !ok {
 			continue
 		}
+
 		for _, sub := range subs {
-			if err := sub.handler(p); err != nil {
-				p.SetError(err)
-				if sub.opts.ErrorHandler != nil {
-					eh = sub.opts.ErrorHandler
-				}
-				if eh != nil {
-					eh(p)
-				} else if m.opts.Logger.V(logger.ErrorLevel) {
-					m.opts.Logger.Error(m.opts.Context, err.Error())
-				}
-			} else if sub.opts.AutoAck {
-				if err := p.Ack(); err != nil {
-					m.opts.Logger.Errorf(m.opts.Context, "ack failed: %v", err)
+			// batch processing
+			if sub.batchhandler != nil {
+				if err = sub.batchhandler(ms); err != nil {
+					ms.SetError(err)
+					if sub.opts.BatchErrorHandler != nil {
+						beh = sub.opts.BatchErrorHandler
+					}
+					if beh != nil {
+						beh(ms)
+					} else if m.opts.Logger.V(logger.ErrorLevel) {
+						m.opts.Logger.Error(m.opts.Context, err.Error())
+					}
+				} else if sub.opts.AutoAck {
+					if err = ms.Ack(); err != nil {
+						m.opts.Logger.Errorf(m.opts.Context, "ack failed: %v", err)
+					}
 				}
 			}
-		}
-	}
-
-	return nil
-}
-
-func (m *memoryBroker) Publish(ctx context.Context, topic string, msg *Message, opts ...PublishOption) error {
-	m.RLock()
-	if !m.connected {
-		m.RUnlock()
-		return ErrNotConnected
-	}
-
-	subs, ok := m.subscribers[topic]
-	m.RUnlock()
-	if !ok {
-		return nil
-	}
-
-	var v interface{}
-	if m.opts.Codec != nil {
-		buf, err := m.opts.Codec.Marshal(msg)
-		if err != nil {
-			return err
-		}
-		v = buf
-	} else {
-		v = msg
-	}
-
-	p := &memoryEvent{
-		topic:   topic,
-		message: v,
-		opts:    m.opts,
-	}
-
-	eh := m.opts.ErrorHandler
-
-	for _, sub := range subs {
-		if err := sub.handler(p); err != nil {
-			p.err = err
-			if sub.opts.ErrorHandler != nil {
-				eh = sub.opts.ErrorHandler
+			// single processing
+			if sub.handler != nil {
+				for _, p := range ms {
+					if err = sub.handler(p); err != nil {
+						p.SetError(err)
+						if sub.opts.ErrorHandler != nil {
+							eh = sub.opts.ErrorHandler
+						}
+						if eh != nil {
+							eh(p)
+						} else if m.opts.Logger.V(logger.ErrorLevel) {
+							m.opts.Logger.Error(m.opts.Context, err.Error())
+						}
+					} else if sub.opts.AutoAck {
+						if err = p.Ack(); err != nil {
+							m.opts.Logger.Errorf(m.opts.Context, "ack failed: %v", err)
+						}
+					}
+				}
 			}
-			if eh != nil {
-				eh(p)
-			} else if m.opts.Logger.V(logger.ErrorLevel) {
-				m.opts.Logger.Error(m.opts.Context, err.Error())
-			}
-			continue
 		}
 	}
 
@@ -268,37 +224,37 @@ func (m *memoryBroker) BatchSubscribe(ctx context.Context, topic string, handler
 	}
 	m.RUnlock()
 
-	options := NewSubscribeOptions(opts...)
-
 	id, err := uuid.NewRandom()
 	if err != nil {
 		return nil, err
 	}
 
-	sub := &memoryBatchSubscriber{
-		exit:    make(chan bool, 1),
-		id:      id.String(),
-		topic:   topic,
-		handler: handler,
-		opts:    options,
-		ctx:     ctx,
+	options := NewSubscribeOptions(opts...)
+
+	sub := &memorySubscriber{
+		exit:         make(chan bool, 1),
+		id:           id.String(),
+		topic:        topic,
+		batchhandler: handler,
+		opts:         options,
+		ctx:          ctx,
 	}
 
 	m.Lock()
-	m.batchsubscribers[topic] = append(m.batchsubscribers[topic], sub)
+	m.subscribers[topic] = append(m.subscribers[topic], sub)
 	m.Unlock()
 
 	go func() {
 		<-sub.exit
 		m.Lock()
-		var newSubscribers []*memoryBatchSubscriber
-		for _, sb := range m.batchsubscribers[topic] {
+		newSubscribers := make([]*memorySubscriber, 0, len(m.subscribers)-1)
+		for _, sb := range m.subscribers[topic] {
 			if sb.id == sub.id {
 				continue
 			}
 			newSubscribers = append(newSubscribers, sb)
 		}
-		m.batchsubscribers[topic] = newSubscribers
+		m.subscribers[topic] = newSubscribers
 		m.Unlock()
 	}()
 
@@ -313,13 +269,13 @@ func (m *memoryBroker) Subscribe(ctx context.Context, topic string, handler Hand
 	}
 	m.RUnlock()
 
-	options := NewSubscribeOptions(opts...)
-
 	id, err := uuid.NewRandom()
 	if err != nil {
 		return nil, err
 	}
 
+	options := NewSubscribeOptions(opts...)
+
 	sub := &memorySubscriber{
 		exit:    make(chan bool, 1),
 		id:      id.String(),
@@ -336,7 +292,7 @@ func (m *memoryBroker) Subscribe(ctx context.Context, topic string, handler Hand
 	go func() {
 		<-sub.exit
 		m.Lock()
-		var newSubscribers []*memorySubscriber
+		newSubscribers := make([]*memorySubscriber, 0, len(m.subscribers)-1)
 		for _, sb := range m.subscribers[topic] {
 			if sb.id == sub.id {
 				continue
@@ -392,19 +348,6 @@ func (m *memoryEvent) SetError(err error) {
 	m.err = err
 }
 
-func (m *memoryBatchSubscriber) Options() SubscribeOptions {
-	return m.opts
-}
-
-func (m *memoryBatchSubscriber) Topic() string {
-	return m.topic
-}
-
-func (m *memoryBatchSubscriber) Unsubscribe(ctx context.Context) error {
-	m.exit <- true
-	return nil
-}
-
 func (m *memorySubscriber) Options() SubscribeOptions {
 	return m.opts
 }
@@ -419,10 +362,9 @@ func (m *memorySubscriber) Unsubscribe(ctx context.Context) error {
 }
 
 // NewBroker return new memory broker
-func NewBroker(opts ...Option) BatchBroker {
+func NewBroker(opts ...Option) Broker {
 	return &memoryBroker{
-		opts:             NewOptions(opts...),
-		subscribers:      make(map[string][]*memorySubscriber),
-		batchsubscribers: make(map[string][]*memoryBatchSubscriber),
+		opts:        NewOptions(opts...),
+		subscribers: make(map[string][]*memorySubscriber),
 	}
 }

broker/options.go

@@ -3,6 +3,7 @@ package broker
 import (
 	"context"
 	"crypto/tls"
+	"time"
 
 	"github.com/unistack-org/micro/v3/codec"
 	"github.com/unistack-org/micro/v3/logger"
@@ -73,11 +74,9 @@ func NewPublishOptions(opts ...PublishOption) PublishOptions {
 	options := PublishOptions{
 		Context: context.Background(),
 	}
-
 	for _, o := range opts {
 		o(&options)
 	}
-
 	return options
 }
 
@@ -95,6 +94,10 @@ type SubscribeOptions struct {
 	AutoAck bool
 	// BodyOnly flag specifies that message contains only body bytes without header
 	BodyOnly bool
+	// BatchSize flag specifies max batch size
+	BatchSize int
+	// BatchWait flag specifies max wait time for batch filling
+	BatchWait time.Duration
 }
 
 // Option func
@@ -117,23 +120,6 @@ func PublishContext(ctx context.Context) PublishOption {
 	}
 }
 
-// SubscribeOption func
-type SubscribeOption func(*SubscribeOptions)
-
-// NewSubscribeOptions creates new SubscribeOptions
-func NewSubscribeOptions(opts ...SubscribeOption) SubscribeOptions {
-	options := SubscribeOptions{
-		AutoAck: true,
-		Context: context.Background(),
-	}
-
-	for _, o := range opts {
-		o(&options)
-	}
-
-	return options
-}
-
 // Addrs sets the host addresses to be used by the broker
 func Addrs(addrs ...string) Option {
 	return func(o *Options) {
@@ -149,28 +135,6 @@ func Codec(c codec.Codec) Option {
 	}
 }
 
-// DisableAutoAck disables auto ack
-func DisableAutoAck() SubscribeOption {
-	return func(o *SubscribeOptions) {
-		o.AutoAck = false
-	}
-}
-
-// SubscribeAutoAck will disable auto acking of messages
-// after they have been handled.
-func SubscribeAutoAck(b bool) SubscribeOption {
-	return func(o *SubscribeOptions) {
-		o.AutoAck = b
-	}
-}
-
-// SubscribeBodyOnly consumes only body of the message
-func SubscribeBodyOnly(b bool) SubscribeOption {
-	return func(o *SubscribeOptions) {
-		o.BodyOnly = b
-	}
-}
-
 // ErrorHandler will catch all broker errors that cant be handled
 // in normal way, for example Codec errors
 func ErrorHandler(h Handler) Option {
@@ -266,3 +230,55 @@ func SubscribeContext(ctx context.Context) SubscribeOption {
 		o.Context = ctx
 	}
 }
+
+// DisableAutoAck disables auto ack
+// Deprecated
+func DisableAutoAck() SubscribeOption {
+	return func(o *SubscribeOptions) {
+		o.AutoAck = false
+	}
+}
+
+// SubscribeAutoAck contol auto acking of messages
+// after they have been handled.
+func SubscribeAutoAck(b bool) SubscribeOption {
+	return func(o *SubscribeOptions) {
+		o.AutoAck = b
+	}
+}
+
+// SubscribeBodyOnly consumes only body of the message
+func SubscribeBodyOnly(b bool) SubscribeOption {
+	return func(o *SubscribeOptions) {
+		o.BodyOnly = b
+	}
+}
+
+// SubscribeBatchSize specifies max batch size
+func SubscribeBatchSize(n int) SubscribeOption {
+	return func(o *SubscribeOptions) {
+		o.BatchSize = n
+	}
+}
+
+// SubscribeBatchWait specifies max batch wait time
+func SubscribeBatchWait(td time.Duration) SubscribeOption {
+	return func(o *SubscribeOptions) {
+		o.BatchWait = td
+	}
+}
+
+// SubscribeOption func
+type SubscribeOption func(*SubscribeOptions)
+
+// NewSubscribeOptions creates new SubscribeOptions
+func NewSubscribeOptions(opts ...SubscribeOption) SubscribeOptions {
+	options := SubscribeOptions{
+		AutoAck: true,
+		Context: context.Background(),
+	}
+	for _, o := range opts {
+		o(&options)
+	}
+	return options
+}

client/client.go

@@ -40,6 +40,7 @@ type Client interface {
 	Call(ctx context.Context, req Request, rsp interface{}, opts ...CallOption) error
 	Stream(ctx context.Context, req Request, opts ...CallOption) (Stream, error)
 	Publish(ctx context.Context, msg Message, opts ...PublishOption) error
+	BatchPublish(ctx context.Context, msg []Message, opts ...PublishOption) error
 	String() string
 }
 

client/noop.go

@@ -173,7 +173,7 @@ func (n *noopClient) NewRequest(service, endpoint string, req interface{}, opts
 }
 
 func (n *noopClient) NewMessage(topic string, msg interface{}, opts ...MessageOption) Message {
-	options := NewMessageOptions(opts...)
+	options := NewMessageOptions(append([]MessageOption{MessageContentType(n.opts.ContentType)}, opts...)...)
 	return &noopMessage{topic: topic, payload: msg, opts: options}
 }
 
@@ -181,47 +181,59 @@ func (n *noopClient) Stream(ctx context.Context, req Request, opts ...CallOption
 	return &noopStream{}, nil
 }
 
-func (n *noopClient) Publish(ctx context.Context, p Message, opts ...PublishOption) error {
-	var body []byte
+func (n *noopClient) BatchPublish(ctx context.Context, ps []Message, opts ...PublishOption) error {
+	return n.publish(ctx, ps, opts...)
+}
 
+func (n *noopClient) Publish(ctx context.Context, p Message, opts ...PublishOption) error {
+	return n.publish(ctx, []Message{p}, opts...)
+}
+
+func (n *noopClient) publish(ctx context.Context, ps []Message, opts ...PublishOption) error {
 	options := NewPublishOptions(opts...)
 
-	md, ok := metadata.FromOutgoingContext(ctx)
-	if !ok {
-		md = metadata.New(0)
-	}
-	md[metadata.HeaderContentType] = p.ContentType()
-	md[metadata.HeaderTopic] = p.Topic()
+	msgs := make([]*broker.Message, 0, len(ps))
 
-	// passed in raw data
-	if d, ok := p.Payload().(*codec.Frame); ok {
-		body = d.Data
-	} else {
-		// use codec for payload
-		cf, err := n.newCodec(p.ContentType())
-		if err != nil {
-			return errors.InternalServerError("go.micro.client", err.Error())
+	for _, p := range ps {
+		md, ok := metadata.FromOutgoingContext(ctx)
+		if !ok {
+			md = metadata.New(0)
+		}
+		md[metadata.HeaderContentType] = p.ContentType()
+
+		topic := p.Topic()
+
+		// get the exchange
+		if len(options.Exchange) > 0 {
+			topic = options.Exchange
 		}
 
-		// set the body
-		b, err := cf.Marshal(p.Payload())
-		if err != nil {
-			return errors.InternalServerError("go.micro.client", err.Error())
+		md[metadata.HeaderTopic] = topic
+
+		var body []byte
+
+		// passed in raw data
+		if d, ok := p.Payload().(*codec.Frame); ok {
+			body = d.Data
+		} else {
+			// use codec for payload
+			cf, err := n.newCodec(p.ContentType())
+			if err != nil {
+				return errors.InternalServerError("go.micro.client", err.Error())
+			}
+
+			// set the body
+			b, err := cf.Marshal(p.Payload())
+			if err != nil {
+				return errors.InternalServerError("go.micro.client", err.Error())
+			}
+			body = b
 		}
-		body = b
+
+		msgs = append(msgs, &broker.Message{Header: md, Body: body})
 	}
 
-	topic := p.Topic()
-
-	// get the exchange
-	if len(options.Exchange) > 0 {
-		topic = options.Exchange
-	}
-
-	return n.opts.Broker.Publish(ctx, topic, &broker.Message{
-		Header: md,
-		Body:   body,
-	},
+	return n.opts.Broker.BatchPublish(ctx, msgs,
 		broker.PublishContext(options.Context),
 		broker.PublishBodyOnly(options.BodyOnly),
 	)

client/options.go

@@ -373,19 +373,35 @@ func DialTimeout(d time.Duration) Option {
 }
 
 // WithExchange sets the exchange to route a message through
+// Deprecated
 func WithExchange(e string) PublishOption {
 	return func(o *PublishOptions) {
 		o.Exchange = e
 	}
 }
 
+// PublishExchange sets the exchange to route a message through
+func PublishExchange(e string) PublishOption {
+	return func(o *PublishOptions) {
+		o.Exchange = e
+	}
+}
+
 // WithBodyOnly publish only message body
+// DERECATED
 func WithBodyOnly(b bool) PublishOption {
 	return func(o *PublishOptions) {
 		o.BodyOnly = b
 	}
 }
 
+// PublishBodyOnly publish only message body
+func PublishBodyOnly(b bool) PublishOption {
+	return func(o *PublishOptions) {
+		o.BodyOnly = b
+	}
+}
+
 // PublishContext sets the context in publish options
 func PublishContext(ctx context.Context) PublishOption {
 	return func(o *PublishOptions) {
@@ -498,12 +514,20 @@ func WithSelectOptions(sops ...selector.SelectOption) CallOption {
 }
 
 // WithMessageContentType sets the message content type
+// Deprecated
 func WithMessageContentType(ct string) MessageOption {
 	return func(o *MessageOptions) {
 		o.ContentType = ct
 	}
 }
 
+// MessageContentType sets the message content type
+func MessageContentType(ct string) MessageOption {
+	return func(o *MessageOptions) {
+		o.ContentType = ct
+	}
+}
+
 // StreamingRequest specifies that request is streaming
 func StreamingRequest(b bool) RequestOption {
 	return func(o *RequestOptions) {

config/config.go

@@ -4,11 +4,15 @@ package config
 import (
 	"context"
 	"errors"
+	"time"
 )
 
 // DefaultConfig default config
 var DefaultConfig Config = NewConfig()
 
+// DefaultWatcherInterval default interval for poll changes
+var DefaultWatcherInterval = 5 * time.Second
+
 var (
 	// ErrCodecMissing is returned when codec needed and not specified
 	ErrCodecMissing = errors.New("codec missing")
@@ -30,15 +34,17 @@ type Config interface {
 	Load(context.Context, ...LoadOption) error
 	// Save config to sources
 	Save(context.Context, ...SaveOption) error
-	// Watch a value for changes
-	//Watch(context.Context) (Watcher, error)
+	// Watch a config for changes
+	Watch(context.Context, ...WatchOption) (Watcher, error)
 	// String returns config type name
 	String() string
 }
 
 // Watcher is the config watcher
 type Watcher interface {
-	// Next() (, error)
+	// Next blocks until update happens or error returned
+	Next() (map[string]interface{}, error)
+	// Stop stops watcher
 	Stop() error
 }
 

config/default.go

@@ -5,6 +5,7 @@ import (
 	"reflect"
 	"strconv"
 	"strings"
+	"time"
 
 	"github.com/imdario/mergo"
 	rutil "github.com/unistack-org/micro/v3/util/reflect"
@@ -41,11 +42,15 @@ func (c *defaultConfig) Load(ctx context.Context, opts ...LoadOption) error {
 		mopts = append(mopts, mergo.WithAppendSlice)
 	}
 
-	src, err := rutil.Zero(c.opts.Struct)
+	dst := c.opts.Struct
+	if options.Struct != nil {
+		dst = options.Struct
+	}
+
+	src, err := rutil.Zero(dst)
 	if err == nil {
-		valueOf := reflect.ValueOf(src)
-		if err = c.fillValues(valueOf); err == nil {
-			err = mergo.Merge(c.opts.Struct, src, mopts...)
+		if err = fillValues(reflect.ValueOf(src), c.opts.StructTag); err == nil {
+			err = mergo.Merge(dst, src, mopts...)
 		}
 	}
 
@@ -63,7 +68,7 @@ func (c *defaultConfig) Load(ctx context.Context, opts ...LoadOption) error {
 }
 
 //nolint:gocyclo
-func (c *defaultConfig) fillValue(value reflect.Value, val string) error {
+func fillValue(value reflect.Value, val string) error {
 	if !rutil.IsEmpty(value) {
 		return nil
 	}
@@ -80,10 +85,10 @@ func (c *defaultConfig) fillValue(value reflect.Value, val string) error {
 			kv := strings.FieldsFunc(nval, func(c rune) bool { return c == '=' })
 			mkey := reflect.Indirect(reflect.New(kt))
 			mval := reflect.Indirect(reflect.New(et))
-			if err := c.fillValue(mkey, kv[0]); err != nil {
+			if err := fillValue(mkey, kv[0]); err != nil {
 				return err
 			}
-			if err := c.fillValue(mval, kv[1]); err != nil {
+			if err := fillValue(mval, kv[1]); err != nil {
 				return err
 			}
 			value.SetMapIndex(mkey, mval)
@@ -93,7 +98,7 @@ func (c *defaultConfig) fillValue(value reflect.Value, val string) error {
 		value.Set(reflect.MakeSlice(reflect.SliceOf(value.Type().Elem()), len(nvals), len(nvals)))
 		for idx, nval := range nvals {
 			nvalue := reflect.Indirect(reflect.New(value.Type().Elem()))
-			if err := c.fillValue(nvalue, nval); err != nil {
+			if err := fillValue(nvalue, nval); err != nil {
 				return err
 			}
 			value.Index(idx).Set(nvalue)
@@ -182,7 +187,7 @@ func (c *defaultConfig) fillValue(value reflect.Value, val string) error {
 	return nil
 }
 
-func (c *defaultConfig) fillValues(valueOf reflect.Value) error {
+func fillValues(valueOf reflect.Value, tname string) error {
 	var values reflect.Value
 
 	if valueOf.Kind() == reflect.Ptr {
@@ -209,7 +214,7 @@ func (c *defaultConfig) fillValues(valueOf reflect.Value) error {
 		switch value.Kind() {
 		case reflect.Struct:
 			value.Set(reflect.Indirect(reflect.New(value.Type())))
-			if err := c.fillValues(value); err != nil {
+			if err := fillValues(value, tname); err != nil {
 				return err
 			}
 			continue
@@ -223,17 +228,17 @@ func (c *defaultConfig) fillValues(valueOf reflect.Value) error {
 				value.Set(reflect.New(value.Type().Elem()))
 			}
 			value = value.Elem()
-			if err := c.fillValues(value); err != nil {
+			if err := fillValues(value, tname); err != nil {
 				return err
 			}
 			continue
 		}
-		tag, ok := field.Tag.Lookup(c.opts.StructTag)
+		tag, ok := field.Tag.Lookup(tname)
 		if !ok {
 			continue
 		}
 
-		if err := c.fillValue(value, tag); err != nil {
+		if err := fillValue(value, tag); err != nil {
 			return err
 		}
 	}
@@ -265,6 +270,20 @@ func (c *defaultConfig) Name() string {
 	return c.opts.Name
 }
 
+func (c *defaultConfig) Watch(ctx context.Context, opts ...WatchOption) (Watcher, error) {
+	w := &defaultWatcher{
+		opts:  c.opts,
+		wopts: NewWatchOptions(opts...),
+		done:  make(chan bool),
+		vchan: make(chan map[string]interface{}),
+		echan: make(chan error),
+	}
+
+	go w.run()
+
+	return w, nil
+}
+
 // NewConfig returns new default config source
 func NewConfig(opts ...Option) Config {
 	options := NewOptions(opts...)
@@ -273,3 +292,73 @@ func NewConfig(opts ...Option) Config {
 	}
 	return &defaultConfig{opts: options}
 }
+
+type defaultWatcher struct {
+	opts   Options
+	wopts  WatchOptions
+	done   chan bool
+	ticker *time.Ticker
+	vchan  chan map[string]interface{}
+	echan  chan error
+}
+
+func (w *defaultWatcher) run() {
+	ticker := time.NewTicker(w.wopts.Interval)
+	defer ticker.Stop()
+
+	src := w.opts.Struct
+	if w.wopts.Struct != nil {
+		src = w.wopts.Struct
+	}
+
+	for {
+		select {
+		case <-w.done:
+			return
+		case <-ticker.C:
+			dst, err := rutil.Zero(src)
+			if err == nil {
+				err = fillValues(reflect.ValueOf(dst), w.opts.StructTag)
+			}
+			if err != nil {
+				w.echan <- err
+				return
+			}
+			srcmp, err := rutil.StructFieldsMap(src)
+			if err != nil {
+				w.echan <- err
+				return
+			}
+			dstmp, err := rutil.StructFieldsMap(dst)
+			if err != nil {
+				w.echan <- err
+				return
+			}
+			for sk, sv := range srcmp {
+				if reflect.DeepEqual(dstmp[sk], sv) {
+					delete(dstmp, sk)
+				}
+			}
+			w.vchan <- dstmp
+			src = dst
+		}
+	}
+}
+
+func (w *defaultWatcher) Next() (map[string]interface{}, error) {
+	select {
+	case <-w.done:
+		break
+	case v, ok := <-w.vchan:
+		if !ok {
+			break
+		}
+		return v, nil
+	}
+	return nil, ErrWatcherStopped
+}
+
+func (w *defaultWatcher) Stop() error {
+	close(w.done)
+	return nil
+}

config/default_test.go

@@ -4,6 +4,7 @@ import (
 	"context"
 	"fmt"
 	"testing"
+	"time"
 
 	"github.com/unistack-org/micro/v3/config"
 )
@@ -17,6 +18,57 @@ type Cfg struct {
 	IntValue int `default:"99"`
 }
 
+func TestWatch(t *testing.T) {
+	ctx := context.Background()
+
+	conf := &Cfg{IntValue: 10}
+
+	cfg := config.NewConfig(config.Struct(conf))
+	if err := cfg.Init(); err != nil {
+		t.Fatal(err)
+	}
+	if err := cfg.Load(ctx); err != nil {
+		t.Fatal(err)
+	}
+
+	w, err := cfg.Watch(ctx, config.WatchInterval(500*time.Millisecond))
+	if err != nil {
+		t.Fatal(err)
+	}
+
+	defer func() {
+		_ = w.Stop()
+	}()
+
+	done := make(chan struct{})
+
+	go func() {
+		for {
+			mp, err := w.Next()
+			if err != nil && err != config.ErrWatcherStopped {
+				t.Fatal(err)
+			} else if err == config.ErrWatcherStopped {
+				return
+			}
+			if len(mp) != 1 {
+				t.Fatal(fmt.Errorf("default watcher err: %v", mp))
+			}
+
+			v, ok := mp["IntValue"]
+			if !ok {
+				t.Fatal(fmt.Errorf("default watcher err: %v", v))
+			}
+			if nv, ok := v.(int); !ok || nv != 99 {
+				t.Fatal(fmt.Errorf("default watcher err: %v", v))
+			}
+			close(done)
+			return
+		}
+	}()
+
+	<-done
+}
+
 func TestDefault(t *testing.T) {
 	ctx := context.Background()
 	conf := &Cfg{IntValue: 10}
@@ -47,6 +99,6 @@ func TestDefault(t *testing.T) {
 	if conf.StringValue != "after_load" {
 		t.Fatal("AfterLoad option not working")
 	}
-
-	t.Logf("%#+v\n", conf)
+	_ = conf
+	//t.Logf("%#+v\n", conf)
 }

config/options.go

@@ -2,6 +2,7 @@ package config
 
 import (
 	"context"
+	"time"
 
 	"github.com/unistack-org/micro/v3/codec"
 	"github.com/unistack-org/micro/v3/logger"
@@ -62,6 +63,7 @@ type LoadOption func(o *LoadOptions)
 
 // LoadOptions struct
 type LoadOptions struct {
+	Struct   interface{}
 	Override bool
 	Append   bool
 }
@@ -88,13 +90,29 @@ func LoadAppend(b bool) LoadOption {
 	}
 }
 
+// LoadStruct override struct for loading
+func LoadStruct(src interface{}) LoadOption {
+	return func(o *LoadOptions) {
+		o.Struct = src
+	}
+}
+
 // SaveOption function signature
 type SaveOption func(o *SaveOptions)
 
 // SaveOptions struct
 type SaveOptions struct {
+	Struct interface{}
 }
 
+// SaveStruct override struct for save to config
+func SaveStruct(src interface{}) SaveOption {
+	return func(o *SaveOptions) {
+		o.Struct = src
+	}
+}
+
+// NewSaveOptions fill SaveOptions struct
 func NewSaveOptions(opts ...SaveOption) SaveOptions {
 	options := SaveOptions{}
 	for _, o := range opts {
@@ -186,3 +204,56 @@ func Name(n string) Option {
 		o.Name = n
 	}
 }
+
+// WatchOptions struuct
+type WatchOptions struct {
+	// Context used by non default options
+	Context context.Context
+	// Coalesce multiple events to one
+	Coalesce bool
+	// Interval to periodically pull changes if config source not supports async notify
+	Interval time.Duration
+	// Struct for filling
+	Struct interface{}
+}
+
+type WatchOption func(*WatchOptions)
+
+func NewWatchOptions(opts ...WatchOption) WatchOptions {
+	options := WatchOptions{
+		Context:  context.Background(),
+		Interval: DefaultWatcherInterval,
+	}
+	for _, o := range opts {
+		o(&options)
+	}
+	return options
+}
+
+// WatchContext pass context
+func WatchContext(ctx context.Context) WatchOption {
+	return func(o *WatchOptions) {
+		o.Context = ctx
+	}
+}
+
+// WatchCoalesce controls watch event combining
+func WatchCoalesce(b bool) WatchOption {
+	return func(o *WatchOptions) {
+		o.Coalesce = b
+	}
+}
+
+// WatchInterval specifies time.Duration for pulling changes
+func WatchInterval(td time.Duration) WatchOption {
+	return func(o *WatchOptions) {
+		o.Interval = td
+	}
+}
+
+// WatchStruct overrides struct for fill
+func WatchStruct(src interface{}) WatchOption {
+	return func(o *WatchOptions) {
+		o.Struct = src
+	}
+}

config/reflect_test.go

@@ -1,26 +0,0 @@
-package config_test
-
-import (
-	"testing"
-
-	rutil "github.com/unistack-org/micro/v3/util/reflect"
-)
-
-type Config struct {
-	SubConfig *SubConfig
-	Config    *Config
-	Value     string
-}
-
-type SubConfig struct {
-	Value string
-}
-
-func TestReflect(t *testing.T) {
-	cfg1 := &Config{Value: "cfg1", Config: &Config{Value: "cfg1_1"}, SubConfig: &SubConfig{Value: "cfg1"}}
-	cfg2, err := rutil.Zero(cfg1)
-	if err != nil {
-		t.Fatal(err)
-	}
-	t.Logf("dst: %#+v\n", cfg2)
-}

go.mod

@@ -3,8 +3,8 @@ module github.com/unistack-org/micro/v3
 go 1.16
 
 require (
-	github.com/dgrijalva/jwt-go v3.2.0+incompatible
 	github.com/ef-ds/deque v1.0.4
+	github.com/golang-jwt/jwt v3.2.1+incompatible
 	github.com/google/uuid v1.3.0
 	github.com/imdario/mergo v0.3.12
 	github.com/patrickmn/go-cache v2.1.0+incompatible

go.sum

@@ -1,7 +1,7 @@
-github.com/dgrijalva/jwt-go v3.2.0+incompatible h1:7qlOGliEKZXTDg6OTjfoBKDXWrumCAMpl/TFQ4/5kLM=
-github.com/dgrijalva/jwt-go v3.2.0+incompatible/go.mod h1:E3ru+11k8xSBh+hMPgOLZmtrrCbhqsmaPHjLKYnJCaQ=
 github.com/ef-ds/deque v1.0.4 h1:iFAZNmveMT9WERAkqLJ+oaABF9AcVQ5AjXem/hroniI=
 github.com/ef-ds/deque v1.0.4/go.mod h1:gXDnTC3yqvBcHbq2lcExjtAcVrOnJCbMcZXmuj8Z4tg=
+github.com/golang-jwt/jwt v3.2.1+incompatible h1:73Z+4BJcrTC+KczS6WvTPvRGOp1WmfEP4Q1lOd9Z/+c=
+github.com/golang-jwt/jwt v3.2.1+incompatible/go.mod h1:8pz2t5EyA70fFQQSrl6XZXzqecmYZeUEB8OUGHkxJ+I=
 github.com/google/uuid v1.3.0 h1:t6JiXgmwXMjEs8VusXIJk2BXHsn+wx8BZdTaoZ5fu7I=
 github.com/google/uuid v1.3.0/go.mod h1:TIyPZe4MgqvfeYDBFedMoGGpEw/LqOeaOT+nhxU+yHo=
 github.com/imdario/mergo v0.3.12 h1:b6R2BslTbIEToALKP7LxUvijTsNI9TAe80pLWN2g/HU=

logger/default.go

@@ -35,9 +35,6 @@ func (l *defaultLogger) Init(opts ...Option) error {
 		o(&l.opts)
 	}
 	l.enc = json.NewEncoder(l.opts.Out)
-
-	l.logFunc = l.Log
-	l.logfFunc = l.Logf
 	// wrap the Log func
 	for i := len(l.opts.Wrappers); i > 0; i-- {
 		l.logFunc = l.opts.Wrappers[i-1].Log(l.logFunc)
@@ -218,7 +215,11 @@ func (l *defaultLogger) Options() Options {
 
 // NewLogger builds a new logger based on options
 func NewLogger(opts ...Option) Logger {
-	l := &defaultLogger{opts: NewOptions(opts...)}
+	l := &defaultLogger{
+		opts: NewOptions(opts...),
+	}
+	l.logFunc = l.Log
+	l.logfFunc = l.Logf
 	l.enc = json.NewEncoder(l.opts.Out)
 	return l
 }

logger/logger.go

@@ -8,6 +8,8 @@ var (
 	DefaultLogger Logger = NewLogger()
 	// DefaultLevel used by logger
 	DefaultLevel Level = InfoLevel
+	// DefaultCallerSkipCount used by logger
+	DefaultCallerSkipCount = 2
 )
 
 // Logger is a generic logging interface

logger/options.go

@@ -33,7 +33,7 @@ func NewOptions(opts ...Option) Options {
 		Level:           DefaultLevel,
 		Fields:          make(map[string]interface{}),
 		Out:             os.Stderr,
-		CallerSkipCount: 0,
+		CallerSkipCount: DefaultCallerSkipCount,
 		Context:         context.Background(),
 	}
 	for _, o := range opts {

server/noop.go

@@ -6,11 +6,13 @@ import (
 	"sync"
 	"time"
 
-	//	cprotorpc "github.com/unistack-org/micro-codec-protorpc"
 	"github.com/unistack-org/micro/v3/broker"
 	"github.com/unistack-org/micro/v3/codec"
 	"github.com/unistack-org/micro/v3/logger"
 	"github.com/unistack-org/micro/v3/register"
+	maddr "github.com/unistack-org/micro/v3/util/addr"
+	mnet "github.com/unistack-org/micro/v3/util/net"
+	"github.com/unistack-org/micro/v3/util/rand"
 )
 
 // DefaultCodecs will be used to encode/decode
@@ -73,8 +75,7 @@ func (n *noopServer) Subscribe(sb Subscriber) error {
 	sub, ok := sb.(*subscriber)
 	if !ok {
 		return fmt.Errorf("invalid subscriber: expected *subscriber")
-	}
-	if len(sub.handlers) == 0 {
+	} else if len(sub.handlers) == 0 {
 		return fmt.Errorf("invalid subscriber: no handler functions")
 	}
 
@@ -107,11 +108,12 @@ func (n *noopServer) Init(opts ...Option) error {
 	}
 
 	if n.handlers == nil {
-		n.handlers = make(map[string]Handler)
+		n.handlers = make(map[string]Handler, 1)
 	}
 	if n.subscribers == nil {
-		n.subscribers = make(map[*subscriber][]broker.Subscriber)
+		n.subscribers = make(map[*subscriber][]broker.Subscriber, 1)
 	}
+
 	if n.exit == nil {
 		n.exit = make(chan chan error)
 	}
@@ -202,26 +204,34 @@ func (n *noopServer) Register() error {
 
 	cx := config.Context
 
-	for sb := range n.subscribers {
-		handler := n.createSubHandler(sb, config)
-		var opts []broker.SubscribeOption
-		if queue := sb.Options().Queue; len(queue) > 0 {
-			opts = append(opts, broker.SubscribeGroup(queue))
-		}
+	var sub broker.Subscriber
 
+	for sb := range n.subscribers {
 		if sb.Options().Context != nil {
 			cx = sb.Options().Context
 		}
 
-		opts = append(opts, broker.SubscribeContext(cx), broker.SubscribeAutoAck(sb.Options().AutoAck))
+		opts := []broker.SubscribeOption{broker.SubscribeContext(cx), broker.SubscribeAutoAck(sb.Options().AutoAck)}
+		if queue := sb.Options().Queue; len(queue) > 0 {
+			opts = append(opts, broker.SubscribeGroup(queue))
+		}
+
+		if sb.Options().Batch {
+			// batch processing handler
+			sub, err = config.Broker.BatchSubscribe(cx, sb.Topic(), n.newBatchSubHandler(sb, config), opts...)
+		} else {
+			// single processing handler
+			sub, err = config.Broker.Subscribe(cx, sb.Topic(), n.newSubHandler(sb, config), opts...)
+		}
+
+		if err != nil {
+			return err
+		}
 
 		if config.Logger.V(logger.InfoLevel) {
 			config.Logger.Infof(n.opts.Context, "subscribing to topic: %s", sb.Topic())
 		}
-		sub, err := config.Broker.Subscribe(cx, sb.Topic(), handler, opts...)
-		if err != nil {
-			return err
-		}
+
 		n.subscribers[sb] = []broker.Subscriber{sub}
 	}
 
@@ -303,9 +313,22 @@ func (n *noopServer) Start() error {
 	config := n.Options()
 	n.RUnlock()
 
+	// use 127.0.0.1 to avoid scan of all network interfaces
+	addr, err := maddr.Extract("127.0.0.1")
+	if err != nil {
+		return err
+	}
+	var rng rand.Rand
+	i := rng.Intn(20000)
+	// set addr with port
+	addr = mnet.HostPort(addr, 10000+i)
+
+	config.Address = addr
+
 	if config.Logger.V(logger.InfoLevel) {
 		config.Logger.Infof(n.opts.Context, "server [noop] Listening on %s", config.Address)
 	}
+
 	n.Lock()
 	if len(config.Advertise) == 0 {
 		config.Advertise = config.Address

server/noop_test.go

@@ -0,0 +1,106 @@
+package server_test
+
+import (
+	"context"
+	"fmt"
+	"testing"
+
+	"github.com/unistack-org/micro/v3/broker"
+	"github.com/unistack-org/micro/v3/client"
+	"github.com/unistack-org/micro/v3/codec"
+	"github.com/unistack-org/micro/v3/metadata"
+	"github.com/unistack-org/micro/v3/server"
+)
+
+type TestHandler struct {
+	t *testing.T
+}
+
+type TestMessage struct {
+	Name string
+}
+
+var (
+	numMsg int = 8
+)
+
+func (h *TestHandler) SingleSubHandler(ctx context.Context, msg *codec.Frame) error {
+	//fmt.Printf("msg %s\n", msg.Data)
+	return nil
+}
+
+func (h *TestHandler) BatchSubHandler(ctxs []context.Context, msgs []*codec.Frame) error {
+	if len(msgs) != 8 {
+		h.t.Fatal("invalid number of messages received")
+	}
+	for idx := 0; idx < len(msgs); idx++ {
+		md, _ := metadata.FromIncomingContext(ctxs[idx])
+		_ = md
+		//	fmt.Printf("msg md %v\n", md)
+	}
+	return nil
+}
+
+func TestNoopSub(t *testing.T) {
+	ctx := context.Background()
+
+	b := broker.NewBroker()
+
+	if err := b.Init(); err != nil {
+		t.Fatal(err)
+	}
+
+	if err := b.Connect(ctx); err != nil {
+		t.Fatal(err)
+	}
+
+	s := server.NewServer(
+		server.Broker(b),
+		server.Codec("application/octet-stream", codec.NewCodec()),
+	)
+	if err := s.Init(); err != nil {
+		t.Fatal(err)
+	}
+
+	c := client.NewClient(
+		client.Broker(b),
+		client.Codec("application/octet-stream", codec.NewCodec()),
+		client.ContentType("application/octet-stream"),
+	)
+	if err := c.Init(); err != nil {
+		t.Fatal(err)
+	}
+	h := &TestHandler{t: t}
+
+	if err := s.Subscribe(s.NewSubscriber("single_topic", h.SingleSubHandler,
+		server.SubscriberQueue("queue"),
+	)); err != nil {
+		t.Fatal(err)
+	}
+
+	if err := s.Subscribe(s.NewSubscriber("batch_topic", h.BatchSubHandler,
+		server.SubscriberQueue("queue"),
+		server.SubscriberBatch(true),
+	)); err != nil {
+		t.Fatal(err)
+	}
+
+	if err := s.Start(); err != nil {
+		t.Fatal(err)
+	}
+
+	msgs := make([]client.Message, 0, 8)
+	for i := 0; i < 8; i++ {
+		msgs = append(msgs, c.NewMessage("batch_topic", &codec.Frame{Data: []byte(fmt.Sprintf(`{"name": "test_name %d"}`, i))}))
+	}
+
+	if err := c.BatchPublish(ctx, msgs); err != nil {
+		t.Fatal(err)
+	}
+
+	defer func() {
+		if err := s.Stop(); err != nil {
+			t.Fatal(err)
+		}
+	}()
+}

server/options.go

@@ -71,6 +71,8 @@ type Options struct {
 	Version string
 	// SubWrappers holds the server subscribe wrappers
 	SubWrappers []SubscriberWrapper
+	// BatchSubWrappers holds the server batch subscribe wrappers
+	BatchSubWrappers []BatchSubscriberWrapper
 	// HdlrWrappers holds the handler wrappers
 	HdlrWrappers []HandlerWrapper
 	// RegisterAttempts holds the number of register attempts before error
@@ -302,6 +304,13 @@ func WrapSubscriber(w SubscriberWrapper) Option {
 	}
 }
 
+// WrapBatchSubscriber adds a batch subscriber Wrapper to a list of options passed into the server
+func WrapBatchSubscriber(w BatchSubscriberWrapper) Option {
+	return func(o *Options) {
+		o.BatchSubWrappers = append(o.BatchSubWrappers, w)
+	}
+}
+
 // MaxConn specifies maximum number of max simultaneous connections to server
 func MaxConn(n int) Option {
 	return func(o *Options) {
@@ -354,6 +363,12 @@ type SubscriberOptions struct {
 	AutoAck bool
 	// BodyOnly flag specifies that message without headers
 	BodyOnly bool
+	// Batch flag specifies that message processed in batches
+	Batch bool
+	// BatchSize flag specifies max size of batch
+	BatchSize int
+	// BatchWait flag specifies max wait time for batch filling
+	BatchWait time.Duration
 }
 
 // NewSubscriberOptions create new SubscriberOptions
@@ -413,3 +428,32 @@ func SubscriberContext(ctx context.Context) SubscriberOption {
 		o.Context = ctx
 	}
 }
+
+// SubscriberAck control auto ack processing for handler
+func SubscriberAck(b bool) SubscriberOption {
+	return func(o *SubscriberOptions) {
+		o.AutoAck = b
+	}
+}
+
+// SubscriberBatch control batch processing for handler
+func SubscriberBatch(b bool) SubscriberOption {
+	return func(o *SubscriberOptions) {
+		o.Batch = b
+	}
+}
+
+// SubscriberBatchSize control batch filling size for handler
+// Batch filling max waiting time controlled by SubscriberBatchWait
+func SubscriberBatchSize(n int) SubscriberOption {
+	return func(o *SubscriberOptions) {
+		o.BatchSize = n
+	}
+}
+
+// SubscriberBatchWait control batch filling wait time for handler
+func SubscriberBatchWait(td time.Duration) SubscriberOption {
+	return func(o *SubscriberOptions) {
+		o.BatchWait = td
+	}
+}

server/subscriber.go

@@ -18,7 +18,8 @@ import (
 )
 
 const (
-	subSig = "func(context.Context, interface{}) error"
+	subSig      = "func(context.Context, interface{}) error"
+	batchSubSig = "func([]context.Context, []interface{}) error"
 )
 
 // Precompute the reflect type for error. Can't use error directly
@@ -57,26 +58,33 @@ func isExportedOrBuiltinType(t reflect.Type) bool {
 	return isExported(t.Name()) || t.PkgPath() == ""
 }
 
-// ValidateSubscriber func
+// ValidateSubscriber func signature
 func ValidateSubscriber(sub Subscriber) error {
 	typ := reflect.TypeOf(sub.Subscriber())
 	var argType reflect.Type
-
 	switch typ.Kind() {
 	case reflect.Func:
 		name := "Func"
 		switch typ.NumIn() {
 		case 2:
 			argType = typ.In(1)
+			if sub.Options().Batch {
+				if argType.Kind() != reflect.Slice {
+					return fmt.Errorf("subscriber %v dont have required signature %s", name, batchSubSig)
+				}
+				if strings.Compare(fmt.Sprintf("%s", argType), "[]interface{}") == 0 {
+					return fmt.Errorf("subscriber %v dont have required signaure %s", name, batchSubSig)
+				}
+			}
 		default:
-			return fmt.Errorf("subscriber %v takes wrong number of args: %v required signature %s", name, typ.NumIn(), subSig)
+			return fmt.Errorf("subscriber %v takes wrong number of args: %v required signature %s or %s", name, typ.NumIn(), subSig, batchSubSig)
 		}
 		if !isExportedOrBuiltinType(argType) {
 			return fmt.Errorf("subscriber %v argument type not exported: %v", name, argType)
 		}
 		if typ.NumOut() != 1 {
-			return fmt.Errorf("subscriber %v has wrong number of outs: %v require signature %s",
-				name, typ.NumOut(), subSig)
+			return fmt.Errorf("subscriber %v has wrong number of return values: %v require signature %s or %s",
+				name, typ.NumOut(), subSig, batchSubSig)
 		}
 		if returnType := typ.Out(0); returnType != typeOfError {
 			return fmt.Errorf("subscriber %v returns %v not error", name, returnType.String())
@@ -87,13 +95,12 @@ func ValidateSubscriber(sub Subscriber) error {
 
 		for m := 0; m < typ.NumMethod(); m++ {
 			method := typ.Method(m)
-
 			switch method.Type.NumIn() {
 			case 3:
 				argType = method.Type.In(2)
 			default:
-				return fmt.Errorf("subscriber %v.%v takes wrong number of args: %v required signature %s",
-					name, method.Name, method.Type.NumIn(), subSig)
+				return fmt.Errorf("subscriber %v.%v takes wrong number of args: %v required signature %s or %s",
+					name, method.Name, method.Type.NumIn(), subSig, batchSubSig)
 			}
 
 			if !isExportedOrBuiltinType(argType) {
@@ -101,8 +108,8 @@ func ValidateSubscriber(sub Subscriber) error {
 			}
 			if method.Type.NumOut() != 1 {
 				return fmt.Errorf(
-					"subscriber %v.%v has wrong number of outs: %v require signature %s",
-					name, method.Name, method.Type.NumOut(), subSig)
+					"subscriber %v.%v has wrong number of return values: %v require signature %s or %s",
+					name, method.Name, method.Type.NumOut(), subSig, batchSubSig)
 			}
 			if returnType := method.Type.Out(0); returnType != typeOfError {
 				return fmt.Errorf("subscriber %v.%v returns %v not error", name, method.Name, returnType.String())
@@ -183,7 +190,125 @@ func newSubscriber(topic string, sub interface{}, opts ...SubscriberOption) Subs
 }
 
 //nolint:gocyclo
-func (n *noopServer) createSubHandler(sb *subscriber, opts Options) broker.Handler {
+func (n *noopServer) newBatchSubHandler(sb *subscriber, opts Options) broker.BatchHandler {
+	return func(ps broker.Events) (err error) {
+		defer func() {
+			if r := recover(); r != nil {
+				n.RLock()
+				config := n.opts
+				n.RUnlock()
+				if config.Logger.V(logger.ErrorLevel) {
+					config.Logger.Error(n.opts.Context, "panic recovered: ", r)
+					config.Logger.Error(n.opts.Context, string(debug.Stack()))
+				}
+				err = errors.InternalServerError(n.opts.Name+".subscriber", "panic recovered: %v", r)
+			}
+		}()
+
+		msgs := make([]Message, 0, len(ps))
+		ctxs := make([]context.Context, 0, len(ps))
+		for _, p := range ps {
+			msg := p.Message()
+			// if we don't have headers, create empty map
+			if msg.Header == nil {
+				msg.Header = metadata.New(2)
+			}
+
+			ct, _ := msg.Header.Get(metadata.HeaderContentType)
+			if len(ct) == 0 {
+				msg.Header.Set(metadata.HeaderContentType, defaultContentType)
+				ct = defaultContentType
+			}
+			hdr := metadata.Copy(msg.Header)
+			topic, _ := msg.Header.Get(metadata.HeaderTopic)
+			ctxs = append(ctxs, metadata.NewIncomingContext(sb.opts.Context, hdr))
+			msgs = append(msgs, &rpcMessage{
+				topic:       topic,
+				contentType: ct,
+				header:      msg.Header,
+				body:        msg.Body,
+			})
+		}
+		results := make(chan error, len(sb.handlers))
+
+		for i := 0; i < len(sb.handlers); i++ {
+			handler := sb.handlers[i]
+
+			var req reflect.Value
+
+			switch handler.reqType.Kind() {
+			case reflect.Ptr:
+				req = reflect.New(handler.reqType.Elem())
+			default:
+				req = reflect.New(handler.reqType.Elem()).Elem()
+			}
+
+			reqType := handler.reqType
+
+			for _, msg := range msgs {
+				cf, err := n.newCodec(msg.ContentType())
+				if err != nil {
+					return err
+				}
+				rb := reflect.New(req.Type().Elem())
+				if err = cf.ReadBody(bytes.NewReader(msg.Body()), rb.Interface()); err != nil {
+					return err
+				}
+				msg.(*rpcMessage).codec = cf
+				msg.(*rpcMessage).payload = rb.Interface()
+			}
+
+			fn := func(ctxs []context.Context, ms []Message) error {
+				var vals []reflect.Value
+				if sb.typ.Kind() != reflect.Func {
+					vals = append(vals, sb.rcvr)
+				}
+				if handler.ctxType != nil {
+					vals = append(vals, reflect.ValueOf(ctxs))
+				}
+				payloads := reflect.MakeSlice(reqType, 0, len(ms))
+				for _, m := range ms {
+					payloads = reflect.Append(payloads, reflect.ValueOf(m.Payload()))
+				}
+				vals = append(vals, payloads)
+
+				returnValues := handler.method.Call(vals)
+				if rerr := returnValues[0].Interface(); rerr != nil {
+					return rerr.(error)
+				}
+				return nil
+			}
+
+			for i := len(opts.BatchSubWrappers); i > 0; i-- {
+				fn = opts.BatchSubWrappers[i-1](fn)
+			}
+
+			if n.wg != nil {
+				n.wg.Add(1)
+			}
+			go func() {
+				if n.wg != nil {
+					defer n.wg.Done()
+				}
+				results <- fn(ctxs, msgs)
+			}()
+		}
+
+		var errors []string
+		for i := 0; i < len(sb.handlers); i++ {
+			if rerr := <-results; rerr != nil {
+				errors = append(errors, rerr.Error())
+			}
+		}
+		if len(errors) > 0 {
+			err = fmt.Errorf("subscriber error: %s", strings.Join(errors, "\n"))
+		}
+		return err
+	}
+}
+
+//nolint:gocyclo
+func (n *noopServer) newSubHandler(sb *subscriber, opts Options) broker.Handler {
 	return func(p broker.Event) (err error) {
 		defer func() {
 			if r := recover(); r != nil {
@@ -201,12 +326,12 @@ func (n *noopServer) createSubHandler(sb *subscriber, opts Options) broker.Handl
 		msg := p.Message()
 		// if we don't have headers, create empty map
 		if msg.Header == nil {
-			msg.Header = make(map[string]string)
+			msg.Header = metadata.New(2)
 		}
 
 		ct := msg.Header["Content-Type"]
 		if len(ct) == 0 {
-			msg.Header["Content-Type"] = defaultContentType
+			msg.Header.Set(metadata.HeaderContentType, defaultContentType)
 			ct = defaultContentType
 		}
 		cf, err := n.newCodec(ct)
@@ -214,12 +339,12 @@ func (n *noopServer) createSubHandler(sb *subscriber, opts Options) broker.Handl
 			return err
 		}
 
-		hdr := make(map[string]string, len(msg.Header))
+		hdr := metadata.New(len(msg.Header))
 		for k, v := range msg.Header {
 			if k == "Content-Type" {
 				continue
 			}
-			hdr[k] = v
+			hdr.Set(k, v)
 		}
 
 		ctx := metadata.NewIncomingContext(sb.opts.Context, hdr)
@@ -294,7 +419,6 @@ func (n *noopServer) createSubHandler(sb *subscriber, opts Options) broker.Handl
 		if len(errors) > 0 {
 			err = fmt.Errorf("subscriber error: %s", strings.Join(errors, "\n"))
 		}
-
 		return err
 	}
 }

server/wrapper.go

@@ -14,12 +14,20 @@ type HandlerFunc func(ctx context.Context, req Request, rsp interface{}) error
 // publication message.
 type SubscriberFunc func(ctx context.Context, msg Message) error
 
+// BatchSubscriberFunc represents a single method of a subscriber. It's used primarily
+// for the wrappers. What's handed to the actual method is the concrete
+// publication message. This func used by batch subscribers
+type BatchSubscriberFunc func(ctxs []context.Context, msgs []Message) error
+
 // HandlerWrapper wraps the HandlerFunc and returns the equivalent
 type HandlerWrapper func(HandlerFunc) HandlerFunc
 
 // SubscriberWrapper wraps the SubscriberFunc and returns the equivalent
 type SubscriberWrapper func(SubscriberFunc) SubscriberFunc
 
+// BatchSubscriberWrapper wraps the SubscriberFunc and returns the equivalent
+type BatchSubscriberWrapper func(BatchSubscriberFunc) BatchSubscriberFunc
+
 // StreamWrapper wraps a Stream interface and returns the equivalent.
 // Because streams exist for the lifetime of a method invocation this
 // is a convenient way to wrap a Stream as its in use for trace, monitoring,

util/router/compile.go

@@ -8,18 +8,18 @@ const (
 
 // Template is a compiled representation of path templates.
 type Template struct {
-	// Verb is a VERB part in the template
-	Verb string
-	// Original template (example: /v1/a_bit_of_everything)
-	Template string
-	// OpCodes is a sequence of operations
+	// Version is the version number of the format.
+	Version int
+	// OpCodes is a sequence of operations.
 	OpCodes []int
 	// Pool is a constant pool
 	Pool []string
-	// Fields is a list of field paths bound in this template
+	// Verb is a VERB part in the template.
+	Verb string
+	// Fields is a list of field paths bound in this template.
 	Fields []string
-	// Version is the version number of the format
-	Version int
+	// Original template (example: /v1/a_bit_of_everything)
+	Template string
 }
 
 // Compiler compiles utilities representation of path templates into marshallable operations.
@@ -29,9 +29,15 @@ type Compiler interface {
 }
 
 type op struct {
-	str     string
-	code    OpCode
-	operand int
+	// code is the opcode of the operation
+	code OpCode
+
+	// str is a string operand of the code.
+	// num is ignored if str is not empty.
+	str string
+
+	// num is a numeric operand of the code.
+	num int
 }
 
 func (w wildcard) compile() []op {
@@ -61,8 +67,8 @@ func (v variable) compile() []op {
 		ops = append(ops, s.compile()...)
 	}
 	ops = append(ops, op{
-		code:    OpConcatN,
-		operand: len(v.segments),
+		code: OpConcatN,
+		num:  len(v.segments),
 	}, op{
 		code: OpCapture,
 		str:  v.path,
@@ -77,7 +83,6 @@ func (t template) Compile() Template {
 		rawOps = append(rawOps, s.compile()...)
 	}
 
-	// ops := make([]int, 0, len(rawOps))
 	var (
 		ops    []int
 		pool   []string
@@ -87,8 +92,12 @@ func (t template) Compile() Template {
 	for _, op := range rawOps {
 		ops = append(ops, int(op.code))
 		if op.str == "" {
-			ops = append(ops, op.operand)
+			ops = append(ops, op.num)
 		} else {
+			// eof segment literal represents the "/" path pattern
+			if op.str == eof {
+				op.str = ""
+			}
 			if _, ok := consts[op.str]; !ok {
 				consts[op.str] = len(pool)
 				pool = append(pool, op.str)

util/router/compile_test.go

@@ -21,6 +21,13 @@ func TestCompile(t *testing.T) {
 		fields []string
 	}{
 		{},
+		{
+			segs: []segment{
+				literal(eof),
+			},
+			ops:  []int{int(OpLitPush), 0},
+			pool: []string{""},
+		},
 		{
 			segs: []segment{
 				wildcard{},

util/router/parse.go

@@ -3,11 +3,8 @@ package router
 // download from https://raw.githubusercontent.com/grpc-ecosystem/grpc-gateway/master/protoc-gen-grpc-gateway/httprule/parse.go
 
 import (
-	"context"
 	"fmt"
 	"strings"
-
-	"github.com/unistack-org/micro/v3/logger"
 )
 
 // InvalidTemplateError indicates that the path template is not valid.
@@ -83,8 +80,30 @@ func tokenize(path string) (tokens []string, verb string) {
 	}
 
 	l := len(tokens)
+	// See
+	// https://github.com/grpc-ecosystem/grpc-gateway/pull/1947#issuecomment-774523693 ;
+	// although normal and backwards-compat logic here is to use the last index
+	// of a colon, if the final segment is a variable followed by a colon, the
+	// part following the colon must be a verb. Hence if the previous token is
+	// an end var marker, we switch the index we're looking for to Index instead
+	// of LastIndex, so that we correctly grab the remaining part of the path as
+	// the verb.
+	var penultimateTokenIsEndVar bool
+	switch l {
+	case 0, 1:
+		// Not enough to be variable so skip this logic and don't result in an
+		// invalid index
+	default:
+		penultimateTokenIsEndVar = tokens[l-2] == "}"
+	}
 	t := tokens[l-1]
-	if idx := strings.LastIndex(t, ":"); idx == 0 {
+	var idx int
+	if penultimateTokenIsEndVar {
+		idx = strings.Index(t, ":")
+	} else {
+		idx = strings.LastIndex(t, ":")
+	}
+	if idx == 0 {
 		tokens, verb = tokens[:l-1], t[1:]
 	} else if idx > 0 {
 		tokens[l-1], verb = t[:idx], t[idx+1:]
@@ -101,22 +120,17 @@ type parser struct {
 
 // topLevelSegments is the target of this parser.
 func (p *parser) topLevelSegments() ([]segment, error) {
-	if logger.V(logger.TraceLevel) {
-		logger.Trace(context.TODO(), "Parsing %q", p.tokens)
+	if _, err := p.accept(typeEOF); err == nil {
+		p.tokens = p.tokens[:0]
+		return []segment{literal(eof)}, nil
 	}
 	segs, err := p.segments()
 	if err != nil {
 		return nil, err
 	}
-	if logger.V(logger.TraceLevel) {
-		logger.Trace(context.TODO(), "accept segments: %q; %q", p.accepted, p.tokens)
-	}
 	if _, err := p.accept(typeEOF); err != nil {
 		return nil, fmt.Errorf("unexpected token %q after segments %q", p.tokens[0], strings.Join(p.accepted, ""))
 	}
-	if logger.V(logger.TraceLevel) {
-		logger.Trace(context.TODO(), "accept eof: %q; %q", p.accepted, p.tokens)
-	}
 	return segs, nil
 }
 
@@ -126,9 +140,6 @@ func (p *parser) segments() ([]segment, error) {
 		return nil, err
 	}
 
-	if logger.V(logger.TraceLevel) {
-		logger.Trace(context.TODO(), "accept segment: %q; %q", p.accepted, p.tokens)
-	}
 	segs := []segment{s}
 	for {
 		if _, err := p.accept("/"); err != nil {
@@ -139,9 +150,6 @@ func (p *parser) segments() ([]segment, error) {
 			return segs, err
 		}
 		segs = append(segs, s)
-		if logger.V(logger.TraceLevel) {
-			logger.Trace(context.TODO(), "accept segment: %q; %q", p.accepted, p.tokens)
-		}
 	}
 }
 

util/router/parse_test.go

@@ -4,7 +4,6 @@ package router
 
 import (
 	"context"
-	"flag"
 	"fmt"
 	"reflect"
 	"testing"
@@ -16,6 +15,7 @@ func TestTokenize(t *testing.T) {
 	for _, spec := range []struct {
 		src    string
 		tokens []string
+		verb   string
 	}{
 		{
 			src:    "",
@@ -84,32 +84,74 @@ func TestTokenize(t *testing.T) {
 				eof,
 			},
 		},
+		{
+			src: "v1/a/{endpoint}:a",
+			tokens: []string{
+				"v1", "/",
+				"a", "/",
+				"{", "endpoint", "}",
+				eof,
+			},
+			verb: "a",
+		},
+		{
+			src: "v1/a/{endpoint}:b:c",
+			tokens: []string{
+				"v1", "/",
+				"a", "/",
+				"{", "endpoint", "}",
+				eof,
+			},
+			verb: "b:c",
+		},
 	} {
 		tokens, verb := tokenize(spec.src)
 		if got, want := tokens, spec.tokens; !reflect.DeepEqual(got, want) {
 			t.Errorf("tokenize(%q) = %q, _; want %q, _", spec.src, got, want)
 		}
-		if got, want := verb, ""; got != want {
-			t.Errorf("tokenize(%q) = _, %q; want _, %q", spec.src, got, want)
-		}
 
-		src := fmt.Sprintf("%s:%s", spec.src, "LOCK")
-		tokens, verb = tokenize(src)
-		if got, want := tokens, spec.tokens; !reflect.DeepEqual(got, want) {
-			t.Errorf("tokenize(%q) = %q, _; want %q, _", src, got, want)
-		}
-		if got, want := verb, "LOCK"; got != want {
-			t.Errorf("tokenize(%q) = _, %q; want _, %q", src, got, want)
+		switch {
+		case spec.verb != "":
+			if got, want := verb, spec.verb; !reflect.DeepEqual(got, want) {
+				t.Errorf("tokenize(%q) = %q, _; want %q, _", spec.src, got, want)
+			}
+
+		default:
+			if got, want := verb, ""; got != want {
+				t.Errorf("tokenize(%q) = _, %q; want _, %q", spec.src, got, want)
+			}
+
+			src := fmt.Sprintf("%s:%s", spec.src, "LOCK")
+			tokens, verb = tokenize(src)
+			if got, want := tokens, spec.tokens; !reflect.DeepEqual(got, want) {
+				t.Errorf("tokenize(%q) = %q, _; want %q, _", src, got, want)
+			}
+			if got, want := verb, "LOCK"; got != want {
+				t.Errorf("tokenize(%q) = _, %q; want _, %q", src, got, want)
+			}
 		}
 	}
 }
 
 func TestParseSegments(t *testing.T) {
-	flag.Set("v", "3")
 	for _, spec := range []struct {
 		tokens []string
 		want   []segment
 	}{
+		{
+			tokens: []string{eof},
+			want: []segment{
+				literal(eof),
+			},
+		},
+		{
+			// Note: this case will never arise as tokenize() will never return such a sequence of tokens
+			// and even if it does it will be treated as [eof]
+			tokens: []string{eof, "v1", eof},
+			want: []segment{
+				literal(eof),
+			},
+		},
 		{
 			tokens: []string{"v1", eof},
 			want: []segment{
@@ -251,7 +293,6 @@ func TestParseSegments(t *testing.T) {
 }
 
 func TestParseSegmentsWithErrors(t *testing.T) {
-	flag.Set("v", "3")
 	for _, spec := range []struct {
 		tokens []string
 	}{
@@ -275,10 +316,6 @@ func TestParseSegmentsWithErrors(t *testing.T) {
 			// invalid percent-encoding
 			tokens: []string{"a%2z", eof},
 		},
-		{
-			// empty segments
-			tokens: []string{eof},
-		},
 		{
 			// unterminated variable
 			tokens: []string{"{", "name", eof},

util/router/runtime.go

@@ -23,9 +23,9 @@ type rop struct {
 	operand int
 }
 
-// Pattern is a template pattern of http request paths defined in github.com/googleapis/googleapis/google/api/http.proto.
+// Pattern is a template pattern of http request paths defined in
+// https://github.com/googleapis/googleapis/blob/master/google/api/http.proto
 type Pattern struct {
-	verb string
 	// ops is a list of operations
 	ops []rop
 	// pool is a constant pool indexed by the operands or vars
@@ -36,32 +36,16 @@ type Pattern struct {
 	stacksize int
 	// tailLen is the length of the fixed-size segments after a deep wildcard
 	tailLen int
-	// assumeColonVerb indicates whether a path suffix after a final
-	// colon may only be interpreted as a verb.
-	assumeColonVerb bool
+	// verb is the VERB part of the path pattern. It is empty if the pattern does not have VERB part.
+	verb string
 }
 
-type patternOptions struct {
-	assumeColonVerb bool
-}
-
-// PatternOpt is an option for creating Patterns.
-type PatternOpt func(*patternOptions)
-
 // NewPattern returns a new Pattern from the given definition values.
 // "ops" is a sequence of op codes. "pool" is a constant pool.
 // "verb" is the verb part of the pattern. It is empty if the pattern does not have the part.
 // "version" must be 1 for now.
 // It returns an error if the given definition is invalid.
-//nolint:gocyclo
-func NewPattern(version int, ops []int, pool []string, verb string, opts ...PatternOpt) (Pattern, error) {
-	options := patternOptions{
-		assumeColonVerb: true,
-	}
-	for _, o := range opts {
-		o(&options)
-	}
-
+func NewPattern(version int, ops []int, pool []string, verb string) (Pattern, error) {
 	if version != 1 {
 		if logger.V(logger.TraceLevel) {
 			logger.Trace(context.TODO(), "unsupported version: %d", version)
@@ -159,13 +143,12 @@ func NewPattern(version int, ops []int, pool []string, verb string, opts ...Patt
 		typedOps = append(typedOps, op)
 	}
 	return Pattern{
-		ops:             typedOps,
-		pool:            pool,
-		vars:            vars,
-		stacksize:       maxstack,
-		tailLen:         tailLen,
-		verb:            verb,
-		assumeColonVerb: options.assumeColonVerb,
+		ops:       typedOps,
+		pool:      pool,
+		vars:      vars,
+		stacksize: maxstack,
+		tailLen:   tailLen,
+		verb:      verb,
 	}, nil
 }
 
@@ -185,7 +168,7 @@ func MustPattern(p Pattern, err error) Pattern {
 //nolint:gocyclo
 func (p Pattern) Match(components []string, verb string) (map[string]string, error) {
 	if p.verb != verb {
-		if p.assumeColonVerb || p.verb != "" {
+		if p.verb != "" {
 			return nil, ErrNotMatch
 		}
 		if len(components) == 0 {
@@ -274,11 +257,3 @@ func (p Pattern) String() string {
 	}
 	return "/" + segs
 }
-
-// AssumeColonVerbOpt indicates whether a path suffix after a final
-// colon may only be interpreted as a verb.
-func AssumeColonVerbOpt(val bool) PatternOpt {
-	return PatternOpt(func(o *patternOptions) {
-		o.assumeColonVerb = val
-	})
-}

util/router/types.go

@@ -8,9 +8,9 @@ import (
 )
 
 type template struct {
+	segments []segment
 	verb     string
 	template string
-	segments []segment
 }
 
 type segment interface {

util/token/jwt/jwt.go

@@ -4,7 +4,7 @@ import (
 	"encoding/base64"
 	"time"
 
-	"github.com/dgrijalva/jwt-go"
+	"github.com/golang-jwt/jwt"
 	"github.com/unistack-org/micro/v3/auth"
 	"github.com/unistack-org/micro/v3/metadata"
 	"github.com/unistack-org/micro/v3/util/token"