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>

broker/broker.go

@@ -3,6 +3,7 @@ package broker
 
 import (
 	"context"
+	"errors"
 
 	"github.com/unistack-org/micro/v3/metadata"
 )
@@ -34,10 +35,31 @@ type Event interface {
 	Error() 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
+	Body   RawMessage        // contains message body
 }
 
 // Subscriber is a convenience return type for the Subscribe method

broker/memory_test.go

@@ -32,7 +32,7 @@ func TestMemoryBroker(t *testing.T) {
 				"foo": "bar",
 				"id":  fmt.Sprintf("%d", i),
 			},
-			Body: []byte(`hello world`),
+			Body: []byte(`"hello world"`),
 		}
 
 		if err := b.Publish(ctx, topic, message); err != nil {

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
 	}
 }
 

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
 		}
 	}

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)
+	}
+}

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}
+}

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()
 }

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
+	}
+}

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

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=

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)
 }
 

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())
+	}
 }

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)
+	}
+}

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)...)
+	}
+}

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)
 			}
 		}

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)

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)
+//	}
+//}

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)...)
+	}
+}
+*/

util/fn/fn.go

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

util/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 {

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