broker: add BatchBroker interface to avoid breaking older brokers

Signed-off-by: Vasiliy Tolstov <v.tolstov@unistack.org>
metadata: add HeaderAuthorization
2021-07-23 12:55:36 +03:00 · 2021-07-23 12:17:00 +03:00 · 2021-07-23 12:03:18 +03:00 · 2021-07-22 22:53:44 +03:00 · 2021-07-22 15:45:44 +03:00 · 2021-07-22 15:45:31 +03:00
30 changed files with 2238 additions and 263 deletions
--- a/broker/broker.go
+++ b/broker/broker.go
@@ -3,46 +3,132 @@ package broker

 import (
 	"context"
+	"errors"

 	"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")
+)
+
+type BatchBroker interface {
+	Broker
+	// BatchPublish messages to broker with multiple topics
+	BatchPublish(ctx context.Context, msgs []*Message, opts ...PublishOption) error
+	// BatchSubscribe subscribes to topic messages via handler
+	BatchSubscribe(ctx context.Context, topic string, h BatchHandler, opts ...SubscribeOption) (Subscriber, error)
+}
+
 // Broker is an interface used for asynchronous messaging.
 type Broker interface {
+	// Name returns broker instance name
 	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)
+	// 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.
+// It implements Marshaler and Unmarshaler and can be used to delay decoding or precompute a encoding.
+type RawMessage []byte
+
+// MarshalJSON returns m as the JSON encoding of m.
+func (m *RawMessage) MarshalJSON() ([]byte, error) {
+	if m == nil {
+		return []byte("null"), nil
+	}
+	return *m, nil
+}
+
+// UnmarshalJSON sets *m to a copy of data.
+func (m *RawMessage) UnmarshalJSON(data []byte) error {
+	if m == nil {
+		return errors.New("RawMessage UnmarshalJSON on nil pointer")
+	}
+	*m = append((*m)[0:0], data...)
+	return nil
 }

 // Message is used to transfer data
 type Message struct {
-	Header metadata.Metadata // contains message metadata
-	Body   []byte            // 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
 }
--- a/broker/memory.go
+++ b/broker/memory.go
@@ -2,20 +2,21 @@ 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
-	addr        string
-	opts        Options
+	subscribers      map[string][]*memorySubscriber
+	batchsubscribers map[string][]*memoryBatchSubscriber
+	addr             string
+	opts             Options
 	sync.RWMutex
 	connected bool
 }
@@ -36,6 +37,15 @@ type memorySubscriber struct {
 	opts    SubscribeOptions
 }

+type memoryBatchSubscriber struct {
+	ctx     context.Context
+	exit    chan bool
+	handler BatchHandler
+	id      string
+	topic   string
+	opts    SubscribeOptions
+}
+
 func (m *memoryBroker) Options() Options {
 	return m.opts
 }
@@ -77,7 +87,6 @@ func (m *memoryBroker) Disconnect(ctx context.Context) error {
 	}

 	m.connected = false
-
 	return nil
 }

@@ -88,14 +97,127 @@ func (m *memoryBroker) Init(opts ...Option) error {
 	return nil
 }

+func (m *memoryBroker) BatchPublish(ctx context.Context, msgs []*Message, opts ...PublishOption) error {
+	m.RLock()
+	if !m.connected {
+		m.RUnlock()
+		return ErrNotConnected
+	}
+	m.RUnlock()
+
+	type msgWrapper struct {
+		topic string
+		body  interface{}
+	}
+
+	vs := make([]msgWrapper, 0, len(msgs))
+	if m.opts.Codec == nil {
+		m.RLock()
+		for _, msg := range msgs {
+			topic, _ := msg.Header.Get(metadata.HeaderTopic)
+			vs = append(vs, msgWrapper{topic: topic, body: m})
+		}
+		m.RUnlock()
+	} else {
+		m.RLock()
+		for _, msg := range msgs {
+			topic, _ := msg.Header.Get(metadata.HeaderTopic)
+			buf, err := m.opts.Codec.Marshal(msg)
+			if err != nil {
+				m.RUnlock()
+				return err
+			}
+			vs = append(vs, msgWrapper{topic: topic, body: buf})
+		}
+		m.RUnlock()
+	}
+
+	if len(m.batchsubscribers) > 0 {
+		eh := m.opts.BatchErrorHandler
+
+		msgTopicMap := make(map[string]Events)
+		for _, v := range vs {
+			p := &memoryEvent{
+				topic:   v.topic,
+				message: v.body,
+				opts:    m.opts,
+			}
+			msgTopicMap[p.topic] = append(msgTopicMap[p.topic], p)
+		}
+
+		for t, ms := range msgTopicMap {
+			m.RLock()
+			subs, ok := m.batchsubscribers[t]
+			m.RUnlock()
+			if !ok {
+				continue
+			}
+			for _, sub := range subs {
+				if err := sub.handler(ms); err != nil {
+					ms.SetError(err)
+					if sub.opts.BatchErrorHandler != nil {
+						eh = sub.opts.BatchErrorHandler
+					}
+					if eh != nil {
+						eh(ms)
+					} else if m.opts.Logger.V(logger.ErrorLevel) {
+						m.opts.Logger.Error(m.opts.Context, err.Error())
+					}
+				} else if sub.opts.AutoAck {
+					if err := ms.Ack(); err != nil {
+						m.opts.Logger.Errorf(m.opts.Context, "ack failed: %v", err)
+					}
+				}
+			}
+		}
+
+	}
+
+	eh := m.opts.ErrorHandler
+
+	for _, v := range vs {
+		p := &memoryEvent{
+			topic:   v.topic,
+			message: v.body,
+			opts:    m.opts,
+		}
+
+		m.RLock()
+		subs, ok := m.subscribers[p.topic]
+		m.RUnlock()
+		if !ok {
+			continue
+		}
+		for _, sub := range subs {
+			if err := sub.handler(p); err != nil {
+				p.SetError(err)
+				if sub.opts.ErrorHandler != nil {
+					eh = sub.opts.ErrorHandler
+				}
+				if eh != nil {
+					eh(p)
+				} else if m.opts.Logger.V(logger.ErrorLevel) {
+					m.opts.Logger.Error(m.opts.Context, err.Error())
+				}
+			} else if sub.opts.AutoAck {
+				if err := p.Ack(); err != nil {
+					m.opts.Logger.Errorf(m.opts.Context, "ack failed: %v", err)
+				}
+			}
+		}
+	}
+
+	return nil
+}
+
 func (m *memoryBroker) Publish(ctx context.Context, topic string, msg *Message, opts ...PublishOption) error {
 	m.RLock()
 	if !m.connected {
 		m.RUnlock()
-		return errors.New("not connected")
+		return ErrNotConnected
 	}

-	subs, ok := m.Subscribers[topic]
+	subs, ok := m.subscribers[topic]
 	m.RUnlock()
 	if !ok {
 		return nil
@@ -138,11 +260,56 @@ func (m *memoryBroker) Publish(ctx context.Context, topic string, msg *Message,
 	return nil
 }

+func (m *memoryBroker) BatchSubscribe(ctx context.Context, topic string, handler BatchHandler, opts ...SubscribeOption) (Subscriber, error) {
+	m.RLock()
+	if !m.connected {
+		m.RUnlock()
+		return nil, ErrNotConnected
+	}
+	m.RUnlock()
+
+	options := NewSubscribeOptions(opts...)
+
+	id, err := uuid.NewRandom()
+	if err != nil {
+		return nil, err
+	}
+
+	sub := &memoryBatchSubscriber{
+		exit:    make(chan bool, 1),
+		id:      id.String(),
+		topic:   topic,
+		handler: handler,
+		opts:    options,
+		ctx:     ctx,
+	}
+
+	m.Lock()
+	m.batchsubscribers[topic] = append(m.batchsubscribers[topic], sub)
+	m.Unlock()
+
+	go func() {
+		<-sub.exit
+		m.Lock()
+		var newSubscribers []*memoryBatchSubscriber
+		for _, sb := range m.batchsubscribers[topic] {
+			if sb.id == sub.id {
+				continue
+			}
+			newSubscribers = append(newSubscribers, sb)
+		}
+		m.batchsubscribers[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, errors.New("not connected")
+		return nil, ErrNotConnected
 	}
 	m.RUnlock()

@@ -163,20 +330,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] {
+		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 +388,23 @@ func (m *memoryEvent) Error() error {
 	return m.err
 }

+func (m *memoryEvent) SetError(err error) {
+	m.err = err
+}
+
+func (m *memoryBatchSubscriber) Options() SubscribeOptions {
+	return m.opts
+}
+
+func (m *memoryBatchSubscriber) Topic() string {
+	return m.topic
+}
+
+func (m *memoryBatchSubscriber) Unsubscribe(ctx context.Context) error {
+	m.exit <- true
+	return nil
+}
+
 func (m *memorySubscriber) Options() SubscribeOptions {
 	return m.opts
 }
@@ -235,9 +419,10 @@ func (m *memorySubscriber) Unsubscribe(ctx context.Context) error {
 }

 // NewBroker return new memory broker
-func NewBroker(opts ...Option) Broker {
+func NewBroker(opts ...Option) BatchBroker {
 	return &memoryBroker{
-		opts:        NewOptions(opts...),
-		Subscribers: make(map[string][]*memorySubscriber),
+		opts:             NewOptions(opts...),
+		subscribers:      make(map[string][]*memorySubscriber),
+		batchsubscribers: make(map[string][]*memoryBatchSubscriber),
 	}
 }
--- a/broker/memory_test.go
+++ b/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`),
+			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)
 	}
--- a/broker/options.go
+++ b/broker/options.go
@@ -29,6 +29,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
@@ -85,6 +87,8 @@ 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
@@ -175,6 +179,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 +195,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 {
--- a/client/noop.go
+++ b/client/noop.go
@@ -190,8 +190,8 @@ func (n *noopClient) Publish(ctx context.Context, p Message, opts ...PublishOpti
 	if !ok {
 		md = metadata.New(0)
 	}
-	md["Content-Type"] = p.ContentType()
-	md["Micro-Topic"] = p.Topic()
+	md[metadata.HeaderContentType] = p.ContentType()
+	md[metadata.HeaderTopic] = p.Topic()

 	// passed in raw data
 	if d, ok := p.Payload().(*codec.Frame); ok {
--- a/client/options.go
+++ b/client/options.go
@@ -96,8 +96,8 @@ type CallOptions struct {
 	RequestTimeout time.Duration
 	// DialTimeout dial timeout
 	DialTimeout time.Duration
-	// AuthToken flag
-	AuthToken bool
+	// AuthToken string
+	AuthToken string
 }

 // Context pass context to client
@@ -463,9 +463,9 @@ func WithDialTimeout(d time.Duration) CallOption {

 // WithAuthToken is a CallOption which overrides the
 // authorization header with the services own auth token
-func WithAuthToken() CallOption {
+func WithAuthToken(t string) CallOption {
 	return func(o *CallOptions) {
-		o.AuthToken = true
+		o.AuthToken = t
 	}
 }

--- a/config/config.go
+++ b/config/config.go
@@ -43,13 +43,13 @@ type Watcher interface {
 }

 // Load loads config from config sources
-func Load(ctx context.Context, cs ...Config) error {
+func Load(ctx context.Context, cs []Config, opts ...LoadOption) error {
 	var err error
 	for _, c := range cs {
 		if err = c.Init(); err != nil {
 			return err
 		}
-		if err = c.Load(ctx); err != nil {
+		if err = c.Load(ctx, opts...); err != nil {
 			return err
 		}
 	}
--- a/flow/context.go
+++ b/flow/context.go
@@ -0,0 +1,34 @@
+package flow
+
+import (
+	"context"
+)
+
+type flowKey struct{}
+
+// FromContext returns Flow from context
+func FromContext(ctx context.Context) (Flow, bool) {
+	if ctx == nil {
+		return nil, false
+	}
+	c, ok := ctx.Value(flowKey{}).(Flow)
+	return c, ok
+}
+
+// NewContext stores Flow to context
+func NewContext(ctx context.Context, f Flow) context.Context {
+	if ctx == nil {
+		ctx = context.Background()
+	}
+	return context.WithValue(ctx, flowKey{}, f)
+}
+
+// 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)
+	}
+}
--- a/flow/default.go
+++ b/flow/default.go
@@ -0,0 +1,588 @@
+package flow
+
+import (
+	"context"
+	"fmt"
+	"path/filepath"
+	"sync"
+
+	"github.com/google/uuid"
+	"github.com/silas/dag"
+	"github.com/unistack-org/micro/v3/client"
+	"github.com/unistack-org/micro/v3/codec"
+	"github.com/unistack-org/micro/v3/logger"
+	"github.com/unistack-org/micro/v3/metadata"
+	"github.com/unistack-org/micro/v3/store"
+)
+
+type microFlow struct {
+	opts Options
+}
+
+type microWorkflow struct {
+	id   string
+	g    *dag.AcyclicGraph
+	init bool
+	sync.RWMutex
+	opts   Options
+	steps  map[string]Step
+	status Status
+}
+
+func (w *microWorkflow) ID() string {
+	return w.id
+}
+
+func (w *microWorkflow) Steps() ([][]Step, error) {
+	return w.getSteps("", false)
+}
+
+func (w *microWorkflow) Status() Status {
+	return w.status
+}
+
+func (w *microWorkflow) AppendSteps(steps ...Step) error {
+	w.Lock()
+
+	for _, s := range steps {
+		w.steps[s.String()] = s
+		w.g.Add(s)
+	}
+
+	for _, dst := range steps {
+		for _, req := range dst.Requires() {
+			src, ok := w.steps[req]
+			if !ok {
+				return ErrStepNotExists
+			}
+			w.g.Connect(dag.BasicEdge(src, dst))
+		}
+	}
+
+	if err := w.g.Validate(); err != nil {
+		w.Unlock()
+		return err
+	}
+
+	w.g.TransitiveReduction()
+
+	w.Unlock()
+
+	return nil
+}
+
+func (w *microWorkflow) RemoveSteps(steps ...Step) error {
+	// TODO: handle case when some step requires or required by removed step
+
+	w.Lock()
+
+	for _, s := range steps {
+		delete(w.steps, s.String())
+		w.g.Remove(s)
+	}
+
+	for _, dst := range steps {
+		for _, req := range dst.Requires() {
+			src, ok := w.steps[req]
+			if !ok {
+				return ErrStepNotExists
+			}
+			w.g.Connect(dag.BasicEdge(src, dst))
+		}
+	}
+
+	if err := w.g.Validate(); err != nil {
+		w.Unlock()
+		return err
+	}
+
+	w.g.TransitiveReduction()
+
+	w.Unlock()
+
+	return nil
+}
+
+func (w *microWorkflow) getSteps(start string, reverse bool) ([][]Step, error) {
+	var steps [][]Step
+	var root dag.Vertex
+	var err error
+
+	fn := func(n dag.Vertex, idx int) error {
+		if idx == 0 {
+			steps = make([][]Step, 1)
+			steps[0] = make([]Step, 0, 1)
+		} else if idx >= len(steps) {
+			tsteps := make([][]Step, idx+1)
+			copy(tsteps, steps)
+			steps = tsteps
+			steps[idx] = make([]Step, 0, 1)
+		}
+		steps[idx] = append(steps[idx], n.(Step))
+		return nil
+	}
+
+	if start != "" {
+		var ok bool
+		w.RLock()
+		root, ok = w.steps[start]
+		w.RUnlock()
+		if !ok {
+			return nil, ErrStepNotExists
+		}
+	} else {
+		root, err = w.g.Root()
+		if err != nil {
+			return nil, err
+		}
+	}
+
+	if reverse {
+		err = w.g.SortedReverseDepthFirstWalk([]dag.Vertex{root}, fn)
+	} else {
+		err = w.g.SortedDepthFirstWalk([]dag.Vertex{root}, fn)
+	}
+	if err != nil {
+		return nil, err
+	}
+
+	return steps, nil
+}
+
+func (w *microWorkflow) Abort(ctx context.Context, eid string) error {
+	workflowStore := store.NewNamespaceStore(w.opts.Store, filepath.Join("workflows", eid))
+	return workflowStore.Write(ctx, "status", &codec.Frame{Data: []byte(StatusAborted.String())})
+}
+
+func (w *microWorkflow) Suspend(ctx context.Context, eid string) error {
+	workflowStore := store.NewNamespaceStore(w.opts.Store, filepath.Join("workflows", eid))
+	return workflowStore.Write(ctx, "status", &codec.Frame{Data: []byte(StatusSuspend.String())})
+}
+
+func (w *microWorkflow) Resume(ctx context.Context, eid string) error {
+	workflowStore := store.NewNamespaceStore(w.opts.Store, filepath.Join("workflows", eid))
+	return workflowStore.Write(ctx, "status", &codec.Frame{Data: []byte(StatusRunning.String())})
+}
+
+func (w *microWorkflow) Execute(ctx context.Context, req *Message, opts ...ExecuteOption) (string, error) {
+	w.Lock()
+	if !w.init {
+		if err := w.g.Validate(); err != nil {
+			w.Unlock()
+			return "", err
+		}
+		w.g.TransitiveReduction()
+		w.init = true
+	}
+	w.Unlock()
+
+	uid, err := uuid.NewRandom()
+	if err != nil {
+		return "", err
+	}
+	eid := uid.String()
+
+	stepStore := store.NewNamespaceStore(w.opts.Store, filepath.Join("steps", eid))
+	workflowStore := store.NewNamespaceStore(w.opts.Store, filepath.Join("workflows", eid))
+
+	options := NewExecuteOptions(opts...)
+
+	steps, err := w.getSteps(options.Start, options.Reverse)
+	if err != nil {
+		if werr := workflowStore.Write(w.opts.Context, "status", &codec.Frame{Data: []byte(StatusPending.String())}); werr != nil {
+			w.opts.Logger.Errorf(w.opts.Context, "store error: %v", werr)
+		}
+		return "", err
+	}
+
+	var wg sync.WaitGroup
+	cherr := make(chan error, 1)
+	chstatus := make(chan Status, 1)
+
+	nctx, cancel := context.WithCancel(ctx)
+	defer cancel()
+
+	nopts := make([]ExecuteOption, 0, len(opts)+5)
+
+	nopts = append(nopts,
+		ExecuteClient(w.opts.Client),
+		ExecuteTracer(w.opts.Tracer),
+		ExecuteLogger(w.opts.Logger),
+		ExecuteMeter(w.opts.Meter),
+	)
+	nopts = append(nopts, opts...)
+	done := make(chan struct{})
+
+	if werr := workflowStore.Write(w.opts.Context, "status", &codec.Frame{Data: []byte(StatusRunning.String())}); werr != nil {
+		w.opts.Logger.Errorf(w.opts.Context, "store error: %v", werr)
+		return eid, werr
+	}
+	for idx := range steps {
+		for nidx := range steps[idx] {
+			cstep := steps[idx][nidx]
+			if werr := stepStore.Write(ctx, filepath.Join(cstep.ID(), "status"), &codec.Frame{Data: []byte(StatusPending.String())}); werr != nil {
+				return eid, werr
+			}
+		}
+	}
+
+	go func() {
+		for idx := range steps {
+			for nidx := range steps[idx] {
+				wStatus := &codec.Frame{}
+				if werr := workflowStore.Read(w.opts.Context, "status", wStatus); werr != nil {
+					cherr <- werr
+					return
+				}
+				if status := StringStatus[string(wStatus.Data)]; status != StatusRunning {
+					chstatus <- status
+					return
+				}
+				if w.opts.Logger.V(logger.TraceLevel) {
+					w.opts.Logger.Tracef(nctx, "will be executed %v", steps[idx][nidx])
+				}
+				cstep := steps[idx][nidx]
+				if len(cstep.Requires()) == 0 {
+					wg.Add(1)
+					go func(step Step) {
+						defer wg.Done()
+						if werr := stepStore.Write(ctx, filepath.Join(step.ID(), "req"), req); werr != nil {
+							cherr <- werr
+							return
+						}
+						if werr := stepStore.Write(ctx, filepath.Join(step.ID(), "status"), &codec.Frame{Data: []byte(StatusRunning.String())}); werr != nil {
+							cherr <- werr
+							return
+						}
+						rsp, serr := step.Execute(nctx, req, nopts...)
+						if serr != nil {
+							step.SetStatus(StatusFailure)
+							if werr := stepStore.Write(ctx, filepath.Join(step.ID(), "rsp"), serr); werr != nil && w.opts.Logger.V(logger.ErrorLevel) {
+								w.opts.Logger.Errorf(ctx, "store write error: %v", werr)
+							}
+							if werr := stepStore.Write(ctx, filepath.Join(step.ID(), "status"), &codec.Frame{Data: []byte(StatusFailure.String())}); werr != nil && w.opts.Logger.V(logger.ErrorLevel) {
+								w.opts.Logger.Errorf(ctx, "store write error: %v", werr)
+							}
+							cherr <- serr
+							return
+						} else {
+							if werr := stepStore.Write(ctx, filepath.Join(step.ID(), "rsp"), rsp); werr != nil {
+								w.opts.Logger.Errorf(ctx, "store write error: %v", werr)
+								cherr <- werr
+								return
+							}
+							if werr := stepStore.Write(ctx, filepath.Join(step.ID(), "status"), &codec.Frame{Data: []byte(StatusSuccess.String())}); werr != nil {
+								w.opts.Logger.Errorf(ctx, "store write error: %v", werr)
+								cherr <- werr
+								return
+							}
+						}
+					}(cstep)
+					wg.Wait()
+				} else {
+					if werr := stepStore.Write(ctx, filepath.Join(cstep.ID(), "req"), req); werr != nil {
+						cherr <- werr
+						return
+					}
+					if werr := stepStore.Write(ctx, filepath.Join(cstep.ID(), "status"), &codec.Frame{Data: []byte(StatusRunning.String())}); werr != nil {
+						cherr <- werr
+						return
+					}
+					rsp, serr := cstep.Execute(nctx, req, nopts...)
+					if serr != nil {
+						cstep.SetStatus(StatusFailure)
+						if werr := stepStore.Write(ctx, filepath.Join(cstep.ID(), "rsp"), serr); werr != nil && w.opts.Logger.V(logger.ErrorLevel) {
+							w.opts.Logger.Errorf(ctx, "store write error: %v", werr)
+						}
+						if werr := stepStore.Write(ctx, filepath.Join(cstep.ID(), "status"), &codec.Frame{Data: []byte(StatusFailure.String())}); werr != nil && w.opts.Logger.V(logger.ErrorLevel) {
+							w.opts.Logger.Errorf(ctx, "store write error: %v", werr)
+						}
+						cherr <- serr
+						return
+					} else {
+						if werr := stepStore.Write(ctx, filepath.Join(cstep.ID(), "rsp"), rsp); werr != nil {
+							w.opts.Logger.Errorf(ctx, "store write error: %v", werr)
+							cherr <- werr
+							return
+						}
+						if werr := stepStore.Write(ctx, filepath.Join(cstep.ID(), "status"), &codec.Frame{Data: []byte(StatusSuccess.String())}); werr != nil {
+							cherr <- werr
+							return
+						}
+					}
+				}
+			}
+		}
+		close(done)
+	}()
+
+	if options.Async {
+		return eid, nil
+	}
+
+	logger.Tracef(ctx, "wait for finish or error")
+	select {
+	case <-nctx.Done():
+		err = nctx.Err()
+	case cerr := <-cherr:
+		err = cerr
+	case <-done:
+		close(cherr)
+	case <-chstatus:
+		close(chstatus)
+		return uid.String(), nil
+	}
+
+	switch {
+	case nctx.Err() != nil:
+		if werr := workflowStore.Write(w.opts.Context, "status", &codec.Frame{Data: []byte(StatusAborted.String())}); werr != nil {
+			w.opts.Logger.Errorf(w.opts.Context, "store error: %v", werr)
+		}
+		break
+	case err == nil:
+		if werr := workflowStore.Write(w.opts.Context, "status", &codec.Frame{Data: []byte(StatusSuccess.String())}); werr != nil {
+			w.opts.Logger.Errorf(w.opts.Context, "store error: %v", werr)
+		}
+		break
+	case err != nil:
+		if werr := workflowStore.Write(w.opts.Context, "status", &codec.Frame{Data: []byte(StatusFailure.String())}); werr != nil {
+			w.opts.Logger.Errorf(w.opts.Context, "store error: %v", werr)
+		}
+		break
+	}
+
+	return uid.String(), err
+}
+
+func NewFlow(opts ...Option) Flow {
+	options := NewOptions(opts...)
+	return &microFlow{opts: options}
+}
+
+func (f *microFlow) Options() Options {
+	return f.opts
+}
+
+func (f *microFlow) Init(opts ...Option) error {
+	for _, o := range opts {
+		o(&f.opts)
+	}
+	if err := f.opts.Client.Init(); err != nil {
+		return err
+	}
+	if err := f.opts.Tracer.Init(); err != nil {
+		return err
+	}
+	if err := f.opts.Logger.Init(); err != nil {
+		return err
+	}
+	if err := f.opts.Meter.Init(); err != nil {
+		return err
+	}
+	if err := f.opts.Store.Init(); err != nil {
+		return err
+	}
+	return nil
+}
+
+func (f *microFlow) WorkflowList(ctx context.Context) ([]Workflow, error) {
+	return nil, nil
+}
+
+func (f *microFlow) WorkflowCreate(ctx context.Context, id string, steps ...Step) (Workflow, error) {
+	w := &microWorkflow{opts: f.opts, id: id, g: &dag.AcyclicGraph{}, steps: make(map[string]Step, len(steps))}
+
+	for _, s := range steps {
+		w.steps[s.String()] = s
+		w.g.Add(s)
+	}
+
+	for _, dst := range steps {
+		for _, req := range dst.Requires() {
+			src, ok := w.steps[req]
+			if !ok {
+				return nil, ErrStepNotExists
+			}
+			w.g.Connect(dag.BasicEdge(src, dst))
+		}
+	}
+
+	if err := w.g.Validate(); err != nil {
+		return nil, err
+	}
+	w.g.TransitiveReduction()
+
+	w.init = true
+
+	return w, nil
+}
+
+func (f *microFlow) WorkflowRemove(ctx context.Context, id string) error {
+	return nil
+}
+
+func (f *microFlow) WorkflowSave(ctx context.Context, w Workflow) error {
+	return nil
+}
+
+func (f *microFlow) WorkflowLoad(ctx context.Context, id string) (Workflow, error) {
+	return nil, nil
+}
+
+type microCallStep struct {
+	opts    StepOptions
+	service string
+	method  string
+	rsp     *Message
+	req     *Message
+	status  Status
+}
+
+func (s *microCallStep) Request() *Message {
+	return s.req
+}
+
+func (s *microCallStep) Response() *Message {
+	return s.rsp
+}
+
+func (s *microCallStep) ID() string {
+	return s.String()
+}
+
+func (s *microCallStep) Options() StepOptions {
+	return s.opts
+}
+
+func (s *microCallStep) Endpoint() string {
+	return s.method
+}
+
+func (s *microCallStep) Requires() []string {
+	return s.opts.Requires
+}
+
+func (s *microCallStep) Require(steps ...Step) error {
+	for _, step := range steps {
+		s.opts.Requires = append(s.opts.Requires, step.String())
+	}
+	return nil
+}
+
+func (s *microCallStep) String() string {
+	if s.opts.ID != "" {
+		return s.opts.ID
+	}
+	return fmt.Sprintf("%s.%s", s.service, s.method)
+}
+
+func (s *microCallStep) Name() string {
+	return s.String()
+}
+
+func (s *microCallStep) Hashcode() interface{} {
+	return s.String()
+}
+
+func (s *microCallStep) GetStatus() Status {
+	return s.status
+}
+
+func (s *microCallStep) SetStatus(status Status) {
+	s.status = status
+}
+
+func (s *microCallStep) Execute(ctx context.Context, req *Message, opts ...ExecuteOption) (*Message, error) {
+	options := NewExecuteOptions(opts...)
+	if options.Client == nil {
+		return nil, ErrMissingClient
+	}
+	rsp := &codec.Frame{}
+	copts := []client.CallOption{client.WithRetries(0)}
+	if options.Timeout > 0 {
+		copts = append(copts, client.WithRequestTimeout(options.Timeout), client.WithDialTimeout(options.Timeout))
+	}
+	nctx := metadata.NewOutgoingContext(ctx, req.Header)
+	err := options.Client.Call(nctx, options.Client.NewRequest(s.service, s.method, &codec.Frame{Data: req.Body}), rsp)
+	if err != nil {
+		return nil, err
+	}
+	md, _ := metadata.FromOutgoingContext(nctx)
+	return &Message{Header: md, Body: rsp.Data}, err
+}
+
+type microPublishStep struct {
+	opts   StepOptions
+	topic  string
+	req    *Message
+	rsp    *Message
+	status Status
+}
+
+func (s *microPublishStep) Request() *Message {
+	return s.req
+}
+
+func (s *microPublishStep) Response() *Message {
+	return s.rsp
+}
+
+func (s *microPublishStep) ID() string {
+	return s.String()
+}
+
+func (s *microPublishStep) Options() StepOptions {
+	return s.opts
+}
+
+func (s *microPublishStep) Endpoint() string {
+	return s.topic
+}
+
+func (s *microPublishStep) Requires() []string {
+	return s.opts.Requires
+}
+
+func (s *microPublishStep) Require(steps ...Step) error {
+	for _, step := range steps {
+		s.opts.Requires = append(s.opts.Requires, step.String())
+	}
+	return nil
+}
+
+func (s *microPublishStep) String() string {
+	if s.opts.ID != "" {
+		return s.opts.ID
+	}
+	return fmt.Sprintf("%s", s.topic)
+}
+
+func (s *microPublishStep) Name() string {
+	return s.String()
+}
+
+func (s *microPublishStep) Hashcode() interface{} {
+	return s.String()
+}
+
+func (s *microPublishStep) GetStatus() Status {
+	return s.status
+}
+
+func (s *microPublishStep) SetStatus(status Status) {
+	s.status = status
+}
+
+func (s *microPublishStep) Execute(ctx context.Context, req *Message, opts ...ExecuteOption) (*Message, error) {
+	return nil, nil
+}
+
+func NewCallStep(service string, name string, method string, opts ...StepOption) Step {
+	options := NewStepOptions(opts...)
+	return &microCallStep{service: service, method: name + "." + method, opts: options}
+}
+
+func NewPublishStep(topic string, opts ...StepOption) Step {
+	options := NewStepOptions(opts...)
+	return &microPublishStep{topic: topic, opts: options}
+}
--- a/flow/flow.go
+++ b/flow/flow.go
@@ -1,17 +1,152 @@
 // Package flow is an interface used for saga pattern microservice workflow
 package flow

+import (
+	"context"
+	"errors"
+	"sync"
+	"sync/atomic"
+
+	"github.com/unistack-org/micro/v3/metadata"
+)
+
+var (
+	ErrStepNotExists = errors.New("step not exists")
+	ErrMissingClient = errors.New("client not set")
+)
+
+// RawMessage is a raw encoded JSON value.
+// It implements Marshaler and Unmarshaler and can be used to delay decoding or precompute a encoding.
+type RawMessage []byte
+
+// MarshalJSON returns m as the JSON encoding of m.
+func (m *RawMessage) MarshalJSON() ([]byte, error) {
+	if m == nil {
+		return []byte("null"), nil
+	}
+	return *m, nil
+}
+
+// UnmarshalJSON sets *m to a copy of data.
+func (m *RawMessage) UnmarshalJSON(data []byte) error {
+	if m == nil {
+		return errors.New("RawMessage UnmarshalJSON on nil pointer")
+	}
+	*m = append((*m)[0:0], data...)
+	return nil
+}
+
+type Message struct {
+	Header metadata.Metadata
+	Body   RawMessage
+}
+
+// Step represents dedicated workflow step
 type Step interface {
+	// ID returns step id
+	ID() string
 	// Endpoint returns rpc endpoint service_name.service_method or broker topic
 	Endpoint() string
+	// Execute step run
+	Execute(ctx context.Context, req *Message, opts ...ExecuteOption) (*Message, error)
+	// Requires returns dependent steps
+	Requires() []string
+	// Options returns step options
+	Options() StepOptions
+	// Require add required steps
+	Require(steps ...Step) error
+	// String
+	String() string
+	// GetStatus returns step status
+	GetStatus() Status
+	// SetStatus sets the step status
+	SetStatus(Status)
+	// Request returns step request message
+	Request() *Message
+	// Response returns step response message
+	Response() *Message
 }

+type Status int
+
+func (status Status) String() string {
+	return StatusString[status]
+}
+
+const (
+	StatusPending Status = iota
+	StatusRunning
+	StatusFailure
+	StatusSuccess
+	StatusAborted
+	StatusSuspend
+)
+
+var (
+	StatusString = map[Status]string{
+		StatusPending: "StatusPending",
+		StatusRunning: "StatusRunning",
+		StatusFailure: "StatusFailure",
+		StatusSuccess: "StatusSuccess",
+		StatusAborted: "StatusAborted",
+		StatusSuspend: "StatusSuspend",
+	}
+	StringStatus = map[string]Status{
+		"StatusPending": StatusPending,
+		"StatusRunning": StatusRunning,
+		"StatusFailure": StatusFailure,
+		"StatusSuccess": StatusSuccess,
+		"StatusAborted": StatusAborted,
+		"StatusSuspend": StatusSuspend,
+	}
+)
+
+// Workflow contains all steps to execute
 type Workflow interface {
-	Steps() [][]Step
-	Stop() error
+	// ID returns id of the workflow
+	ID() string
+	// Execute workflow with args, return execution id and error
+	Execute(ctx context.Context, req *Message, opts ...ExecuteOption) (string, error)
+	// RemoveSteps remove steps from workflow
+	RemoveSteps(steps ...Step) error
+	// AppendSteps append steps to workflow
+	AppendSteps(steps ...Step) error
+	// Status returns workflow status
+	Status() Status
+	// Steps returns steps slice where parallel steps returned on the same level
+	Steps() ([][]Step, error)
+	// Suspend suspends execution
+	Suspend(ctx context.Context, eid string) error
+	// Resume resumes execution
+	Resume(ctx context.Context, eid string) error
+	// Abort abort execution
+	Abort(ctx context.Context, eid string) error
 }

+// Flow the base interface to interact with workflows
 type Flow interface {
-	Start(Workflow) error
-	Stop(Workflow)
+	// Options returns options
+	Options() Options
+	// Init initialize
+	Init(...Option) error
+	// WorkflowCreate creates new workflow with specific id and steps
+	WorkflowCreate(ctx context.Context, id string, steps ...Step) (Workflow, error)
+	// WorkflowSave saves workflow
+	WorkflowSave(ctx context.Context, w Workflow) error
+	// WorkflowLoad loads workflow with specific id
+	WorkflowLoad(ctx context.Context, id string) (Workflow, error)
+	// WorkflowList lists all workflows
+	WorkflowList(ctx context.Context) ([]Workflow, error)
+}
+
+var (
+	flowMu      sync.Mutex
+	atomicSteps atomic.Value
+)
+
+func RegisterStep(step Step) {
+	flowMu.Lock()
+	steps, _ := atomicSteps.Load().([]Step)
+	atomicSteps.Store(append(steps, step))
+	flowMu.Unlock()
 }
--- a/flow/options.go
+++ b/flow/options.go
@@ -0,0 +1,230 @@
+package flow
+
+import (
+	"context"
+	"time"
+
+	"github.com/unistack-org/micro/v3/client"
+	"github.com/unistack-org/micro/v3/logger"
+	"github.com/unistack-org/micro/v3/meter"
+	"github.com/unistack-org/micro/v3/store"
+	"github.com/unistack-org/micro/v3/tracer"
+)
+
+// Option func
+type Option func(*Options)
+
+// Options server struct
+type Options struct {
+	// Context holds the external options and can be used for flow shutdown
+	Context context.Context
+	// Client holds the client.Client
+	Client client.Client
+	// Tracer holds the tracer
+	Tracer tracer.Tracer
+	// Logger holds the logger
+	Logger logger.Logger
+	// Meter holds the meter
+	Meter meter.Meter
+	// Store used for intermediate results
+	Store store.Store
+}
+
+// NewOptions returns new options struct with default or passed values
+func NewOptions(opts ...Option) Options {
+	options := Options{
+		Context: context.Background(),
+		Logger:  logger.DefaultLogger,
+		Meter:   meter.DefaultMeter,
+		Tracer:  tracer.DefaultTracer,
+		Client:  client.DefaultClient,
+	}
+
+	for _, o := range opts {
+		o(&options)
+	}
+
+	return options
+}
+
+// Logger sets the logger option
+func Logger(l logger.Logger) Option {
+	return func(o *Options) {
+		o.Logger = l
+	}
+}
+
+// Meter sets the meter option
+func Meter(m meter.Meter) Option {
+	return func(o *Options) {
+		o.Meter = m
+	}
+}
+
+// Client to use for sync/async communication
+func Client(c client.Client) Option {
+	return func(o *Options) {
+		o.Client = c
+	}
+}
+
+// Context specifies a context for the service.
+// Can be used to signal shutdown of the flow
+// Can be used for extra option values.
+func Context(ctx context.Context) Option {
+	return func(o *Options) {
+		o.Context = ctx
+	}
+}
+
+// Tracer mechanism for distributed tracking
+func Tracer(t tracer.Tracer) Option {
+	return func(o *Options) {
+		o.Tracer = t
+	}
+}
+
+// Store used for intermediate results
+func Store(s store.Store) Option {
+	return func(o *Options) {
+		o.Store = s
+	}
+}
+
+// WorflowOption signature
+type WorkflowOption func(*WorkflowOptions)
+
+// WorkflowOptions holds workflow options
+type WorkflowOptions struct {
+	ID      string
+	Context context.Context
+}
+
+// WorkflowID set workflow id
+func WorkflowID(id string) WorkflowOption {
+	return func(o *WorkflowOptions) {
+		o.ID = id
+	}
+}
+
+type ExecuteOptions struct {
+	// Client holds the client.Client
+	Client client.Client
+	// Tracer holds the tracer
+	Tracer tracer.Tracer
+	// Logger holds the logger
+	Logger logger.Logger
+	// Meter holds the meter
+	Meter meter.Meter
+	// Context can be used to abort execution or pass additional opts
+	Context context.Context
+	// Start step
+	Start string
+	// Reverse execution
+	Reverse bool
+	// Timeout for execution
+	Timeout time.Duration
+	// Async enables async execution
+	Async bool
+}
+
+type ExecuteOption func(*ExecuteOptions)
+
+func ExecuteClient(c client.Client) ExecuteOption {
+	return func(o *ExecuteOptions) {
+		o.Client = c
+	}
+}
+
+func ExecuteTracer(t tracer.Tracer) ExecuteOption {
+	return func(o *ExecuteOptions) {
+		o.Tracer = t
+	}
+}
+
+func ExecuteLogger(l logger.Logger) ExecuteOption {
+	return func(o *ExecuteOptions) {
+		o.Logger = l
+	}
+}
+
+func ExecuteMeter(m meter.Meter) ExecuteOption {
+	return func(o *ExecuteOptions) {
+		o.Meter = m
+	}
+}
+
+func ExecuteContext(ctx context.Context) ExecuteOption {
+	return func(o *ExecuteOptions) {
+		o.Context = ctx
+	}
+}
+
+func ExecuteReverse(b bool) ExecuteOption {
+	return func(o *ExecuteOptions) {
+		o.Reverse = b
+	}
+}
+
+func ExecuteTimeout(td time.Duration) ExecuteOption {
+	return func(o *ExecuteOptions) {
+		o.Timeout = td
+	}
+}
+
+func ExecuteAsync(b bool) ExecuteOption {
+	return func(o *ExecuteOptions) {
+		o.Async = b
+	}
+}
+
+func NewExecuteOptions(opts ...ExecuteOption) ExecuteOptions {
+	options := ExecuteOptions{
+		Client:  client.DefaultClient,
+		Logger:  logger.DefaultLogger,
+		Tracer:  tracer.DefaultTracer,
+		Meter:   meter.DefaultMeter,
+		Context: context.Background(),
+	}
+	for _, o := range opts {
+		o(&options)
+	}
+	return options
+}
+
+type StepOptions struct {
+	ID       string
+	Context  context.Context
+	Requires []string
+	Fallback string
+}
+
+type StepOption func(*StepOptions)
+
+func NewStepOptions(opts ...StepOption) StepOptions {
+	options := StepOptions{
+		Context: context.Background(),
+	}
+	for _, o := range opts {
+		o(&options)
+	}
+	return options
+}
+
+func StepID(id string) StepOption {
+	return func(o *StepOptions) {
+		o.ID = id
+	}
+}
+
+func StepRequires(steps ...string) StepOption {
+	return func(o *StepOptions) {
+		o.Requires = steps
+	}
+}
+
+func StepFallback(step string) StepOption {
+	return func(o *StepOptions) {
+		o.Fallback = step
+	}
+}
--- a/go.mod
+++ b/go.mod
@@ -5,7 +5,7 @@ go 1.16
 require (
 	github.com/dgrijalva/jwt-go v3.2.0+incompatible
 	github.com/ef-ds/deque v1.0.4
-	github.com/google/uuid v1.2.0
+	github.com/google/uuid v1.3.0
 	github.com/imdario/mergo v0.3.12
 	github.com/patrickmn/go-cache v2.1.0+incompatible
 	github.com/silas/dag v0.0.0-20210121180416-41cf55125c34
--- a/go.sum
+++ b/go.sum
@@ -2,8 +2,8 @@ github.com/dgrijalva/jwt-go v3.2.0+incompatible h1:7qlOGliEKZXTDg6OTjfoBKDXWrumC
 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/google/uuid v1.2.0 h1:qJYtXnJRWmpe7m/3XlyhrsLrEURqHRM2kxzoxXqyUDs=
-github.com/google/uuid v1.2.0/go.mod h1:TIyPZe4MgqvfeYDBFedMoGGpEw/LqOeaOT+nhxU+yHo=
+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=
 github.com/imdario/mergo v0.3.12/go.mod h1:jmQim1M+e3UYxmgPu/WyfjB3N3VflVyUjjjwH0dnCYA=
 github.com/patrickmn/go-cache v2.1.0+incompatible h1:HRMgzkcYKYpi3C8ajMPV8OFXaaRUnok+kx1WdO15EQc=
--- a/logger/default.go
+++ b/logger/default.go
@@ -24,6 +24,8 @@ type defaultLogger struct {
 	enc  *json.Encoder
 	opts Options
 	sync.RWMutex
+	logFunc  LogFunc
+	logfFunc LogfFunc
 }

 // Init(opts...) should only overwrite provided options
@@ -33,6 +35,15 @@ 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)
+		l.logfFunc = l.opts.Wrappers[i-1].Logf(l.logfFunc)
+	}
+
 	l.Unlock()
 	return nil
 }
@@ -121,27 +132,27 @@ func (l *defaultLogger) Fatal(ctx context.Context, args ...interface{}) {
 }

 func (l *defaultLogger) Infof(ctx context.Context, msg string, args ...interface{}) {
-	l.Logf(ctx, InfoLevel, msg, args...)
+	l.logfFunc(ctx, InfoLevel, msg, args...)
 }

 func (l *defaultLogger) Errorf(ctx context.Context, msg string, args ...interface{}) {
-	l.Logf(ctx, ErrorLevel, msg, args...)
+	l.logfFunc(ctx, ErrorLevel, msg, args...)
 }

 func (l *defaultLogger) Debugf(ctx context.Context, msg string, args ...interface{}) {
-	l.Logf(ctx, DebugLevel, msg, args...)
+	l.logfFunc(ctx, DebugLevel, msg, args...)
 }

 func (l *defaultLogger) Warnf(ctx context.Context, msg string, args ...interface{}) {
-	l.Logf(ctx, WarnLevel, msg, args...)
+	l.logfFunc(ctx, WarnLevel, msg, args...)
 }

 func (l *defaultLogger) Tracef(ctx context.Context, msg string, args ...interface{}) {
-	l.Logf(ctx, TraceLevel, msg, args...)
+	l.logfFunc(ctx, TraceLevel, msg, args...)
 }

 func (l *defaultLogger) Fatalf(ctx context.Context, msg string, args ...interface{}) {
-	l.Logf(ctx, FatalLevel, msg, args...)
+	l.logfFunc(ctx, FatalLevel, msg, args...)
 	os.Exit(1)
 }

--- a/logger/logger_test.go
+++ b/logger/logger_test.go
@@ -1,13 +1,15 @@
 package logger

 import (
+	"bytes"
 	"context"
 	"testing"
 )

 func TestLogger(t *testing.T) {
 	ctx := context.TODO()
-	l := NewLogger(WithLevel(TraceLevel))
+	buf := bytes.NewBuffer(nil)
+	l := NewLogger(WithLevel(TraceLevel), WithOutput(buf))
 	if err := l.Init(); err != nil {
 		t.Fatal(err)
 	}
@@ -15,4 +17,52 @@ func TestLogger(t *testing.T) {
 	l.Warn(ctx, "warn_msg1")
 	l.Fields(map[string]interface{}{"error": "test"}).Info(ctx, "error message")
 	l.Warn(ctx, "first", " ", "second")
+	if !bytes.Contains(buf.Bytes(), []byte(`"level":"trace","msg":"trace_msg1"`)) {
+		t.Fatalf("logger error, buf %s", buf.Bytes())
+	}
+	if !bytes.Contains(buf.Bytes(), []byte(`"warn","msg":"warn_msg1"`)) {
+		t.Fatalf("logger error, buf %s", buf.Bytes())
+	}
+	if !bytes.Contains(buf.Bytes(), []byte(`"error":"test","level":"info","msg":"error message"`)) {
+		t.Fatalf("logger error, buf %s", buf.Bytes())
+	}
+	if !bytes.Contains(buf.Bytes(), []byte(`"level":"warn","msg":"first second"`)) {
+		t.Fatalf("logger error, buf %s", buf.Bytes())
+	}
+}
+
+func TestLoggerWrapper(t *testing.T) {
+	ctx := context.TODO()
+	buf := bytes.NewBuffer(nil)
+	l := NewLogger(WithLevel(TraceLevel), WithOutput(buf))
+	if err := l.Init(WrapLogger(NewOmitWrapper())); err != nil {
+		t.Fatal(err)
+	}
+	type secret struct {
+		Name  string
+		Passw string `logger:"omit"`
+	}
+	s := &secret{Name: "name", Passw: "secret"}
+	l.Errorf(ctx, "test %#+v", s)
+	if !bytes.Contains(buf.Bytes(), []byte(`logger.secret{Name:\"name\", Passw:\"\"}"`)) {
+		t.Fatalf("omit not works, struct: %v, output: %s", s, buf.Bytes())
+	}
+}
+
+func TestOmitLoggerWrapper(t *testing.T) {
+	ctx := context.TODO()
+	buf := bytes.NewBuffer(nil)
+	l := NewOmitLogger(NewLogger(WithLevel(TraceLevel), WithOutput(buf)))
+	if err := l.Init(); err != nil {
+		t.Fatal(err)
+	}
+	type secret struct {
+		Name  string
+		Passw string `logger:"omit"`
+	}
+	s := &secret{Name: "name", Passw: "secret"}
+	l.Errorf(ctx, "test %#+v", s)
+	if !bytes.Contains(buf.Bytes(), []byte(`logger.secret{Name:\"name\", Passw:\"\"}"`)) {
+		t.Fatalf("omit not works, struct: %v, output: %s", s, buf.Bytes())
+	}
 }
--- a/logger/options.go
+++ b/logger/options.go
@@ -23,6 +23,8 @@ type Options struct {
 	CallerSkipCount int
 	// The logging level the logger should log
 	Level Level
+	// Wrappers logger wrapper that called before actual Log/Logf function
+	Wrappers []Wrapper
 }

 // NewOptions creates new options struct
@@ -81,3 +83,10 @@ func WithName(n string) Option {
 		o.Name = n
 	}
 }
+
+// WrapLogger adds a logger Wrapper to a list of options passed into the logger
+func WrapLogger(w Wrapper) Option {
+	return func(o *Options) {
+		o.Wrappers = append(o.Wrappers, w)
+	}
+}
--- a/logger/wrapper.go
+++ b/logger/wrapper.go
@@ -0,0 +1,154 @@
+package logger
+
+import (
+	"context"
+	"reflect"
+
+	rutil "github.com/unistack-org/micro/v3/util/reflect"
+)
+
+// LogFunc function used for Log method
+type LogFunc func(ctx context.Context, level Level, args ...interface{})
+
+// LogfFunc function used for Logf method
+type LogfFunc func(ctx context.Context, level Level, msg string, args ...interface{})
+
+type Wrapper interface {
+	// Log logs message with needed level
+	Log(LogFunc) LogFunc
+	// Logf logs message with needed level
+	Logf(LogfFunc) LogfFunc
+}
+
+var (
+	_ Logger = &OmitLogger{}
+)
+
+type OmitLogger struct {
+	l Logger
+}
+
+func NewOmitLogger(l Logger) Logger {
+	return &OmitLogger{l: l}
+}
+
+func (w *OmitLogger) Init(opts ...Option) error {
+	return w.l.Init(append(opts, WrapLogger(NewOmitWrapper()))...)
+}
+
+func (w *OmitLogger) V(level Level) bool {
+	return w.l.V(level)
+}
+
+func (w *OmitLogger) Options() Options {
+	return w.l.Options()
+}
+
+func (w *OmitLogger) Fields(fields map[string]interface{}) Logger {
+	return w.l.Fields(fields)
+}
+
+func (w *OmitLogger) Info(ctx context.Context, args ...interface{}) {
+	w.l.Info(ctx, args...)
+}
+
+func (w *OmitLogger) Trace(ctx context.Context, args ...interface{}) {
+	w.l.Trace(ctx, args...)
+}
+
+func (w *OmitLogger) Debug(ctx context.Context, args ...interface{}) {
+	w.l.Debug(ctx, args...)
+}
+
+func (w *OmitLogger) Warn(ctx context.Context, args ...interface{}) {
+	w.l.Warn(ctx, args...)
+}
+
+func (w *OmitLogger) Error(ctx context.Context, args ...interface{}) {
+	w.l.Error(ctx, args...)
+}
+
+func (w *OmitLogger) Fatal(ctx context.Context, args ...interface{}) {
+	w.l.Fatal(ctx, args...)
+}
+
+func (w *OmitLogger) Infof(ctx context.Context, msg string, args ...interface{}) {
+	w.l.Infof(ctx, msg, args...)
+}
+
+func (w *OmitLogger) Tracef(ctx context.Context, msg string, args ...interface{}) {
+	w.l.Tracef(ctx, msg, args...)
+}
+
+func (w *OmitLogger) Debugf(ctx context.Context, msg string, args ...interface{}) {
+	w.l.Debugf(ctx, msg, args...)
+}
+
+func (w *OmitLogger) Warnf(ctx context.Context, msg string, args ...interface{}) {
+	w.l.Warnf(ctx, msg, args...)
+}
+
+func (w *OmitLogger) Errorf(ctx context.Context, msg string, args ...interface{}) {
+	w.l.Errorf(ctx, msg, args...)
+}
+
+func (w *OmitLogger) Fatalf(ctx context.Context, msg string, args ...interface{}) {
+	w.l.Fatalf(ctx, msg, args...)
+}
+
+func (w *OmitLogger) Log(ctx context.Context, level Level, args ...interface{}) {
+	w.l.Log(ctx, level, args...)
+}
+
+func (w *OmitLogger) Logf(ctx context.Context, level Level, msg string, args ...interface{}) {
+	w.l.Logf(ctx, level, msg, args...)
+}
+
+func (w *OmitLogger) String() string {
+	return w.l.String()
+}
+
+type OmitWrapper struct{}
+
+func NewOmitWrapper() Wrapper {
+	return &OmitWrapper{}
+}
+
+func getArgs(args []interface{}) []interface{} {
+	nargs := make([]interface{}, 0, len(args))
+	var err error
+	for _, arg := range args {
+		val := reflect.ValueOf(arg)
+		switch val.Kind() {
+		case reflect.Ptr:
+			val = val.Elem()
+		}
+		narg := arg
+		if val.Kind() == reflect.Struct {
+			if narg, err = rutil.Zero(arg); err == nil {
+				rutil.CopyDefaults(narg, arg)
+				if flds, ferr := rutil.StructFields(narg); ferr == nil {
+					for _, fld := range flds {
+						if tv, ok := fld.Field.Tag.Lookup("logger"); ok && tv == "omit" {
+							fld.Value.Set(reflect.Zero(fld.Value.Type()))
+						}
+					}
+				}
+			}
+		}
+		nargs = append(nargs, narg)
+	}
+	return nargs
+}
+
+func (w *OmitWrapper) Log(fn LogFunc) LogFunc {
+	return func(ctx context.Context, level Level, args ...interface{}) {
+		fn(ctx, level, getArgs(args)...)
+	}
+}
+
+func (w *OmitWrapper) Logf(fn LogfFunc) LogfFunc {
+	return func(ctx context.Context, level Level, msg string, args ...interface{}) {
+		fn(ctx, level, msg, getArgs(args)...)
+	}
+}
--- a/metadata/metadata.go
+++ b/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
--- a/meter/meter.go
+++ b/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()
 }
--- a/meter/meter_test.go
+++ b/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)
+		}
 	}
 }
--- a/network/tunnel/broker/broker.go
+++ b/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
--- a/server/server.go
+++ b/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
--- a/store/memory.go
+++ b/store/memory.go
@@ -36,16 +36,6 @@ func (m *memoryStore) key(prefix, key string) string {
 	return filepath.Join(prefix, key)
 }

-func (m *memoryStore) prefix(database, table string) string {
-	if len(database) == 0 {
-		database = m.opts.Database
-	}
-	if len(table) == 0 {
-		table = m.opts.Table
-	}
-	return filepath.Join(database, table)
-}
-
 func (m *memoryStore) exists(prefix, key string) error {
 	key = m.key(prefix, key)

@@ -80,15 +70,17 @@ func (m *memoryStore) delete(prefix, key string) {

 func (m *memoryStore) list(prefix string, limit, offset uint) []string {
 	allItems := m.store.Items()
-	allKeys := make([]string, len(allItems))
-	i := 0
+	allKeys := make([]string, 0, len(allItems))

 	for k := range allItems {
-		if !strings.HasPrefix(k, prefix+"/") {
+		if !strings.HasPrefix(k, prefix) {
 			continue
 		}
-		allKeys[i] = strings.TrimPrefix(k, prefix+"/")
-		i++
+		k = strings.TrimPrefix(k, prefix)
+		if k[0] == '/' {
+			k = k[1:]
+		}
+		allKeys = append(allKeys, k)
 	}

 	if limit != 0 || offset != 0 {
@@ -107,7 +99,6 @@ func (m *memoryStore) list(prefix string, limit, offset uint) []string {
 		}
 		return allKeys[offset:end]
 	}
-
 	return allKeys
 }

@@ -127,37 +118,48 @@ func (m *memoryStore) Name() string {
 }

 func (m *memoryStore) Exists(ctx context.Context, key string, opts ...ExistsOption) error {
-	prefix := m.prefix(m.opts.Database, m.opts.Table)
-	return m.exists(prefix, key)
+	options := NewExistsOptions(opts...)
+	if options.Namespace == "" {
+		options.Namespace = m.opts.Namespace
+	}
+	return m.exists(options.Namespace, key)
 }

 func (m *memoryStore) Read(ctx context.Context, key string, val interface{}, opts ...ReadOption) error {
-	readOpts := NewReadOptions(opts...)
-	prefix := m.prefix(readOpts.Database, readOpts.Table)
-	return m.get(prefix, key, val)
+	options := NewReadOptions(opts...)
+	if options.Namespace == "" {
+		options.Namespace = m.opts.Namespace
+	}
+	return m.get(options.Namespace, key, val)
 }

 func (m *memoryStore) Write(ctx context.Context, key string, val interface{}, opts ...WriteOption) error {
-	writeOpts := NewWriteOptions(opts...)
+	options := NewWriteOptions(opts...)
+	if options.Namespace == "" {
+		options.Namespace = m.opts.Namespace
+	}
+	if options.TTL == 0 {
+		options.TTL = cache.NoExpiration
+	}

-	prefix := m.prefix(writeOpts.Database, writeOpts.Table)
-
-	key = m.key(prefix, key)
+	key = m.key(options.Namespace, key)

 	buf, err := m.opts.Codec.Marshal(val)
 	if err != nil {
 		return err
 	}

-	m.store.Set(key, buf, writeOpts.TTL)
+	m.store.Set(key, buf, options.TTL)
 	return nil
 }

 func (m *memoryStore) Delete(ctx context.Context, key string, opts ...DeleteOption) error {
-	deleteOptions := NewDeleteOptions(opts...)
+	options := NewDeleteOptions(opts...)
+	if options.Namespace == "" {
+		options.Namespace = m.opts.Namespace
+	}

-	prefix := m.prefix(deleteOptions.Database, deleteOptions.Table)
-	m.delete(prefix, key)
+	m.delete(options.Namespace, key)
 	return nil
 }

@@ -166,25 +168,27 @@ func (m *memoryStore) Options() Options {
 }

 func (m *memoryStore) List(ctx context.Context, opts ...ListOption) ([]string, error) {
-	listOptions := NewListOptions(opts...)
+	options := NewListOptions(opts...)
+	if options.Namespace == "" {
+		options.Namespace = m.opts.Namespace
+	}

-	prefix := m.prefix(listOptions.Database, listOptions.Table)
-	keys := m.list(prefix, listOptions.Limit, listOptions.Offset)
+	keys := m.list(options.Namespace, options.Limit, options.Offset)

-	if len(listOptions.Prefix) > 0 {
+	if len(options.Prefix) > 0 {
 		var prefixKeys []string
 		for _, k := range keys {
-			if strings.HasPrefix(k, listOptions.Prefix) {
+			if strings.HasPrefix(k, options.Prefix) {
 				prefixKeys = append(prefixKeys, k)
 			}
 		}
 		keys = prefixKeys
 	}

-	if len(listOptions.Suffix) > 0 {
+	if len(options.Suffix) > 0 {
 		var suffixKeys []string
 		for _, k := range keys {
-			if strings.HasSuffix(k, listOptions.Suffix) {
+			if strings.HasSuffix(k, options.Suffix) {
 				suffixKeys = append(suffixKeys, k)
 			}
 		}
--- a/store/memory_test.go
+++ b/store/memory_test.go
@@ -9,11 +9,11 @@ import (
 )

 func TestMemoryReInit(t *testing.T) {
-	s := store.NewStore(store.Table("aaa"))
-	if err := s.Init(store.Table("")); err != nil {
+	s := store.NewStore(store.Namespace("aaa"))
+	if err := s.Init(store.Namespace("")); err != nil {
 		t.Fatal(err)
 	}
-	if len(s.Options().Table) > 0 {
+	if len(s.Options().Namespace) > 0 {
 		t.Error("Init didn't reinitialise the store")
 	}
 }
@@ -28,7 +28,7 @@ func TestMemoryBasic(t *testing.T) {

 func TestMemoryPrefix(t *testing.T) {
 	s := store.NewStore()
-	if err := s.Init(store.Table("some-prefix")); err != nil {
+	if err := s.Init(store.Namespace("some-prefix")); err != nil {
 		t.Fatal(err)
 	}
 	basictest(s, t)
@@ -36,7 +36,7 @@ func TestMemoryPrefix(t *testing.T) {

 func TestMemoryNamespace(t *testing.T) {
 	s := store.NewStore()
-	if err := s.Init(store.Database("some-namespace")); err != nil {
+	if err := s.Init(store.Namespace("some-namespace")); err != nil {
 		t.Fatal(err)
 	}
 	basictest(s, t)
@@ -44,7 +44,7 @@ func TestMemoryNamespace(t *testing.T) {

 func TestMemoryNamespacePrefix(t *testing.T) {
 	s := store.NewStore()
-	if err := s.Init(store.Table("some-prefix"), store.Database("some-namespace")); err != nil {
+	if err := s.Init(store.Namespace("some-namespace")); err != nil {
 		t.Fatal(err)
 	}
 	basictest(s, t)
--- a/store/options.go
+++ b/store/options.go
@@ -28,13 +28,12 @@ type Options struct {
 	TLSConfig *tls.Config
 	// Name specifies store name
 	Name string
-	// Database specifies store database
-	Database string
-	// Table specifies store table
-	Table string
-	// Nodes contains store address
-	// TODO: replace with Addrs
-	Nodes []string
+	// Namespace of the records
+	Namespace string
+	// Addrs contains store address
+	Addrs []string
+	//Wrappers store wrapper that called before actual functions
+	//Wrappers []Wrapper
 }

 // NewOptions creates options struct
@@ -90,13 +89,20 @@ func Meter(m meter.Meter) Option {
 	}
 }

-// Name the name
+// Name the name of the store
 func Name(n string) Option {
 	return func(o *Options) {
 		o.Name = n
 	}
 }

+// Namespace sets namespace of the store
+func Namespace(ns string) Option {
+	return func(o *Options) {
+		o.Namespace = ns
+	}
+}
+
 // Tracer sets the tracer
 func Tracer(t tracer.Tracer) Option {
 	return func(o *Options) {
@@ -104,27 +110,21 @@ func Tracer(t tracer.Tracer) Option {
 	}
 }

-// Nodes contains the addresses or other connection information of the backing storage.
+// Addrs contains the addresses or other connection information of the backing storage.
 // For example, an etcd implementation would contain the nodes of the cluster.
 // A SQL implementation could contain one or more connection strings.
-func Nodes(a ...string) Option {
+func Addrs(addrs ...string) Option {
 	return func(o *Options) {
-		o.Nodes = a
+		o.Addrs = addrs
 	}
 }

-// Database allows multiple isolated stores to be kept in one backend, if supported.
-func Database(db string) Option {
-	return func(o *Options) {
-		o.Database = db
-	}
-}
-
-// Table is analag for a table in database backends or a key prefix in KV backends
-func Table(t string) Option {
-	return func(o *Options) {
-		o.Table = t
-	}
+// ReadOptions configures an individual Read operation
+type ReadOptions struct {
+	// Context holds external options
+	Context context.Context
+	// Namespace holds namespace
+	Namespace string
 }

 // NewReadOptions fills ReadOptions struct with opts slice
@@ -136,29 +136,35 @@ func NewReadOptions(opts ...ReadOption) ReadOptions {
 	return options
 }

-// ReadOptions configures an individual Read operation
-type ReadOptions struct {
-	// Context holds external options
-	Context context.Context
-	// Database holds the database name
-	Database string
-	// Table holds table name
-	Table string
-	// Namespace holds namespace
-	Namespace string
-}
-
 // ReadOption sets values in ReadOptions
 type ReadOption func(r *ReadOptions)

-// ReadFrom the database and table
-func ReadFrom(database, table string) ReadOption {
-	return func(r *ReadOptions) {
-		r.Database = database
-		r.Table = table
+// ReadContext pass context.Context to ReadOptions
+func ReadContext(ctx context.Context) ReadOption {
+	return func(o *ReadOptions) {
+		o.Context = ctx
 	}
 }

+// ReadNamespace pass namespace to ReadOptions
+func ReadNamespace(ns string) ReadOption {
+	return func(o *ReadOptions) {
+		o.Namespace = ns
+	}
+}
+
+// WriteOptions configures an individual Write operation
+type WriteOptions struct {
+	// Context holds external options
+	Context context.Context
+	// Metadata contains additional metadata
+	Metadata metadata.Metadata
+	// Namespace holds namespace
+	Namespace string
+	// TTL specifies key TTL
+	TTL time.Duration
+}
+
 // NewWriteOptions fills WriteOptions struct with opts slice
 func NewWriteOptions(opts ...WriteOption) WriteOptions {
 	options := WriteOptions{}
@@ -168,47 +174,45 @@ func NewWriteOptions(opts ...WriteOption) WriteOptions {
 	return options
 }

-// WriteOptions configures an individual Write operation
-type WriteOptions struct {
-	// Context holds external options
-	Context context.Context
-	// Metadata contains additional metadata
-	Metadata metadata.Metadata
-	// Database holds database name
-	Database string
-	// Table holds table name
-	Table string
-	// Namespace holds namespace
-	Namespace string
-	// TTL specifies key TTL
-	TTL time.Duration
-}
-
 // WriteOption sets values in WriteOptions
 type WriteOption func(w *WriteOptions)

-// WriteTo the database and table
-func WriteTo(database, table string) WriteOption {
-	return func(w *WriteOptions) {
-		w.Database = database
-		w.Table = table
-	}
-}
-
-// WriteTTL is the time the record expires
-func WriteTTL(d time.Duration) WriteOption {
-	return func(w *WriteOptions) {
-		w.TTL = d
+// WriteContext pass context.Context to wirte options
+func WriteContext(ctx context.Context) WriteOption {
+	return func(o *WriteOptions) {
+		o.Context = ctx
 	}
 }

 // WriteMetadata add metadata.Metadata
 func WriteMetadata(md metadata.Metadata) WriteOption {
-	return func(w *WriteOptions) {
-		w.Metadata = metadata.Copy(md)
+	return func(o *WriteOptions) {
+		o.Metadata = metadata.Copy(md)
 	}
 }

+// WriteTTL is the time the record expires
+func WriteTTL(d time.Duration) WriteOption {
+	return func(o *WriteOptions) {
+		o.TTL = d
+	}
+}
+
+// WriteNamespace pass namespace to write options
+func WriteNamespace(ns string) WriteOption {
+	return func(o *WriteOptions) {
+		o.Namespace = ns
+	}
+}
+
+// DeleteOptions configures an individual Delete operation
+type DeleteOptions struct {
+	// Context holds external options
+	Context context.Context
+	// Namespace holds namespace
+	Namespace string
+}
+
 // NewDeleteOptions fills DeleteOptions struct with opts slice
 func NewDeleteOptions(opts ...DeleteOption) DeleteOptions {
 	options := DeleteOptions{}
@@ -218,29 +222,33 @@ func NewDeleteOptions(opts ...DeleteOption) DeleteOptions {
 	return options
 }

-// DeleteOptions configures an individual Delete operation
-type DeleteOptions struct {
-	// Context holds external options
-	Context context.Context
-	// Database holds database name
-	Database string
-	// Table holds table name
-	Table string
-	// Namespace holds namespace
-	Namespace string
-}
-
 // DeleteOption sets values in DeleteOptions
 type DeleteOption func(d *DeleteOptions)

-// DeleteFrom the database and table
-func DeleteFrom(database, table string) DeleteOption {
-	return func(d *DeleteOptions) {
-		d.Database = database
-		d.Table = table
+// DeleteContext pass context.Context to delete options
+func DeleteContext(ctx context.Context) DeleteOption {
+	return func(o *DeleteOptions) {
+		o.Context = ctx
 	}
 }

+// DeleteNamespace pass namespace to delete options
+func DeleteNamespace(ns string) DeleteOption {
+	return func(o *DeleteOptions) {
+		o.Namespace = ns
+	}
+}
+
+// ListOptions configures an individual List operation
+type ListOptions struct {
+	Context   context.Context
+	Prefix    string
+	Suffix    string
+	Namespace string
+	Limit     uint
+	Offset    uint
+}
+
 // NewListOptions fills ListOptions struct with opts slice
 func NewListOptions(opts ...ListOption) ListOptions {
 	options := ListOptions{}
@@ -250,59 +258,50 @@ func NewListOptions(opts ...ListOption) ListOptions {
 	return options
 }

-// ListOptions configures an individual List operation
-type ListOptions struct {
-	Context   context.Context
-	Database  string
-	Prefix    string
-	Suffix    string
-	Namespace string
-	Table     string
-	Limit     uint
-	Offset    uint
-}
-
 // ListOption sets values in ListOptions
 type ListOption func(l *ListOptions)

-// ListFrom the database and table
-func ListFrom(database, table string) ListOption {
-	return func(l *ListOptions) {
-		l.Database = database
-		l.Table = table
+// ListContext pass context.Context to list options
+func ListContext(ctx context.Context) ListOption {
+	return func(o *ListOptions) {
+		o.Context = ctx
 	}
 }

 // ListPrefix returns all keys that are prefixed with key
-func ListPrefix(p string) ListOption {
-	return func(l *ListOptions) {
-		l.Prefix = p
+func ListPrefix(s string) ListOption {
+	return func(o *ListOptions) {
+		o.Prefix = s
 	}
 }

 // ListSuffix returns all keys that end with key
 func ListSuffix(s string) ListOption {
-	return func(l *ListOptions) {
-		l.Suffix = s
+	return func(o *ListOptions) {
+		o.Suffix = s
 	}
 }

-// ListLimit limits the number of returned keys to l
-func ListLimit(l uint) ListOption {
-	return func(lo *ListOptions) {
-		lo.Limit = l
+// ListLimit limits the number of returned keys
+func ListLimit(n uint) ListOption {
+	return func(o *ListOptions) {
+		o.Limit = n
 	}
 }

-// ListOffset starts returning responses from o. Use in conjunction with Limit for pagination.
-func ListOffset(o uint) ListOption {
-	return func(l *ListOptions) {
-		l.Offset = o
+// ListOffset use with Limit for pagination
+func ListOffset(n uint) ListOption {
+	return func(o *ListOptions) {
+		o.Offset = n
 	}
 }

-// ExistsOption specifies Exists call options
-type ExistsOption func(*ExistsOptions)
+// ListNamespace pass namespace to list options
+func ListNamespace(ns string) ListOption {
+	return func(o *ListOptions) {
+		o.Namespace = ns
+	}
+}

 // ExistsOptions holds options for Exists method
 type ExistsOptions struct {
@@ -312,6 +311,9 @@ type ExistsOptions struct {
 	Namespace string
 }

+// ExistsOption specifies Exists call options
+type ExistsOption func(*ExistsOptions)
+
 // NewExistsOptions helper for Exists method
 func NewExistsOptions(opts ...ExistsOption) ExistsOptions {
 	options := ExistsOptions{
@@ -322,3 +324,24 @@ func NewExistsOptions(opts ...ExistsOption) ExistsOptions {
 	}
 	return options
 }
+
+// ExistsContext pass context.Context to exist options
+func ExistsContext(ctx context.Context) ExistsOption {
+	return func(o *ExistsOptions) {
+		o.Context = ctx
+	}
+}
+
+// ExistsNamespace pass namespace to exist options
+func ExistsNamespace(ns string) ExistsOption {
+	return func(o *ExistsOptions) {
+		o.Namespace = ns
+	}
+}
+
+// WrapStore adds a store Wrapper to a list of options passed into the store
+//func WrapStore(w Wrapper) Option {
+//	return func(o *Options) {
+//		o.Wrappers = append(o.Wrappers, w)
+//	}
+//}
--- a/store/wrapper.go
+++ b/store/wrapper.go
@@ -0,0 +1,84 @@
+package store
+
+import (
+	"context"
+)
+
+// LogfFunc function used for Logf method
+//type LogfFunc func(ctx context.Context, level Level, msg string, args ...interface{})
+
+//type Wrapper interface {
+// Logf logs message with needed level
+//Logf(LogfFunc) LogfFunc
+//}
+
+type NamespaceStore struct {
+	s  Store
+	ns string
+}
+
+var (
+	_ Store = &NamespaceStore{}
+)
+
+func NewNamespaceStore(s Store, ns string) Store {
+	return &NamespaceStore{s: s, ns: ns}
+}
+
+func (w *NamespaceStore) Init(opts ...Option) error {
+	return w.s.Init(opts...)
+}
+
+func (w *NamespaceStore) Connect(ctx context.Context) error {
+	return w.s.Connect(ctx)
+}
+
+func (w *NamespaceStore) Disconnect(ctx context.Context) error {
+	return w.s.Disconnect(ctx)
+}
+
+func (w *NamespaceStore) Read(ctx context.Context, key string, val interface{}, opts ...ReadOption) error {
+	return w.s.Read(ctx, key, val, append(opts, ReadNamespace(w.ns))...)
+}
+
+func (w *NamespaceStore) Write(ctx context.Context, key string, val interface{}, opts ...WriteOption) error {
+	return w.s.Write(ctx, key, val, append(opts, WriteNamespace(w.ns))...)
+}
+
+func (w *NamespaceStore) Delete(ctx context.Context, key string, opts ...DeleteOption) error {
+	return w.s.Delete(ctx, key, append(opts, DeleteNamespace(w.ns))...)
+}
+
+func (w *NamespaceStore) Exists(ctx context.Context, key string, opts ...ExistsOption) error {
+	return w.s.Exists(ctx, key, append(opts, ExistsNamespace(w.ns))...)
+}
+
+func (w *NamespaceStore) List(ctx context.Context, opts ...ListOption) ([]string, error) {
+	return w.s.List(ctx, append(opts, ListNamespace(w.ns))...)
+}
+
+func (w *NamespaceStore) Options() Options {
+	return w.s.Options()
+}
+
+func (w *NamespaceStore) Name() string {
+	return w.s.Name()
+}
+
+func (w *NamespaceStore) String() string {
+	return w.s.String()
+}
+
+//type NamespaceWrapper struct{}
+
+//func NewNamespaceWrapper() Wrapper {
+//	return &NamespaceWrapper{}
+//}
+
+/*
+func (w *OmitWrapper) Logf(fn LogfFunc) LogfFunc {
+	return func(ctx context.Context, level Level, msg string, args ...interface{}) {
+		fn(ctx, level, msg, getArgs(args)...)
+	}
+}
+*/
--- a/util/reflect/struct.go
+++ b/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
 		}

--- a/util/reflect/struct_test.go
+++ b/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"`
--- a/util/router/parse.go
+++ b/util/router/parse.go
@@ -102,7 +102,7 @@ type parser struct {
 // topLevelSegments is the target of this parser.
 func (p *parser) topLevelSegments() ([]segment, error) {
 	if logger.V(logger.TraceLevel) {
-		logger.Debug(context.TODO(), "Parsing %q", p.tokens)
+		logger.Trace(context.TODO(), "Parsing %q", p.tokens)
 	}
 	segs, err := p.segments()
 	if err != nil {
--- a/util/router/runtime.go
+++ b/util/router/runtime.go
@@ -63,16 +63,16 @@ func NewPattern(version int, ops []int, pool []string, verb string, opts ...Patt
 	}

 	if version != 1 {
-		if logger.V(logger.DebugLevel) {
-			logger.Debug(context.TODO(), "unsupported version: %d", version)
+		if logger.V(logger.TraceLevel) {
+			logger.Trace(context.TODO(), "unsupported version: %d", version)
 		}
 		return Pattern{}, ErrInvalidPattern
 	}

 	l := len(ops)
 	if l%2 != 0 {
-		if logger.V(logger.DebugLevel) {
-			logger.Debug(context.TODO(), "odd number of ops codes: %d", l)
+		if logger.V(logger.TraceLevel) {
+			logger.Trace(context.TODO(), "odd number of ops codes: %d", l)
 		}
 		return Pattern{}, ErrInvalidPattern
 	}
@@ -141,13 +141,13 @@ func NewPattern(version int, ops []int, pool []string, verb string, opts ...Patt
 			vars = append(vars, v)
 			stack--
 			if stack < 0 {
-				if logger.V(logger.DebugLevel) {
+				if logger.V(logger.TraceLevel) {
 					logger.Trace(context.TODO(), "stack underflow")
 				}
 				return Pattern{}, ErrInvalidPattern
 			}
 		default:
-			if logger.V(logger.DebugLevel) {
+			if logger.V(logger.TraceLevel) {
 				logger.Trace(context.TODO(), "invalid opcode: %d", op.code)
 			}
 			return Pattern{}, ErrInvalidPattern
Author	SHA1	Message	Date
Vasiliy Tolstov	e64269b2a8	broker: add BatchBroker interface to avoid breaking older brokers Signed-off-by: Vasiliy Tolstov <v.tolstov@unistack.org>	2021-07-23 12:55:36 +03:00
Vasiliy Tolstov	d18429e024	metadata: add HeaderAuthorization Signed-off-by: Vasiliy Tolstov <v.tolstov@unistack.org>	2021-07-23 12:17:00 +03:00
Vasiliy Tolstov	675e121049	metadata: add default headers Signed-off-by: Vasiliy Tolstov <v.tolstov@unistack.org>	2021-07-23 12:03:18 +03:00
Vasiliy Tolstov	d357fb1e0d	WIP: broker batch processing Signed-off-by: Vasiliy Tolstov <v.tolstov@unistack.org>	2021-07-22 22:53:44 +03:00
Vasiliy Tolstov	e4673bcc50	remove old cruft Signed-off-by: Vasiliy Tolstov <v.tolstov@unistack.org>	2021-07-22 15:45:44 +03:00
Vasiliy Tolstov	a839f75a2f	util/reflect: add new funcs Signed-off-by: Vasiliy Tolstov <v.tolstov@unistack.org>	2021-07-22 15:45:31 +03:00
Vasiliy Tolstov	a7e6d61b95	meter: fast path for only one label Signed-off-by: Vasiliy Tolstov <v.tolstov@unistack.org>	2021-07-21 14:29:13 +03:00
Vasiliy Tolstov	650d167313	meter: add BuildLabels func that sorts and deletes duplicates Signed-off-by: Vasiliy Tolstov <v.tolstov@unistack.org>	2021-07-21 14:10:20 +03:00
Vasiliy Tolstov	c6ba2a91e6	meter: BuildName func to combine metric name with labels into string Signed-off-by: Vasiliy Tolstov <v.tolstov@unistack.org>	2021-07-21 12:39:59 +03:00
Vasiliy Tolstov	7ece08896f	server: use 127.0.0.1:0 if no address provided Signed-off-by: Vasiliy Tolstov <v.tolstov@unistack.org>	2021-07-17 01:57:39 +06:00
dependabot[bot]	57f6f23294	build(deps): bump github.com/google/uuid from 1.2.0 to 1.3.0 (#53 ) Bumps [github.com/google/uuid](https://github.com/google/uuid) from 1.2.0 to 1.3.0. - [Release notes](https://github.com/google/uuid/releases) - [Commits](https://github.com/google/uuid/compare/v1.2.0...v1.3.0) --- updated-dependencies: - dependency-name: github.com/google/uuid dependency-type: direct:production update-type: version-update:semver-minor ... Signed-off-by: dependabot[bot] <support@github.com> Co-authored-by: dependabot[bot] <49699333+dependabot[bot]@users.noreply.github.com>	2021-07-16 00:27:56 +03:00
Vasiliy Tolstov	09e6fa2fed	flow: implement new methods, add Async ExecutionOption Signed-off-by: Vasiliy Tolstov <v.tolstov@unistack.org>	2021-07-16 00:17:16 +03:00
Vasiliy Tolstov	10a09a5c6f	flow: improve store Signed-off-by: Vasiliy Tolstov <v.tolstov@unistack.org>	2021-07-15 22:56:34 +03:00
Vasiliy Tolstov	b4e5d9462a	util/router: move some messages to Trace level Signed-off-by: Vasiliy Tolstov <v.tolstov@unistack.org>	2021-07-15 22:56:34 +03:00
Vasiliy Tolstov	96aa0b6906	store/memory: fix List Signed-off-by: Vasiliy Tolstov <v.tolstov@unistack.org>	2021-07-15 22:53:12 +03:00
Vasiliy Tolstov	f54658830d	store/memory: fixup Signed-off-by: Vasiliy Tolstov <v.tolstov@unistack.org>	2021-07-15 12:11:55 +03:00
Vasiliy Tolstov	1e43122660	store/memory: small fixups for flow usage Signed-off-by: Vasiliy Tolstov <v.tolstov@unistack.org>	2021-07-15 11:59:35 +03:00
Vasiliy Tolstov	42800fa247	flow: improve steps handling Signed-off-by: Vasiliy Tolstov <v.tolstov@unistack.org>	2021-07-14 17:12:54 +03:00
Vasiliy Tolstov	5b9c810653	logger: add compile time test for interface compat Signed-off-by: Vasiliy Tolstov <v.tolstov@unistack.org>	2021-07-14 17:12:09 +03:00
Vasiliy Tolstov	c3def24bf4	store: add Wrappers support, create Namespace wrapper Signed-off-by: Vasiliy Tolstov <v.tolstov@unistack.org>	2021-07-14 17:11:37 +03:00
Vasiliy Tolstov	0d1ef31764	client: change AuthToken option signature Signed-off-by: Vasiliy Tolstov <v.tolstov@unistack.org>	2021-07-09 10:47:40 +03:00
Vasiliy Tolstov	d49afa230f	logger: add omit logger Signed-off-by: Vasiliy Tolstov <v.tolstov@unistack.org>	2021-07-05 23:04:20 +03:00
Vasiliy Tolstov	e545eb4e13	logger: add wrapper support Signed-off-by: Vasiliy Tolstov <v.tolstov@unistack.org>	2021-07-05 22:32:47 +03:00
Vasiliy Tolstov	f28b107372	broker: fix RawMessage marshal Signed-off-by: Vasiliy Tolstov <v.tolstov@unistack.org>	2021-07-01 23:23:01 +03:00
Vasiliy Tolstov	c592fabe2a	minor fixes Signed-off-by: Vasiliy Tolstov <v.tolstov@unistack.org>	2021-07-01 15:56:22 +03:00
Vasiliy Tolstov	eb107020c7	broker: replace Message.Body []byte slice to RawMessage Signed-off-by: Vasiliy Tolstov <v.tolstov@unistack.org>	2021-07-01 15:11:17 +03:00
Vasiliy Tolstov	bd4d4c363e	flow improvements (#52 ) * flow improvements Signed-off-by: Vasiliy Tolstov <v.tolstov@unistack.org>	2021-06-30 17:50:58 +03:00