broker: refactor (#396)

* remove subscribe from server
* remove publish from client
* broker package refactoring

Co-authored-by: vtolstov <vtolstov@users.noreply.github.com>
Reviewed-on: #396
Co-authored-by: Vasiliy Tolstov <v.tolstov@unistack.org>
Co-committed-by: Vasiliy Tolstov <v.tolstov@unistack.org>

broker/broker.go

@@ -18,6 +18,10 @@ var (
 	ErrNotConnected = errors.New("broker not connected")
 	// ErrDisconnected returns when broker disconnected
 	ErrDisconnected = errors.New("broker disconnected")
+	// ErrInvalidMessage returns when invalid Message passed
+	ErrInvalidMessage = errors.New("invalid message")
+	// ErrInvalidHandler returns when subscriber passed to Subscribe
+	ErrInvalidHandler = errors.New("invalid handler")
 	// DefaultGracefulTimeout
 	DefaultGracefulTimeout = 5 * time.Second
 )
@@ -36,14 +40,12 @@ type Broker interface {
 	Connect(ctx context.Context) error
 	// Disconnect disconnect from broker
 	Disconnect(ctx context.Context) error
+	// NewMessage create new broker message to publish.
+	NewMessage(ctx context.Context, hdr metadata.Metadata, body interface{}, opts ...PublishOption) (Message, error)
 	// Publish message to broker topic
-	Publish(ctx context.Context, topic string, msg *Message, opts ...PublishOption) error
+	Publish(ctx context.Context, topic string, messages ...Message) 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)
+	Subscribe(ctx context.Context, topic string, handler interface{}, opts ...SubscribeOption) (Subscriber, error)
 	// String type of broker
 	String() string
 	// Live returns broker liveness
@@ -55,72 +57,27 @@ type Broker interface {
 }
 
 type (
-	FuncPublish        func(ctx context.Context, topic string, msg *Message, opts ...PublishOption) error
-	HookPublish        func(next FuncPublish) FuncPublish
-	FuncBatchPublish   func(ctx context.Context, msgs []*Message, opts ...PublishOption) error
-	HookBatchPublish   func(next FuncBatchPublish) FuncBatchPublish
-	FuncSubscribe      func(ctx context.Context, topic string, h Handler, opts ...SubscribeOption) (Subscriber, error)
-	HookSubscribe      func(next FuncSubscribe) FuncSubscribe
-	FuncBatchSubscribe func(ctx context.Context, topic string, h BatchHandler, opts ...SubscribeOption) (Subscriber, error)
-	HookBatchSubscribe func(next FuncBatchSubscribe) FuncBatchSubscribe
+	FuncPublish   func(ctx context.Context, topic string, messages ...Message) error
+	HookPublish   func(next FuncPublish) FuncPublish
+	FuncSubscribe func(ctx context.Context, topic string, handler interface{}, opts ...SubscribeOption) (Subscriber, error)
+	HookSubscribe func(next FuncSubscribe) FuncSubscribe
 )
 
-// Handler is used to process messages via a subscription of a topic.
-type Handler func(Event) error
-
-// Events contains multiple events
-type Events []Event
-
-// Ack try to ack all events and return
-func (evs Events) Ack() error {
-	var err error
-	for _, ev := range evs {
-		if err = ev.Ack(); err != nil {
-			return err
-		}
-	}
-	return nil
-}
-
-// SetError sets error on event
-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 {
-	// Context return context.Context for event
+// Message is given to a subscription handler for processing
+type Message interface {
+	// Context for the message.
 	Context() context.Context
-	// Topic returns event topic
+	// Topic returns message destination topic.
 	Topic() string
-	// Message returns broker message
-	Message() *Message
-	// Ack acknowledge message
+	// Header returns message headers.
+	Header() metadata.Metadata
+	// Body returns broker message []byte slice
+	Body() []byte
+	// Unmarshal try to decode message body to dst.
+	// This is helper method that uses codec.Unmarshal.
+	Unmarshal(dst interface{}, opts ...codec.Option) error
+	// Ack acknowledge message if supported.
 	Ack() error
-	// Error returns message error (like decoding errors or some other)
-	Error() error
-	// SetError set event processing error
-	SetError(err error)
-}
-
-// Message is used to transfer data
-type Message struct {
-	// Header contains message metadata
-	Header metadata.Metadata
-	// Body contains message body
-	Body codec.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

broker/context.go

@@ -51,13 +51,3 @@ func SetOption(k, v interface{}) Option {
 		o.Context = context.WithValue(o.Context, k, v)
 	}
 }
-
-// SetPublishOption returns a function to setup a context with given value
-func SetPublishOption(k, v interface{}) PublishOption {
-	return func(o *PublishOptions) {
-		if o.Context == nil {
-			o.Context = context.Background()
-		}
-		o.Context = context.WithValue(o.Context, k, v)
-	}
-}

broker/context_test.go

@@ -49,17 +49,6 @@ func TestSetSubscribeOption(t *testing.T) {
 	}
 }
 
-func TestSetPublishOption(t *testing.T) {
-	type key struct{}
-	o := SetPublishOption(key{}, "test")
-	opts := &PublishOptions{}
-	o(opts)
-
-	if v, ok := opts.Context.Value(key{}).(string); !ok || v == "" {
-		t.Fatal("SetPublishOption not works")
-	}
-}
-
 func TestSetOption(t *testing.T) {
 	type key struct{}
 	o := SetOption(key{}, "test")

broker/memory/memory.go

@@ -2,9 +2,11 @@ package broker
 
 import (
 	"context"
+	"strings"
 	"sync"
 
 	"go.unistack.org/micro/v4/broker"
+	"go.unistack.org/micro/v4/codec"
 	"go.unistack.org/micro/v4/logger"
 	"go.unistack.org/micro/v4/metadata"
 	"go.unistack.org/micro/v4/options"
@@ -14,54 +16,90 @@ import (
 	"go.unistack.org/micro/v4/util/rand"
 )
 
-type memoryBroker struct {
-	funcPublish        broker.FuncPublish
-	funcBatchPublish   broker.FuncBatchPublish
-	funcSubscribe      broker.FuncSubscribe
-	funcBatchSubscribe broker.FuncBatchSubscribe
-	subscribers        map[string][]*memorySubscriber
-	addr               string
-	opts               broker.Options
+type Broker struct {
+	funcPublish   broker.FuncPublish
+	funcSubscribe broker.FuncSubscribe
+	subscribers   map[string][]*Subscriber
+	addr          string
+	opts          broker.Options
 	sync.RWMutex
 	connected bool
 }
 
-type memoryEvent struct {
-	err     error
-	message interface{}
+type memoryMessage struct {
+	c     codec.Codec
+	topic string
+	ctx   context.Context
+	body  []byte
+	hdr   metadata.Metadata
+	opts  broker.PublishOptions
+}
+
+func (m *memoryMessage) Ack() error {
+	return nil
+}
+
+func (m *memoryMessage) Body() []byte {
+	return m.body
+}
+
+func (m *memoryMessage) Header() metadata.Metadata {
+	return m.hdr
+}
+
+func (m *memoryMessage) Context() context.Context {
+	return m.ctx
+}
+
+func (m *memoryMessage) Topic() string {
+	return ""
+}
+
+func (m *memoryMessage) Unmarshal(dst interface{}, opts ...codec.Option) error {
+	return m.c.Unmarshal(m.body, dst)
+}
+
+type Subscriber struct {
+	ctx     context.Context
+	exit    chan bool
+	handler interface{}
+	id      string
 	topic   string
-	opts    broker.Options
+	opts    broker.SubscribeOptions
 }
 
-type memorySubscriber struct {
-	ctx          context.Context
-	exit         chan bool
-	handler      broker.Handler
-	batchhandler broker.BatchHandler
-	id           string
-	topic        string
-	opts         broker.SubscribeOptions
+func (b *Broker) newCodec(ct string) (codec.Codec, error) {
+	if idx := strings.IndexRune(ct, ';'); idx >= 0 {
+		ct = ct[:idx]
+	}
+	b.RLock()
+	c, ok := b.opts.Codecs[ct]
+	b.RUnlock()
+	if ok {
+		return c, nil
+	}
+	return nil, codec.ErrUnknownContentType
 }
 
-func (m *memoryBroker) Options() broker.Options {
-	return m.opts
+func (b *Broker) Options() broker.Options {
+	return b.opts
 }
 
-func (m *memoryBroker) Address() string {
-	return m.addr
+func (b *Broker) Address() string {
+	return b.addr
 }
 
-func (m *memoryBroker) Connect(ctx context.Context) error {
+func (b *Broker) Connect(ctx context.Context) error {
 	select {
 	case <-ctx.Done():
 		return ctx.Err()
 	default:
 	}
 
-	m.Lock()
-	defer m.Unlock()
+	b.Lock()
+	defer b.Unlock()
 
-	if m.connected {
+	if b.connected {
 		return nil
 	}
 
@@ -75,154 +113,126 @@ func (m *memoryBroker) Connect(ctx context.Context) error {
 	// set addr with port
 	addr = mnet.HostPort(addr, 10000+i)
 
-	m.addr = addr
-	m.connected = true
+	b.addr = addr
+	b.connected = true
 
 	return nil
 }
 
-func (m *memoryBroker) Disconnect(ctx context.Context) error {
+func (b *Broker) Disconnect(ctx context.Context) error {
 	select {
 	case <-ctx.Done():
 		return ctx.Err()
 	default:
 	}
 
-	m.Lock()
-	defer m.Unlock()
+	b.Lock()
+	defer b.Unlock()
 
-	if !m.connected {
+	if !b.connected {
 		return nil
 	}
 
-	m.connected = false
+	b.connected = false
 	return nil
 }
 
-func (m *memoryBroker) Init(opts ...broker.Option) error {
+func (b *Broker) Init(opts ...broker.Option) error {
 	for _, o := range opts {
-		o(&m.opts)
+		o(&b.opts)
 	}
 
-	m.funcPublish = m.fnPublish
-	m.funcBatchPublish = m.fnBatchPublish
-	m.funcSubscribe = m.fnSubscribe
-	m.funcBatchSubscribe = m.fnBatchSubscribe
+	b.funcPublish = b.fnPublish
+	b.funcSubscribe = b.fnSubscribe
 
-	m.opts.Hooks.EachPrev(func(hook options.Hook) {
+	b.opts.Hooks.EachPrev(func(hook options.Hook) {
 		switch h := hook.(type) {
 		case broker.HookPublish:
-			m.funcPublish = h(m.funcPublish)
-		case broker.HookBatchPublish:
-			m.funcBatchPublish = h(m.funcBatchPublish)
+			b.funcPublish = h(b.funcPublish)
 		case broker.HookSubscribe:
-			m.funcSubscribe = h(m.funcSubscribe)
-		case broker.HookBatchSubscribe:
-			m.funcBatchSubscribe = h(m.funcBatchSubscribe)
+			b.funcSubscribe = h(b.funcSubscribe)
 		}
 	})
 
 	return nil
 }
 
-func (m *memoryBroker) Publish(ctx context.Context, topic string, msg *broker.Message, opts ...broker.PublishOption) error {
-	return m.funcPublish(ctx, topic, msg, opts...)
+func (b *Broker) NewMessage(ctx context.Context, hdr metadata.Metadata, body interface{}, opts ...broker.PublishOption) (broker.Message, error) {
+	options := broker.NewPublishOptions(opts...)
+	m := &memoryMessage{ctx: ctx, hdr: hdr, opts: options}
+	c, err := b.newCodec(m.opts.ContentType)
+	if err == nil {
+		m.body, err = c.Marshal(body)
+	}
+	if err != nil {
+		return nil, err
+	}
+
+	return m, nil
 }
 
-func (m *memoryBroker) fnPublish(ctx context.Context, topic string, msg *broker.Message, opts ...broker.PublishOption) error {
-	msg.Header.Set(metadata.HeaderTopic, topic)
-	return m.publish(ctx, []*broker.Message{msg}, opts...)
+func (b *Broker) Publish(ctx context.Context, topic string, messages ...broker.Message) error {
+	return b.funcPublish(ctx, topic, messages...)
 }
 
-func (m *memoryBroker) BatchPublish(ctx context.Context, msgs []*broker.Message, opts ...broker.PublishOption) error {
-	return m.funcBatchPublish(ctx, msgs, opts...)
+func (b *Broker) fnPublish(ctx context.Context, topic string, messages ...broker.Message) error {
+	return b.publish(ctx, topic, messages...)
 }
 
-func (m *memoryBroker) fnBatchPublish(ctx context.Context, msgs []*broker.Message, opts ...broker.PublishOption) error {
-	return m.publish(ctx, msgs, opts...)
-}
-
-func (m *memoryBroker) publish(ctx context.Context, msgs []*broker.Message, opts ...broker.PublishOption) error {
-	m.RLock()
-	if !m.connected {
-		m.RUnlock()
+func (b *Broker) publish(ctx context.Context, topic string, messages ...broker.Message) error {
+	b.RLock()
+	if !b.connected {
+		b.RUnlock()
 		return broker.ErrNotConnected
 	}
-	m.RUnlock()
-
-	var err error
+	b.RUnlock()
 
 	select {
 	case <-ctx.Done():
 		return ctx.Err()
 	default:
-		options := broker.NewPublishOptions(opts...)
+	}
 
-		msgTopicMap := make(map[string]broker.Events)
-		for _, v := range msgs {
-			p := &memoryEvent{opts: m.opts}
+	b.RLock()
+	subs, ok := b.subscribers[topic]
+	b.RUnlock()
+	if !ok {
+		return nil
+	}
 
-			if m.opts.Codec == nil || options.BodyOnly {
-				p.topic, _ = v.Header.Get(metadata.HeaderTopic)
-				p.message = v.Body
-			} else {
-				p.topic, _ = v.Header.Get(metadata.HeaderTopic)
-				p.message, err = m.opts.Codec.Marshal(v)
-				if err != nil {
-					return err
-				}
+	var err error
+
+	for _, sub := range subs {
+		switch s := sub.handler.(type) {
+		default:
+			if b.opts.Logger.V(logger.ErrorLevel) {
+				b.opts.Logger.Error(ctx, "broker  handler error", broker.ErrInvalidHandler)
 			}
-			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[t]
-			m.RUnlock()
-			if !ok {
-				continue
-			}
-
-			for _, sub := range subs {
-				if sub.opts.BatchErrorHandler != nil {
-					beh = sub.opts.BatchErrorHandler
-				}
-				if sub.opts.ErrorHandler != nil {
-					eh = sub.opts.ErrorHandler
-				}
-
-				switch {
-				// batch processing
-				case sub.batchhandler != nil:
-					if err = sub.batchhandler(ms); err != nil {
-						ms.SetError(err)
-						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.Error(m.opts.Context, "broker ack error", err)
-						}
+		case func(broker.Message) error:
+			for _, message := range messages {
+				msg, ok := message.(*memoryMessage)
+				if !ok {
+					if b.opts.Logger.V(logger.ErrorLevel) {
+						b.opts.Logger.Error(ctx, "broker handler error", broker.ErrInvalidMessage)
 					}
-					// single processing
-				case sub.handler != nil:
-					for _, p := range ms {
-						if err = sub.handler(p); err != nil {
-							p.SetError(err)
-							if eh != nil {
-								_ = eh(p)
-							} else if m.opts.Logger.V(logger.ErrorLevel) {
-								m.opts.Logger.Error(m.opts.Context, "broker handler error", err)
-							}
-						} else if sub.opts.AutoAck {
-							if err = p.Ack(); err != nil {
-								m.opts.Logger.Error(m.opts.Context, "broker ack error", err)
-							}
+				}
+				msg.topic = topic
+				if err = s(msg); err == nil && sub.opts.AutoAck {
+					err = msg.Ack()
+				}
+				if err != nil {
+					if b.opts.Logger.V(logger.ErrorLevel) {
+						b.opts.Logger.Error(ctx, "broker handler error", err)
+					}
+				}
+			}
+		case func([]broker.Message) error:
+			if err = s(messages); err == nil && sub.opts.AutoAck {
+				for _, message := range messages {
+					err = message.Ack()
+					if err != nil {
+						if b.opts.Logger.V(logger.ErrorLevel) {
+							b.opts.Logger.Error(ctx, "broker handler error", err)
 						}
 					}
 				}
@@ -233,17 +243,21 @@ func (m *memoryBroker) publish(ctx context.Context, msgs []*broker.Message, opts
 	return nil
 }
 
-func (m *memoryBroker) BatchSubscribe(ctx context.Context, topic string, handler broker.BatchHandler, opts ...broker.SubscribeOption) (broker.Subscriber, error) {
-	return m.funcBatchSubscribe(ctx, topic, handler, opts...)
+func (b *Broker) Subscribe(ctx context.Context, topic string, handler interface{}, opts ...broker.SubscribeOption) (broker.Subscriber, error) {
+	return b.funcSubscribe(ctx, topic, handler, opts...)
 }
 
-func (m *memoryBroker) fnBatchSubscribe(ctx context.Context, topic string, handler broker.BatchHandler, opts ...broker.SubscribeOption) (broker.Subscriber, error) {
-	m.RLock()
-	if !m.connected {
-		m.RUnlock()
+func (b *Broker) fnSubscribe(ctx context.Context, topic string, handler interface{}, opts ...broker.SubscribeOption) (broker.Subscriber, error) {
+	if err := broker.IsValidHandler(handler); err != nil {
+		return nil, err
+	}
+
+	b.RLock()
+	if !b.connected {
+		b.RUnlock()
 		return nil, broker.ErrNotConnected
 	}
-	m.RUnlock()
+	b.RUnlock()
 
 	sid, err := id.New()
 	if err != nil {
@@ -252,56 +266,7 @@ func (m *memoryBroker) fnBatchSubscribe(ctx context.Context, topic string, handl
 
 	options := broker.NewSubscribeOptions(opts...)
 
-	sub := &memorySubscriber{
-		exit:         make(chan bool, 1),
-		id:           sid,
-		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 broker.Handler, opts ...broker.SubscribeOption) (broker.Subscriber, error) {
-	return m.funcSubscribe(ctx, topic, handler, opts...)
-}
-
-func (m *memoryBroker) fnSubscribe(ctx context.Context, topic string, handler broker.Handler, opts ...broker.SubscribeOption) (broker.Subscriber, error) {
-	m.RLock()
-	if !m.connected {
-		m.RUnlock()
-		return nil, broker.ErrNotConnected
-	}
-	m.RUnlock()
-
-	sid, err := id.New()
-	if err != nil {
-		return nil, err
-	}
-
-	options := broker.NewSubscribeOptions(opts...)
-
-	sub := &memorySubscriber{
+	sub := &Subscriber{
 		exit:    make(chan bool, 1),
 		id:      sid,
 		topic:   topic,
@@ -310,102 +275,64 @@ func (m *memoryBroker) fnSubscribe(ctx context.Context, topic string, handler br
 		ctx:     ctx,
 	}
 
-	m.Lock()
-	m.subscribers[topic] = append(m.subscribers[topic], sub)
-	m.Unlock()
+	b.Lock()
+	b.subscribers[topic] = append(b.subscribers[topic], sub)
+	b.Unlock()
 
 	go func() {
 		<-sub.exit
-		m.Lock()
-		newSubscribers := make([]*memorySubscriber, 0, len(m.subscribers)-1)
-		for _, sb := range m.subscribers[topic] {
+		b.Lock()
+		newSubscribers := make([]*Subscriber, 0, len(b.subscribers)-1)
+		for _, sb := range b.subscribers[topic] {
 			if sb.id == sub.id {
 				continue
 			}
 			newSubscribers = append(newSubscribers, sb)
 		}
-		m.subscribers[topic] = newSubscribers
-		m.Unlock()
+		b.subscribers[topic] = newSubscribers
+		b.Unlock()
 	}()
 
 	return sub, nil
 }
 
-func (m *memoryBroker) String() string {
+func (b *Broker) String() string {
 	return "memory"
 }
 
-func (m *memoryBroker) Name() string {
-	return m.opts.Name
+func (b *Broker) Name() string {
+	return b.opts.Name
 }
 
-func (m *memoryBroker) Live() bool {
+func (b *Broker) Live() bool {
 	return true
 }
 
-func (m *memoryBroker) Ready() bool {
+func (b *Broker) Ready() bool {
 	return true
 }
 
-func (m *memoryBroker) Health() bool {
+func (b *Broker) Health() bool {
 	return true
 }
 
-func (m *memoryEvent) Topic() string {
-	return m.topic
-}
-
-func (m *memoryEvent) Message() *broker.Message {
-	switch v := m.message.(type) {
-	case *broker.Message:
-		return v
-	case []byte:
-		msg := &broker.Message{}
-		if err := m.opts.Codec.Unmarshal(v, msg); err != nil {
-			if m.opts.Logger.V(logger.ErrorLevel) {
-				m.opts.Logger.Error(m.opts.Context, "[memory]: failed to unmarshal: %v", err)
-			}
-			return nil
-		}
-		return msg
-	}
-
-	return nil
-}
-
-func (m *memoryEvent) Ack() error {
-	return nil
-}
-
-func (m *memoryEvent) Error() error {
-	return m.err
-}
-
-func (m *memoryEvent) SetError(err error) {
-	m.err = err
-}
-
-func (m *memoryEvent) Context() context.Context {
-	return m.opts.Context
-}
-
-func (m *memorySubscriber) Options() broker.SubscribeOptions {
+func (m *Subscriber) Options() broker.SubscribeOptions {
 	return m.opts
 }
 
-func (m *memorySubscriber) Topic() string {
+func (m *Subscriber) Topic() string {
 	return m.topic
 }
 
-func (m *memorySubscriber) Unsubscribe(ctx context.Context) error {
+func (m *Subscriber) Unsubscribe(ctx context.Context) error {
 	m.exit <- true
 	return nil
 }
 
 // NewBroker return new memory broker
 func NewBroker(opts ...broker.Option) broker.Broker {
-	return &memoryBroker{
+	return &Broker{
 		opts:        broker.NewOptions(opts...),
-		subscribers: make(map[string][]*memorySubscriber),
+		subscribers: make(map[string][]*Subscriber),
 	}
 }

broker/memory/memory_test.go

@@ -5,62 +5,23 @@ import (
 	"fmt"
 	"testing"
 
+	"go.uber.org/atomic"
 	"go.unistack.org/micro/v4/broker"
+	"go.unistack.org/micro/v4/codec"
 	"go.unistack.org/micro/v4/metadata"
 )
 
-func TestMemoryBatchBroker(t *testing.T) {
-	b := NewBroker()
-	ctx := context.Background()
+type hldr struct {
+	c atomic.Int64
+}
 
-	if err := b.Init(); err != nil {
-		t.Fatalf("Unexpected init error %v", err)
-	}
-
-	if err := b.Connect(ctx); err != nil {
-		t.Fatalf("Unexpected connect error %v", err)
-	}
-
-	topic := "test"
-	count := 10
-
-	fn := func(evts broker.Events) error {
-		return evts.Ack()
-	}
-
-	sub, err := b.BatchSubscribe(ctx, topic, fn)
-	if err != nil {
-		t.Fatalf("Unexpected error subscribing %v", err)
-	}
-
-	msgs := make([]*broker.Message, 0, count)
-	for i := 0; i < count; i++ {
-		message := &broker.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 (h *hldr) Handler(m broker.Message) error {
+	h.c.Add(1)
+	return nil
 }
 
 func TestMemoryBroker(t *testing.T) {
-	b := NewBroker()
+	b := NewBroker(broker.Codec("application/octet-stream", codec.NewCodec()))
 	ctx := context.Background()
 
 	if err := b.Init(); err != nil {
@@ -72,38 +33,33 @@ func TestMemoryBroker(t *testing.T) {
 	}
 
 	topic := "test"
-	count := 10
+	count := int64(10)
 
-	fn := func(_ broker.Event) error {
-		return nil
-	}
+	h := &hldr{}
 
-	sub, err := b.Subscribe(ctx, topic, fn)
+	sub, err := b.Subscribe(ctx, topic, h.Handler)
 	if err != nil {
 		t.Fatalf("Unexpected error subscribing %v", err)
 	}
 
-	msgs := make([]*broker.Message, 0, count)
-	for i := 0; i < count; i++ {
-		message := &broker.Message{
-			Header: map[string]string{
-				metadata.HeaderTopic: topic,
-				"foo":                "bar",
-				"id":                 fmt.Sprintf("%d", i),
-			},
-			Body: []byte(`"hello world"`),
+	for i := int64(0); i < count; i++ {
+		message, err := b.NewMessage(ctx,
+			metadata.Pairs(
+				"foo", "bar",
+				"id", fmt.Sprintf("%d", i),
+			),
+			[]byte(`"hello world"`),
+			broker.PublishContentType("application/octet-stream"),
+		)
+		if err != nil {
+			t.Fatal(err)
 		}
-		msgs = append(msgs, message)
 
 		if err := b.Publish(ctx, topic, message); err != nil {
 			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)
 	}
@@ -111,4 +67,8 @@ func TestMemoryBroker(t *testing.T) {
 	if err := b.Disconnect(ctx); err != nil {
 		t.Fatalf("Unexpected connect error %v", err)
 	}
+
+	if h.c.Load() != count {
+		t.Fatal("invalid messages count received")
+	}
 }

broker/noop.go

@@ -3,24 +3,37 @@ package broker
 import (
 	"context"
 	"strings"
+	"sync"
 
+	"go.unistack.org/micro/v4/codec"
+	"go.unistack.org/micro/v4/metadata"
 	"go.unistack.org/micro/v4/options"
 )
 
 type NoopBroker struct {
-	funcPublish        FuncPublish
-	funcBatchPublish   FuncBatchPublish
-	funcSubscribe      FuncSubscribe
-	funcBatchSubscribe FuncBatchSubscribe
-	opts               Options
+	funcPublish   FuncPublish
+	funcSubscribe FuncSubscribe
+	opts          Options
+	sync.RWMutex
+}
+
+func (b *NoopBroker) newCodec(ct string) (codec.Codec, error) {
+	if idx := strings.IndexRune(ct, ';'); idx >= 0 {
+		ct = ct[:idx]
+	}
+	b.RLock()
+	c, ok := b.opts.Codecs[ct]
+	b.RUnlock()
+	if ok {
+		return c, nil
+	}
+	return nil, codec.ErrUnknownContentType
 }
 
 func NewBroker(opts ...Option) *NoopBroker {
 	b := &NoopBroker{opts: NewOptions(opts...)}
 	b.funcPublish = b.fnPublish
-	b.funcBatchPublish = b.fnBatchPublish
 	b.funcSubscribe = b.fnSubscribe
-	b.funcBatchSubscribe = b.fnBatchSubscribe
 
 	return b
 }
@@ -55,20 +68,14 @@ func (b *NoopBroker) Init(opts ...Option) error {
 	}
 
 	b.funcPublish = b.fnPublish
-	b.funcBatchPublish = b.fnBatchPublish
 	b.funcSubscribe = b.fnSubscribe
-	b.funcBatchSubscribe = b.fnBatchSubscribe
 
 	b.opts.Hooks.EachPrev(func(hook options.Hook) {
 		switch h := hook.(type) {
 		case HookPublish:
 			b.funcPublish = h(b.funcPublish)
-		case HookBatchPublish:
-			b.funcBatchPublish = h(b.funcBatchPublish)
 		case HookSubscribe:
 			b.funcSubscribe = h(b.funcSubscribe)
-		case HookBatchSubscribe:
-			b.funcBatchSubscribe = h(b.funcBatchSubscribe)
 		}
 	})
 
@@ -87,43 +94,72 @@ func (b *NoopBroker) Address() string {
 	return strings.Join(b.opts.Addrs, ",")
 }
 
-func (b *NoopBroker) fnBatchPublish(_ context.Context, _ []*Message, _ ...PublishOption) error {
+type noopMessage struct {
+	c    codec.Codec
+	ctx  context.Context
+	body []byte
+	hdr  metadata.Metadata
+	opts PublishOptions
+}
+
+func (m *noopMessage) Ack() error {
 	return nil
 }
 
-func (b *NoopBroker) BatchPublish(ctx context.Context, msgs []*Message, opts ...PublishOption) error {
-	return b.funcBatchPublish(ctx, msgs, opts...)
+func (m *noopMessage) Body() []byte {
+	return m.body
 }
 
-func (b *NoopBroker) fnPublish(_ context.Context, _ string, _ *Message, _ ...PublishOption) error {
+func (m *noopMessage) Header() metadata.Metadata {
+	return m.hdr
+}
+
+func (m *noopMessage) Context() context.Context {
+	return m.ctx
+}
+
+func (m *noopMessage) Topic() string {
+	return ""
+}
+
+func (m *noopMessage) Unmarshal(dst interface{}, opts ...codec.Option) error {
+	return m.c.Unmarshal(m.body, dst)
+}
+
+func (b *NoopBroker) NewMessage(ctx context.Context, hdr metadata.Metadata, body interface{}, opts ...PublishOption) (Message, error) {
+	options := NewPublishOptions(opts...)
+	m := &noopMessage{ctx: ctx, hdr: hdr, opts: options}
+	c, err := b.newCodec(m.opts.ContentType)
+	if err == nil {
+		m.body, err = c.Marshal(body)
+	}
+	if err != nil {
+		return nil, err
+	}
+
+	return m, nil
+}
+
+func (b *NoopBroker) fnPublish(_ context.Context, _ string, _ ...Message) error {
 	return nil
 }
 
-func (b *NoopBroker) Publish(ctx context.Context, topic string, msg *Message, opts ...PublishOption) error {
-	return b.funcPublish(ctx, topic, msg, opts...)
+func (b *NoopBroker) Publish(ctx context.Context, topic string, msg ...Message) error {
+	return b.funcPublish(ctx, topic, msg...)
 }
 
 type NoopSubscriber struct {
-	ctx          context.Context
-	topic        string
-	handler      Handler
-	batchHandler BatchHandler
-	opts         SubscribeOptions
+	ctx     context.Context
+	topic   string
+	handler interface{}
+	opts    SubscribeOptions
 }
 
-func (b *NoopBroker) fnBatchSubscribe(ctx context.Context, topic string, handler BatchHandler, opts ...SubscribeOption) (Subscriber, error) {
-	return &NoopSubscriber{ctx: ctx, topic: topic, opts: NewSubscribeOptions(opts...), batchHandler: handler}, nil
-}
-
-func (b *NoopBroker) BatchSubscribe(ctx context.Context, topic string, handler BatchHandler, opts ...SubscribeOption) (Subscriber, error) {
-	return b.funcBatchSubscribe(ctx, topic, handler, opts...)
-}
-
-func (b *NoopBroker) fnSubscribe(ctx context.Context, topic string, handler Handler, opts ...SubscribeOption) (Subscriber, error) {
+func (b *NoopBroker) fnSubscribe(ctx context.Context, topic string, handler interface{}, opts ...SubscribeOption) (Subscriber, error) {
 	return &NoopSubscriber{ctx: ctx, topic: topic, opts: NewSubscribeOptions(opts...), handler: handler}, nil
 }
 
-func (b *NoopBroker) Subscribe(ctx context.Context, topic string, handler Handler, opts ...SubscribeOption) (Subscriber, error) {
+func (b *NoopBroker) Subscribe(ctx context.Context, topic string, handler interface{}, opts ...SubscribeOption) (Subscriber, error) {
 	return b.funcSubscribe(ctx, topic, handler, opts...)
 }
 

broker/noop_test.go

@@ -10,9 +10,9 @@ type testHook struct {
 }
 
 func (t *testHook) Publish1(fn FuncPublish) FuncPublish {
-	return func(ctx context.Context, topic string, msg *Message, opts ...PublishOption) error {
+	return func(ctx context.Context, topic string, messages ...Message) error {
 		t.f = true
-		return fn(ctx, topic, msg, opts...)
+		return fn(ctx, topic, messages...)
 	}
 }
 

broker/options.go

@@ -23,8 +23,8 @@ type Options struct {
 	Tracer tracer.Tracer
 	// Register can be used for clustering
 	Register register.Register
-	// Codec holds the codec for marshal/unmarshal
-	Codec codec.Codec
+	// Codecs holds the codecs for marshal/unmarshal based on content-type
+	Codecs map[string]codec.Codec
 	// Logger used for logging
 	Logger logger.Logger
 	// Meter used for metrics
@@ -37,11 +37,6 @@ type Options struct {
 	// TLSConfig holds tls.TLSConfig options
 	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
-
 	// Addrs holds the broker address
 	Addrs []string
 	// Hooks can be run before broker Publish/BatchPublish and
@@ -59,10 +54,11 @@ func NewOptions(opts ...Option) Options {
 		Logger:          logger.DefaultLogger,
 		Context:         context.Background(),
 		Meter:           meter.DefaultMeter,
-		Codec:           codec.DefaultCodec,
+		Codecs:          make(map[string]codec.Codec),
 		Tracer:          tracer.DefaultTracer,
 		GracefulTimeout: DefaultGracefulTimeout,
 	}
+
 	for _, o := range opts {
 		o(&options)
 	}
@@ -78,17 +74,16 @@ func Context(ctx context.Context) Option {
 
 // PublishOptions struct
 type PublishOptions struct {
-	// Context holds external options
-	Context context.Context
-	// BodyOnly flag says the message contains raw body bytes
+	// ContentType for message body
+	ContentType string
+	// BodyOnly flag says the message contains raw body bytes and don't need
+	// codec Marshal method
 	BodyOnly bool
 }
 
 // NewPublishOptions creates PublishOptions struct
 func NewPublishOptions(opts ...PublishOption) PublishOptions {
-	options := PublishOptions{
-		Context: context.Background(),
-	}
+	options := PublishOptions{}
 	for _, o := range opts {
 		o(&options)
 	}
@@ -99,10 +94,6 @@ func NewPublishOptions(opts ...PublishOption) PublishOptions {
 type SubscribeOptions struct {
 	// Context holds external options
 	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
@@ -121,6 +112,13 @@ type Option func(*Options)
 // PublishOption func
 type PublishOption func(*PublishOptions)
 
+// PublishContentType sets message content-type that used to Marshal
+func PublishContentType(ct string) PublishOption {
+	return func(o *PublishOptions) {
+		o.ContentType = ct
+	}
+}
+
 // PublishBodyOnly publish only body of the message
 func PublishBodyOnly(b bool) PublishOption {
 	return func(o *PublishOptions) {
@@ -128,13 +126,6 @@ func PublishBodyOnly(b bool) PublishOption {
 	}
 }
 
-// PublishContext sets the context
-func PublishContext(ctx context.Context) PublishOption {
-	return func(o *PublishOptions) {
-		o.Context = ctx
-	}
-}
-
 // Addrs sets the host addresses to be used by the broker
 func Addrs(addrs ...string) Option {
 	return func(o *Options) {
@@ -142,51 +133,10 @@ func Addrs(addrs ...string) Option {
 	}
 }
 
-// Codec sets the codec used for encoding/decoding used where
-// a broker does not support headers
-func Codec(c codec.Codec) Option {
+// Codec sets the codec used for encoding/decoding messages
+func Codec(ct string, c codec.Codec) Option {
 	return func(o *Options) {
-		o.Codec = c
-	}
-}
-
-// ErrorHandler will catch all broker errors that cant be handled
-// in normal way, for example Codec errors
-func ErrorHandler(h Handler) Option {
-	return func(o *Options) {
-		o.ErrorHandler = h
-	}
-}
-
-// 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 {
-	return func(o *SubscribeOptions) {
-		o.ErrorHandler = h
-	}
-}
-
-// 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 {
-	return func(o *SubscribeOptions) {
-		o.Group = name
+		o.Codecs[ct] = c
 	}
 }
 
@@ -253,14 +203,6 @@ func SubscribeContext(ctx context.Context) SubscribeOption {
 	}
 }
 
-// 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 {

broker/subscriber.go

@@ -0,0 +1,87 @@
+package broker
+
+import (
+	"fmt"
+	"reflect"
+	"unicode"
+	"unicode/utf8"
+)
+
+const (
+	messageSig  = "func(broker.Message) error"
+	messagesSig = "func([]broker.Message) error"
+)
+
+// Precompute the reflect type for error. Can't use error directly
+// because Typeof takes an empty interface value. This is annoying.
+var typeOfError = reflect.TypeOf((*error)(nil)).Elem()
+
+// Is this an exported - upper case - name?
+func isExported(name string) bool {
+	r, _ := utf8.DecodeRuneInString(name)
+	return unicode.IsUpper(r)
+}
+
+// Is this type exported or a builtin?
+func isExportedOrBuiltinType(t reflect.Type) bool {
+	for t.Kind() == reflect.Ptr {
+		t = t.Elem()
+	}
+	// PkgPath will be non-empty even for an exported type,
+	// so we need to check the type name as well.
+	return isExported(t.Name()) || t.PkgPath() == ""
+}
+
+// IsValidHandler func signature
+func IsValidHandler(sub interface{}) error {
+	typ := reflect.TypeOf(sub)
+	var argType reflect.Type
+	switch typ.Kind() {
+	case reflect.Func:
+		name := "Func"
+		switch typ.NumIn() {
+		case 1:
+			argType = typ.In(0)
+		default:
+			return fmt.Errorf("subscriber %v takes wrong number of args: %v required signature %s", name, typ.NumIn(), messageSig)
+		}
+		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 return values: %v require signature %s",
+				name, typ.NumOut(), messageSig)
+		}
+		if returnType := typ.Out(0); returnType != typeOfError {
+			return fmt.Errorf("subscriber %v returns %v not error", name, returnType.String())
+		}
+	default:
+		hdlr := reflect.ValueOf(sub)
+		name := reflect.Indirect(hdlr).Type().Name()
+
+		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(), messageSig)
+			}
+
+			if !isExportedOrBuiltinType(argType) {
+				return fmt.Errorf("%v argument type not exported: %v", name, argType)
+			}
+			if method.Type.NumOut() != 1 {
+				return fmt.Errorf(
+					"subscriber %v.%v has wrong number of return values: %v require signature %s",
+					name, method.Name, method.Type.NumOut(), messageSig)
+			}
+			if returnType := method.Type.Out(0); returnType != typeOfError {
+				return fmt.Errorf("subscriber %v.%v returns %v not error", name, method.Name, returnType.String())
+			}
+		}
+	}
+
+	return nil
+}