move options to dedicated package

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

broker/broker.go

@@ -5,12 +5,12 @@ import (
 	"context"
 	"errors"
 
-	"go.unistack.org/micro/v4/codec"
 	"go.unistack.org/micro/v4/metadata"
+	"go.unistack.org/micro/v4/options"
 )
 
 // DefaultBroker default memory broker
-var DefaultBroker Broker // =  NewBroker()
+var DefaultBroker Broker = NewBroker()
 
 var (
 	// ErrNotConnected returns when broker used but not connected yet
@@ -26,7 +26,7 @@ type Broker interface {
 	// Name returns broker instance name
 	Name() string
 	// Init initilize broker
-	Init(opts ...Option) error
+	Init(opts ...options.Option) error
 	// Options returns broker options
 	Options() Options
 	// Address return configured address
@@ -35,12 +35,10 @@ type Broker interface {
 	Connect(ctx context.Context) error
 	// Disconnect disconnect from broker
 	Disconnect(ctx context.Context) error
-	// NewMessage creates new broker message
-	NewMessage(endpoint string, req interface{}, opts ...MessageOption) Message
-	// Publish message to broker topic
-	Publish(ctx context.Context, msg interface{}, opts ...PublishOption) error
+	// Publish message, msg can be single broker.Message or []broker.Message
+	Publish(ctx context.Context, msg interface{}, opts ...options.Option) error
 	// Subscribe subscribes to topic message via handler
-	Subscribe(ctx context.Context, topic string, handler interface{}, opts ...SubscribeOption) (Subscriber, error)
+	Subscribe(ctx context.Context, topic string, handler interface{}, opts ...options.Option) (Subscriber, error)
 	// String type of broker
 	String() string
 }
@@ -49,9 +47,11 @@ type Broker interface {
 type Message interface {
 	// Context for the message
 	Context() context.Context
-	// Topic returns event topic
+	// Topic
 	Topic() string
-	// Body returns broker message
+	// Header returns message headers
+	Header() metadata.Metadata
+	// Body returns broker message may be []byte slice or some go struct
 	Body() interface{}
 	// Ack acknowledge message
 	Ack() error
@@ -60,21 +60,6 @@ type Message interface {
 	Error() error
 }
 
-// RawMessage is used to transfer data
-type RawMessage struct {
-	// Header contains message metadata
-	Header metadata.Metadata
-	// Body contains message body
-	Body codec.RawMessage
-}
-
-// NewMessage create broker message with topic filled
-func NewRawMessage(topic string) *RawMessage {
-	m := &RawMessage{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
@@ -84,3 +69,9 @@ type Subscriber interface {
 	// Unsubscribe from topic
 	Unsubscribe(ctx context.Context) error
 }
+
+// MessageHandler func signature for single message processing
+type MessageHandler func(Message) error
+
+// MessagesHandler func signature for batch message processing
+type MessagesHandler func([]Message) error

broker/context.go

@@ -22,33 +22,3 @@ func NewContext(ctx context.Context, s Broker) context.Context {
 	}
 	return context.WithValue(ctx, brokerKey{}, s)
 }
-
-// SetSubscribeOption returns a function to setup a context with given value
-func SetSubscribeOption(k, v interface{}) SubscribeOption {
-	return func(o *SubscribeOptions) {
-		if o.Context == nil {
-			o.Context = context.Background()
-		}
-		o.Context = context.WithValue(o.Context, k, v)
-	}
-}
-
-// SetOption returns a function to setup a context with given value
-func SetOption(k, v interface{}) Option {
-	return func(o *Options) {
-		if o.Context == nil {
-			o.Context = context.Background()
-		}
-		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

@@ -37,36 +37,3 @@ func TestNewNilContext(t *testing.T) {
 		t.Fatal("NewContext not works")
 	}
 }
-
-func TestSetSubscribeOption(t *testing.T) {
-	type key struct{}
-	o := SetSubscribeOption(key{}, "test")
-	opts := &SubscribeOptions{}
-	o(opts)
-
-	if v, ok := opts.Context.Value(key{}).(string); !ok || v == "" {
-		t.Fatal("SetSubscribeOption not works")
-	}
-}
-
-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")
-	opts := &Options{}
-	o(opts)
-
-	if v, ok := opts.Context.Value(key{}).(string); !ok || v == "" {
-		t.Fatal("SetOption not works")
-	}
-}

broker/memory.go

@@ -1,20 +1,23 @@
-//go:build ignore
-
 package broker
 
 import (
 	"context"
+	"fmt"
 	"sync"
 	"time"
 
+	"go.unistack.org/micro/v4/codec"
 	"go.unistack.org/micro/v4/logger"
+	"go.unistack.org/micro/v4/metadata"
+	"go.unistack.org/micro/v4/options"
+	"go.unistack.org/micro/v4/semconv"
 	maddr "go.unistack.org/micro/v4/util/addr"
 	"go.unistack.org/micro/v4/util/id"
 	mnet "go.unistack.org/micro/v4/util/net"
 	"go.unistack.org/micro/v4/util/rand"
 )
 
-type memoryBroker struct {
+type MemoryBroker struct {
 	subscribers map[string][]*memorySubscriber
 	addr        string
 	opts        Options
@@ -22,15 +25,15 @@ type memoryBroker struct {
 	connected bool
 }
 
-func (m *memoryBroker) Options() Options {
+func (m *MemoryBroker) Options() Options {
 	return m.opts
 }
 
-func (m *memoryBroker) Address() string {
+func (m *MemoryBroker) Address() string {
 	return m.addr
 }
 
-func (m *memoryBroker) Connect(ctx context.Context) error {
+func (m *MemoryBroker) Connect(ctx context.Context) error {
 	m.Lock()
 	defer m.Unlock()
 
@@ -54,30 +57,33 @@ func (m *memoryBroker) Connect(ctx context.Context) error {
 	return nil
 }
 
-func (m *memoryBroker) Disconnect(ctx context.Context) error {
+func (m *MemoryBroker) Disconnect(ctx context.Context) error {
 	m.Lock()
 	defer m.Unlock()
 
-	if !m.connected {
-		return nil
+	select {
+	case <-ctx.Done():
+		return ctx.Err()
+	default:
+		if m.connected {
+			m.connected = false
+		}
 	}
 
-	m.connected = false
 	return nil
 }
 
-func (m *memoryBroker) Init(opts ...Option) error {
+func (m *MemoryBroker) Init(opts ...options.Option) error {
+	var err error
 	for _, o := range opts {
-		o(&m.opts)
+		if err = o(&m.opts); err != nil {
+			return err
+		}
 	}
 	return nil
 }
 
-func (m *memoryBroker) NewMessage(endpoint string, req interface{}, opts ...MessageOption) Message {
-	return &memoryMessage{}
-}
-
-func (m *memoryBroker) Publish(ctx context.Context, message interface{}, opts ...PublishOption) error {
+func (m *MemoryBroker) Publish(ctx context.Context, message interface{}, opts ...options.Option) error {
 	m.RLock()
 	if !m.connected {
 		m.RUnlock()
@@ -92,30 +98,34 @@ func (m *memoryBroker) Publish(ctx context.Context, message interface{}, opts ..
 		return ctx.Err()
 	default:
 		options := NewPublishOptions(opts...)
-		var msgs []*memoryMessage
+		var msgs []Message
 		switch v := message.(type) {
-		case *memoryMessage:
-			msgs = []*memoryMessage{v}
-		case []*memoryMessage:
+		case []Message:
 			msgs = v
+		case Message:
+			msgs = append(msgs, v)
 		default:
 			return ErrInvalidMessage
 		}
 		msgTopicMap := make(map[string][]*memoryMessage)
 		for _, msg := range msgs {
 			p := &memoryMessage{opts: options}
-			/*
-				if mb, ok := msg.Body().(*codec.Frame); ok {
-					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
-					}
+			p.topic, _ = msg.Header().Get(metadata.HeaderTopic)
+			if v, ok := msg.Body().(*codec.Frame); ok {
+				p.body = msg.Body()
+			} else if len(m.opts.Codecs) == 0 {
+				p.body = msg.Body()
+			} else {
+				cf, ok := m.opts.Codecs[options.ContentType]
+				if !ok {
+					return fmt.Errorf("%s: %s", codec.ErrUnknownContentType, options.ContentType)
 				}
-			*/
-			msgTopicMap[msg.Topic()] = append(msgTopicMap[p.topic], p)
+				p.body, err = cf.Marshal(v)
+				if err != nil {
+					return err
+				}
+			}
+			msgTopicMap[p.topic] = append(msgTopicMap[p.topic], p)
 		}
 
 		eh := m.opts.ErrorHandler
@@ -123,55 +133,83 @@ func (m *memoryBroker) Publish(ctx context.Context, message interface{}, opts ..
 		for t, ms := range msgTopicMap {
 			ts := time.Now()
 
-			m.opts.Meter.Counter(PublishMessageInflight, "endpoint", t).Add(len(ms))
-			m.opts.Meter.Counter(SubscribeMessageInflight, "endpoint", t).Add(len(ms))
+			m.opts.Meter.Counter(semconv.PublishMessageInflight, "endpoint", t).Add(len(ms))
+			m.opts.Meter.Counter(semconv.SubscribeMessageInflight, "endpoint", t).Add(len(ms))
 
 			m.RLock()
 			subs, ok := m.subscribers[t]
 			m.RUnlock()
 			if !ok {
-				m.opts.Meter.Counter(PublishMessageTotal, "endpoint", t, "status", "failure").Add(len(ms))
-				m.opts.Meter.Counter(PublishMessageInflight, "endpoint", t).Add(-len(ms))
-				m.opts.Meter.Counter(SubscribeMessageInflight, "endpoint", t).Add(-len(ms))
+				m.opts.Meter.Counter(semconv.PublishMessageTotal, "endpoint", t, "status", "failure").Add(len(ms))
+				m.opts.Meter.Counter(semconv.PublishMessageInflight, "endpoint", t).Add(-len(ms))
+				m.opts.Meter.Counter(semconv.SubscribeMessageInflight, "endpoint", t).Add(-len(ms))
 				continue
 			}
 
-			m.opts.Meter.Counter(PublishMessageTotal, "endpoint", t, "status", "success").Add(len(ms))
+			m.opts.Meter.Counter(semconv.PublishMessageTotal, "endpoint", t, "status", "success").Add(len(ms))
 			for _, sub := range subs {
 				if sub.opts.ErrorHandler != nil {
 					eh = sub.opts.ErrorHandler
 				}
 
-				for _, p := range ms {
-					if err = sub.handler(p); err != nil {
-						m.opts.Meter.Counter(SubscribeMessageTotal, "endpoint", t, "status", "failure").Inc()
+				switch mh := sub.handler.(type) {
+				case MessagesHandler:
+					mhs := make([]Message, 0, len(ms))
+					for _, m := range ms {
+						mhs = append(mhs, m)
+					}
+					if err = mh(mhs); err != nil {
+						m.opts.Meter.Counter(semconv.SubscribeMessageTotal, "endpoint", t, "status", "failure").Add(len(ms))
 						if eh != nil {
-							_ = eh(p)
+							switch meh := eh.(type) {
+							case MessagesHandler:
+								_ = meh(mhs)
+							case MessageHandler:
+								for _, me := range mhs {
+									_ = meh(me)
+								}
+							}
 						} 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)
-								m.opts.Meter.Counter(SubscribeMessageTotal, "endpoint", t, "status", "failure").Inc()
-							} else {
-								m.opts.Meter.Counter(SubscribeMessageTotal, "endpoint", t, "status", "success").Inc()
+					}
+				case MessageHandler:
+					for _, p := range ms {
+						if err = mh(p); err != nil {
+							m.opts.Meter.Counter(semconv.SubscribeMessageTotal, "endpoint", t, "status", "failure").Inc()
+							if eh != nil {
+								switch meh := eh.(type) {
+								case MessageHandler:
+									_ = meh(p)
+								case MessagesHandler:
+									_ = meh([]Message{p})
+								}
+							} else if m.opts.Logger.V(logger.ErrorLevel) {
+								m.opts.Logger.Error(m.opts.Context, err.Error())
 							}
 						} else {
-							m.opts.Meter.Counter(SubscribeMessageTotal, "endpoint", t, "status", "success").Inc()
+							if sub.opts.AutoAck {
+								if err = p.Ack(); err != nil {
+									m.opts.Logger.Errorf(m.opts.Context, "ack failed: %v", err)
+									m.opts.Meter.Counter(semconv.SubscribeMessageTotal, "endpoint", t, "status", "failure").Inc()
+								} else {
+									m.opts.Meter.Counter(semconv.SubscribeMessageTotal, "endpoint", t, "status", "success").Inc()
+								}
+							} else {
+								m.opts.Meter.Counter(semconv.SubscribeMessageTotal, "endpoint", t, "status", "success").Inc()
+							}
 						}
+						m.opts.Meter.Counter(semconv.PublishMessageInflight, "endpoint", t).Add(-1)
+						m.opts.Meter.Counter(semconv.SubscribeMessageInflight, "endpoint", t).Add(-1)
 					}
-					m.opts.Meter.Counter(PublishMessageInflight, "endpoint", t).Add(-1)
-					m.opts.Meter.Counter(SubscribeMessageInflight, "endpoint", t).Add(-1)
 				}
-
 			}
+
 			te := time.Since(ts)
-			m.opts.Meter.Summary(PublishMessageLatencyMicroseconds, "endpoint", t).Update(te.Seconds())
-			m.opts.Meter.Histogram(PublishMessageDurationSeconds, "endpoint", t).Update(te.Seconds())
-			m.opts.Meter.Summary(SubscribeMessageLatencyMicroseconds, "endpoint", t).Update(te.Seconds())
-			m.opts.Meter.Histogram(SubscribeMessageDurationSeconds, "endpoint", t).Update(te.Seconds())
+			m.opts.Meter.Summary(semconv.PublishMessageLatencyMicroseconds, "endpoint", t).Update(te.Seconds())
+			m.opts.Meter.Histogram(semconv.PublishMessageDurationSeconds, "endpoint", t).Update(te.Seconds())
+			m.opts.Meter.Summary(semconv.SubscribeMessageLatencyMicroseconds, "endpoint", t).Update(te.Seconds())
+			m.opts.Meter.Histogram(semconv.SubscribeMessageDurationSeconds, "endpoint", t).Update(te.Seconds())
 		}
 
 	}
@@ -179,7 +217,7 @@ func (m *memoryBroker) Publish(ctx context.Context, message interface{}, opts ..
 	return nil
 }
 
-func (m *memoryBroker) Subscribe(ctx context.Context, topic string, handler interface{}, opts ...SubscribeOption) (Subscriber, error) {
+func (m *MemoryBroker) Subscribe(ctx context.Context, topic string, handler interface{}, opts ...options.Option) (Subscriber, error) {
 	m.RLock()
 	if !m.connected {
 		m.RUnlock()
@@ -224,26 +262,31 @@ func (m *memoryBroker) Subscribe(ctx context.Context, topic string, handler inte
 	return sub, nil
 }
 
-func (m *memoryBroker) String() string {
+func (m *MemoryBroker) String() string {
 	return "memory"
 }
 
-func (m *memoryBroker) Name() string {
+func (m *MemoryBroker) Name() string {
 	return m.opts.Name
 }
 
 type memoryMessage struct {
-	err   error
-	body  interface{}
-	topic string
-	opts  PublishOptions
-	ctx   context.Context
+	err    error
+	body   interface{}
+	topic  string
+	header metadata.Metadata
+	opts   PublishOptions
+	ctx    context.Context
 }
 
 func (m *memoryMessage) Topic() string {
 	return m.topic
 }
 
+func (m *memoryMessage) Header() metadata.Metadata {
+	return m.header
+}
+
 func (m *memoryMessage) Body() interface{} {
 	return m.body
 }
@@ -283,8 +326,8 @@ func (m *memorySubscriber) Unsubscribe(ctx context.Context) error {
 }
 
 // NewBroker return new memory broker
-func NewBroker(opts ...Option) *memoryBroker {
-	return &memoryBroker{
+func NewBroker(opts ...options.Option) *MemoryBroker {
+	return &MemoryBroker{
 		opts:        NewOptions(opts...),
 		subscribers: make(map[string][]*memorySubscriber),
 	}

broker/memory_test.go

@@ -19,29 +19,35 @@ func TestMemoryBatchBroker(t *testing.T) {
 	topic := "test"
 	count := 10
 
-	fn := func(evts Events) error {
-		return evts.Ack()
+	fn := func(evts []Message) error {
+		var err error
+		for _, evt := range evts {
+			if err = evt.Ack(); err != nil {
+				return err
+			}
+		}
+		return nil
 	}
 
-	sub, err := b.BatchSubscribe(ctx, topic, fn)
+	sub, err := b.Subscribe(ctx, topic, fn)
 	if err != nil {
 		t.Fatalf("Unexpected error subscribing %v", err)
 	}
 
-	msgs := make([]*Message, 0, count)
+	msgs := make([]Message, 0, count)
 	for i := 0; i < count; i++ {
-		message := &Message{
-			Header: map[string]string{
+		message := &memoryMessage{
+			header: map[string]string{
 				metadata.HeaderTopic: topic,
 				"foo":                "bar",
 				"id":                 fmt.Sprintf("%d", i),
 			},
-			Body: []byte(`"hello world"`),
+			body: []byte(`"hello world"`),
 		}
 		msgs = append(msgs, message)
 	}
 
-	if err := b.BatchPublish(ctx, msgs); err != nil {
+	if err := b.Publish(ctx, msgs); err != nil {
 		t.Fatalf("Unexpected error publishing %v", err)
 	}
 
@@ -65,8 +71,8 @@ func TestMemoryBroker(t *testing.T) {
 	topic := "test"
 	count := 10
 
-	fn := func(p Event) error {
-		return nil
+	fn := func(p Message) error {
+		return p.Ack()
 	}
 
 	sub, err := b.Subscribe(ctx, topic, fn)
@@ -74,24 +80,20 @@ func TestMemoryBroker(t *testing.T) {
 		t.Fatalf("Unexpected error subscribing %v", err)
 	}
 
-	msgs := make([]*Message, 0, count)
+	msgs := make([]Message, 0, count)
 	for i := 0; i < count; i++ {
-		message := &Message{
-			Header: map[string]string{
+		message := &memoryMessage{
+			header: map[string]string{
 				metadata.HeaderTopic: topic,
 				"foo":                "bar",
 				"id":                 fmt.Sprintf("%d", i),
 			},
-			Body: []byte(`"hello world"`),
+			body: []byte(`"hello world"`),
 		}
 		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 {
+	if err := b.Publish(ctx, msgs); err != nil {
 		t.Fatalf("Unexpected error publishing %v", err)
 	}
 

broker/options.go

@@ -9,29 +9,11 @@ import (
 	"go.unistack.org/micro/v4/logger"
 	"go.unistack.org/micro/v4/metadata"
 	"go.unistack.org/micro/v4/meter"
+	"go.unistack.org/micro/v4/options"
 	"go.unistack.org/micro/v4/register"
 	"go.unistack.org/micro/v4/tracer"
 )
 
-var (
-	// PublishMessageDurationSeconds specifies meter metric name
-	PublishMessageDurationSeconds = "publish_message_duration_seconds"
-	// PublishMessageLatencyMicroseconds specifies meter metric name
-	PublishMessageLatencyMicroseconds = "publish_message_latency_microseconds"
-	// PublishMessageTotal specifies meter metric name
-	PublishMessageTotal = "publish_message_total"
-	// PublishMessageInflight specifies meter metric name
-	PublishMessageInflight = "publish_message_inflight"
-	// SubscribeMessageDurationSeconds specifies meter metric name
-	SubscribeMessageDurationSeconds = "subscribe_message_duration_seconds"
-	// SubscribeMessageLatencyMicroseconds specifies meter metric name
-	SubscribeMessageLatencyMicroseconds = "subscribe_message_latency_microseconds"
-	// SubscribeMessageTotal specifies meter metric name
-	SubscribeMessageTotal = "subscribe_message_total"
-	// SubscribeMessageInflight specifies meter metric name
-	SubscribeMessageInflight = "subscribe_message_inflight"
-)
-
 // Options struct
 type Options struct {
 	// Tracer used for tracing
@@ -48,19 +30,16 @@ type Options struct {
 	Context context.Context
 	// TLSConfig holds tls.TLSConfig options
 	TLSConfig *tls.Config
-	// ErrorHandler used when broker can't unmarshal incoming message
-	ErrorHandler func(Message)
+	// ErrorHandler used when broker have error while processing message
+	ErrorHandler interface{}
 	// Name holds the broker name
 	Name string
-	// Addrs holds the broker address
-	Addrs []string
+	// Address holds the broker address
+	Address []string
 }
 
-// Option func
-type Option func(*Options)
-
 // NewOptions create new Options
-func NewOptions(opts ...Option) Options {
+func NewOptions(opts ...options.Option) Options {
 	options := Options{
 		Register: register.DefaultRegister,
 		Logger:   logger.DefaultLogger,
@@ -75,49 +54,22 @@ func NewOptions(opts ...Option) Options {
 	return options
 }
 
-// Context sets the context option
-func Context(ctx context.Context) Option {
-	return func(o *Options) {
-		o.Context = ctx
-	}
-}
-
-// MessageOption func
-type MessageOption func(*MessageOptions)
-
-// MessageOptions struct
-type MessageOptions struct {
-	Metadata    metadata.Metadata
-	ContentType string
-}
-
-// MessageMetadata pass additional message metadata
-func MessageMetadata(md metadata.Metadata) MessageOption {
-	return func(o *MessageOptions) {
-		o.Metadata = md
-	}
-}
-
-// MessageContentType pass ContentType for message data
-func MessageContentType(ct string) MessageOption {
-	return func(o *MessageOptions) {
-		o.ContentType = ct
-	}
-}
-
-// PublishOption func
-type PublishOption func(*PublishOptions)
-
 // PublishOptions struct
 type PublishOptions struct {
 	// Context holds external options
 	Context context.Context
 	// BodyOnly flag says the message contains raw body bytes
 	BodyOnly bool
+	// Message metadata usually passed as message headers
+	Metadata metadata.Metadata
+	// Content-Type of message for marshal
+	ContentType string
+	// Topic destination
+	Topic string
 }
 
 // NewPublishOptions creates PublishOptions struct
-func NewPublishOptions(opts ...PublishOption) PublishOptions {
+func NewPublishOptions(opts ...options.Option) PublishOptions {
 	options := PublishOptions{
 		Context: context.Background(),
 	}
@@ -127,12 +79,19 @@ func NewPublishOptions(opts ...PublishOption) PublishOptions {
 	return options
 }
 
+// PublishTopic pass topic for messages
+func PublishTopic(t string) options.Option {
+	return func(src interface{}) error {
+		return options.Set(src, t, ".Topic")
+	}
+}
+
 // SubscribeOptions struct
 type SubscribeOptions struct {
 	// Context holds external options
 	Context context.Context
-	// ErrorHandler used when broker can't unmarshal incoming message
-	ErrorHandler func(Message)
+	// ErrorHandler used when broker have error while processing message
+	ErrorHandler interface{}
 	// QueueGroup holds consumer group
 	QueueGroup string
 	// AutoAck flag specifies auto ack of incoming message when no error happens
@@ -145,99 +104,16 @@ type SubscribeOptions struct {
 	BatchWait time.Duration
 }
 
-// PublishBodyOnly publish only body of the message
-func PublishBodyOnly(b bool) PublishOption {
-	return func(o *PublishOptions) {
-		o.BodyOnly = b
-	}
-}
-
-// 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) {
-		o.Addrs = addrs
-	}
-}
-
-// Codec sets the codec used for encoding/decoding used where
-// a broker does not support headers
-// Codec to be used to encode/decode requests for a given content type
-func Codec(contentType string, c codec.Codec) Option {
-	return func(o *Options) {
-		o.Codecs[contentType] = c
-	}
-}
-
 // ErrorHandler will catch all broker errors that cant be handled
 // in normal way, for example Codec errors
-func ErrorHandler(h func(Message)) Option {
-	return func(o *Options) {
-		o.ErrorHandler = h
+func ErrorHandler(h interface{}) options.Option {
+	return func(src interface{}) error {
+		return options.Set(src, h, ".ErrorHandler")
 	}
 }
 
-// SubscribeErrorHandler will catch all broker errors that cant be handled
-// in normal way, for example Codec errors
-func SubscribeErrorHandler(h func(Message)) SubscribeOption {
-	return func(o *SubscribeOptions) {
-		o.ErrorHandler = h
-	}
-}
-
-// Register sets register option
-func Register(r register.Register) Option {
-	return func(o *Options) {
-		o.Register = r
-	}
-}
-
-// TLSConfig sets the TLS Config
-func TLSConfig(t *tls.Config) Option {
-	return func(o *Options) {
-		o.TLSConfig = t
-	}
-}
-
-// Logger sets the logger
-func Logger(l logger.Logger) Option {
-	return func(o *Options) {
-		o.Logger = l
-	}
-}
-
-// Tracer to be used for tracing
-func Tracer(t tracer.Tracer) Option {
-	return func(o *Options) {
-		o.Tracer = t
-	}
-}
-
-// Meter sets the meter
-func Meter(m meter.Meter) Option {
-	return func(o *Options) {
-		o.Meter = m
-	}
-}
-
-// Name sets the name
-func Name(n string) Option {
-	return func(o *Options) {
-		o.Name = n
-	}
-}
-
-// SubscribeOption func signature
-type SubscribeOption func(*SubscribeOptions)
-
 // NewSubscribeOptions creates new SubscribeOptions
-func NewSubscribeOptions(opts ...SubscribeOption) SubscribeOptions {
+func NewSubscribeOptions(opts ...options.Option) SubscribeOptions {
 	options := SubscribeOptions{
 		AutoAck: true,
 		Context: context.Background(),
@@ -248,52 +124,38 @@ func NewSubscribeOptions(opts ...SubscribeOption) SubscribeOptions {
 	return options
 }
 
-// SubscribeContext set context
-func SubscribeContext(ctx context.Context) SubscribeOption {
-	return func(o *SubscribeOptions) {
-		o.Context = ctx
-	}
-}
-
 // SubscribeAutoAck contol auto acking of messages
 // after they have been handled.
-func SubscribeAutoAck(b bool) SubscribeOption {
-	return func(o *SubscribeOptions) {
-		o.AutoAck = b
+func SubscribeAutoAck(b bool) options.Option {
+	return func(src interface{}) error {
+		return options.Set(src, b, ".AutoAck")
 	}
 }
 
-// SubscribeBodyOnly consumes only body of the message
-func SubscribeBodyOnly(b bool) SubscribeOption {
-	return func(o *SubscribeOptions) {
-		o.BodyOnly = b
+// BodyOnly transfer only body without
+func BodyOnly(b bool) options.Option {
+	return func(src interface{}) error {
+		return options.Set(src, b, ".BodyOnly")
 	}
 }
 
 // SubscribeBatchSize specifies max batch size
-func SubscribeBatchSize(n int) SubscribeOption {
-	return func(o *SubscribeOptions) {
-		o.BatchSize = n
+func SubscribeBatchSize(n int) options.Option {
+	return func(src interface{}) error {
+		return options.Set(src, n, ".BatchSize")
 	}
 }
 
 // SubscribeBatchWait specifies max batch wait time
-func SubscribeBatchWait(td time.Duration) SubscribeOption {
-	return func(o *SubscribeOptions) {
-		o.BatchWait = td
+func SubscribeBatchWait(td time.Duration) options.Option {
+	return func(src interface{}) error {
+		return options.Set(src, td, ".BatchWait")
 	}
 }
 
 // SubscribeQueueGroup sets the shared queue name distributed messages across subscribers
-func SubscribeQueueGroup(n string) SubscribeOption {
-	return func(o *SubscribeOptions) {
-		o.QueueGroup = n
-	}
-}
-
-// SubscribeAutoAck control auto ack processing for handler
-func SubscribeAuthAck(b bool) SubscribeOption {
-	return func(o *SubscribeOptions) {
-		o.AutoAck = b
+func SubscribeQueueGroup(n string) options.Option {
+	return func(src interface{}) error {
+		return options.Set(src, n, ".QueueGroup")
 	}
 }