config: cleanup tests

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

broker/broker.go

@@ -8,31 +8,79 @@ import (
 	"github.com/unistack-org/micro/v3/metadata"
 )
 
-// DefaultBroker default broker
+// DefaultBroker default memory broker
 var DefaultBroker Broker = NewBroker()
 
+var (
+	// ErrNotConnected returns when broker used but not connected yet
+	ErrNotConnected = errors.New("broker not connected")
+	// ErrDisconnected returns when broker disconnected
+	ErrDisconnected = errors.New("broker disconnected")
+)
+
 // Broker is an interface used for asynchronous messaging.
 type Broker interface {
+	// Name returns broker instance name
 	Name() string
-	Init(...Option) error
+	// Init initilize broker
+	Init(opts ...Option) error
+	// Options returns broker options
 	Options() Options
+	// Address return configured address
 	Address() string
-	Connect(context.Context) error
-	Disconnect(context.Context) error
-	Publish(context.Context, string, *Message, ...PublishOption) error
-	Subscribe(context.Context, string, Handler, ...SubscribeOption) (Subscriber, error)
+	// Connect connects to broker
+	Connect(ctx context.Context) error
+	// Disconnect disconnect from broker
+	Disconnect(ctx context.Context) error
+	// Publish message to broker topic
+	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
 }
 
 // Handler is used to process messages via a subscription of a topic.
 type Handler func(Event) error
 
+// Events contains multiple events
+type Events []Event
+
+func (evs Events) Ack() error {
+	var err error
+	for _, ev := range evs {
+		if err = ev.Ack(); err != nil {
+			return err
+		}
+	}
+	return nil
+}
+
+func (evs Events) SetError(err error) {
+	for _, ev := range evs {
+		ev.SetError(err)
+	}
+}
+
+// BatchHandler is used to process messages in batches via a subscription of a topic.
+type BatchHandler func(Events) error
+
 // Event is given to a subscription handler for processing
 type Event interface {
+	// Topic returns event topic
 	Topic() string
+	// Message returns broker message
 	Message() *Message
+	// Ack acknowledge message
 	Ack() error
+	// Error returns message error (like decoding errors or some other)
 	Error() error
+	// SetError set event processing error
+	SetError(err error)
 }
 
 // RawMessage is a raw encoded JSON value.
@@ -58,13 +106,25 @@ func (m *RawMessage) UnmarshalJSON(data []byte) error {
 
 // Message is used to transfer data
 type Message struct {
-	Header metadata.Metadata // contains message metadata
-	Body   RawMessage        // contains message body
+	// Header contains message metadata
+	Header metadata.Metadata
+	// Body contains message body
+	Body RawMessage
+}
+
+// NewMessage create broker message with topic filled
+func NewMessage(topic string) *Message {
+	m := &Message{Header: metadata.New(2)}
+	m.Header.Set(metadata.HeaderTopic, topic)
+	return m
 }
 
 // Subscriber is a convenience return type for the Subscribe method
 type Subscriber interface {
+	// Options returns subscriber options
 	Options() SubscribeOptions
+	// Topic returns topic for subscription
 	Topic() string
-	Unsubscribe(context.Context) error
+	// Unsubscribe from topic
+	Unsubscribe(ctx context.Context) error
 }

broker/memory.go

@@ -2,18 +2,18 @@ package broker
 
 import (
 	"context"
-	"errors"
 	"sync"
 
 	"github.com/google/uuid"
 	"github.com/unistack-org/micro/v3/logger"
+	"github.com/unistack-org/micro/v3/metadata"
 	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"
 )
 
 type memoryBroker struct {
-	Subscribers map[string][]*memorySubscriber
+	subscribers map[string][]*memorySubscriber
 	addr        string
 	opts        Options
 	sync.RWMutex
@@ -28,12 +28,13 @@ type memoryEvent struct {
 }
 
 type memorySubscriber struct {
-	ctx     context.Context
-	exit    chan bool
-	handler Handler
-	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 {
@@ -77,7 +78,6 @@ func (m *memoryBroker) Disconnect(ctx context.Context) error {
 	}
 
 	m.connected = false
-
 	return nil
 }
 
@@ -92,67 +92,190 @@ func (m *memoryBroker) Publish(ctx context.Context, topic string, msg *Message,
 	m.RLock()
 	if !m.connected {
 		m.RUnlock()
-		return errors.New("not connected")
+		return ErrNotConnected
 	}
-
-	subs, ok := m.Subscribers[topic]
 	m.RUnlock()
-	if !ok {
-		return nil
-	}
 
-	var v interface{}
-	if m.opts.Codec != nil {
+	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
 		}
-		v = buf
+		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 {
+		m.RUnlock()
+		return ErrNotConnected
+	}
+	m.RUnlock()
+
+	options := NewPublishOptions(opts...)
+	vs := make([]msgWrapper, 0, len(msgs))
+	if m.opts.Codec == nil || options.BodyOnly {
+		for _, msg := range msgs {
+			topic, _ := msg.Header.Get(metadata.HeaderTopic)
+			vs = append(vs, msgWrapper{topic: topic, body: msg})
+		}
 	} else {
-		v = msg
+		for _, msg := range msgs {
+			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})
+		}
 	}
 
-	p := &memoryEvent{
-		topic:   topic,
-		message: v,
-		opts:    m.opts,
+	return m.publish(ctx, vs, opts...)
+}
+
+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 _, sub := range subs {
-		if err := sub.handler(p); err != nil {
-			p.err = 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())
-			}
+	for t, ms := range msgTopicMap {
+		m.RLock()
+		subs, ok := m.subscribers[t]
+		m.RUnlock()
+		if !ok {
 			continue
 		}
+
+		for _, sub := range subs {
+			// 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)
+					}
+				}
+			}
+			// 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)
+						}
+					}
+				}
+			}
+		}
 	}
 
 	return nil
 }
 
-func (m *memoryBroker) Subscribe(ctx context.Context, topic string, handler Handler, opts ...SubscribeOption) (Subscriber, error) {
+func (m *memoryBroker) BatchSubscribe(ctx context.Context, topic string, handler BatchHandler, opts ...SubscribeOption) (Subscriber, error) {
 	m.RLock()
 	if !m.connected {
 		m.RUnlock()
-		return nil, errors.New("not connected")
+		return nil, ErrNotConnected
 	}
 	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(),
+		topic:        topic,
+		batchhandler: handler,
+		opts:         options,
+		ctx:          ctx,
+	}
+
+	m.Lock()
+	m.subscribers[topic] = append(m.subscribers[topic], sub)
+	m.Unlock()
+
+	go func() {
+		<-sub.exit
+		m.Lock()
+		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.subscribers[topic] = newSubscribers
+		m.Unlock()
+	}()
+
+	return sub, nil
+}
+
+func (m *memoryBroker) Subscribe(ctx context.Context, topic string, handler Handler, opts ...SubscribeOption) (Subscriber, error) {
+	m.RLock()
+	if !m.connected {
+		m.RUnlock()
+		return nil, ErrNotConnected
+	}
+	m.RUnlock()
+
+	id, err := uuid.NewRandom()
+	if err != nil {
+		return nil, err
+	}
+
+	options := NewSubscribeOptions(opts...)
+
 	sub := &memorySubscriber{
 		exit:    make(chan bool, 1),
 		id:      id.String(),
@@ -163,20 +286,20 @@ func (m *memoryBroker) Subscribe(ctx context.Context, topic string, handler Hand
 	}
 
 	m.Lock()
-	m.Subscribers[topic] = append(m.Subscribers[topic], sub)
+	m.subscribers[topic] = append(m.subscribers[topic], sub)
 	m.Unlock()
 
 	go func() {
 		<-sub.exit
 		m.Lock()
-		var newSubscribers []*memorySubscriber
-		for _, sb := range m.Subscribers[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.Subscribers[topic] = newSubscribers
+		m.subscribers[topic] = newSubscribers
 		m.Unlock()
 	}()
 
@@ -221,6 +344,10 @@ func (m *memoryEvent) Error() error {
 	return m.err
 }
 
+func (m *memoryEvent) SetError(err error) {
+	m.err = err
+}
+
 func (m *memorySubscriber) Options() SubscribeOptions {
 	return m.opts
 }
@@ -238,6 +365,6 @@ func (m *memorySubscriber) Unsubscribe(ctx context.Context) error {
 func NewBroker(opts ...Option) Broker {
 	return &memoryBroker{
 		opts:        NewOptions(opts...),
-		Subscribers: make(map[string][]*memorySubscriber),
+		subscribers: make(map[string][]*memorySubscriber),
 	}
 }

broker/memory_test.go

@@ -4,8 +4,55 @@ import (
 	"context"
 	"fmt"
 	"testing"
+
+	"github.com/unistack-org/micro/v3/metadata"
 )
 
+func TestMemoryBatchBroker(t *testing.T) {
+	b := NewBroker()
+	ctx := context.Background()
+
+	if err := b.Connect(ctx); err != nil {
+		t.Fatalf("Unexpected connect error %v", err)
+	}
+
+	topic := "test"
+	count := 10
+
+	fn := func(evts Events) error {
+		return evts.Ack()
+	}
+
+	sub, err := b.BatchSubscribe(ctx, topic, fn)
+	if err != nil {
+		t.Fatalf("Unexpected error subscribing %v", err)
+	}
+
+	msgs := make([]*Message, 0, 0)
+	for i := 0; i < count; i++ {
+		message := &Message{
+			Header: map[string]string{
+				metadata.HeaderTopic: topic,
+				"foo":                "bar",
+				"id":                 fmt.Sprintf("%d", i),
+			},
+			Body: []byte(`"hello world"`),
+		}
+		msgs = append(msgs, message)
+	}
+
+	if err := b.BatchPublish(ctx, msgs); err != nil {
+		t.Fatalf("Unexpected error publishing %v", err)
+	}
+
+	if err := sub.Unsubscribe(ctx); err != nil {
+		t.Fatalf("Unexpected error unsubscribing from %s: %v", topic, err)
+	}
+
+	if err := b.Disconnect(ctx); err != nil {
+		t.Fatalf("Unexpected connect error %v", err)
+	}
+}
 func TestMemoryBroker(t *testing.T) {
 	b := NewBroker()
 	ctx := context.Background()
@@ -26,20 +73,27 @@ func TestMemoryBroker(t *testing.T) {
 		t.Fatalf("Unexpected error subscribing %v", err)
 	}
 
+	msgs := make([]*Message, 0, 0)
 	for i := 0; i < count; i++ {
 		message := &Message{
 			Header: map[string]string{
-				"foo": "bar",
-				"id":  fmt.Sprintf("%d", i),
+				metadata.HeaderTopic: topic,
+				"foo":                "bar",
+				"id":                 fmt.Sprintf("%d", i),
 			},
 			Body: []byte(`"hello world"`),
 		}
+		msgs = append(msgs, message)
 
 		if err := b.Publish(ctx, topic, message); err != nil {
-			t.Fatalf("Unexpected error publishing %d", i)
+			t.Fatalf("Unexpected error publishing %d err: %v", i, err)
 		}
 	}
 
+	if err := b.BatchPublish(ctx, msgs); err != nil {
+		t.Fatalf("Unexpected error publishing %v", err)
+	}
+
 	if err := sub.Unsubscribe(ctx); err != nil {
 		t.Fatalf("Unexpected error unsubscribing from %s: %v", topic, err)
 	}

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"
@@ -29,6 +30,8 @@ type Options struct {
 	TLSConfig *tls.Config
 	// ErrorHandler used when broker can't unmarshal incoming message
 	ErrorHandler Handler
+	// BatchErrorHandler used when broker can't unmashal incoming messages
+	BatchErrorHandler BatchHandler
 	// Name holds the broker name
 	Name string
 	// Addrs holds the broker address
@@ -71,11 +74,9 @@ func NewPublishOptions(opts ...PublishOption) PublishOptions {
 	options := PublishOptions{
 		Context: context.Background(),
 	}
-
 	for _, o := range opts {
 		o(&options)
 	}
-
 	return options
 }
 
@@ -85,12 +86,18 @@ type SubscribeOptions struct {
 	Context context.Context
 	// ErrorHandler used when broker can't unmarshal incoming message
 	ErrorHandler Handler
+	// BatchErrorHandler used when broker can't unmashal incoming messages
+	BatchErrorHandler BatchHandler
 	// Group holds consumer group
 	Group string
 	// AutoAck flag specifies auto ack of incoming message when no error happens
 	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
@@ -113,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) {
@@ -145,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 {
@@ -175,6 +143,14 @@ func ErrorHandler(h Handler) Option {
 	}
 }
 
+// BatchErrorHandler will catch all broker errors that cant be handled
+// in normal way, for example Codec errors
+func BatchErrorHandler(h BatchHandler) Option {
+	return func(o *Options) {
+		o.BatchErrorHandler = h
+	}
+}
+
 // SubscribeErrorHandler will catch all broker errors that cant be handled
 // in normal way, for example Codec errors
 func SubscribeErrorHandler(h Handler) SubscribeOption {
@@ -183,6 +159,14 @@ func SubscribeErrorHandler(h Handler) SubscribeOption {
 	}
 }
 
+// SubscribeBatchErrorHandler will catch all broker errors that cant be handled
+// in normal way, for example Codec errors
+func SubscribeBatchErrorHandler(h BatchHandler) SubscribeOption {
+	return func(o *SubscribeOptions) {
+		o.BatchErrorHandler = h
+	}
+}
+
 // Queue sets the subscribers queue
 // Deprecated
 func Queue(name string) SubscribeOption {
@@ -246,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["Content-Type"] = p.ContentType()
-	md["Micro-Topic"] = 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/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 {

metadata/metadata.go

@@ -6,8 +6,20 @@ import (
 	"sort"
 )
 
-// HeaderPrefix for all headers passed
-var HeaderPrefix = "Micro-"
+var (
+	// HeaderTopic is the header name that contains topic name
+	HeaderTopic = "Micro-Topic"
+	// HeaderContentType specifies content type of message
+	HeaderContentType = "Content-Type"
+	// HeaderEndpoint specifies endpoint in service
+	HeaderEndpoint = "Micro-Endpoint"
+	// HeaderService specifies service
+	HeaderService = "Micro-Service"
+	// HeaderTimeout specifies timeout of operation
+	HeaderTimeout = "Micro-Timeout"
+	// HeaderAuthorization specifies Authorization header
+	HeaderAuthorization = "Authorization"
+)
 
 // Metadata is our way of representing request headers internally.
 // They're used at the RPC level and translate back and forth

meter/meter.go

@@ -3,8 +3,9 @@ package meter
 
 import (
 	"io"
-	"reflect"
 	"sort"
+	"strconv"
+	"strings"
 	"time"
 )
 
@@ -77,36 +78,62 @@ type Summary interface {
 	UpdateDuration(time.Time)
 }
 
+// sort labels alphabeticaly by label name
 type byKey []string
 
 func (k byKey) Len() int           { return len(k) / 2 }
 func (k byKey) Less(i, j int) bool { return k[i*2] < k[j*2] }
 func (k byKey) Swap(i, j int) {
-	k[i*2], k[i*2+1], k[j*2], k[j*2+1] = k[j*2], k[j*2+1], k[i*2], k[i*2+1]
+	k[i*2], k[j*2] = k[j*2], k[i*2]
+	k[i*2+1], k[j*2+1] = k[j*2+1], k[i*2+1]
 }
 
-func Sort(slice *[]string) {
-	bk := byKey(*slice)
-	if bk.Len() <= 1 {
-		return
+// BuildLables used to sort labels and delete duplicates.
+// Last value wins in case of duplicate label keys.
+func BuildLabels(labels ...string) []string {
+	if len(labels)%2 == 1 {
+		labels = labels[:len(labels)-1]
 	}
-	sort.Sort(bk)
-	v := reflect.ValueOf(slice).Elem()
-	cnt := 0
-	key := 0
-	val := 1
-	for key < v.Len() {
-		if len(bk) > key+2 && bk[key] == bk[key+2] {
-			key += 2
-			val += 2
-			continue
-		}
-		v.Index(cnt).Set(v.Index(key))
-		cnt++
-		v.Index(cnt).Set(v.Index(val))
-		cnt++
-		key += 2
-		val += 2
-	}
-	v.SetLen(cnt)
+	sort.Sort(byKey(labels))
+	return labels
+}
+
+// BuildName used to combine metric with labels.
+// If labels count is odd, drop last element
+func BuildName(name string, labels ...string) string {
+	if len(labels)%2 == 1 {
+		labels = labels[:len(labels)-1]
+	}
+
+	if len(labels) > 2 {
+		sort.Sort(byKey(labels))
+
+		idx := 0
+		for {
+			if labels[idx] == labels[idx+2] {
+				copy(labels[idx:], labels[idx+2:])
+				labels = labels[:len(labels)-2]
+			} else {
+				idx += 2
+			}
+			if idx+2 >= len(labels) {
+				break
+			}
+		}
+	}
+
+	var b strings.Builder
+	_, _ = b.WriteString(name)
+	_, _ = b.WriteRune('{')
+	for idx := 0; idx < len(labels); idx += 2 {
+		if idx > 0 {
+			_, _ = b.WriteRune(',')
+		}
+		_, _ = b.WriteString(labels[idx])
+		_, _ = b.WriteString(`=`)
+		_, _ = b.WriteString(strconv.Quote(labels[idx+1]))
+	}
+	_, _ = b.WriteRune('}')
+
+	return b.String()
 }

meter/meter_test.go

@@ -14,11 +14,57 @@ func TestNoopMeter(t *testing.T) {
 	cnt.Inc()
 }
 
-func TestLabelsSort(t *testing.T) {
-	ls := []string{"server", "http", "register", "mdns", "broker", "broker1", "broker", "broker2", "server", "tcp"}
-	Sort(&ls)
+func testEq(a, b []string) bool {
+	if len(a) != len(b) {
+		return false
+	}
+	for i := range a {
+		if a[i] != b[i] {
+			return false
+		}
+	}
+	return true
+}
 
-	if ls[0] != "broker" || ls[1] != "broker2" {
-		t.Fatalf("sort error: %v", ls)
+func TestBuildLabels(t *testing.T) {
+	type testData struct {
+		src []string
+		dst []string
+	}
+
+	data := []testData{
+		testData{
+			src: []string{"zerolabel", "value3", "firstlabel", "value2"},
+			dst: []string{"firstlabel", "value2", "zerolabel", "value3"},
+		},
+	}
+
+	for _, d := range data {
+		if !testEq(d.dst, BuildLabels(d.src...)) {
+			t.Fatalf("slices not properly sorted: %v %v", d.dst, d.src)
+		}
+	}
+}
+
+func TestBuildName(t *testing.T) {
+	data := map[string][]string{
+		`my_metric{firstlabel="value2",zerolabel="value3"}`: []string{
+			"my_metric",
+			"zerolabel", "value3", "firstlabel", "value2",
+		},
+		`my_metric{broker="broker2",register="mdns",server="tcp"}`: []string{
+			"my_metric",
+			"broker", "broker1", "broker", "broker2", "server", "http", "server", "tcp", "register", "mdns",
+		},
+		`my_metric{aaa="aaa"}`: []string{
+			"my_metric",
+			"aaa", "aaa",
+		},
+	}
+
+	for e, d := range data {
+		if x := BuildName(d[0], d[1:]...); x != e {
+			t.Fatalf("expect: %s, result: %s", e, x)
+		}
 	}
 }

network/tunnel/broker/broker.go

@@ -24,9 +24,18 @@ type tunSubscriber struct {
 	opts     broker.SubscribeOptions
 }
 
+type tunBatchSubscriber struct {
+	listener tunnel.Listener
+	handler  broker.BatchHandler
+	closed   chan bool
+	topic    string
+	opts     broker.SubscribeOptions
+}
+
 type tunEvent struct {
 	message *broker.Message
 	topic   string
+	err     error
 }
 
 // used to access tunnel from options context
@@ -62,6 +71,36 @@ func (t *tunBroker) Disconnect(ctx context.Context) error {
 	return t.tunnel.Close(ctx)
 }
 
+func (t *tunBroker) BatchPublish(ctx context.Context, msgs []*broker.Message, opts ...broker.PublishOption) error {
+	// TODO: this is probably inefficient, we might want to just maintain an open connection
+	// it may be easier to add broadcast to the tunnel
+	topicMap := make(map[string]tunnel.Session)
+
+	var err error
+	for _, msg := range msgs {
+		topic, _ := msg.Header.Get(metadata.HeaderTopic)
+		c, ok := topicMap[topic]
+		if !ok {
+			c, err := t.tunnel.Dial(ctx, topic, tunnel.DialMode(tunnel.Multicast))
+			if err != nil {
+				return err
+			}
+			defer c.Close()
+			topicMap[topic] = c
+		}
+
+		if err = c.Send(&transport.Message{
+			Header: msg.Header,
+			Body:   msg.Body,
+		}); err != nil {
+			//	msg.SetError(err)
+			return err
+		}
+	}
+
+	return nil
+}
+
 func (t *tunBroker) Publish(ctx context.Context, topic string, m *broker.Message, opts ...broker.PublishOption) error {
 	// TODO: this is probably inefficient, we might want to just maintain an open connection
 	// it may be easier to add broadcast to the tunnel
@@ -77,6 +116,26 @@ func (t *tunBroker) Publish(ctx context.Context, topic string, m *broker.Message
 	})
 }
 
+func (t *tunBroker) BatchSubscribe(ctx context.Context, topic string, h broker.BatchHandler, opts ...broker.SubscribeOption) (broker.Subscriber, error) {
+	l, err := t.tunnel.Listen(ctx, topic, tunnel.ListenMode(tunnel.Multicast))
+	if err != nil {
+		return nil, err
+	}
+
+	tunSub := &tunBatchSubscriber{
+		topic:    topic,
+		handler:  h,
+		opts:     broker.NewSubscribeOptions(opts...),
+		closed:   make(chan bool),
+		listener: l,
+	}
+
+	// start processing
+	go tunSub.run()
+
+	return tunSub, nil
+}
+
 func (t *tunBroker) Subscribe(ctx context.Context, topic string, h broker.Handler, opts ...broker.SubscribeOption) (broker.Subscriber, error) {
 	l, err := t.tunnel.Listen(ctx, topic, tunnel.ListenMode(tunnel.Multicast))
 	if err != nil {
@@ -101,6 +160,49 @@ func (t *tunBroker) String() string {
 	return "tunnel"
 }
 
+func (t *tunBatchSubscriber) run() {
+	for {
+		// accept a new connection
+		c, err := t.listener.Accept()
+		if err != nil {
+			select {
+			case <-t.closed:
+				return
+			default:
+				continue
+			}
+		}
+
+		// receive message
+		m := new(transport.Message)
+		if err := c.Recv(m); err != nil {
+			if logger.V(logger.ErrorLevel) {
+				logger.Error(t.opts.Context, err.Error())
+			}
+			if err = c.Close(); err != nil {
+				if logger.V(logger.ErrorLevel) {
+					logger.Error(t.opts.Context, err.Error())
+				}
+			}
+			continue
+		}
+
+		// close the connection
+		c.Close()
+
+		evts := broker.Events{&tunEvent{
+			topic: t.topic,
+			message: &broker.Message{
+				Header: m.Header,
+				Body:   m.Body,
+			},
+		}}
+		// handle the message
+		go t.handler(evts)
+
+	}
+}
+
 func (t *tunSubscriber) run() {
 	for {
 		// accept a new connection
@@ -142,6 +244,24 @@ func (t *tunSubscriber) run() {
 	}
 }
 
+func (t *tunBatchSubscriber) Options() broker.SubscribeOptions {
+	return t.opts
+}
+
+func (t *tunBatchSubscriber) Topic() string {
+	return t.topic
+}
+
+func (t *tunBatchSubscriber) Unsubscribe(ctx context.Context) error {
+	select {
+	case <-t.closed:
+		return nil
+	default:
+		close(t.closed)
+		return t.listener.Close()
+	}
+}
+
 func (t *tunSubscriber) Options() broker.SubscribeOptions {
 	return t.opts
 }
@@ -173,7 +293,11 @@ func (t *tunEvent) Ack() error {
 }
 
 func (t *tunEvent) Error() error {
-	return nil
+	return t.err
+}
+
+func (t *tunEvent) SetError(err error) {
+	t.err = err
 }
 
 // NewBroker returns new tunnel broker

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/server.go

@@ -15,8 +15,8 @@ import (
 var DefaultServer Server = NewServer()
 
 var (
-	// DefaultAddress will be used if no address passed
-	DefaultAddress = ":0"
+	// DefaultAddress will be used if no address passed, use secure localhost
+	DefaultAddress = "127.0.0.1:0"
 	// DefaultName will be used if no name passed
 	DefaultName = "server"
 	// DefaultVersion will be used if no version passed

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/fn/fn.go

@@ -1,15 +0,0 @@
-package fn
-
-type Initer interface {
-	Init(opts ...interface{}) error
-}
-
-func Init(ifaces ...Initer) error {
-	var err error
-	for _, iface := range ifaces {
-		if err = iface.Init(); err != nil {
-			return err
-		}
-	}
-	return nil
-}

util/reflect/struct.go

@@ -7,7 +7,6 @@ import (
 	"reflect"
 	"regexp"
 	"strings"
-	"time"
 )
 
 // ErrInvalidParam specifies invalid url query params
@@ -15,11 +14,12 @@ var ErrInvalidParam = errors.New("invalid url query param provided")
 
 var bracketSplitter = regexp.MustCompile(`\[|\]`)
 
-var timeKind = reflect.TypeOf(time.Time{}).Kind()
+//var timeKind = reflect.ValueOf(time.Time{}).Kind()
 
 type StructField struct {
 	Field reflect.StructField
 	Value reflect.Value
+	Path  string
 }
 
 func StructFieldByTag(src interface{}, tkey string, tval string) (interface{}, error) {
@@ -35,7 +35,7 @@ func StructFieldByTag(src interface{}, tkey string, tval string) (interface{}, e
 	for idx := 0; idx < typ.NumField(); idx++ {
 		fld := typ.Field(idx)
 		val := sv.Field(idx)
-		if !val.CanSet() || len(fld.PkgPath) != 0 {
+		if len(fld.PkgPath) != 0 {
 			continue
 		}
 
@@ -66,6 +66,17 @@ func StructFieldByTag(src interface{}, tkey string, tval string) (interface{}, e
 	return nil, ErrNotFound
 }
 
+func StructFieldByPath(src interface{}, path string) (interface{}, error) {
+	var err error
+	for _, p := range strings.Split(path, ".") {
+		src, err = StructFieldByName(src, p)
+		if err != nil {
+			return nil, err
+		}
+	}
+	return src, err
+}
+
 func StructFieldByName(src interface{}, tkey string) (interface{}, error) {
 	sv := reflect.ValueOf(src)
 	if sv.Kind() == reflect.Ptr {
@@ -79,7 +90,7 @@ func StructFieldByName(src interface{}, tkey string) (interface{}, error) {
 	for idx := 0; idx < typ.NumField(); idx++ {
 		fld := typ.Field(idx)
 		val := sv.Field(idx)
-		if !val.CanSet() || len(fld.PkgPath) != 0 {
+		if len(fld.PkgPath) != 0 {
 			continue
 		}
 		if fld.Name == tkey {
@@ -105,6 +116,19 @@ func StructFieldByName(src interface{}, tkey string) (interface{}, error) {
 	return nil, ErrNotFound
 }
 
+// StructFieldsMap returns map[string]interface{} or error
+func StructFieldsMap(src interface{}) (map[string]interface{}, error) {
+	fields, err := StructFields(src)
+	if err != nil {
+		return nil, err
+	}
+	mp := make(map[string]interface{}, len(fields))
+	for _, field := range fields {
+		mp[field.Path] = field.Value.Interface()
+	}
+	return mp, nil
+}
+
 // StructFields returns slice of struct fields
 func StructFields(src interface{}) ([]StructField, error) {
 	var fields []StructField
@@ -116,25 +140,29 @@ func StructFields(src interface{}) ([]StructField, error) {
 	if sv.Kind() != reflect.Struct {
 		return nil, ErrInvalidStruct
 	}
-
 	typ := sv.Type()
 	for idx := 0; idx < typ.NumField(); idx++ {
 		fld := typ.Field(idx)
 		val := sv.Field(idx)
-		if !val.CanSet() || len(fld.PkgPath) != 0 {
+		if !val.IsValid() || len(fld.PkgPath) != 0 {
 			continue
 		}
+
 		switch val.Kind() {
-		case timeKind:
-			fields = append(fields, StructField{Field: fld, Value: val})
+		//case timeKind:
+		//fmt.Printf("GGG\n")
+		//fields = append(fields, StructField{Field: fld, Value: val, Path: fld.Name})
 		case reflect.Struct:
 			infields, err := StructFields(val.Interface())
 			if err != nil {
 				return nil, err
 			}
-			fields = append(fields, infields...)
+			for _, infield := range infields {
+				infield.Path = fmt.Sprintf("%s.%s", fld.Name, infield.Path)
+				fields = append(fields, infield)
+			}
 		default:
-			fields = append(fields, StructField{Field: fld, Value: val})
+			fields = append(fields, StructField{Field: fld, Value: val, Path: fld.Name})
 		}
 	}
 
@@ -194,7 +222,7 @@ func StructURLValues(src interface{}, pref string, tags []string) (url.Values, e
 	for idx := 0; idx < typ.NumField(); idx++ {
 		fld := typ.Field(idx)
 		val := sv.Field(idx)
-		if !val.CanSet() || len(fld.PkgPath) != 0 || !val.IsValid() {
+		if len(fld.PkgPath) != 0 || !val.IsValid() {
 			continue
 		}
 

util/reflect/struct_test.go

@@ -2,9 +2,80 @@ package reflect
 
 import (
 	"net/url"
+	"reflect"
+	rfl "reflect"
 	"testing"
 )
 
+func TestStructFieldsMap(t *testing.T) {
+	type NestedStr struct {
+		BBB string
+		CCC int
+	}
+	type Str struct {
+		Name   []string `json:"name" codec:"flatten"`
+		XXX    string   `json:"xxx"`
+		Nested NestedStr
+	}
+
+	val := &Str{Name: []string{"first", "second"}, XXX: "ttt", Nested: NestedStr{BBB: "ddd", CCC: 9}}
+	fields, err := StructFieldsMap(val)
+	if err != nil {
+		t.Fatal(err)
+	}
+	if v, ok := fields["Nested.BBB"]; !ok || v != "ddd" {
+		t.Fatalf("invalid field from %v", fields)
+	}
+}
+
+func TestStructFields(t *testing.T) {
+	type NestedStr struct {
+		BBB string
+		CCC int
+	}
+	type Str struct {
+		Name   []string `json:"name" codec:"flatten"`
+		XXX    string   `json:"xxx"`
+		Nested NestedStr
+	}
+
+	val := &Str{Name: []string{"first", "second"}, XXX: "ttt", Nested: NestedStr{BBB: "ddd", CCC: 9}}
+	fields, err := StructFields(val)
+	if err != nil {
+		t.Fatal(err)
+	}
+	var ok bool
+	for _, field := range fields {
+		if field.Path == "Nested.CCC" {
+			ok = true
+		}
+	}
+	if !ok {
+		t.Fatalf("struct fields returns invalid path: %v", fields)
+	}
+}
+
+func TestStructByPath(t *testing.T) {
+	type NestedStr struct {
+		BBB string
+		CCC int
+	}
+	type Str struct {
+		Name   []string `json:"name" codec:"flatten"`
+		XXX    string   `json:"xxx"`
+		Nested NestedStr
+	}
+
+	val := &Str{Name: []string{"first", "second"}, XXX: "ttt", Nested: NestedStr{BBB: "ddd", CCC: 9}}
+	field, err := StructFieldByPath(val, "Nested.CCC")
+	if err != nil {
+		t.Fatal(err)
+	}
+	if rfl.Indirect(reflect.ValueOf(field)).Int() != 9 {
+		t.Fatalf("invalid elem returned: %v", field)
+	}
+}
+
 func TestStructByTag(t *testing.T) {
 	type Str struct {
 		Name []string `json:"name" codec:"flatten"`

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"