Merge pull request #127 from unistack-org/cover

add more cover stuff

.github/workflows/autoapprove.yml

@@ -3,6 +3,10 @@ name: "autoapprove"
 on:
   pull_request_target:
     types: [assigned, opened, synchronize, reopened]
+  workflow_run:
+    workflows: ["prbuild"]
+    types:
+      - completed
 
 permissions:
   pull-requests: write
@@ -17,4 +21,4 @@ jobs:
         if: github.actor == 'vtolstov' || github.actor == 'dependabot[bot]'
         id: approve
         with:
-          github-token: ${{ secrets.GITHUB_TOKEN }}
+          github-token: ${{ secrets.GITHUB_TOKEN }}

.github/workflows/build.yml

@@ -10,7 +10,7 @@ jobs:
     runs-on: ubuntu-latest
     steps:
     - name: setup
-      uses: actions/setup-go@v2
+      uses: actions/setup-go@v3
       with:
         go-version: 1.17
     - name: checkout

.github/workflows/codecov.yml

@@ -0,0 +1,39 @@
+name: "codecov"
+
+on:
+  workflow_run:
+    workflows: ["build"]
+    types:
+      - completed
+  push:
+    branches: [ v3 ]
+  pull_request:
+    branches: [ v3 ]
+  schedule:
+    - cron: '34 1 * * 0'
+
+jobs:
+  codecov:
+    name: codecov
+    runs-on: ubuntu-latest
+    permissions:
+      actions: read
+      contents: read
+      security-events: write
+
+    strategy:
+      fail-fast: false
+      matrix:
+        language: [ 'go' ]
+
+    steps:
+    - name: checkout
+      uses: actions/checkout@v3
+    - name: setup
+      uses: actions/setup-go@v3
+      with:
+        go-version: 1.17
+    - name: Run coverage
+      run: go test -v -race -coverprofile=coverage.out -covermode=atomic ./...
+    - name: codecov
+      uses: codecov/codecov-action@v3.1.0

.github/workflows/codeql-analysis.yml

@@ -45,12 +45,12 @@ jobs:
     - name: checkout
       uses: actions/checkout@v3
     - name: setup
-      uses: actions/setup-go@v2
+      uses: actions/setup-go@v3
       with:
         go-version: 1.17
     # Initializes the CodeQL tools for scanning.
     - name: init
-      uses: github/codeql-action/init@v1
+      uses: github/codeql-action/init@v2
       with:
         languages: ${{ matrix.language }}
         # If you wish to specify custom queries, you can do so here or in a config file.
@@ -61,7 +61,7 @@ jobs:
     # Autobuild attempts to build any compiled languages  (C/C++, C#, or Java).
     # If this step fails, then you should remove it and run the build manually (see below)
     - name: autobuild
-      uses: github/codeql-action/autobuild@v1
+      uses: github/codeql-action/autobuild@v2
 
     # ℹ️ Command-line programs to run using the OS shell.
     # 📚 https://git.io/JvXDl
@@ -75,4 +75,4 @@ jobs:
     #   make release
 
     - name: analyze
-      uses: github/codeql-action/analyze@v1
+      uses: github/codeql-action/analyze@v2

.github/workflows/dependabot-automerge.yml

@@ -15,7 +15,7 @@ jobs:
     steps:
       - name: metadata
         id: metadata
-        uses: dependabot/fetch-metadata@v1.3.0
+        uses: dependabot/fetch-metadata@v1.3.1
         with:
           github-token: "${{ secrets.TOKEN }}"
       - name: merge

.github/workflows/pr.yml

@@ -10,7 +10,7 @@ jobs:
     runs-on: ubuntu-latest
     steps:
     - name: setup
-      uses: actions/setup-go@v2
+      uses: actions/setup-go@v3
       with:
         go-version: 1.17
     - name: checkout

README.md

@@ -1,4 +1,4 @@
-# Micro [![License](https://img.shields.io/:license-apache-blue.svg)](https://opensource.org/licenses/Apache-2.0) [![Doc](https://img.shields.io/badge/go.dev-reference-007d9c?logo=go&logoColor=white&style=flat-square)](https://pkg.go.dev/github.com/unistack-org/micro/v3?tab=overview) [![Status](https://github.com/unistack-org/micro/workflows/build/badge.svg?branch=master)](https://github.com/unistack-org/micro/actions?query=workflow%3Abuild+branch%3Amaster+event%3Apush) [![Lint](https://goreportcard.com/badge/go.unistack.org/micro/v3)](https://goreportcard.com/report/go.unistack.org/micro/v3) [![Slack](https://img.shields.io/static/v1?label=micro&message=slack&color=blueviolet)](https://unistack-org.slack.com/messages/default)
+# Micro [![License](https://img.shields.io/:license-apache-blue.svg)](https://opensource.org/licenses/Apache-2.0) [![Doc](https://img.shields.io/badge/go.dev-reference-007d9c?logo=go&logoColor=white&style=flat-square)](https://pkg.go.dev/github.com/unistack-org/micro/v3?tab=overview) [![Status](https://github.com/unistack-org/micro/workflows/build/badge.svg?branch=master)](https://github.com/unistack-org/micro/actions?query=workflow%3Abuild+branch%3Amaster+event%3Apush) [![Lint](https://goreportcard.com/badge/go.unistack.org/micro/v3)](https://goreportcard.com/report/go.unistack.org/micro/v3) [![Coverage](https://codecov.io/gh/unistack-org/micro/branch/v3/graph/badge.svg?token=OZPO2LP7VS)](https://codecov.io/gh/unistack-org/micro)
 
 Micro is a standard library for microservices.
 

broker/broker.go

@@ -9,7 +9,7 @@ import (
 )
 
 // DefaultBroker default memory broker
-var DefaultBroker Broker = NewBroker()
+var DefaultBroker = NewBroker()
 
 var (
 	// ErrNotConnected returns when broker used but not connected yet

broker/context_test.go

@@ -0,0 +1,57 @@
+package broker
+
+import (
+	"context"
+	"testing"
+)
+
+func TestFromContext(t *testing.T) {
+	ctx := context.WithValue(context.TODO(), brokerKey{}, NewBroker())
+
+	c, ok := FromContext(ctx)
+	if c == nil || !ok {
+		t.Fatal("FromContext not works")
+	}
+}
+
+func TestNewContext(t *testing.T) {
+	ctx := NewContext(context.TODO(), NewBroker())
+
+	c, ok := FromContext(ctx)
+	if c == nil || !ok {
+		t.Fatal("NewContext not works")
+	}
+}
+
+func TestSetSubscribeOption(t *testing.T) {
+	type key struct{}
+	o := SetSubscribeOption(key{}, "test")
+	opts := &SubscribeOptions{}
+	o(opts)
+
+	if v, ok := opts.Context.Value(key{}).(string); !ok || v == "" {
+		t.Fatal("SetSubscribeOption not works")
+	}
+}
+
+func TestSetPublishOption(t *testing.T) {
+	type key struct{}
+	o := SetPublishOption(key{}, "test")
+	opts := &PublishOptions{}
+	o(opts)
+
+	if v, ok := opts.Context.Value(key{}).(string); !ok || v == "" {
+		t.Fatal("SetPublishOption not works")
+	}
+}
+
+func TestSetOption(t *testing.T) {
+	type key struct{}
+	o := SetOption(key{}, "test")
+	opts := &Options{}
+	o(opts)
+
+	if v, ok := opts.Context.Value(key{}).(string); !ok || v == "" {
+		t.Fatal("SetOption not works")
+	}
+}

client/backoff.go

@@ -19,7 +19,7 @@ func BackoffExp(_ context.Context, _ Request, attempts int) (time.Duration, erro
 // BackoffInterval specifies randomization interval for backoff func
 func BackoffInterval(min time.Duration, max time.Duration) BackoffFunc {
 	return func(_ context.Context, _ Request, attempts int) (time.Duration, error) {
-		td := time.Duration(time.Duration(math.Pow(float64(attempts), math.E)) * time.Millisecond * 100)
+		td := time.Duration(math.Pow(float64(attempts), math.E)) * time.Millisecond * 100
 		if td < min {
 			return min, nil
 		} else if td > max {

client/backoff_test.go

@@ -6,7 +6,7 @@ import (
 	"time"
 )
 
-func TestBackoff(t *testing.T) {
+func TestBackoffExp(t *testing.T) {
 	results := []time.Duration{
 		0 * time.Second,
 		100 * time.Millisecond,
@@ -32,3 +32,25 @@ func TestBackoff(t *testing.T) {
 		}
 	}
 }
+
+func TestBackoffInterval(t *testing.T) {
+	min := 100 * time.Millisecond
+	max := 300 * time.Millisecond
+
+	r := &testRequest{
+		service: "test",
+		method:  "test",
+	}
+
+	fn := BackoffInterval(min, max)
+	for i := 0; i < 5; i++ {
+		d, err := fn(context.TODO(), r, i)
+		if err != nil {
+			t.Fatal(err)
+		}
+
+		if d < min || d > max {
+			t.Fatalf("Expected %v < %v < %v", min, d, max)
+		}
+	}
+}

client/client.go

@@ -11,7 +11,7 @@ import (
 
 var (
 	// DefaultClient is the global default client
-	DefaultClient Client = NewClient()
+	DefaultClient = NewClient()
 	// DefaultContentType is the default content-type if not specified
 	DefaultContentType = "application/json"
 	// DefaultBackoff is the default backoff function for retries (minimum 10 millisecond and maximum 5 second)

client/client_call_options_test.go

@@ -0,0 +1,26 @@
+package client
+
+import (
+	"context"
+	"testing"
+	"time"
+)
+
+func TestNewClientCallOptions(t *testing.T) {
+	var flag bool
+	w := func(fn CallFunc) CallFunc {
+		flag = true
+		return fn
+	}
+	c := NewClientCallOptions(NewClient(),
+		WithAddress("127.0.0.1"),
+		WithCallWrapper(w),
+		WithRequestTimeout(1*time.Millisecond),
+		WithRetries(0),
+		WithBackoff(BackoffInterval(10*time.Millisecond, 100*time.Millisecond)),
+	)
+	_ = c.Call(context.TODO(), c.NewRequest("service", "endpoint", nil), nil)
+	if !flag {
+		t.Fatalf("NewClientCallOptions not works")
+	}
+}

client/context_test.go

@@ -0,0 +1,57 @@
+package client
+
+import (
+	"context"
+	"testing"
+)
+
+func TestFromContext(t *testing.T) {
+	ctx := context.WithValue(context.TODO(), clientKey{}, NewClient())
+
+	c, ok := FromContext(ctx)
+	if c == nil || !ok {
+		t.Fatal("FromContext not works")
+	}
+}
+
+func TestNewContext(t *testing.T) {
+	ctx := NewContext(context.TODO(), NewClient())
+
+	c, ok := FromContext(ctx)
+	if c == nil || !ok {
+		t.Fatal("NewContext not works")
+	}
+}
+
+func TestSetPublishOption(t *testing.T) {
+	type key struct{}
+	o := SetPublishOption(key{}, "test")
+	opts := &PublishOptions{}
+	o(opts)
+
+	if v, ok := opts.Context.Value(key{}).(string); !ok || v == "" {
+		t.Fatal("SetPublishOption not works")
+	}
+}
+
+func TestSetCallOption(t *testing.T) {
+	type key struct{}
+	o := SetCallOption(key{}, "test")
+	opts := &CallOptions{}
+	o(opts)
+
+	if v, ok := opts.Context.Value(key{}).(string); !ok || v == "" {
+		t.Fatal("SetCallOption not works")
+	}
+}
+
+func TestSetOption(t *testing.T) {
+	type key struct{}
+	o := SetOption(key{}, "test")
+	opts := &Options{}
+	o(opts)
+
+	if v, ok := opts.Context.Value(key{}).(string); !ok || v == "" {
+		t.Fatal("SetOption not works")
+	}
+}

client/noop.go

@@ -2,6 +2,8 @@ package client
 
 import (
 	"context"
+	"fmt"
+	"time"
 
 	"go.unistack.org/micro/v3/broker"
 	"go.unistack.org/micro/v3/codec"
@@ -181,6 +183,133 @@ func (n *noopClient) String() string {
 }
 
 func (n *noopClient) Call(ctx context.Context, req Request, rsp interface{}, opts ...CallOption) error {
+	// make a copy of call opts
+	callOpts := n.opts.CallOptions
+	for _, opt := range opts {
+		opt(&callOpts)
+	}
+
+	// check if we already have a deadline
+	d, ok := ctx.Deadline()
+	if !ok {
+		var cancel context.CancelFunc
+		// no deadline so we create a new one
+		ctx, cancel = context.WithTimeout(ctx, callOpts.RequestTimeout)
+		defer cancel()
+	} else {
+		// got a deadline so no need to setup context
+		// but we need to set the timeout we pass along
+		opt := WithRequestTimeout(time.Until(d))
+		opt(&callOpts)
+	}
+
+	// should we noop right here?
+	select {
+	case <-ctx.Done():
+		return errors.New("go.micro.client", fmt.Sprintf("%v", ctx.Err()), 408)
+	default:
+	}
+
+	// make copy of call method
+	hcall := n.call
+
+	// wrap the call in reverse
+	for i := len(callOpts.CallWrappers); i > 0; i-- {
+		hcall = callOpts.CallWrappers[i-1](hcall)
+	}
+
+	// use the router passed as a call option, or fallback to the rpc clients router
+	if callOpts.Router == nil {
+		callOpts.Router = n.opts.Router
+	}
+
+	if callOpts.Selector == nil {
+		callOpts.Selector = n.opts.Selector
+	}
+
+	// inject proxy address
+	// TODO: don't even bother using Lookup/Select in this case
+	if len(n.opts.Proxy) > 0 {
+		callOpts.Address = []string{n.opts.Proxy}
+	}
+
+	// lookup the route to send the reques to
+	// TODO apply any filtering here
+	routes, err := n.opts.Lookup(ctx, req, callOpts)
+	if err != nil {
+		return errors.InternalServerError("go.micro.client", err.Error())
+	}
+
+	// balance the list of nodes
+	next, err := callOpts.Selector.Select(routes)
+	if err != nil {
+		return err
+	}
+
+	// return errors.New("go.micro.client", "request timeout", 408)
+	call := func(i int) error {
+		// call backoff first. Someone may want an initial start delay
+		t, err := callOpts.Backoff(ctx, req, i)
+		if err != nil {
+			return errors.InternalServerError("go.micro.client", err.Error())
+		}
+
+		// only sleep if greater than 0
+		if t.Seconds() > 0 {
+			time.Sleep(t)
+		}
+
+		node := next()
+
+		// make the call
+		err = hcall(ctx, node, req, rsp, callOpts)
+		// record the result of the call to inform future routing decisions
+		if verr := n.opts.Selector.Record(node, err); verr != nil {
+			return verr
+		}
+
+		// try and transform the error to a go-micro error
+		if verr, ok := err.(*errors.Error); ok {
+			return verr
+		}
+
+		return err
+	}
+
+	ch := make(chan error, callOpts.Retries)
+	var gerr error
+
+	for i := 0; i <= callOpts.Retries; i++ {
+		go func() {
+			ch <- call(i)
+		}()
+
+		select {
+		case <-ctx.Done():
+			return errors.New("go.micro.client", fmt.Sprintf("%v", ctx.Err()), 408)
+		case err := <-ch:
+			// if the call succeeded lets bail early
+			if err == nil {
+				return nil
+			}
+
+			retry, rerr := callOpts.Retry(ctx, req, i, err)
+			if rerr != nil {
+				return rerr
+			}
+
+			if !retry {
+				return err
+			}
+
+			gerr = err
+		}
+	}
+
+	return gerr
+}
+
+func (n *noopClient) call(ctx context.Context, addr string, req Request, rsp interface{}, opts CallOptions) error {
 	return nil
 }
 
@@ -194,6 +323,139 @@ func (n *noopClient) NewMessage(topic string, msg interface{}, opts ...MessageOp
 }
 
 func (n *noopClient) Stream(ctx context.Context, req Request, opts ...CallOption) (Stream, error) {
+	// make a copy of call opts
+	callOpts := n.opts.CallOptions
+	for _, o := range opts {
+		o(&callOpts)
+	}
+
+	// check if we already have a deadline
+	d, ok := ctx.Deadline()
+	if !ok && callOpts.StreamTimeout > time.Duration(0) {
+		var cancel context.CancelFunc
+		// no deadline so we create a new one
+		ctx, cancel = context.WithTimeout(ctx, callOpts.StreamTimeout)
+		defer cancel()
+	} else {
+		// got a deadline so no need to setup context
+		// but we need to set the timeout we pass along
+		o := WithStreamTimeout(time.Until(d))
+		o(&callOpts)
+	}
+
+	// should we noop right here?
+	select {
+	case <-ctx.Done():
+		return nil, errors.New("go.micro.client", fmt.Sprintf("%v", ctx.Err()), 408)
+	default:
+	}
+
+	/*
+		// make copy of call method
+		hstream := h.stream
+		// wrap the call in reverse
+		for i := len(callOpts.CallWrappers); i > 0; i-- {
+			hstream = callOpts.CallWrappers[i-1](hstream)
+		}
+	*/
+
+	// use the router passed as a call option, or fallback to the rpc clients router
+	if callOpts.Router == nil {
+		callOpts.Router = n.opts.Router
+	}
+
+	if callOpts.Selector == nil {
+		callOpts.Selector = n.opts.Selector
+	}
+
+	// inject proxy address
+	// TODO: don't even bother using Lookup/Select in this case
+	if len(n.opts.Proxy) > 0 {
+		callOpts.Address = []string{n.opts.Proxy}
+	}
+
+	// lookup the route to send the reques to
+	// TODO apply any filtering here
+	routes, err := n.opts.Lookup(ctx, req, callOpts)
+	if err != nil {
+		return nil, errors.InternalServerError("go.micro.client", err.Error())
+	}
+
+	// balance the list of nodes
+	next, err := callOpts.Selector.Select(routes)
+	if err != nil {
+		return nil, err
+	}
+
+	call := func(i int) (Stream, error) {
+		// call backoff first. Someone may want an initial start delay
+		t, cerr := callOpts.Backoff(ctx, req, i)
+		if cerr != nil {
+			return nil, errors.InternalServerError("go.micro.client", cerr.Error())
+		}
+
+		// only sleep if greater than 0
+		if t.Seconds() > 0 {
+			time.Sleep(t)
+		}
+
+		node := next()
+
+		stream, cerr := n.stream(ctx, node, req, callOpts)
+
+		// record the result of the call to inform future routing decisions
+		if verr := n.opts.Selector.Record(node, cerr); verr != nil {
+			return nil, verr
+		}
+
+		// try and transform the error to a go-micro error
+		if verr, ok := cerr.(*errors.Error); ok {
+			return nil, verr
+		}
+
+		return stream, cerr
+	}
+
+	type response struct {
+		stream Stream
+		err    error
+	}
+
+	ch := make(chan response, callOpts.Retries)
+	var grr error
+
+	for i := 0; i <= callOpts.Retries; i++ {
+		go func() {
+			s, cerr := call(i)
+			ch <- response{s, cerr}
+		}()
+
+		select {
+		case <-ctx.Done():
+			return nil, errors.New("go.micro.client", fmt.Sprintf("%v", ctx.Err()), 408)
+		case rsp := <-ch:
+			// if the call succeeded lets bail early
+			if rsp.err == nil {
+				return rsp.stream, nil
+			}
+
+			retry, rerr := callOpts.Retry(ctx, req, i, err)
+			if rerr != nil {
+				return nil, rerr
+			}
+
+			if !retry {
+				return nil, rsp.err
+			}
+
+			grr = rsp.err
+		}
+	}
+
+	return nil, grr
+}
+
+func (n *noopClient) stream(ctx context.Context, addr string, req Request, opts CallOptions) (Stream, error) {
 	return &noopStream{}, nil
 }
 

client/retry_test.go

@@ -0,0 +1,70 @@
+package client
+
+import (
+	"context"
+	"fmt"
+	"testing"
+
+	"go.unistack.org/micro/v3/errors"
+)
+
+func TestRetryAlways(t *testing.T) {
+	tests := []error{
+		nil,
+		errors.InternalServerError("test", "%s", "test"),
+		fmt.Errorf("test"),
+	}
+
+	for _, e := range tests {
+		ok, er := RetryAlways(context.TODO(), nil, 1, e)
+		if !ok || er != nil {
+			t.Fatal("RetryAlways not works properly")
+		}
+	}
+}
+
+func TestRetryNever(t *testing.T) {
+	tests := []error{
+		nil,
+		errors.InternalServerError("test", "%s", "test"),
+		fmt.Errorf("test"),
+	}
+
+	for _, e := range tests {
+		ok, er := RetryNever(context.TODO(), nil, 1, e)
+		if ok || er != nil {
+			t.Fatal("RetryNever not works properly")
+		}
+	}
+}
+
+func TestRetryOnError(t *testing.T) {
+	tests := []error{
+		fmt.Errorf("test"),
+		errors.NotFound("test", "%s", "test"),
+		errors.Timeout("test", "%s", "test"),
+	}
+
+	for i, e := range tests {
+		ok, er := RetryOnError(context.TODO(), nil, 1, e)
+		if i == 2 && (!ok || er != nil) {
+			t.Fatal("RetryOnError not works properly")
+		}
+	}
+}
+
+func TestRetryOnErrors(t *testing.T) {
+	tests := []error{
+		fmt.Errorf("test"),
+		errors.NotFound("test", "%s", "test"),
+		errors.Timeout("test", "%s", "test"),
+	}
+
+	fn := RetryOnErrors(404)
+	for i, e := range tests {
+		ok, er := fn(context.TODO(), nil, 1, e)
+		if i == 1 && (!ok || er != nil) {
+			t.Fatal("RetryOnErrors not works properly")
+		}
+	}
+}

codec/context_test.go

@@ -0,0 +1,35 @@
+package codec
+
+import (
+	"context"
+	"testing"
+)
+
+func TestFromContext(t *testing.T) {
+	ctx := context.WithValue(context.TODO(), codecKey{}, NewCodec())
+
+	c, ok := FromContext(ctx)
+	if c == nil || !ok {
+		t.Fatal("FromContext not works")
+	}
+}
+
+func TestNewContext(t *testing.T) {
+	ctx := NewContext(context.TODO(), NewCodec())
+
+	c, ok := FromContext(ctx)
+	if c == nil || !ok {
+		t.Fatal("NewContext not works")
+	}
+}
+
+func TestSetOption(t *testing.T) {
+	type key struct{}
+	o := SetOption(key{}, "test")
+	opts := &Options{}
+	o(opts)
+
+	if v, ok := opts.Context.Value(key{}).(string); !ok || v == "" {
+		t.Fatal("SetOption not works")
+	}
+}

config/context_test.go

@@ -0,0 +1,68 @@
+package config
+
+import (
+	"context"
+	"testing"
+)
+
+func TestFromContext(t *testing.T) {
+	ctx := context.WithValue(context.TODO(), configKey{}, NewConfig())
+
+	c, ok := FromContext(ctx)
+	if c == nil || !ok {
+		t.Fatal("FromContext not works")
+	}
+}
+
+func TestNewContext(t *testing.T) {
+	ctx := NewContext(context.TODO(), NewConfig())
+
+	c, ok := FromContext(ctx)
+	if c == nil || !ok {
+		t.Fatal("NewContext not works")
+	}
+}
+
+func TestSetOption(t *testing.T) {
+	type key struct{}
+	o := SetOption(key{}, "test")
+	opts := &Options{}
+	o(opts)
+
+	if v, ok := opts.Context.Value(key{}).(string); !ok || v == "" {
+		t.Fatal("SetOption not works")
+	}
+}
+
+func TestSetSaveOption(t *testing.T) {
+	type key struct{}
+	o := SetSaveOption(key{}, "test")
+	opts := &SaveOptions{}
+	o(opts)
+
+	if v, ok := opts.Context.Value(key{}).(string); !ok || v == "" {
+		t.Fatal("SetSaveOption not works")
+	}
+}
+
+func TestSetLoadOption(t *testing.T) {
+	type key struct{}
+	o := SetLoadOption(key{}, "test")
+	opts := &LoadOptions{}
+	o(opts)
+
+	if v, ok := opts.Context.Value(key{}).(string); !ok || v == "" {
+		t.Fatal("SetLoadOption not works")
+	}
+}
+
+func TestSetWatchOption(t *testing.T) {
+	type key struct{}
+	o := SetWatchOption(key{}, "test")
+	opts := &WatchOptions{}
+	o(opts)
+
+	if v, ok := opts.Context.Value(key{}).(string); !ok || v == "" {
+		t.Fatal("SetWatchOption not works")
+	}
+}

context_test.go

@@ -0,0 +1,24 @@
+package micro
+
+import (
+	"context"
+	"testing"
+)
+
+func TestFromContext(t *testing.T) {
+	ctx := context.WithValue(context.TODO(), serviceKey{}, NewService())
+
+	c, ok := FromContext(ctx)
+	if c == nil || !ok {
+		t.Fatal("FromContext not works")
+	}
+}
+
+func TestNewContext(t *testing.T) {
+	ctx := NewContext(context.TODO(), NewService())
+
+	c, ok := FromContext(ctx)
+	if c == nil || !ok {
+		t.Fatal("NewContext not works")
+	}
+}

errors/errors.go

@@ -233,6 +233,27 @@ func Equal(err1 error, err2 error) bool {
 	return true
 }
 
+// CodeIn return true if err has specified code
+func CodeIn(err interface{}, codes ...int32) bool {
+	var code int32
+	switch verr := err.(type) {
+	case *Error:
+		code = verr.Code
+	case int32:
+		code = verr
+	default:
+		return false
+	}
+
+	for _, check := range codes {
+		if code == check {
+			return true
+		}
+	}
+
+	return false
+}
+
 // FromError try to convert go error to *Error
 func FromError(err error) *Error {
 	if verr, ok := err.(*Error); ok && verr != nil {

errors/errors_test.go

@@ -96,3 +96,19 @@ func TestErrors(t *testing.T) {
 		}
 	}
 }
+
+func TestCodeIn(t *testing.T) {
+	err := InternalServerError("id", "%s", "msg")
+
+	if ok := CodeIn(err, 400, 500); !ok {
+		t.Fatalf("CodeIn not works: %v", err)
+	}
+
+	if ok := CodeIn(err.(*Error).Code, 500); !ok {
+		t.Fatalf("CodeIn not works: %v", err)
+	}
+
+	if ok := CodeIn(err, 100); ok {
+		t.Fatalf("CodeIn not works: %v", err)
+	}
+}

flow/context_test.go

@@ -0,0 +1,35 @@
+package flow
+
+import (
+	"context"
+	"testing"
+)
+
+func TestFromContext(t *testing.T) {
+	ctx := context.WithValue(context.TODO(), flowKey{}, NewFlow())
+
+	c, ok := FromContext(ctx)
+	if c == nil || !ok {
+		t.Fatal("FromContext not works")
+	}
+}
+
+func TestNewContext(t *testing.T) {
+	ctx := NewContext(context.TODO(), NewFlow())
+
+	c, ok := FromContext(ctx)
+	if c == nil || !ok {
+		t.Fatal("NewContext not works")
+	}
+}
+
+func TestSetOption(t *testing.T) {
+	type key struct{}
+	o := SetOption(key{}, "test")
+	opts := &Options{}
+	o(opts)
+
+	if v, ok := opts.Context.Value(key{}).(string); !ok || v == "" {
+		t.Fatal("SetOption not works")
+	}
+}

fsm/fsm.go

@@ -0,0 +1,181 @@
+package fsm // import "go.unistack.org/micro/v3/fsm"
+
+import (
+	"context"
+	"errors"
+	"fmt"
+	"sync"
+)
+
+var (
+	ErrInvalidState = errors.New("does not exists")
+	StateEnd        = "end"
+)
+
+// Options struct holding fsm options
+type Options struct {
+	// DryRun mode
+	DryRun bool
+	// Initial state
+	Initial string
+	// HooksBefore func slice runs in order before state
+	HooksBefore []HookBeforeFunc
+	// HooksAfter func slice runs in order after state
+	HooksAfter []HookAfterFunc
+}
+
+// HookBeforeFunc func signature
+type HookBeforeFunc func(ctx context.Context, state string, args interface{})
+
+// HookAfterFunc func signature
+type HookAfterFunc func(ctx context.Context, state string, args interface{})
+
+// Option func signature
+type Option func(*Options)
+
+// StateOptions holds state options
+type StateOptions struct {
+	DryRun bool
+}
+
+// StateDryRun says that state executes in dry run mode
+func StateDryRun(b bool) StateOption {
+	return func(o *StateOptions) {
+		o.DryRun = b
+	}
+}
+
+// StateOption func signature
+type StateOption func(*StateOptions)
+
+// InitialState sets init state for state machine
+func InitialState(initial string) Option {
+	return func(o *Options) {
+		o.Initial = initial
+	}
+}
+
+// HookBefore provides hook func slice
+func HookBefore(fns ...HookBeforeFunc) Option {
+	return func(o *Options) {
+		o.HooksBefore = fns
+	}
+}
+
+// HookAfter provides hook func slice
+func HookAfter(fns ...HookAfterFunc) Option {
+	return func(o *Options) {
+		o.HooksAfter = fns
+	}
+}
+
+// StateFunc called on state transition and return next step and error
+type StateFunc func(ctx context.Context, args interface{}, opts ...StateOption) (string, interface{}, error)
+
+// FSM is a finite state machine
+type FSM struct {
+	mu          sync.Mutex
+	statesMap   map[string]StateFunc
+	statesOrder []string
+	opts        *Options
+	current     string
+}
+
+// New creates a new finite state machine having the specified initial state
+// with specified options
+func New(opts ...Option) *FSM {
+	options := &Options{}
+
+	for _, opt := range opts {
+		opt(options)
+	}
+
+	return &FSM{
+		statesMap: map[string]StateFunc{},
+		opts:      options,
+	}
+}
+
+// Current returns the current state
+func (f *FSM) Current() string {
+	f.mu.Lock()
+	defer f.mu.Unlock()
+	return f.current
+}
+
+// Current returns the current state
+func (f *FSM) Reset() {
+	f.mu.Lock()
+	f.current = f.opts.Initial
+	f.mu.Unlock()
+}
+
+// State adds state to fsm
+func (f *FSM) State(state string, fn StateFunc) {
+	f.mu.Lock()
+	f.statesMap[state] = fn
+	f.statesOrder = append(f.statesOrder, state)
+	f.mu.Unlock()
+}
+
+// Init initialize fsm and check states
+
+// Start runs state machine with provided data
+func (f *FSM) Start(ctx context.Context, args interface{}, opts ...Option) (interface{}, error) {
+	var err error
+	var ok bool
+	var fn StateFunc
+	var nstate string
+
+	f.mu.Lock()
+	options := f.opts
+
+	for _, opt := range opts {
+		opt(options)
+	}
+
+	sopts := []StateOption{StateDryRun(options.DryRun)}
+
+	cstate := options.Initial
+	states := make(map[string]StateFunc, len(f.statesMap))
+	for k, v := range f.statesMap {
+		states[k] = v
+	}
+	f.current = cstate
+	f.mu.Unlock()
+
+	for {
+		select {
+		case <-ctx.Done():
+			return nil, ctx.Err()
+		default:
+			fn, ok = states[cstate]
+			if !ok {
+				return nil, fmt.Errorf(`state "%s" %w`, cstate, ErrInvalidState)
+			}
+			f.mu.Lock()
+			f.current = cstate
+			f.mu.Unlock()
+			for _, fn := range options.HooksBefore {
+				fn(ctx, cstate, args)
+			}
+			nstate, args, err = fn(ctx, args, sopts...)
+			for _, fn := range options.HooksAfter {
+				fn(ctx, cstate, args)
+			}
+			switch {
+			case err != nil:
+				return args, err
+			case nstate == StateEnd:
+				return args, nil
+			case nstate == "":
+				for idx := range f.statesOrder {
+					if f.statesOrder[idx] == cstate && len(f.statesOrder) > idx+1 {
+						nstate = f.statesOrder[idx+1]
+					}
+				}
+			}
+			cstate = nstate
+		}
+	}
+}

fsm/fsm_test.go

@@ -0,0 +1,63 @@
+package fsm
+
+import (
+	"bytes"
+	"context"
+	"fmt"
+	"testing"
+)
+
+func TestFSMStart(t *testing.T) {
+	ctx := context.TODO()
+	buf := bytes.NewBuffer(nil)
+	pfb := func(_ context.Context, state string, _ interface{}) {
+		fmt.Fprintf(buf, "before state %s\n", state)
+	}
+	pfa := func(_ context.Context, state string, _ interface{}) {
+		fmt.Fprintf(buf, "after state %s\n", state)
+	}
+	f := New(InitialState("1"), HookBefore(pfb), HookAfter(pfa))
+	f1 := func(_ context.Context, req interface{}, _ ...StateOption) (string, interface{}, error) {
+		args := req.(map[string]interface{})
+		if v, ok := args["request"].(string); !ok || v == "" {
+			return "", nil, fmt.Errorf("empty request")
+		}
+		return "2", map[string]interface{}{"response": "test2"}, nil
+	}
+	f2 := func(_ context.Context, req interface{}, _ ...StateOption) (string, interface{}, error) {
+		args := req.(map[string]interface{})
+		if v, ok := args["response"].(string); !ok || v == "" {
+			return "", nil, fmt.Errorf("empty response")
+		}
+		return "", map[string]interface{}{"response": "test"}, nil
+	}
+	f3 := func(_ context.Context, req interface{}, _ ...StateOption) (string, interface{}, error) {
+		args := req.(map[string]interface{})
+		if v, ok := args["response"].(string); !ok || v == "" {
+			return "", nil, fmt.Errorf("empty response")
+		}
+		return StateEnd, map[string]interface{}{"response": "test_last"}, nil
+	}
+	f.State("1", f1)
+	f.State("2", f2)
+	f.State("3", f3)
+	rsp, err := f.Start(ctx, map[string]interface{}{"request": "test1"})
+	if err != nil {
+		t.Fatal(err)
+	}
+	args := rsp.(map[string]interface{})
+	if v, ok := args["response"].(string); !ok || v == "" {
+		t.Fatalf("nil rsp: %#+v", args)
+	} else if v != "test_last" {
+		t.Fatalf("invalid rsp %#+v", args)
+	}
+
+	if !bytes.Contains(buf.Bytes(), []byte(`before state 1`)) ||
+		!bytes.Contains(buf.Bytes(), []byte(`before state 2`)) ||
+		!bytes.Contains(buf.Bytes(), []byte(`after state 1`)) ||
+		!bytes.Contains(buf.Bytes(), []byte(`after state 2`)) ||
+		!bytes.Contains(buf.Bytes(), []byte(`after state 3`)) ||
+		!bytes.Contains(buf.Bytes(), []byte(`after state 3`)) {
+		t.Fatalf("fsm not works properly or hooks error, buf: %s", buf.Bytes())
+	}
+}

go.mod

@@ -4,10 +4,16 @@ go 1.16
 
 require (
 	github.com/ef-ds/deque v1.0.4
-	github.com/golang-jwt/jwt/v4 v4.4.0
+	github.com/golang-jwt/jwt/v4 v4.4.1
+	github.com/google/gnostic v0.6.8 // indirect
+	github.com/google/go-cmp v0.5.7 // indirect
 	github.com/imdario/mergo v0.3.12
+	github.com/kr/pretty v0.2.1 // indirect
+	github.com/kr/text v0.2.0 // indirect
 	github.com/patrickmn/go-cache v2.1.0+incompatible
 	github.com/silas/dag v0.0.0-20211117232152-9d50aa809f35
 	go.unistack.org/micro-proto/v3 v3.2.7
-	golang.org/x/net v0.0.0-20210928044308-7d9f5e0b762b
+	golang.org/x/net v0.0.0-20220225172249-27dd8689420f
+	google.golang.org/protobuf v1.28.0 // indirect
+	gopkg.in/yaml.v2 v2.4.0 // indirect
 )

go.sum

@@ -10,6 +10,7 @@ github.com/client9/misspell v0.3.4/go.mod h1:qj6jICC3Q7zFZvVWo7KLAzC3yx5G7kyvSDk
 github.com/cncf/udpa/go v0.0.0-20191209042840-269d4d468f6f/go.mod h1:M8M6+tZqaGXZJjfX53e64911xZQV5JYwmTeXPW+k8Sc=
 github.com/cncf/udpa/go v0.0.0-20201120205902-5459f2c99403/go.mod h1:WmhPx2Nbnhtbo57+VJT5O0JRkEi1Wbu0z5j0R8u5Hbk=
 github.com/cncf/xds/go v0.0.0-20210312221358-fbca930ec8ed/go.mod h1:eXthEFrGJvWHgFFCl3hGmgk+/aYT6PnTQLykKQRLhEs=
+github.com/creack/pty v1.1.9/go.mod h1:oKZEueFk5CKHvIhNR5MUki03XCEU+Q6VDXinZuGJ33E=
 github.com/davecgh/go-spew v1.1.0/go.mod h1:J7Y8YcW2NihsgmVo/mv3lAwl/skON4iLHjSsI+c5H38=
 github.com/davecgh/go-spew v1.1.1/go.mod h1:J7Y8YcW2NihsgmVo/mv3lAwl/skON4iLHjSsI+c5H38=
 github.com/docopt/docopt-go v0.0.0-20180111231733-ee0de3bc6815/go.mod h1:WwZ+bS3ebgob9U8Nd0kOddGdZWjyMGR8Wziv+TBNwSE=
@@ -23,8 +24,8 @@ github.com/envoyproxy/go-control-plane v0.9.9-0.20210512163311-63b5d3c536b0/go.m
 github.com/envoyproxy/protoc-gen-validate v0.1.0/go.mod h1:iSmxcyjqTsJpI2R4NaDN7+kN2VEUnK/pcBlmesArF7c=
 github.com/flowstack/go-jsonschema v0.1.1/go.mod h1:yL7fNggx1o8rm9RlgXv7hTBWxdBM0rVwpMwimd3F3N0=
 github.com/ghodss/yaml v1.0.0/go.mod h1:4dBDuWmgqj2HViK6kFavaiC9ZROes6MMH2rRYeMEF04=
-github.com/golang-jwt/jwt/v4 v4.4.0 h1:EmVIxB5jzbllGIjiCV5JG4VylbK3KE400tLGLI1cdfU=
-github.com/golang-jwt/jwt/v4 v4.4.0/go.mod h1:/xlHOz8bRuivTWchD4jCa+NbatV+wEUSzwAxVc6locg=
+github.com/golang-jwt/jwt/v4 v4.4.1 h1:pC5DB52sCeK48Wlb9oPcdhnjkz1TKt1D/P7WKJ0kUcQ=
+github.com/golang-jwt/jwt/v4 v4.4.1/go.mod h1:m21LjoU+eqJr34lmDMbreY2eSTRJ1cv77w39/MY0Ch0=
 github.com/golang/glog v0.0.0-20160126235308-23def4e6c14b/go.mod h1:SBH7ygxi8pfUlaOkMMuAQtPIUF8ecWP5IEl/CR7VP2Q=
 github.com/golang/mock v1.1.1/go.mod h1:oTYuIxOrZwtPieC+H1uAHpcLFnEyAGVDL/k47Jfbm0A=
 github.com/golang/protobuf v1.2.0/go.mod h1:6lQm79b+lXiMfvg/cZm0SGofjICqVBUtrP5yJMmIC1U=
@@ -41,24 +42,28 @@ github.com/golang/protobuf v1.4.3/go.mod h1:oDoupMAO8OvCJWAcko0GGGIgR6R6ocIYbsSw
 github.com/golang/protobuf v1.5.0/go.mod h1:FsONVRAS9T7sI+LIUmWTfcYkHO4aIWwzhcaSAoJOfIk=
 github.com/golang/protobuf v1.5.2 h1:ROPKBNFfQgOUMifHyP+KYbvpjbdoFNs+aK7DXlji0Tw=
 github.com/golang/protobuf v1.5.2/go.mod h1:XVQd3VNwM+JqD3oG2Ue2ip4fOMUkwXdXDdiuN0vRsmY=
-github.com/google/gnostic v0.6.6 h1:MVSM2r2j9aRUvYNym66JGW96Ddd5MN4sTi59yktb6yk=
 github.com/google/gnostic v0.6.6/go.mod h1:Nm8234We1lq6iB9OmlgNv3nH91XLLVZHCDayfA3xq+E=
+github.com/google/gnostic v0.6.8 h1:bT56GPYBWh1tvBuBEd94qcS3+60b+y0HQur0ITkGuCk=
+github.com/google/gnostic v0.6.8/go.mod h1:Nm8234We1lq6iB9OmlgNv3nH91XLLVZHCDayfA3xq+E=
 github.com/google/go-cmp v0.2.0/go.mod h1:oXzfMopK8JAjlY9xF4vHSVASa0yLyX7SntLO5aqRK0M=
 github.com/google/go-cmp v0.3.0/go.mod h1:8QqcDgzrUqlUb/G2PQTWiueGozuR1884gddMywk6iLU=
 github.com/google/go-cmp v0.3.1/go.mod h1:8QqcDgzrUqlUb/G2PQTWiueGozuR1884gddMywk6iLU=
 github.com/google/go-cmp v0.4.0/go.mod h1:v8dTdLbMG2kIc/vJvl+f65V22dbkXbowE6jgT/gNBxE=
 github.com/google/go-cmp v0.5.0/go.mod h1:v8dTdLbMG2kIc/vJvl+f65V22dbkXbowE6jgT/gNBxE=
-github.com/google/go-cmp v0.5.5 h1:Khx7svrCpmxxtHBq5j2mp/xVjsi8hQMfNLvJFAlrGgU=
 github.com/google/go-cmp v0.5.5/go.mod h1:v8dTdLbMG2kIc/vJvl+f65V22dbkXbowE6jgT/gNBxE=
+github.com/google/go-cmp v0.5.7 h1:81/ik6ipDQS2aGcBfIN5dHDB36BwrStyeAQquSYCV4o=
+github.com/google/go-cmp v0.5.7/go.mod h1:n+brtR0CgQNWTVd5ZUFpTBC8YFBDLK/h/bpaJ8/DtOE=
 github.com/google/uuid v1.1.2/go.mod h1:TIyPZe4MgqvfeYDBFedMoGGpEw/LqOeaOT+nhxU+yHo=
 github.com/grpc-ecosystem/grpc-gateway v1.16.0/go.mod h1:BDjrQk3hbvj6Nolgz8mAMFbcEtjT1g+wF4CSlocrBnw=
 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/kr/pretty v0.2.0 h1:s5hAObm+yFO5uHYt5dYjxi2rXrsnmRpJx4OYvIWUaQs=
 github.com/kr/pretty v0.2.0/go.mod h1:ipq/a2n7PKx3OHsz4KJII5eveXtPO4qwEXGdVfWzfnI=
+github.com/kr/pretty v0.2.1 h1:Fmg33tUaq4/8ym9TJN1x7sLJnHVwhP33CNkpYV/7rwI=
+github.com/kr/pretty v0.2.1/go.mod h1:ipq/a2n7PKx3OHsz4KJII5eveXtPO4qwEXGdVfWzfnI=
 github.com/kr/pty v1.1.1/go.mod h1:pFQYn66WHrOpPYNljwOMqo10TkYh1fy3cYio2l3bCsQ=
-github.com/kr/text v0.1.0 h1:45sCR5RtlFHMR4UwH9sdQ5TC8v0qDQCHnXt+kaKSTVE=
 github.com/kr/text v0.1.0/go.mod h1:4Jbv+DJW3UT/LiOwJeYQe1efqtUx/iVham/4vfdArNI=
+github.com/kr/text v0.2.0 h1:5Nx0Ya0ZqY2ygV366QzturHI13Jq95ApcVaJBhpS+AY=
+github.com/kr/text v0.2.0/go.mod h1:eLer722TekiGuMkidMxC/pM04lWEeraHUUmBw8l2grE=
 github.com/patrickmn/go-cache v2.1.0+incompatible h1:HRMgzkcYKYpi3C8ajMPV8OFXaaRUnok+kx1WdO15EQc=
 github.com/patrickmn/go-cache v2.1.0+incompatible/go.mod h1:3Qf8kWWT7OJRJbdiICTKqZju1ZixQ/KpMGzzAfe6+WQ=
 github.com/pmezard/go-difflib v1.0.0/go.mod h1:iKH77koFhYxTK1pcRnkKkqfTogsbg7gZNVY4sRDYZ/4=
@@ -92,8 +97,8 @@ golang.org/x/net v0.0.0-20190404232315-eb5bcb51f2a3/go.mod h1:t9HGtf8HONx5eT2rtn
 golang.org/x/net v0.0.0-20200822124328-c89045814202/go.mod h1:/O7V0waA8r7cgGh81Ro3o1hOxt32SMVPicZroKQ2sZA=
 golang.org/x/net v0.0.0-20210405180319-a5a99cb37ef4/go.mod h1:p54w0d4576C0XHj96bSt6lcn1PtDYWL6XObtHCRCNQM=
 golang.org/x/net v0.0.0-20210805182204-aaa1db679c0d/go.mod h1:9nx3DQGgdP8bBQD5qxJ1jj9UTztislL4KSBs9R2vV5Y=
-golang.org/x/net v0.0.0-20210928044308-7d9f5e0b762b h1:eB48h3HiRycXNy8E0Gf5e0hv7YT6Kt14L/D73G1fuwo=
-golang.org/x/net v0.0.0-20210928044308-7d9f5e0b762b/go.mod h1:9nx3DQGgdP8bBQD5qxJ1jj9UTztislL4KSBs9R2vV5Y=
+golang.org/x/net v0.0.0-20220225172249-27dd8689420f h1:oA4XRj0qtSt8Yo1Zms0CUlsT3KG69V2UGQWPBxujDmc=
+golang.org/x/net v0.0.0-20220225172249-27dd8689420f/go.mod h1:CfG3xpIq0wQ8r1q4Su4UZFWDARRcnwPjda9FqA0JpMk=
 golang.org/x/oauth2 v0.0.0-20180821212333-d2e6202438be/go.mod h1:N/0e6XlmueqKjAGxoOufVs8QHGRruUQn6yWY3a++T0U=
 golang.org/x/oauth2 v0.0.0-20200107190931-bf48bf16ab8d/go.mod h1:gOpvHmFTYa4IltrdGE7lF6nIHvwfUNPOp7c8zoXwtLw=
 golang.org/x/sync v0.0.0-20180314180146-1d60e4601c6f/go.mod h1:RxMgew5VJxzue5/jJTE5uejpjVlOe/izrB70Jof72aM=
@@ -108,11 +113,15 @@ golang.org/x/sys v0.0.0-20201119102817-f84b799fce68/go.mod h1:h1NjWce9XRLGQEsW7w
 golang.org/x/sys v0.0.0-20210330210617-4fbd30eecc44/go.mod h1:h1NjWce9XRLGQEsW7wpKNCjG9DtNlClVuFLEZdDNbEs=
 golang.org/x/sys v0.0.0-20210423082822-04245dca01da/go.mod h1:h1NjWce9XRLGQEsW7wpKNCjG9DtNlClVuFLEZdDNbEs=
 golang.org/x/sys v0.0.0-20210510120138-977fb7262007/go.mod h1:oPkhp1MJrh7nUepCBck5+mAzfO9JrbApNNgaTdGDITg=
+golang.org/x/sys v0.0.0-20210615035016-665e8c7367d1/go.mod h1:oPkhp1MJrh7nUepCBck5+mAzfO9JrbApNNgaTdGDITg=
+golang.org/x/sys v0.0.0-20211216021012-1d35b9e2eb4e/go.mod h1:oPkhp1MJrh7nUepCBck5+mAzfO9JrbApNNgaTdGDITg=
 golang.org/x/term v0.0.0-20201126162022-7de9c90e9dd1/go.mod h1:bj7SfCRtBDWHUb9snDiAeCFNEtKQo2Wmx5Cou7ajbmo=
+golang.org/x/term v0.0.0-20210927222741-03fcf44c2211/go.mod h1:jbD1KX2456YbFQfuXm/mYQcufACuNUgVhRMnK/tPxf8=
 golang.org/x/text v0.3.0/go.mod h1:NqM8EUOU14njkJ3fqMW+pc6Ldnwhi/IjpwHt7yyuwOQ=
 golang.org/x/text v0.3.3/go.mod h1:5Zoc/QRtKVWzQhOtBMvqHzDpF6irO9z98xDceosuGiQ=
 golang.org/x/text v0.3.5/go.mod h1:5Zoc/QRtKVWzQhOtBMvqHzDpF6irO9z98xDceosuGiQ=
 golang.org/x/text v0.3.6/go.mod h1:5Zoc/QRtKVWzQhOtBMvqHzDpF6irO9z98xDceosuGiQ=
+golang.org/x/text v0.3.7/go.mod h1:u+2+/6zg+i71rQMx5EYifcz6MCKuco9NR6JIITiCfzQ=
 golang.org/x/tools v0.0.0-20180917221912-90fa682c2a6e/go.mod h1:n7NCudcB/nEzxVGmLbDWY5pfWTLqBcC2KZ6jyYvM4mQ=
 golang.org/x/tools v0.0.0-20190114222345-bf090417da8b/go.mod h1:n7NCudcB/nEzxVGmLbDWY5pfWTLqBcC2KZ6jyYvM4mQ=
 golang.org/x/tools v0.0.0-20190226205152-f727befe758c/go.mod h1:9Yl7xja0Znq3iFh3HoIrodX9oNMXvdceNzlUR8zjMvY=
@@ -146,15 +155,17 @@ google.golang.org/protobuf v1.23.1-0.20200526195155-81db48ad09cc/go.mod h1:EGpAD
 google.golang.org/protobuf v1.25.0/go.mod h1:9JNX74DMeImyA3h4bdi1ymwjUzf21/xIlbajtzgsN7c=
 google.golang.org/protobuf v1.26.0-rc.1/go.mod h1:jlhhOSvTdKEhbULTjvd4ARK9grFBp09yW+WbY/TyQbw=
 google.golang.org/protobuf v1.26.0/go.mod h1:9q0QmTI4eRPtz6boOQmLYwt+qCgq0jsYwAQnmE0givc=
-google.golang.org/protobuf v1.27.1 h1:SnqbnDw1V7RiZcXPx5MEeqPv2s79L9i7BJUlG/+RurQ=
 google.golang.org/protobuf v1.27.1/go.mod h1:9q0QmTI4eRPtz6boOQmLYwt+qCgq0jsYwAQnmE0givc=
+google.golang.org/protobuf v1.28.0 h1:w43yiav+6bVFTBQFZX0r7ipe9JQ1QsbMgHwbBziscLw=
+google.golang.org/protobuf v1.28.0/go.mod h1:HV8QOd/L58Z+nl8r43ehVNZIU/HEI6OcFqwMG9pJV4I=
 gopkg.in/check.v1 v0.0.0-20161208181325-20d25e280405/go.mod h1:Co6ibVJAznAaIkqp8huTwlJQCZ016jof/cbN4VW5Yz0=
 gopkg.in/check.v1 v1.0.0-20190902080502-41f04d3bba15 h1:YR8cESwS4TdDjEe65xsg0ogRM/Nc3DYOhEAlW+xobZo=
 gopkg.in/check.v1 v1.0.0-20190902080502-41f04d3bba15/go.mod h1:Co6ibVJAznAaIkqp8huTwlJQCZ016jof/cbN4VW5Yz0=
 gopkg.in/yaml.v2 v2.2.2/go.mod h1:hI93XBmqTisBFMUTm0b8Fm+jr3Dg1NNxqwp+5A1VGuI=
 gopkg.in/yaml.v2 v2.2.3/go.mod h1:hI93XBmqTisBFMUTm0b8Fm+jr3Dg1NNxqwp+5A1VGuI=
-gopkg.in/yaml.v2 v2.3.0 h1:clyUAQHOM3G0M3f5vQj7LuJrETvjVot3Z5el9nffUtU=
 gopkg.in/yaml.v2 v2.3.0/go.mod h1:hI93XBmqTisBFMUTm0b8Fm+jr3Dg1NNxqwp+5A1VGuI=
+gopkg.in/yaml.v2 v2.4.0 h1:D8xgwECY7CYvx+Y2n4sBz93Jn9JRvxdiyyo8CTfuKaY=
+gopkg.in/yaml.v2 v2.4.0/go.mod h1:RDklbk79AGWmwhnvt/jBztapEOGDOx6ZbXqjP6csGnQ=
 gopkg.in/yaml.v3 v3.0.0-20200615113413-eeeca48fe776/go.mod h1:K4uyk7z7BCEPqu6E+C64Yfv1cQ7kz7rIZviUmN+EgEM=
 gopkg.in/yaml.v3 v3.0.0-20210107192922-496545a6307b h1:h8qDotaEPuJATrMmW04NCwg7v22aHH28wwpauUhK9Oo=
 gopkg.in/yaml.v3 v3.0.0-20210107192922-496545a6307b/go.mod h1:K4uyk7z7BCEPqu6E+C64Yfv1cQ7kz7rIZviUmN+EgEM=

logger/context_test.go

@@ -0,0 +1,35 @@
+package logger
+
+import (
+	"context"
+	"testing"
+)
+
+func TestFromContext(t *testing.T) {
+	ctx := context.WithValue(context.TODO(), loggerKey{}, NewLogger())
+
+	c, ok := FromContext(ctx)
+	if c == nil || !ok {
+		t.Fatal("FromContext not works")
+	}
+}
+
+func TestNewContext(t *testing.T) {
+	ctx := NewContext(context.TODO(), NewLogger())
+
+	c, ok := FromContext(ctx)
+	if c == nil || !ok {
+		t.Fatal("NewContext not works")
+	}
+}
+
+func TestSetOption(t *testing.T) {
+	type key struct{}
+	o := SetOption(key{}, "test")
+	opts := &Options{}
+	o(opts)
+
+	if v, ok := opts.Context.Value(key{}).(string); !ok || v == "" {
+		t.Fatal("SetOption not works")
+	}
+}

metadata/context_test.go

@@ -0,0 +1,122 @@
+package metadata
+
+import (
+	"context"
+	"testing"
+)
+
+func TestFromContext(t *testing.T) {
+	ctx := context.WithValue(context.TODO(), mdKey{}, &rawMetadata{New(0)})
+
+	c, ok := FromContext(ctx)
+	if c == nil || !ok {
+		t.Fatal("FromContext not works")
+	}
+}
+
+func TestNewContext(t *testing.T) {
+	ctx := NewContext(context.TODO(), New(0))
+
+	c, ok := FromContext(ctx)
+	if c == nil || !ok {
+		t.Fatal("NewContext not works")
+	}
+}
+
+func TestFromIncomingContext(t *testing.T) {
+	ctx := context.WithValue(context.TODO(), mdIncomingKey{}, &rawMetadata{New(0)})
+
+	c, ok := FromIncomingContext(ctx)
+	if c == nil || !ok {
+		t.Fatal("FromIncomingContext not works")
+	}
+}
+
+func TestFromOutgoingContext(t *testing.T) {
+	ctx := context.WithValue(context.TODO(), mdOutgoingKey{}, &rawMetadata{New(0)})
+
+	c, ok := FromOutgoingContext(ctx)
+	if c == nil || !ok {
+		t.Fatal("FromOutgoingContext not works")
+	}
+}
+
+func TestSetIncomingContext(t *testing.T) {
+	md := New(1)
+	md.Set("key", "val")
+	ctx := context.WithValue(context.TODO(), mdIncomingKey{}, &rawMetadata{})
+	if !SetIncomingContext(ctx, md) {
+		t.Fatal("SetIncomingContext not works")
+	}
+	md, ok := FromIncomingContext(ctx)
+	if md == nil || !ok {
+		t.Fatal("SetIncomingContext not works")
+	} else if v, ok := md.Get("key"); !ok || v != "val" {
+		t.Fatal("SetIncomingContext not works")
+	}
+}
+
+func TestSetOutgoingContext(t *testing.T) {
+	md := New(1)
+	md.Set("key", "val")
+	ctx := context.WithValue(context.TODO(), mdOutgoingKey{}, &rawMetadata{})
+	if !SetOutgoingContext(ctx, md) {
+		t.Fatal("SetOutgoingContext not works")
+	}
+	md, ok := FromOutgoingContext(ctx)
+	if md == nil || !ok {
+		t.Fatal("SetOutgoingContext not works")
+	} else if v, ok := md.Get("key"); !ok || v != "val" {
+		t.Fatal("SetOutgoingContext not works")
+	}
+}
+
+func TestNewIncomingContext(t *testing.T) {
+	md := New(1)
+	md.Set("key", "val")
+	ctx := NewIncomingContext(context.TODO(), md)
+
+	c, ok := FromIncomingContext(ctx)
+	if c == nil || !ok {
+		t.Fatal("NewIncomingContext not works")
+	}
+}
+
+func TestNewOutgoingContext(t *testing.T) {
+	md := New(1)
+	md.Set("key", "val")
+	ctx := NewOutgoingContext(context.TODO(), md)
+
+	c, ok := FromOutgoingContext(ctx)
+	if c == nil || !ok {
+		t.Fatal("NewOutgoingContext not works")
+	}
+}
+
+func TestAppendIncomingContext(t *testing.T) {
+	md := New(1)
+	md.Set("key1", "val1")
+	ctx := AppendIncomingContext(context.TODO(), "key2", "val2")
+
+	nmd, ok := FromIncomingContext(ctx)
+	if nmd == nil || !ok {
+		t.Fatal("AppendIncomingContext not works")
+	}
+	if v, ok := nmd.Get("key2"); !ok || v != "val2" {
+		t.Fatal("AppendIncomingContext not works")
+	}
+}
+
+func TestAppendOutgoingContext(t *testing.T) {
+	md := New(1)
+	md.Set("key1", "val1")
+	ctx := AppendOutgoingContext(context.TODO(), "key2", "val2")
+
+	nmd, ok := FromOutgoingContext(ctx)
+	if nmd == nil || !ok {
+		t.Fatal("AppendOutgoingContext not works")
+	}
+	if v, ok := nmd.Get("key2"); !ok || v != "val2" {
+		t.Fatal("AppendOutgoingContext not works")
+	}
+}

meter/context_test.go

@@ -0,0 +1,35 @@
+package meter
+
+import (
+	"context"
+	"testing"
+)
+
+func TestFromContext(t *testing.T) {
+	ctx := context.WithValue(context.TODO(), meterKey{}, NewMeter())
+
+	c, ok := FromContext(ctx)
+	if c == nil || !ok {
+		t.Fatal("FromContext not works")
+	}
+}
+
+func TestNewContext(t *testing.T) {
+	ctx := NewContext(context.TODO(), NewMeter())
+
+	c, ok := FromContext(ctx)
+	if c == nil || !ok {
+		t.Fatal("NewContext not works")
+	}
+}
+
+func TestSetOption(t *testing.T) {
+	type key struct{}
+	o := SetOption(key{}, "test")
+	opts := &Options{}
+	o(opts)
+
+	if v, ok := opts.Context.Value(key{}).(string); !ok || v == "" {
+		t.Fatal("SetOption not works")
+	}
+}

register/context_test.go

@@ -0,0 +1,35 @@
+package register
+
+import (
+	"context"
+	"testing"
+)
+
+func TestFromContext(t *testing.T) {
+	ctx := context.WithValue(context.TODO(), registerKey{}, NewRegister())
+
+	c, ok := FromContext(ctx)
+	if c == nil || !ok {
+		t.Fatal("FromContext not works")
+	}
+}
+
+func TestNewContext(t *testing.T) {
+	ctx := NewContext(context.TODO(), NewRegister())
+
+	c, ok := FromContext(ctx)
+	if c == nil || !ok {
+		t.Fatal("NewContext not works")
+	}
+}
+
+func TestSetOption(t *testing.T) {
+	type key struct{}
+	o := SetOption(key{}, "test")
+	opts := &Options{}
+	o(opts)
+
+	if v, ok := opts.Context.Value(key{}).(string); !ok || v == "" {
+		t.Fatal("SetOption not works")
+	}
+}

register/memory.go

@@ -29,10 +29,10 @@ type record struct {
 }
 
 type memory struct {
+	sync.RWMutex
 	records  map[string]services
 	watchers map[string]*watcher
 	opts     Options
-	sync.RWMutex
 }
 
 // services is a KV map with service name as the key and a map of records as the value

resolver/dns/dns.go

@@ -12,10 +12,9 @@ import (
 
 // Resolver is a DNS network resolve
 type Resolver struct {
-	goresolver *net.Resolver
-	// Address of resolver to use
-	Address string
 	sync.RWMutex
+	goresolver *net.Resolver
+	Address    string
 }
 
 // Resolve tries to resolve endpoint address
@@ -47,7 +46,7 @@ func (r *Resolver) Resolve(name string) ([]*resolver.Record, error) {
 	if goresolver == nil {
 		r.Lock()
 		r.goresolver = &net.Resolver{
-			Dial: func(ctx context.Context, network, address string) (net.Conn, error) {
+			Dial: func(ctx context.Context, _ string, _ string) (net.Conn, error) {
 				d := net.Dialer{
 					Timeout: time.Millisecond * time.Duration(100),
 				}

router/context.go

@@ -0,0 +1,34 @@
+package router
+
+import (
+	"context"
+)
+
+type routerKey struct{}
+
+// FromContext get router from context
+func FromContext(ctx context.Context) (Router, bool) {
+	if ctx == nil {
+		return nil, false
+	}
+	c, ok := ctx.Value(routerKey{}).(Router)
+	return c, ok
+}
+
+// NewContext put router in context
+func NewContext(ctx context.Context, c Router) context.Context {
+	if ctx == nil {
+		ctx = context.Background()
+	}
+	return context.WithValue(ctx, routerKey{}, c)
+}
+
+// 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)
+	}
+}

router/context_test.go

@@ -0,0 +1,35 @@
+package router
+
+import (
+	"context"
+	"testing"
+)
+
+func TestFromContext(t *testing.T) {
+	ctx := context.WithValue(context.TODO(), routerKey{}, NewRouter())
+
+	c, ok := FromContext(ctx)
+	if c == nil || !ok {
+		t.Fatal("FromContext not works")
+	}
+}
+
+func TestNewContext(t *testing.T) {
+	ctx := NewContext(context.TODO(), NewRouter())
+
+	c, ok := FromContext(ctx)
+	if c == nil || !ok {
+		t.Fatal("NewContext not works")
+	}
+}
+
+func TestSetOption(t *testing.T) {
+	type key struct{}
+	o := SetOption(key{}, "test")
+	opts := &Options{}
+	o(opts)
+
+	if v, ok := opts.Context.Value(key{}).(string); !ok || v == "" {
+		t.Fatal("SetOption not works")
+	}
+}

server/context_test.go

@@ -0,0 +1,46 @@
+package server
+
+import (
+	"context"
+	"testing"
+)
+
+func TestFromContext(t *testing.T) {
+	ctx := context.WithValue(context.TODO(), serverKey{}, NewServer())
+
+	c, ok := FromContext(ctx)
+	if c == nil || !ok {
+		t.Fatal("FromContext not works")
+	}
+}
+
+func TestNewContext(t *testing.T) {
+	ctx := NewContext(context.TODO(), NewServer())
+
+	c, ok := FromContext(ctx)
+	if c == nil || !ok {
+		t.Fatal("NewContext not works")
+	}
+}
+
+func TestSetOption(t *testing.T) {
+	type key struct{}
+	o := SetOption(key{}, "test")
+	opts := &Options{}
+	o(opts)
+
+	if v, ok := opts.Context.Value(key{}).(string); !ok || v == "" {
+		t.Fatal("SetOption not works")
+	}
+}
+
+func TestSetSubscriberOption(t *testing.T) {
+	type key struct{}
+	o := SetSubscriberOption(key{}, "test")
+	opts := &SubscriberOptions{}
+	o(opts)
+
+	if v, ok := opts.Context.Value(key{}).(string); !ok || v == "" {
+		t.Fatal("SetSubscriberOption not works")
+	}
+}

server/noop.go

@@ -277,23 +277,24 @@ func (n *noopServer) Deregister() error {
 
 	wg := sync.WaitGroup{}
 	for sb, subs := range n.subscribers {
-		for _, sub := range subs {
+		for idx := range subs {
 			if sb.Options().Context != nil {
 				cx = sb.Options().Context
 			}
 
+			ncx := cx
 			wg.Add(1)
 			go func(s broker.Subscriber) {
 				defer wg.Done()
 				if config.Logger.V(logger.InfoLevel) {
 					config.Logger.Infof(n.opts.Context, "unsubscribing from topic: %s", s.Topic())
 				}
-				if err := s.Unsubscribe(cx); err != nil {
+				if err := s.Unsubscribe(ncx); err != nil {
 					if config.Logger.V(logger.ErrorLevel) {
 						config.Logger.Errorf(n.opts.Context, "unsubscribing from topic: %s err: %v", s.Topic(), err)
 					}
 				}
-			}(sub)
+			}(subs[idx])
 		}
 		n.subscribers[sb] = nil
 	}

server/noop_test.go

@@ -8,6 +8,7 @@ import (
 	"go.unistack.org/micro/v3/broker"
 	"go.unistack.org/micro/v3/client"
 	"go.unistack.org/micro/v3/codec"
+	"go.unistack.org/micro/v3/logger"
 	"go.unistack.org/micro/v3/metadata"
 	"go.unistack.org/micro/v3/server"
 )
@@ -50,6 +51,7 @@ func TestNoopSub(t *testing.T) {
 		t.Fatal(err)
 	}
 
+	logger.DefaultLogger.Init(logger.WithLevel(logger.ErrorLevel))
 	s := server.NewServer(
 		server.Broker(b),
 		server.Codec("application/octet-stream", codec.NewCodec()),

server/server.go

@@ -11,7 +11,7 @@ import (
 )
 
 // DefaultServer default server
-var DefaultServer Server = NewServer()
+var DefaultServer = NewServer()
 
 var (
 	// DefaultAddress will be used if no address passed, use secure localhost

server/subscriber.go

@@ -36,11 +36,11 @@ type handler struct {
 type subscriber struct {
 	typ        reflect.Type
 	subscriber interface{}
-	rcvr       reflect.Value
 	topic      string
 	endpoints  []*register.Endpoint
 	handlers   []*handler
 	opts       SubscriberOptions
+	rcvr       reflect.Value
 }
 
 // Is this an exported - upper case - name?

service.go

@@ -58,6 +58,8 @@ type Service interface {
 	Run() error
 	// Start the service
 	Start() error
+	// Stop the service
+	Stop() error
 	// The service implementation
 	String() string
 }
@@ -73,9 +75,8 @@ func RegisterSubscriber(topic string, s server.Server, h interface{}, opts ...se
 }
 
 type service struct {
-	opts Options
 	sync.RWMutex
-	//	once sync.Once
+	opts Options
 }
 
 // NewService creates and returns a new Service based on the packages within.
@@ -108,11 +109,6 @@ func (s *service) Init(opts ...Option) error {
 		if err = cfg.Init(config.Context(cfg.Options().Context)); err != nil {
 			return err
 		}
-
-		if err = cfg.Load(cfg.Options().Context); err != nil {
-			return err
-		}
-
 	}
 
 	for _, log := range s.opts.Loggers {
@@ -247,7 +243,7 @@ func (s *service) Meter(names ...string) meter.Meter {
 }
 
 func (s *service) String() string {
-	return "micro"
+	return s.opts.Name
 }
 
 //nolint:gocyclo
@@ -258,16 +254,6 @@ func (s *service) Start() error {
 	config := s.opts
 	s.RUnlock()
 
-	if config.Loggers[0].V(logger.InfoLevel) {
-		config.Loggers[0].Infof(s.opts.Context, "starting [service] %s version %s", s.Options().Name, s.Options().Version)
-	}
-
-	for _, fn := range s.opts.BeforeStart {
-		if err = fn(s.opts.Context); err != nil {
-			return err
-		}
-	}
-
 	for _, cfg := range s.opts.Configs {
 		if cfg.Options().Struct == nil {
 			// skip config as the struct not passed
@@ -279,6 +265,16 @@ func (s *service) Start() error {
 		}
 	}
 
+	for _, fn := range s.opts.BeforeStart {
+		if err = fn(s.opts.Context); err != nil {
+			return err
+		}
+	}
+
+	if config.Loggers[0].V(logger.InfoLevel) {
+		config.Loggers[0].Infof(s.opts.Context, "starting [service] %s version %s", s.Options().Name, s.Options().Version)
+	}
+
 	if len(s.opts.Servers) == 0 {
 		return fmt.Errorf("cant start nil server")
 	}

service_test.go

@@ -1,7 +1,21 @@
 package micro
 
 import (
+	"context"
+	"reflect"
 	"testing"
+
+	"go.unistack.org/micro/v3/auth"
+	"go.unistack.org/micro/v3/broker"
+	"go.unistack.org/micro/v3/client"
+	"go.unistack.org/micro/v3/config"
+	"go.unistack.org/micro/v3/logger"
+	"go.unistack.org/micro/v3/meter"
+	"go.unistack.org/micro/v3/register"
+	"go.unistack.org/micro/v3/router"
+	"go.unistack.org/micro/v3/server"
+	"go.unistack.org/micro/v3/store"
+	"go.unistack.org/micro/v3/tracer"
 )
 
 type testItem struct {
@@ -20,3 +34,743 @@ func TestGetNameIndex(t *testing.T) {
 		t.Fatalf("getNameIndex func error, item not found")
 	}
 }
+
+func TestRegisterHandler(t *testing.T) {
+	type args struct {
+		s    server.Server
+		h    interface{}
+		opts []server.HandlerOption
+	}
+	h := struct{}{}
+	tests := []struct {
+		name    string
+		args    args
+		wantErr bool
+	}{
+		{
+			name: "RegisterHandler",
+			args: args{
+				s:    server.DefaultServer,
+				h:    h,
+				opts: nil,
+			},
+			wantErr: false,
+		},
+	}
+	for _, tt := range tests {
+		t.Run(tt.name, func(t *testing.T) {
+			if err := RegisterHandler(tt.args.s, tt.args.h, tt.args.opts...); (err != nil) != tt.wantErr {
+				t.Errorf("RegisterHandler() error = %v, wantErr %v", err, tt.wantErr)
+			}
+		})
+	}
+}
+
+func TestRegisterSubscriber(t *testing.T) {
+	type args struct {
+		topic string
+		s     server.Server
+		h     interface{}
+		opts  []server.SubscriberOption
+	}
+	h := func(_ context.Context, _ interface{}) error {
+		return nil
+	}
+	tests := []struct {
+		name    string
+		args    args
+		wantErr bool
+	}{
+		{
+			name: "RegisterSubscriber",
+			args: args{
+				topic: "test",
+				s:     server.DefaultServer,
+				h:     h,
+				opts:  nil,
+			},
+			wantErr: false,
+		},
+	}
+	for _, tt := range tests {
+		t.Run(tt.name, func(t *testing.T) {
+			if err := RegisterSubscriber(tt.args.topic, tt.args.s, tt.args.h, tt.args.opts...); (err != nil) != tt.wantErr {
+				t.Errorf("RegisterSubscriber() error = %v, wantErr %v", err, tt.wantErr)
+			}
+		})
+	}
+}
+
+func TestNewService(t *testing.T) {
+	type args struct {
+		opts []Option
+	}
+	tests := []struct {
+		name string
+		args args
+		want Service
+	}{
+		{
+			name: "NewService",
+			args: args{
+				opts: []Option{Name("test")},
+			},
+			want: NewService(Name("test")),
+		},
+	}
+	for _, tt := range tests {
+		t.Run(tt.name, func(t *testing.T) {
+			if got := NewService(tt.args.opts...); !reflect.DeepEqual(got, tt.want) {
+				t.Errorf("NewService() = %v, want %v", got.Options().Name, tt.want.Options().Name)
+			}
+		})
+	}
+}
+
+func Test_service_Name(t *testing.T) {
+	type fields struct {
+		opts Options
+	}
+	tests := []struct {
+		name   string
+		fields fields
+		want   string
+	}{
+		{
+			name: "Test_service_Name",
+			fields: fields{
+				opts: Options{Name: "test"},
+			},
+			want: "test",
+		},
+	}
+	for _, tt := range tests {
+		t.Run(tt.name, func(t *testing.T) {
+			s := &service{
+				opts: tt.fields.opts,
+			}
+			if got := s.Name(); got != tt.want {
+				t.Errorf("service.Name() = %v, want %v", got, tt.want)
+			}
+		})
+	}
+}
+
+func Test_service_Init(t *testing.T) {
+	type fields struct {
+		opts Options
+	}
+	type args struct {
+		opts []Option
+	}
+	tests := []struct {
+		name    string
+		fields  fields
+		args    args
+		wantErr bool
+	}{
+		{
+			name: "service.Init()",
+			fields: fields{
+				opts: Options{},
+			},
+			args: args{
+				opts: []Option{},
+			},
+			wantErr: false,
+		},
+	}
+	for _, tt := range tests {
+		t.Run(tt.name, func(t *testing.T) {
+			s := &service{
+				opts: tt.fields.opts,
+			}
+			if err := s.Init(tt.args.opts...); (err != nil) != tt.wantErr {
+				t.Errorf("service.Init() error = %v, wantErr %v", err, tt.wantErr)
+			}
+		})
+	}
+}
+
+func Test_service_Options(t *testing.T) {
+	opts := Options{Name: "test"}
+	type fields struct {
+		opts Options
+	}
+	tests := []struct {
+		name   string
+		fields fields
+		want   Options
+	}{
+		{
+			name: "service.Options",
+			fields: fields{
+				opts: opts,
+			},
+			want: opts,
+		},
+	}
+	for _, tt := range tests {
+		t.Run(tt.name, func(t *testing.T) {
+			s := &service{
+				opts: tt.fields.opts,
+			}
+			if got := s.Options(); !reflect.DeepEqual(got, tt.want) {
+				t.Errorf("service.Options() = %v, want %v", got, tt.want)
+			}
+		})
+	}
+}
+
+func Test_service_Broker(t *testing.T) {
+	b := broker.NewBroker()
+	type fields struct {
+		opts Options
+	}
+	type args struct {
+		names []string
+	}
+	tests := []struct {
+		name   string
+		fields fields
+		args   args
+		want   broker.Broker
+	}{
+		{
+			name: "service.Broker",
+			fields: fields{
+				opts: Options{Brokers: []broker.Broker{b}},
+			},
+			args: args{
+				names: []string{"noop"},
+			},
+			want: b,
+		},
+	}
+	for _, tt := range tests {
+		t.Run(tt.name, func(t *testing.T) {
+			s := &service{
+				opts: tt.fields.opts,
+			}
+			if got := s.Broker(tt.args.names...); !reflect.DeepEqual(got, tt.want) {
+				t.Errorf("service.Broker() = %v, want %v", got, tt.want)
+			}
+		})
+	}
+}
+
+/*
+func TestServiceBroker(t *testing.T) {
+	b := broker.NewBroker(broker.Name("test"))
+
+	srv := server.NewServer()
+
+	svc := NewService(Server(srv),Broker(b))
+
+	if err := svc.Init(); err != nil {
+		t.Fatalf("failed to init service")
+	}
+
+	if brk := svc.Server().Options().Broker; brk.Name() != "test" {
+		t.Fatalf("server broker not set: %v", svc.Server().Options().Broker)
+	}
+
+}
+*/
+
+func Test_service_Tracer(t *testing.T) {
+	tr := tracer.NewTracer()
+	type fields struct {
+		opts Options
+	}
+	type args struct {
+		names []string
+	}
+	tests := []struct {
+		name   string
+		fields fields
+		args   args
+		want   tracer.Tracer
+	}{
+		{
+			name: "service.Tracer",
+			fields: fields{
+				opts: Options{Tracers: []tracer.Tracer{tr}},
+			},
+			args: args{
+				names: []string{"noop"},
+			},
+			want: tr,
+		},
+	}
+	for _, tt := range tests {
+		t.Run(tt.name, func(t *testing.T) {
+			s := &service{
+				opts: tt.fields.opts,
+			}
+			if got := s.Tracer(tt.args.names...); !reflect.DeepEqual(got, tt.want) {
+				t.Errorf("service.Tracer() = %v, want %v", got, tt.want)
+			}
+		})
+	}
+}
+
+func Test_service_Config(t *testing.T) {
+	c := config.NewConfig()
+	type fields struct {
+		opts Options
+	}
+	type args struct {
+		names []string
+	}
+	tests := []struct {
+		name   string
+		fields fields
+		args   args
+		want   config.Config
+	}{
+		{
+			name: "service.Config",
+			fields: fields{
+				opts: Options{Configs: []config.Config{c}},
+			},
+			args: args{
+				names: []string{"noop"},
+			},
+			want: c,
+		},
+	}
+	for _, tt := range tests {
+		t.Run(tt.name, func(t *testing.T) {
+			s := &service{
+				opts: tt.fields.opts,
+			}
+			if got := s.Config(tt.args.names...); !reflect.DeepEqual(got, tt.want) {
+				t.Errorf("service.Config() = %v, want %v", got, tt.want)
+			}
+		})
+	}
+}
+
+func Test_service_Client(t *testing.T) {
+	c := client.NewClient()
+	type fields struct {
+		opts Options
+	}
+	type args struct {
+		names []string
+	}
+	tests := []struct {
+		name   string
+		fields fields
+		args   args
+		want   client.Client
+	}{
+		{
+			name: "service.Client",
+			fields: fields{
+				opts: Options{Clients: []client.Client{c}},
+			},
+			args: args{
+				names: []string{"noop"},
+			},
+			want: c,
+		},
+	}
+	for _, tt := range tests {
+		t.Run(tt.name, func(t *testing.T) {
+			s := &service{
+				opts: tt.fields.opts,
+			}
+			if got := s.Client(tt.args.names...); !reflect.DeepEqual(got, tt.want) {
+				t.Errorf("service.Client() = %v, want %v", got, tt.want)
+			}
+		})
+	}
+}
+
+func Test_service_Server(t *testing.T) {
+	s := server.NewServer()
+	type fields struct {
+		opts Options
+	}
+	type args struct {
+		names []string
+	}
+	tests := []struct {
+		name   string
+		fields fields
+		args   args
+		want   server.Server
+	}{
+		{
+			name: "service.Server",
+			fields: fields{
+				opts: Options{Servers: []server.Server{s}},
+			},
+			args: args{
+				names: []string{"noop"},
+			},
+			want: s,
+		},
+	}
+	for _, tt := range tests {
+		t.Run(tt.name, func(t *testing.T) {
+			s := &service{
+				opts: tt.fields.opts,
+			}
+			if got := s.Server(tt.args.names...); !reflect.DeepEqual(got, tt.want) {
+				t.Errorf("service.Server() = %v, want %v", got, tt.want)
+			}
+		})
+	}
+}
+
+func Test_service_Store(t *testing.T) {
+	s := store.NewStore()
+	type fields struct {
+		opts Options
+	}
+	type args struct {
+		names []string
+	}
+	tests := []struct {
+		name   string
+		fields fields
+		args   args
+		want   store.Store
+	}{
+		{
+			name: "service.Store",
+			fields: fields{
+				opts: Options{Stores: []store.Store{s}},
+			},
+			args: args{
+				names: []string{"noop"},
+			},
+			want: s,
+		},
+	}
+	for _, tt := range tests {
+		t.Run(tt.name, func(t *testing.T) {
+			s := &service{
+				opts: tt.fields.opts,
+			}
+			if got := s.Store(tt.args.names...); !reflect.DeepEqual(got, tt.want) {
+				t.Errorf("service.Store() = %v, want %v", got, tt.want)
+			}
+		})
+	}
+}
+
+func Test_service_Register(t *testing.T) {
+	r := register.NewRegister()
+	type fields struct {
+		opts Options
+	}
+	type args struct {
+		names []string
+	}
+	tests := []struct {
+		name   string
+		fields fields
+		args   args
+		want   register.Register
+	}{
+		{
+			name: "service.Register",
+			fields: fields{
+				opts: Options{Registers: []register.Register{r}},
+			},
+			args: args{
+				names: []string{"noop"},
+			},
+			want: r,
+		},
+	}
+	for _, tt := range tests {
+		t.Run(tt.name, func(t *testing.T) {
+			s := &service{
+				opts: tt.fields.opts,
+			}
+			if got := s.Register(tt.args.names...); !reflect.DeepEqual(got, tt.want) {
+				t.Errorf("service.Register() = %v, want %v", got, tt.want)
+			}
+		})
+	}
+}
+
+func Test_service_Logger(t *testing.T) {
+	l := logger.NewLogger()
+	type fields struct {
+		opts Options
+	}
+	type args struct {
+		names []string
+	}
+	tests := []struct {
+		name   string
+		fields fields
+		args   args
+		want   logger.Logger
+	}{
+		{
+			name: "service.Logger",
+			fields: fields{
+				opts: Options{Loggers: []logger.Logger{l}},
+			},
+			args: args{
+				names: []string{"noop"},
+			},
+			want: l,
+		},
+	}
+	for _, tt := range tests {
+		t.Run(tt.name, func(t *testing.T) {
+			s := &service{
+				opts: tt.fields.opts,
+			}
+			if got := s.Logger(tt.args.names...); !reflect.DeepEqual(got, tt.want) {
+				t.Errorf("service.Logger() = %v, want %v", got, tt.want)
+			}
+		})
+	}
+}
+
+func Test_service_Auth(t *testing.T) {
+	a := auth.NewAuth()
+	type fields struct {
+		opts Options
+	}
+	type args struct {
+		names []string
+	}
+	tests := []struct {
+		name   string
+		fields fields
+		args   args
+		want   auth.Auth
+	}{
+		{
+			name: "service.Auth",
+			fields: fields{
+				opts: Options{Auths: []auth.Auth{a}},
+			},
+			args: args{
+				names: []string{"noop"},
+			},
+			want: a,
+		},
+	}
+	for _, tt := range tests {
+		t.Run(tt.name, func(t *testing.T) {
+			s := &service{
+				opts: tt.fields.opts,
+			}
+			if got := s.Auth(tt.args.names...); !reflect.DeepEqual(got, tt.want) {
+				t.Errorf("service.Auth() = %v, want %v", got, tt.want)
+			}
+		})
+	}
+}
+
+func Test_service_Router(t *testing.T) {
+	r := router.NewRouter()
+	type fields struct {
+		opts Options
+	}
+	type args struct {
+		names []string
+	}
+	tests := []struct {
+		name   string
+		fields fields
+		args   args
+		want   router.Router
+	}{
+		{
+			name: "service.Router",
+			fields: fields{
+				opts: Options{Routers: []router.Router{r}},
+			},
+			args: args{
+				names: []string{"noop"},
+			},
+			want: r,
+		},
+	}
+	for _, tt := range tests {
+		t.Run(tt.name, func(t *testing.T) {
+			s := &service{
+				opts: tt.fields.opts,
+			}
+			if got := s.Router(tt.args.names...); !reflect.DeepEqual(got, tt.want) {
+				t.Errorf("service.Router() = %v, want %v", got, tt.want)
+			}
+		})
+	}
+}
+
+func Test_service_Meter(t *testing.T) {
+	m := meter.NewMeter()
+	type fields struct {
+		opts Options
+	}
+	type args struct {
+		names []string
+	}
+	tests := []struct {
+		name   string
+		fields fields
+		args   args
+		want   meter.Meter
+	}{
+		{
+			name: "service.Meter",
+			fields: fields{
+				opts: Options{Meters: []meter.Meter{m}},
+			},
+			args: args{
+				names: []string{"noop"},
+			},
+			want: m,
+		},
+	}
+	for _, tt := range tests {
+		t.Run(tt.name, func(t *testing.T) {
+			s := &service{
+				opts: tt.fields.opts,
+			}
+			if got := s.Meter(tt.args.names...); !reflect.DeepEqual(got, tt.want) {
+				t.Errorf("service.Meter() = %v, want %v", got, tt.want)
+			}
+		})
+	}
+}
+
+func Test_service_String(t *testing.T) {
+	type fields struct {
+		opts Options
+	}
+	tests := []struct {
+		name   string
+		fields fields
+		want   string
+	}{
+		{
+			name: "service.String",
+			fields: fields{
+				opts: Options{Name: "noop"},
+			},
+			want: "noop",
+		},
+	}
+	for _, tt := range tests {
+		t.Run(tt.name, func(t *testing.T) {
+			s := &service{
+				opts: tt.fields.opts,
+			}
+			if got := s.String(); got != tt.want {
+				t.Errorf("service.String() = %v, want %v", got, tt.want)
+			}
+		})
+	}
+}
+
+/*
+func Test_service_Start(t *testing.T) {
+	type fields struct {
+		RWMutex sync.RWMutex
+		opts    Options
+	}
+	tests := []struct {
+		name    string
+		fields  fields
+		wantErr bool
+	}{
+		// TODO: Add test cases.
+	}
+	for _, tt := range tests {
+		t.Run(tt.name, func(t *testing.T) {
+			s := &service{
+				RWMutex: tt.fields.RWMutex,
+				opts:    tt.fields.opts,
+			}
+			if err := s.Start(); (err != nil) != tt.wantErr {
+				t.Errorf("service.Start() error = %v, wantErr %v", err, tt.wantErr)
+			}
+		})
+	}
+}
+
+func Test_service_Stop(t *testing.T) {
+	type fields struct {
+		RWMutex sync.RWMutex
+		opts    Options
+	}
+	tests := []struct {
+		name    string
+		fields  fields
+		wantErr bool
+	}{
+		// TODO: Add test cases.
+	}
+	for _, tt := range tests {
+		t.Run(tt.name, func(t *testing.T) {
+			s := &service{
+				RWMutex: tt.fields.RWMutex,
+				opts:    tt.fields.opts,
+			}
+			if err := s.Stop(); (err != nil) != tt.wantErr {
+				t.Errorf("service.Stop() error = %v, wantErr %v", err, tt.wantErr)
+			}
+		})
+	}
+}
+
+func Test_service_Run(t *testing.T) {
+	type fields struct {
+		RWMutex sync.RWMutex
+		opts    Options
+	}
+	tests := []struct {
+		name    string
+		fields  fields
+		wantErr bool
+	}{
+		// TODO: Add test cases.
+	}
+	for _, tt := range tests {
+		t.Run(tt.name, func(t *testing.T) {
+			s := &service{
+				RWMutex: tt.fields.RWMutex,
+				opts:    tt.fields.opts,
+			}
+			if err := s.Run(); (err != nil) != tt.wantErr {
+				t.Errorf("service.Run() error = %v, wantErr %v", err, tt.wantErr)
+			}
+		})
+	}
+}
+
+func Test_getNameIndex(t *testing.T) {
+	type args struct {
+		n      string
+		ifaces interface{}
+	}
+	tests := []struct {
+		name string
+		args args
+		want int
+	}{
+		// TODO: Add test cases.
+	}
+	for _, tt := range tests {
+		t.Run(tt.name, func(t *testing.T) {
+			if got := getNameIndex(tt.args.n, tt.args.ifaces); got != tt.want {
+				t.Errorf("getNameIndex() = %v, want %v", got, tt.want)
+			}
+		})
+	}
+}
+*/

store/context_test.go

@@ -0,0 +1,35 @@
+package store
+
+import (
+	"context"
+	"testing"
+)
+
+func TestFromContext(t *testing.T) {
+	ctx := context.WithValue(context.TODO(), storeKey{}, NewStore())
+
+	c, ok := FromContext(ctx)
+	if c == nil || !ok {
+		t.Fatal("FromContext not works")
+	}
+}
+
+func TestNewContext(t *testing.T) {
+	ctx := NewContext(context.TODO(), NewStore())
+
+	c, ok := FromContext(ctx)
+	if c == nil || !ok {
+		t.Fatal("NewContext not works")
+	}
+}
+
+func TestSetOption(t *testing.T) {
+	type key struct{}
+	o := SetOption(key{}, "test")
+	opts := &Options{}
+	o(opts)
+
+	if v, ok := opts.Context.Value(key{}).(string); !ok || v == "" {
+		t.Fatal("SetOption not works")
+	}
+}

sync/memory.go

@@ -6,9 +6,9 @@ import (
 )
 
 type memorySync struct {
+	mtx     gosync.RWMutex
 	locks   map[string]*memoryLock
 	options Options
-	mtx     gosync.RWMutex
 }
 
 type memoryLock struct {

tracer/context_test.go

@@ -0,0 +1,24 @@
+package tracer
+
+import (
+	"context"
+	"testing"
+)
+
+func TestFromContext(t *testing.T) {
+	ctx := context.WithValue(context.TODO(), tracerKey{}, NewTracer())
+
+	c, ok := FromContext(ctx)
+	if c == nil || !ok {
+		t.Fatal("FromContext not works")
+	}
+}
+
+func TestNewContext(t *testing.T) {
+	ctx := NewContext(context.TODO(), NewTracer())
+
+	c, ok := FromContext(ctx)
+	if c == nil || !ok {
+		t.Fatal("NewContext not works")
+	}
+}

util/jitter/random.go

@@ -13,3 +13,8 @@ func Random(d time.Duration) time.Duration {
 	v := rng.Float64() * float64(d.Nanoseconds())
 	return time.Duration(v)
 }
+
+func RandomInterval(min, max time.Duration) time.Duration {
+	var rng rand.Rand
+	return time.Duration(rng.Int63n(max.Nanoseconds()-min.Nanoseconds())+min.Nanoseconds()) * time.Nanosecond
+}

util/reflect/path_test.go

@@ -4,7 +4,7 @@ import (
 	"testing"
 )
 
-func TestPath(t *testing.T) {
+func TestLookup(t *testing.T) {
 	type Nested2 struct {
 		Name string
 	}

util/reflect/reflect_test.go

@@ -0,0 +1,135 @@
+package reflect
+
+import (
+	"testing"
+)
+
+func TestFieldName(t *testing.T) {
+	src := "SomeVar"
+	chk := "some_var"
+	dst := FieldName(src)
+	if dst != chk {
+		t.Fatalf("FieldName error %s != %s", src, chk)
+	}
+}
+
+func TestMergeBool(t *testing.T) {
+	type str struct {
+		Bool bool `json:"bool"`
+	}
+
+	mp := make(map[string]interface{})
+	mp["bool"] = "true"
+	s := &str{}
+
+	if err := Merge(s, mp, Tags([]string{"json"})); err != nil {
+		t.Fatal(err)
+	}
+
+	if !s.Bool {
+		t.Fatalf("merge bool error: %#+v\n", s)
+	}
+
+	mp["bool"] = "false"
+
+	if err := Merge(s, mp, Tags([]string{"json"})); err != nil {
+		t.Fatal(err)
+	}
+
+	if s.Bool {
+		t.Fatalf("merge bool error: %#+v\n", s)
+	}
+
+	mp["bool"] = 1
+
+	if err := Merge(s, mp, Tags([]string{"json"})); err != nil {
+		t.Fatal(err)
+	}
+
+	if !s.Bool {
+		t.Fatalf("merge bool error: %#+v\n", s)
+	}
+}
+
+func TestMergeString(t *testing.T) {
+	type str struct {
+		Bool string `json:"bool"`
+	}
+
+	mp := make(map[string]interface{})
+	mp["bool"] = true
+	s := &str{}
+
+	if err := Merge(s, mp, Tags([]string{"json"})); err != nil {
+		t.Fatalf("merge with true err: %v", err)
+	}
+
+	if s.Bool != "true" {
+		t.Fatalf("merge bool error: %#+v\n", s)
+	}
+
+	mp["bool"] = false
+	if err := Merge(s, mp, Tags([]string{"json"})); err != nil {
+		t.Fatalf("merge with falst err: %v", err)
+	}
+
+	if s.Bool != "false" {
+		t.Fatalf("merge bool error: %#+v\n", s)
+	}
+}
+
+func TestMergeNested(t *testing.T) {
+	type CallReqNested struct {
+		StringArgs []string       `json:"string_args"`
+		Uint64Args []uint64       `json:"uint64_args"`
+		Nested     *CallReqNested `json:"nested2"`
+	}
+
+	type CallReq struct {
+		Name   string         `json:"name"`
+		Req    string         `json:"req"`
+		Arg2   int            `json:"arg2"`
+		Nested *CallReqNested `json:"nested"`
+	}
+
+	dst := &CallReq{
+		Name: "name_old",
+		Req:  "req_old",
+	}
+
+	mp := make(map[string]interface{})
+	mp["name"] = "name_new"
+	mp["req"] = "req_new"
+	mp["arg2"] = 1
+	mp["nested.string_args"] = []string{"args1", "args2"}
+	mp["nested.uint64_args"] = []uint64{1, 2, 3}
+	mp["nested.nested2.uint64_args"] = []uint64{1, 2, 3}
+
+	mp = FlattenMap(mp)
+
+	if err := Merge(dst, mp, Tags([]string{"json"})); err != nil {
+		t.Fatal(err)
+	}
+
+	if dst.Name != "name_new" || dst.Req != "req_new" || dst.Arg2 != 1 {
+		t.Fatalf("merge error: %#+v", dst)
+	}
+
+	if dst.Nested == nil || len(dst.Nested.Uint64Args) != 3 ||
+		len(dst.Nested.StringArgs) != 2 || dst.Nested.StringArgs[0] != "args1" ||
+		len(dst.Nested.Uint64Args) != 3 || dst.Nested.Uint64Args[2] != 3 {
+		t.Fatalf("merge error: %#+v", dst.Nested)
+	}
+
+	nmp := make(map[string]interface{})
+	nmp["nested.uint64_args"] = []uint64{4}
+	nmp = FlattenMap(nmp)
+
+	if err := Merge(dst, nmp, SliceAppend(true), Tags([]string{"json"})); err != nil {
+		t.Fatal(err)
+	}
+
+	if dst.Nested == nil || len(dst.Nested.Uint64Args) != 4 || dst.Nested.Uint64Args[3] != 4 {
+		t.Fatalf("merge error: %#+v", dst.Nested)
+	}
+}

util/reflect/struct_test.go

@@ -9,7 +9,34 @@ import (
 	rutil "go.unistack.org/micro/v3/util/reflect"
 )
 
-func TestStructfields(t *testing.T) {
+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 := rutil.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 TestStructFieldsNested(t *testing.T) {
 	type NestedConfig struct {
 		Value string
 	}
@@ -130,34 +157,7 @@ func TestStructFieldsMap(t *testing.T) {
 	}
 }
 
-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 := rutil.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) {
+func TestStructFieldByPath(t *testing.T) {
 	type NestedStr struct {
 		BBB string
 		CCC int
@@ -178,7 +178,7 @@ func TestStructByPath(t *testing.T) {
 	}
 }
 
-func TestStructByTag(t *testing.T) {
+func TestStructFieldByTag(t *testing.T) {
 	type Str struct {
 		Name []string `json:"name" codec:"flatten"`
 	}
@@ -197,7 +197,7 @@ func TestStructByTag(t *testing.T) {
 	}
 }
 
-func TestStructByName(t *testing.T) {
+func TestStructFieldByName(t *testing.T) {
 	type Str struct {
 		Name []string `json:"name" codec:"flatten"`
 	}
@@ -260,7 +260,7 @@ func TestURLSliceVars(t *testing.T) {
 	}
 }
 
-func TestURLVars(t *testing.T) {
+func TestURLMap(t *testing.T) {
 	u, err := url.Parse("http://localhost/v1/test/call/my_name?req=key&arg1=arg1&arg2=12345&nested.string_args=str1&nested.string_args=str2&arg2=54321")
 	if err != nil {
 		t.Fatal(err)

util/ring/buffer.go

@@ -10,16 +10,16 @@ import (
 
 // Buffer is ring buffer
 type Buffer struct {
+	sync.RWMutex
 	streams map[string]*Stream
 	vals    []*Entry
 	size    int
-	sync.RWMutex
 }
 
 // Entry is ring buffer data entry
 type Entry struct {
-	Value     interface{}
 	Timestamp time.Time
+	Value     interface{}
 }
 
 // Stream is used to stream the buffer

util/socket/pool.go

@@ -6,8 +6,8 @@ import (
 
 // Pool holds the socket pool
 type Pool struct {
-	pool map[string]*Socket
 	sync.RWMutex
+	pool map[string]*Socket
 }
 
 // Get socket from pool

util/stream/stream.go

@@ -20,10 +20,10 @@ type Stream interface {
 }
 
 type stream struct {
+	sync.RWMutex
 	Stream
 	err     error
 	request *request
-	sync.RWMutex
 }
 
 type request struct {