Merge pull request #127 from unistack-org/cover

add more cover stuff

.github/workflows/autoapprove.yml

@@ -0,0 +1,24 @@
+name: "autoapprove"
+
+on:
+  pull_request_target:
+    types: [assigned, opened, synchronize, reopened]
+  workflow_run:
+    workflows: ["prbuild"]
+    types:
+      - completed
+
+permissions:
+  pull-requests: write
+  contents: write
+
+jobs:
+  autoapprove:
+    runs-on: ubuntu-latest
+    steps:
+      - name: approve
+        uses: hmarr/auto-approve-action@v2
+        if: github.actor == 'vtolstov' || github.actor == 'dependabot[bot]'
+        id: approve
+        with:
+          github-token: ${{ secrets.GITHUB_TOKEN }}

.github/workflows/automerge.yml

@@ -0,0 +1,21 @@
+name: "automerge"
+
+on:
+  pull_request_target:
+    types: [assigned, opened, synchronize, reopened]
+
+permissions:
+  pull-requests: write
+  contents: write
+
+jobs:
+  automerge:
+    runs-on: ubuntu-latest
+    if: github.actor == 'vtolstov'
+    steps:
+      - name: merge
+        id: merge
+        run: gh pr merge --auto --merge "$PR_URL"
+        env:
+          PR_URL: ${{github.event.pull_request.html_url}}
+          GITHUB_TOKEN: ${{secrets.TOKEN}}

.github/workflows/build.yml

@@ -1,6 +1,6 @@
 name: build
 on:
-  push:
+ push:
     branches:
     - master
     - v3
@@ -10,51 +10,35 @@ jobs:
     runs-on: ubuntu-latest
     steps:
     - name: setup
-      uses: actions/setup-go@v2
+      uses: actions/setup-go@v3
       with:
-        go-version: 1.16
+        go-version: 1.17
+    - name: checkout
+      uses: actions/checkout@v3
     - name: cache
-      uses: actions/cache@v2
+      uses: actions/cache@v3
       with:
         path: ~/go/pkg/mod
         key: ${{ runner.os }}-go-${{ hashFiles('**/go.sum') }}
         restore-keys: ${{ runner.os }}-go-
-    - name: sdk checkout
-      uses: actions/checkout@v2
-    - name: sdk deps
+    - name: deps
       run: go get -v -t -d ./...
-    - name: sdk test
+    - name: test
       env:
         INTEGRATION_TESTS: yes
       run: go test -mod readonly -v ./...
-    - name: tests checkout
-      uses: actions/checkout@v2
-      with:
-        repository: unistack-org/micro-tests
-        ref: refs/heads/master
-        path: micro-tests
-        fetch-depth: 1
-    - name: tests deps
-      run: |
-        cd micro-tests
-        go mod edit -replace="github.com/unistack-org/micro/v3=../"
-        go get -v -t -d ./...
-    - name: tests test
-      env:
-        INTEGRATION_TESTS: yes
-      run: cd micro-tests && go test -mod readonly -v ./...
   lint:
     name: lint
     runs-on: ubuntu-latest
     steps:
       - name: checkout
-        uses: actions/checkout@v2
+        uses: actions/checkout@v3
       - name: lint
-        uses: golangci/golangci-lint-action@v2
+        uses: golangci/golangci-lint-action@v3.1.0
         continue-on-error: true
         with:
           # Required: the version of golangci-lint is required and must be specified without patch version: we always use the latest patch version.
-          version: v1.39
+          version: v1.30
           # Optional: working directory, useful for monorepos
           # working-directory: somedir
           # Optional: golangci-lint command line arguments.

.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

@@ -43,14 +43,14 @@ jobs:
 
     steps:
     - name: checkout
-      uses: actions/checkout@v2
+      uses: actions/checkout@v3
     - name: setup
-      uses: actions/setup-go@v2
+      uses: actions/setup-go@v3
       with:
-        go-version: 1.16
+        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

@@ -1,4 +1,4 @@
-name: "prautomerge"
+name: "dependabot-automerge"
 
 on:
   pull_request_target:
@@ -9,21 +9,17 @@ permissions:
   contents: write
 
 jobs:
-  dependabot:
+  automerge:
     runs-on: ubuntu-latest
-    if: ${{ github.actor == 'dependabot[bot]' }}
+    if: github.actor == 'dependabot[bot]'
     steps:
       - name: metadata
         id: metadata
-        uses: dependabot/fetch-metadata@v1.1.1
+        uses: dependabot/fetch-metadata@v1.3.1
         with:
           github-token: "${{ secrets.TOKEN }}"
-      - name: approve
-        run: gh pr review --approve "$PR_URL"
-        env:
-          PR_URL: ${{github.event.pull_request.html_url}}
-          GITHUB_TOKEN: ${{secrets.TOKEN}}
       - name: merge
+        id: merge
         if: ${{contains(steps.metadata.outputs.dependency-names, 'go.unistack.org')}}
         run: gh pr merge --auto --merge "$PR_URL"
         env:

.github/workflows/pr.yml

@@ -10,51 +10,35 @@ jobs:
     runs-on: ubuntu-latest
     steps:
     - name: setup
-      uses: actions/setup-go@v2
+      uses: actions/setup-go@v3
       with:
-        go-version: 1.16
+        go-version: 1.17
+    - name: checkout
+      uses: actions/checkout@v3
     - name: cache
-      uses: actions/cache@v2
+      uses: actions/cache@v3
       with:
-        path: ~/go/pkg
+        path: ~/go/pkg/mod
         key: ${{ runner.os }}-go-${{ hashFiles('**/go.sum') }}
         restore-keys: ${{ runner.os }}-go-
-    - name: sdk checkout
-      uses: actions/checkout@v2
-    - name: sdk deps
+    - name: deps
       run: go get -v -t -d ./...
-    - name: sdk test
+    - name: test
       env:
         INTEGRATION_TESTS: yes
       run: go test -mod readonly -v ./...
-    - name: tests checkout
-      uses: actions/checkout@v2
-      with:
-        repository: unistack-org/micro-tests
-        ref: refs/heads/master
-        path: micro-tests
-        fetch-depth: 1
-    - name: tests deps
-      run: |
-        cd micro-tests
-        go mod edit -replace="github.com/unistack-org/micro/v3=../"
-        go get -v -t -d ./...
-    - name: tests test
-      env:
-        INTEGRATION_TESTS: yes
-      run: cd micro-tests && go test -mod readonly -v ./...
   lint:
     name: lint
     runs-on: ubuntu-latest
     steps:
       - name: checkout
-        uses: actions/checkout@v2
+        uses: actions/checkout@v3
       - name: lint
-        uses: golangci/golangci-lint-action@v2
+        uses: golangci/golangci-lint-action@v3.1.0
         continue-on-error: true
         with:
           # Required: the version of golangci-lint is required and must be specified without patch version: we always use the latest patch version.
-          version: v1.39
+          version: v1.30
           # Optional: working directory, useful for monorepos
           # working-directory: somedir
           # Optional: golangci-lint command line arguments.

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/report/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
@@ -50,6 +50,7 @@ type Handler func(Event) error
 // Events contains multiple events
 type Events []Event
 
+// Ack try to ack all events and return
 func (evs Events) Ack() error {
 	var err error
 	for _, ev := range evs {
@@ -60,6 +61,7 @@ func (evs Events) Ack() error {
 	return nil
 }
 
+// SetError sets error on event
 func (evs Events) SetError(err error) {
 	for _, ev := range evs {
 		ev.SetError(err)

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

@@ -2,6 +2,7 @@ package client
 
 import (
 	"context"
+	"math"
 	"time"
 
 	"go.unistack.org/micro/v3/util/backoff"
@@ -10,6 +11,20 @@ import (
 // BackoffFunc is the backoff call func
 type BackoffFunc func(ctx context.Context, req Request, attempts int) (time.Duration, error)
 
-func exponentialBackoff(ctx context.Context, req Request, attempts int) (time.Duration, error) {
+// BackoffExp using exponential backoff func
+func BackoffExp(_ context.Context, _ Request, attempts int) (time.Duration, error) {
 	return backoff.Do(attempts), nil
 }
+
+// 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(math.Pow(float64(attempts), math.E)) * time.Millisecond * 100
+		if td < min {
+			return min, nil
+		} else if td > max {
+			return max, nil
+		}
+		return td, nil
+	}
+}

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,
@@ -22,7 +22,7 @@ func TestBackoff(t *testing.T) {
 	}
 
 	for i := 0; i < 5; i++ {
-		d, err := exponentialBackoff(context.TODO(), r, i)
+		d, err := BackoffExp(context.TODO(), r, i)
 		if err != nil {
 			t.Fatal(err)
 		}
@@ -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,11 +11,11 @@ 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
-	DefaultBackoff = exponentialBackoff
+	// DefaultBackoff is the default backoff function for retries (minimum 10 millisecond and maximum 5 second)
+	DefaultBackoff = BackoffInterval(10*time.Millisecond, 5*time.Second)
 	// DefaultRetry is the default check-for-retry function for retries
 	DefaultRetry = RetryNever
 	// DefaultRetries is the default number of times a request is tried

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.go

@@ -19,18 +19,32 @@ func RetryNever(ctx context.Context, req Request, retryCount int, err error) (bo
 	return false, nil
 }
 
-// RetryOnError retries a request on a 500 or timeout error
+// RetryOnError retries a request on a 500 or 408 (timeout) error
 func RetryOnError(_ context.Context, _ Request, _ int, err error) (bool, error) {
 	if err == nil {
 		return false, nil
 	}
-
 	me := errors.FromError(err)
 	switch me.Code {
 	// retry on timeout or internal server error
 	case 408, 500:
 		return true, nil
 	}
-
 	return false, nil
 }
+
+// RetryOnErrors retries a request on specified error codes
+func RetryOnErrors(codes ...int32) RetryFunc {
+	return func(_ context.Context, _ Request, _ int, err error) (bool, error) {
+		if err == nil {
+			return false, nil
+		}
+		me := errors.FromError(err)
+		for _, code := range codes {
+			if me.Code == code {
+				return true, nil
+			}
+		}
+		return false, 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/config.go

@@ -4,11 +4,16 @@ package config // import "go.unistack.org/micro/v3/config"
 import (
 	"context"
 	"errors"
+	"reflect"
 	"time"
 )
 
+type Validator interface {
+	Validate() error
+}
+
 // DefaultConfig default config
-var DefaultConfig Config = NewConfig()
+var DefaultConfig = NewConfig()
 
 // DefaultWatcherMinInterval default min interval for poll changes
 var DefaultWatcherMinInterval = 5 * time.Second
@@ -67,7 +72,59 @@ func Load(ctx context.Context, cs []Config, opts ...LoadOption) error {
 	return nil
 }
 
+// Validate runs Validate() error func for each struct field
+func Validate(ctx context.Context, cfg interface{}) error {
+	if cfg == nil {
+		return nil
+	}
+
+	if v, ok := cfg.(Validator); ok {
+		if err := v.Validate(); err != nil {
+			return err
+		}
+	}
+
+	sv := reflect.ValueOf(cfg)
+	if sv.Kind() == reflect.Ptr {
+		sv = sv.Elem()
+	}
+	if sv.Kind() != reflect.Struct {
+		return nil
+	}
+
+	typ := sv.Type()
+	for idx := 0; idx < typ.NumField(); idx++ {
+		fld := typ.Field(idx)
+		val := sv.Field(idx)
+		if !val.IsValid() || len(fld.PkgPath) != 0 {
+			continue
+		}
+
+		if v, ok := val.Interface().(Validator); ok {
+			if err := v.Validate(); err != nil {
+				return err
+			}
+		}
+
+		switch val.Kind() {
+		case reflect.Ptr:
+			if reflect.Indirect(val).Kind() == reflect.Struct {
+				if err := Validate(ctx, val.Interface()); err != nil {
+					return err
+				}
+			}
+		case reflect.Struct:
+			if err := Validate(ctx, val.Interface()); err != nil {
+				return err
+			}
+		}
+	}
+
+	return nil
+}
+
 var (
+	// DefaultAfterLoad default func that runs after config load
 	DefaultAfterLoad = func(ctx context.Context, c Config) error {
 		for _, fn := range c.Options().AfterLoad {
 			if err := fn(ctx, c); err != nil {
@@ -79,7 +136,7 @@ var (
 		}
 		return nil
 	}
-
+	// DefaultAfterSave default func that runs after config save
 	DefaultAfterSave = func(ctx context.Context, c Config) error {
 		for _, fn := range c.Options().AfterSave {
 			if err := fn(ctx, c); err != nil {
@@ -91,7 +148,7 @@ var (
 		}
 		return nil
 	}
-
+	// DefaultBeforeLoad default func that runs before config load
 	DefaultBeforeLoad = func(ctx context.Context, c Config) error {
 		for _, fn := range c.Options().BeforeLoad {
 			if err := fn(ctx, c); err != nil {
@@ -103,11 +160,11 @@ var (
 		}
 		return nil
 	}
-
+	// DefaultBeforeSave default func that runs befora config save
 	DefaultBeforeSave = func(ctx context.Context, c Config) error {
 		for _, fn := range c.Options().BeforeSave {
 			if err := fn(ctx, c); err != nil {
-				c.Options().Logger.Errorf(ctx, "%s BeforeSavec err: %v", c.String(), err)
+				c.Options().Logger.Errorf(ctx, "%s BeforeSave err: %v", c.String(), err)
 				if !c.Options().AllowFail {
 					return err
 				}

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

config/default_test.go

@@ -8,30 +8,46 @@ import (
 	"go.unistack.org/micro/v3/config"
 )
 
-type Cfg struct {
+type cfg struct {
 	StringValue string `default:"string_value"`
 	IgnoreValue string `json:"-"`
-	StructValue struct {
-		StringValue string `default:"string_value"`
+	StructValue *cfgStructValue
+	IntValue    int `default:"99"`
+}
+
+type cfgStructValue struct {
+	StringValue string `default:"string_value"`
+}
+
+func (c *cfg) Validate() error {
+	if c.IntValue != 10 {
+		return fmt.Errorf("invalid IntValue %d != %d", 10, c.IntValue)
 	}
-	IntValue int `default:"99"`
+	return nil
+}
+
+func (c *cfgStructValue) Validate() error {
+	if c.StringValue != "string_value" {
+		return fmt.Errorf("invalid StringValue %s != %s", "string_value", c.StringValue)
+	}
+	return nil
 }
 
 func TestDefault(t *testing.T) {
 	ctx := context.Background()
-	conf := &Cfg{IntValue: 10}
-	blfn := func(ctx context.Context, cfg config.Config) error {
-		nconf, ok := cfg.Options().Struct.(*Cfg)
+	conf := &cfg{IntValue: 10}
+	blfn := func(_ context.Context, c config.Config) error {
+		nconf, ok := c.Options().Struct.(*cfg)
 		if !ok {
-			return fmt.Errorf("failed to get Struct from options: %v", cfg.Options())
+			return fmt.Errorf("failed to get Struct from options: %v", c.Options())
 		}
 		nconf.StringValue = "before_load"
 		return nil
 	}
-	alfn := func(ctx context.Context, cfg config.Config) error {
-		nconf, ok := cfg.Options().Struct.(*Cfg)
+	alfn := func(_ context.Context, c config.Config) error {
+		nconf, ok := c.Options().Struct.(*cfg)
 		if !ok {
-			return fmt.Errorf("failed to get Struct from options: %v", cfg.Options())
+			return fmt.Errorf("failed to get Struct from options: %v", c.Options())
 		}
 		nconf.StringValue = "after_load"
 		return nil
@@ -50,3 +66,19 @@ func TestDefault(t *testing.T) {
 	_ = conf
 	// t.Logf("%#+v\n", conf)
 }
+
+func TestValidate(t *testing.T) {
+	ctx := context.Background()
+	conf := &cfg{IntValue: 10}
+	cfg := config.NewConfig(config.Struct(conf))
+	if err := cfg.Init(); err != nil {
+		t.Fatal(err)
+	}
+	if err := cfg.Load(ctx); err != nil {
+		t.Fatal(err)
+	}
+
+	if err := config.Validate(ctx, conf); err != nil {
+		t.Fatal(err)
+	}
+}

config/options.go

@@ -69,6 +69,7 @@ type LoadOptions struct {
 	Context  context.Context
 }
 
+// NewLoadOptions create LoadOptions struct with provided opts
 func NewLoadOptions(opts ...LoadOption) LoadOptions {
 	options := LoadOptions{}
 	for _, o := range opts {
@@ -221,8 +222,10 @@ type WatchOptions struct {
 	Coalesce bool
 }
 
+// WatchOption func signature
 type WatchOption func(*WatchOptions)
 
+// NewWatchOptions create WatchOptions struct with provided opts
 func NewWatchOptions(opts ...WatchOption) WatchOptions {
 	options := WatchOptions{
 		Context:     context.Background(),

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

flow/flow.go

@@ -11,7 +11,9 @@ import (
 )
 
 var (
+	// ErrStepNotExists returns when step not found
 	ErrStepNotExists = errors.New("step not exists")
+	// ErrMissingClient returns when client.Client is missing
 	ErrMissingClient = errors.New("client not set")
 )
 
@@ -36,6 +38,7 @@ func (m *RawMessage) UnmarshalJSON(data []byte) error {
 	return nil
 }
 
+// Message used to transfer data between steps
 type Message struct {
 	Header metadata.Metadata
 	Body   RawMessage
@@ -67,6 +70,7 @@ type Step interface {
 	Response() *Message
 }
 
+// Status contains step current status
 type Status int
 
 func (status Status) String() string {
@@ -74,15 +78,22 @@ func (status Status) String() string {
 }
 
 const (
+	// StatusPending step waiting to start
 	StatusPending Status = iota
+	// StatusRunning step is running
 	StatusRunning
+	// StatusFailure step competed with error
 	StatusFailure
+	// StatusSuccess step completed without error
 	StatusSuccess
+	// StatusAborted step aborted while it running
 	StatusAborted
+	// StatusSuspend step suspended
 	StatusSuspend
 )
 
 var (
+	// StatusString contains map status => string
 	StatusString = map[Status]string{
 		StatusPending: "StatusPending",
 		StatusRunning: "StatusRunning",
@@ -91,6 +102,7 @@ var (
 		StatusAborted: "StatusAborted",
 		StatusSuspend: "StatusSuspend",
 	}
+	// StringStatus contains map string => status
 	StringStatus = map[string]Status{
 		"StatusPending": StatusPending,
 		"StatusRunning": StatusRunning,
@@ -144,6 +156,7 @@ var (
 	atomicSteps atomic.Value
 )
 
+// RegisterStep register own step with workflow
 func RegisterStep(step Step) {
 	flowMu.Lock()
 	steps, _ := atomicSteps.Load().([]Step)

flow/options.go

@@ -91,7 +91,7 @@ func Store(s store.Store) Option {
 	}
 }
 
-// WorflowOption func signature
+// WorkflowOption func signature
 type WorkflowOption func(*WorkflowOptions)
 
 // WorkflowOptions holds workflow options

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.2.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.1
-	golang.org/x/net v0.0.0-20210928044308-7d9f5e0b762b
+	go.unistack.org/micro-proto/v3 v3.2.7
+	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.2.0 h1:besgBTC8w8HjP6NzQdxwKH9Z5oQMZ24ThTrHp3cZ8eU=
-github.com/golang-jwt/jwt/v4 v4.2.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=
@@ -39,22 +40,30 @@ github.com/golang/protobuf v1.4.1/go.mod h1:U8fpvMrcmy5pZrNK1lt4xCsGvpyWQ/VVv6QD
 github.com/golang/protobuf v1.4.2/go.mod h1:oDoupMAO8OvCJWAcko0GGGIgR6R6ocIYbsSw735rRwI=
 github.com/golang/protobuf v1.4.3/go.mod h1:oDoupMAO8OvCJWAcko0GGGIgR6R6ocIYbsSw735rRwI=
 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/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/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/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=
@@ -71,8 +80,8 @@ github.com/xeipuuv/gojsonpointer v0.0.0-20180127040702-4e3ac2762d5f/go.mod h1:N2
 github.com/xeipuuv/gojsonreference v0.0.0-20180127040603-bd5ef7bd5415/go.mod h1:GwrjFmJcFw6At/Gs6z4yjiIwzuJ1/+UwLxMQDVQXShQ=
 github.com/xeipuuv/gojsonschema v1.2.0/go.mod h1:anYRn/JVcOK2ZgGU+IjEV4nwlhoK5sQluxsYJ78Id3Y=
 go.opentelemetry.io/proto/otlp v0.7.0/go.mod h1:PqfVotwruBrMGOCsRd/89rSnXhoiJIqeYNgFYFoEGnI=
-go.unistack.org/micro-proto/v3 v3.2.1 h1:z7+V97LcAwMbBiYStmf8b6fFk2UPJTzni+rxNqk4NrI=
-go.unistack.org/micro-proto/v3 v3.2.1/go.mod h1:ZltVWNECD5yK+40+OCONzGw4OtmSdTpVi8/KFgo9dqM=
+go.unistack.org/micro-proto/v3 v3.2.7 h1:zG6d69kHc+oij2lwQ3AfrCgdjiEVRG2A7TlsxjusWs4=
+go.unistack.org/micro-proto/v3 v3.2.7/go.mod h1:ZltVWNECD5yK+40+OCONzGw4OtmSdTpVi8/KFgo9dqM=
 golang.org/x/crypto v0.0.0-20190308221718-c2843e01d9a2/go.mod h1:djNgcEr1/C05ACkg1iLfiJU5Ep61QUkGW8qpdssI0+w=
 golang.org/x/crypto v0.0.0-20200622213623-75b288015ac9/go.mod h1:LzIPMQfyMNhhGPhUkYOs5KpL4U8rLKemX1yGLhDgUto=
 golang.org/x/exp v0.0.0-20190121172915-509febef88a4/go.mod h1:CJ0aWSM057203Lf6IL+f9T1iT9GByDxfZKAQTCR3kQA=
@@ -88,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=
@@ -104,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=
@@ -142,15 +155,19 @@ 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=
 honnef.co/go/tools v0.0.0-20190102054323-c2f93a96b099/go.mod h1:rf3lG4BRIbNafJWhAfAdb/ePZxsR/4RtNHQocxwk9r4=
 honnef.co/go/tools v0.0.0-20190523083050-ea95bdfd59fc/go.mod h1:rf3lG4BRIbNafJWhAfAdb/ePZxsR/4RtNHQocxwk9r4=

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

logger/wrapper/wrapper.go

@@ -11,6 +11,7 @@ import (
 )
 
 var (
+	// DefaultClientCallObserver called by wrapper in client Call
 	DefaultClientCallObserver = func(ctx context.Context, req client.Request, rsp interface{}, opts []client.CallOption, err error) []string {
 		labels := []string{"service", req.Service(), "endpoint", req.Endpoint()}
 		if err != nil {
@@ -19,6 +20,7 @@ var (
 		return labels
 	}
 
+	// DefaultClientStreamObserver called by wrapper in client Stream
 	DefaultClientStreamObserver = func(ctx context.Context, req client.Request, opts []client.CallOption, stream client.Stream, err error) []string {
 		labels := []string{"service", req.Service(), "endpoint", req.Endpoint()}
 		if err != nil {
@@ -27,6 +29,7 @@ var (
 		return labels
 	}
 
+	// DefaultClientPublishObserver called by wrapper in client Publish
 	DefaultClientPublishObserver = func(ctx context.Context, msg client.Message, opts []client.PublishOption, err error) []string {
 		labels := []string{"endpoint", msg.Topic()}
 		if err != nil {
@@ -35,6 +38,7 @@ var (
 		return labels
 	}
 
+	// DefaultServerHandlerObserver called by wrapper in server Handler
 	DefaultServerHandlerObserver = func(ctx context.Context, req server.Request, rsp interface{}, err error) []string {
 		labels := []string{"service", req.Service(), "endpoint", req.Endpoint()}
 		if err != nil {
@@ -43,6 +47,7 @@ var (
 		return labels
 	}
 
+	// DefaultServerSubscriberObserver called by wrapper in server Subscriber
 	DefaultServerSubscriberObserver = func(ctx context.Context, msg server.Message, err error) []string {
 		labels := []string{"endpoint", msg.Topic()}
 		if err != nil {
@@ -51,6 +56,7 @@ var (
 		return labels
 	}
 
+	// DefaultClientCallFuncObserver called by wrapper in client CallFunc
 	DefaultClientCallFuncObserver = func(ctx context.Context, addr string, req client.Request, rsp interface{}, opts client.CallOptions, err error) []string {
 		labels := []string{"service", req.Service(), "endpoint", req.Endpoint()}
 		if err != nil {
@@ -59,6 +65,7 @@ var (
 		return labels
 	}
 
+	// DefaultSkipEndpoints wrapper not called for this endpoints
 	DefaultSkipEndpoints = []string{"Meter.Metrics"}
 )
 
@@ -71,11 +78,17 @@ type lWrapper struct {
 }
 
 type (
-	ClientCallObserver       func(context.Context, client.Request, interface{}, []client.CallOption, error) []string
-	ClientStreamObserver     func(context.Context, client.Request, []client.CallOption, client.Stream, error) []string
-	ClientPublishObserver    func(context.Context, client.Message, []client.PublishOption, error) []string
-	ClientCallFuncObserver   func(context.Context, string, client.Request, interface{}, client.CallOptions, error) []string
-	ServerHandlerObserver    func(context.Context, server.Request, interface{}, error) []string
+	// ClientCallObserver func signature
+	ClientCallObserver func(context.Context, client.Request, interface{}, []client.CallOption, error) []string
+	// ClientStreamObserver func signature
+	ClientStreamObserver func(context.Context, client.Request, []client.CallOption, client.Stream, error) []string
+	// ClientPublishObserver func signature
+	ClientPublishObserver func(context.Context, client.Message, []client.PublishOption, error) []string
+	// ClientCallFuncObserver func signature
+	ClientCallFuncObserver func(context.Context, string, client.Request, interface{}, client.CallOptions, error) []string
+	// ServerHandlerObserver func signature
+	ServerHandlerObserver func(context.Context, server.Request, interface{}, error) []string
+	// ServerSubscriberObserver func signature
 	ServerSubscriberObserver func(context.Context, server.Message, error) []string
 )
 

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

meter/meter.go

@@ -102,7 +102,7 @@ func (k byKey) Swap(i, j int) {
 	k[i*2+1], k[j*2+1] = k[j*2+1], k[i*2+1]
 }
 
-// BuildLables used to sort labels and delete duplicates.
+// BuildLabels used to sort labels and delete duplicates.
 // Last value wins in case of duplicate label keys.
 func BuildLabels(labels ...string) []string {
 	if len(labels)%2 == 1 {

meter/wrapper/wrapper.go

@@ -104,7 +104,7 @@ func Meter(m meter.Meter) Option {
 	}
 }
 
-// SkipEndpoint add endpoint to skip
+// SkipEndoints add endpoint to skip
 func SkipEndoints(eps ...string) Option {
 	return func(o *Options) {
 		o.SkipEndpoints = append(o.SkipEndpoints, eps...)
@@ -294,7 +294,7 @@ func (w *wrapper) HandlerFunc(fn server.HandlerFunc) server.HandlerFunc {
 	}
 }
 
-// NewSubscribeWrapper create server subscribe wrapper
+// NewSubscriberWrapper create server subscribe wrapper
 func NewSubscriberWrapper(opts ...Option) server.SubscriberWrapper {
 	handler := &wrapper{
 		opts: NewOptions(opts...),

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

@@ -2,7 +2,6 @@ package register
 
 import (
 	"context"
-	"errors"
 	"sync"
 	"time"
 
@@ -30,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
@@ -438,7 +437,7 @@ func (m *watcher) Next() (*Result, error) {
 				return r, nil
 			}
 		case <-m.exit:
-			return nil, errors.New("watcher stopped")
+			return nil, ErrWatcherStopped
 		}
 	}
 }

register/memory_test.go

@@ -4,6 +4,7 @@ import (
 	"context"
 	"fmt"
 	"os"
+	"sync"
 	"testing"
 	"time"
 )
@@ -284,29 +285,39 @@ func TestMemoryWildcard(t *testing.T) {
 }
 
 func TestWatcher(t *testing.T) {
-	w := &watcher{
-		id:   "test",
-		res:  make(chan *Result),
-		exit: make(chan bool),
-		wo: WatchOptions{
-			Domain: WildcardDomain,
-		},
-	}
+	testSrv := &Service{Name: "foo", Version: "1.0.0"}
 
+	ctx := context.TODO()
+	m := NewRegister()
+	m.Init()
+	m.Connect(ctx)
+	wc, err := m.Watch(ctx)
+	if err != nil {
+		t.Fatalf("cant watch: %v", err)
+	}
+	defer wc.Stop()
+
+	var wg sync.WaitGroup
+	wg.Add(1)
 	go func() {
-		w.res <- &Result{
-			Service: &Service{Name: "foo"},
+		for {
+			ch, err := wc.Next()
+			if err != nil {
+				t.Fatal("unexpected err", err)
+			}
+			t.Logf("changes %#+v", ch.Service)
+			wc.Stop()
+			wg.Done()
+			return
 		}
 	}()
 
-	_, err := w.Next()
-	if err != nil {
-		t.Fatal("unexpected err", err)
+	if err := m.Register(ctx, testSrv); err != nil {
+		t.Fatalf("Register err: %v", err)
 	}
 
-	w.Stop()
-
-	if _, err := w.Next(); err == nil {
+	wg.Wait()
+	if _, err := wc.Next(); err == nil {
 		t.Fatal("expected error on Next()")
 	}
 }

register/options.go

@@ -44,9 +44,8 @@ func NewOptions(opts ...Option) Options {
 	return options
 }
 
-// nolint: golint,revive
 // RegisterOptions holds options for register method
-type RegisterOptions struct {
+type RegisterOptions struct { // nolint: golint,revive
 	Context  context.Context
 	Domain   string
 	TTL      time.Duration
@@ -197,33 +196,29 @@ func TLSConfig(t *tls.Config) Option {
 	}
 }
 
-// nolint: golint,revive
 // RegisterAttempts specifies register atempts count
-func RegisterAttempts(t int) RegisterOption {
+func RegisterAttempts(t int) RegisterOption { // nolint: golint,revive
 	return func(o *RegisterOptions) {
 		o.Attempts = t
 	}
 }
 
-// nolint: golint,revive
 // RegisterTTL specifies register ttl
-func RegisterTTL(t time.Duration) RegisterOption {
+func RegisterTTL(t time.Duration) RegisterOption { // nolint: golint,revive
 	return func(o *RegisterOptions) {
 		o.TTL = t
 	}
 }
 
-// nolint: golint,revive
 // RegisterContext sets the register context
-func RegisterContext(ctx context.Context) RegisterOption {
+func RegisterContext(ctx context.Context) RegisterOption { // nolint: golint,revive
 	return func(o *RegisterOptions) {
 		o.Context = ctx
 	}
 }
 
-// nolint: golint,revive
 // RegisterDomain secifies register domain
-func RegisterDomain(d string) RegisterOption {
+func RegisterDomain(d string) RegisterOption { // nolint: golint,revive
 	return func(o *RegisterOptions) {
 		o.Domain = d
 	}

register/register.go

@@ -69,9 +69,8 @@ type Endpoint struct {
 // Option func signature
 type Option func(*Options)
 
-// nolint: golint,revive
 // RegisterOption option is used to register service
-type RegisterOption func(*RegisterOptions)
+type RegisterOption func(*RegisterOptions) // nolint: golint,revive
 
 // WatchOption option is used to watch service changes
 type WatchOption func(*WatchOptions)

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/request.go

@@ -22,7 +22,7 @@ func (r *rpcMessage) Topic() string {
 	return r.topic
 }
 
-func (r *rpcMessage) Payload() interface{} {
+func (r *rpcMessage) Body() interface{} {
 	return r.payload
 }
 
@@ -30,10 +30,6 @@ func (r *rpcMessage) Header() metadata.Metadata {
 	return r.header
 }
 
-func (r *rpcMessage) Body() []byte {
-	return r.body
-}
-
 func (r *rpcMessage) Codec() codec.Codec {
 	return r.codec
 }

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
@@ -75,13 +75,11 @@ type Message interface {
 	// Topic of the message
 	Topic() string
 	// The decoded payload value
-	Payload() interface{}
+	Body() interface{}
 	// The content type of the payload
 	ContentType() string
 	// The raw headers of the message
 	Header() metadata.Metadata
-	// The raw body of the message
-	Body() []byte
 	// Codec used to decode the message
 	Codec() codec.Codec
 }

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?
@@ -252,7 +252,7 @@ func (n *noopServer) newBatchSubHandler(sb *subscriber, opts Options) broker.Bat
 					return err
 				}
 				rb := reflect.New(req.Type().Elem())
-				if err = cf.ReadBody(bytes.NewReader(msg.Body()), rb.Interface()); err != nil {
+				if err = cf.ReadBody(bytes.NewReader(msg.(*rpcMessage).body), rb.Interface()); err != nil {
 					return err
 				}
 				msg.(*rpcMessage).codec = cf
@@ -269,7 +269,7 @@ func (n *noopServer) newBatchSubHandler(sb *subscriber, opts Options) broker.Bat
 				}
 				payloads := reflect.MakeSlice(reqType, 0, len(ms))
 				for _, m := range ms {
-					payloads = reflect.Append(payloads, reflect.ValueOf(m.Payload()))
+					payloads = reflect.Append(payloads, reflect.ValueOf(m.Body()))
 				}
 				vals = append(vals, payloads)
 
@@ -381,7 +381,7 @@ func (n *noopServer) newSubHandler(sb *subscriber, opts Options) broker.Handler
 					vals = append(vals, reflect.ValueOf(ctx))
 				}
 
-				vals = append(vals, reflect.ValueOf(msg.Payload()))
+				vals = append(vals, reflect.ValueOf(msg.Body()))
 
 				returnValues := handler.method.Call(vals)
 				if rerr := returnValues[0].Interface(); rerr != nil {
@@ -406,7 +406,6 @@ func (n *noopServer) newSubHandler(sb *subscriber, opts Options) broker.Handler
 					contentType: ct,
 					payload:     req.Interface(),
 					header:      msg.Header,
-					body:        msg.Body,
 				})
 				results <- cerr
 			}()

service.go

@@ -54,8 +54,12 @@ type Service interface {
 	// Runtime(string) (runtime.Runtime, bool)
 	// Profile
 	// Profile(string) (profile.Profile, bool)
-	// Run the service
+	// Run the service and wait
 	Run() error
+	// Start the service
+	Start() error
+	// Stop the service
+	Stop() error
 	// The service implementation
 	String() string
 }
@@ -71,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.
@@ -106,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 {
@@ -245,7 +243,7 @@ func (s *service) Meter(names ...string) meter.Meter {
 }
 
 func (s *service) String() string {
-	return "micro"
+	return s.opts.Name
 }
 
 //nolint:gocyclo
@@ -256,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", s.Name())
-	}
-
-	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
@@ -277,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")
+	}
+}

tracer/options.go

@@ -2,12 +2,16 @@ package tracer
 
 import "go.unistack.org/micro/v3/logger"
 
+// SpanOptions contains span option
 type SpanOptions struct{}
 
+// SpanOption func signature
 type SpanOption func(o *SpanOptions)
 
+// EventOptions contains event options
 type EventOptions struct{}
 
+// EventOption func signature
 type EventOption func(o *EventOptions)
 
 // Options struct
@@ -18,7 +22,7 @@ type Options struct {
 	Name string
 }
 
-// Option func
+// Option func signature
 type Option func(o *Options)
 
 // Logger sets the logger

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/jitter/ticker.go

@@ -1,6 +1,7 @@
-package jitter
+package jitter // import "go.unistack.org/micro/v3/util/jitter"
 
 import (
+	"context"
 	"time"
 
 	"go.unistack.org/micro/v3/util/rand"
@@ -10,13 +11,31 @@ import (
 // the min and max duration values (stored internally as int64 nanosecond
 // counts).
 type Ticker struct {
-	C    chan time.Time
+	ctx  context.Context
 	done chan chan struct{}
+	C    chan time.Time
 	min  int64
 	max  int64
+	exp  int64
+	exit bool
 	rng  rand.Rand
 }
 
+// NewTickerContext returns a pointer to an initialized instance of the Ticker.
+// It works like NewTicker except that it has ability to close via context.
+// Also it works fine with context.WithTimeout to handle max time to run ticker.
+func NewTickerContext(ctx context.Context, min, max time.Duration) *Ticker {
+	ticker := &Ticker{
+		C:    make(chan time.Time),
+		done: make(chan chan struct{}),
+		min:  min.Nanoseconds(),
+		max:  max.Nanoseconds(),
+		ctx:  ctx,
+	}
+	go ticker.run()
+	return ticker
+}
+
 // NewTicker returns a pointer to an initialized instance of the Ticker.
 // Min and max are durations of the shortest and longest allowed
 // ticks. Ticker will run in a goroutine until explicitly stopped.
@@ -26,6 +45,7 @@ func NewTicker(min, max time.Duration) *Ticker {
 		done: make(chan chan struct{}),
 		min:  min.Nanoseconds(),
 		max:  max.Nanoseconds(),
+		ctx:  context.Background(),
 	}
 	go ticker.run()
 	return ticker
@@ -33,9 +53,14 @@ func NewTicker(min, max time.Duration) *Ticker {
 
 // Stop terminates the ticker goroutine and closes the C channel.
 func (ticker *Ticker) Stop() {
+	if ticker.exit {
+		return
+	}
 	c := make(chan struct{})
 	ticker.done <- c
 	<-c
+	// close(ticker.C)
+	ticker.exit = true
 }
 
 func (ticker *Ticker) run() {
@@ -44,6 +69,8 @@ func (ticker *Ticker) run() {
 	for {
 		// either a stop signal or a timeout
 		select {
+		case <-ticker.ctx.Done():
+			t.Stop()
 		case c := <-ticker.done:
 			t.Stop()
 			close(c)

util/jitter/ticker_test.go

@@ -0,0 +1,62 @@
+package jitter
+
+import (
+	"context"
+	"testing"
+	"time"
+)
+
+func TestNewTickerContext(t *testing.T) {
+	t.Parallel()
+	ctx, cancel := context.WithTimeout(context.Background(), 500*time.Millisecond)
+	defer cancel()
+
+	ticker := NewTickerContext(ctx, 600*time.Millisecond, 1000*time.Millisecond)
+	defer ticker.Stop()
+loop:
+	for {
+		select {
+		case <-ctx.Done():
+			ticker.Stop()
+			break loop
+		case v, ok := <-ticker.C:
+			if ok {
+				t.Fatalf("context must be closed %s", v)
+			}
+			break loop
+		}
+	}
+}
+
+func TestTicker(t *testing.T) {
+	t.Parallel()
+
+	min := time.Duration(10)
+	max := time.Duration(20)
+
+	// tick can take a little longer since we're not adjusting it to account for
+	// processing.
+	precision := time.Duration(4)
+
+	rt := NewTicker(min*time.Millisecond, max*time.Millisecond)
+	for i := 0; i < 5; i++ {
+		t0 := time.Now()
+		t1 := <-rt.C
+		td := t1.Sub(t0)
+		if td < min*time.Millisecond {
+			t.Fatalf("tick was shorter than expected: %s", td)
+		} else if td > (max+precision)*time.Millisecond {
+			t.Fatalf("tick was longer than expected: %s", td)
+		}
+	}
+	rt.Stop()
+	time.Sleep((max + precision) * time.Millisecond)
+	select {
+	case v, ok := <-rt.C:
+		if ok || !v.IsZero() {
+			t.Fatal("ticker did not shut down")
+		}
+	default:
+		t.Fatal("expected to receive close channel signal")
+	}
+}

util/reflect/path.go

@@ -9,18 +9,30 @@ import (
 )
 
 const (
-	SplitToken     = "."
+	// SplitToken used to detect path components
+	SplitToken = "."
+	// IndexCloseChar used to detect index end
 	IndexCloseChar = "]"
-	IndexOpenChar  = "["
+	// IndexOpenChar used to detect index start
+	IndexOpenChar = "["
 )
 
 var (
-	ErrMalformedIndex    = errors.New("Malformed index key")
-	ErrInvalidIndexUsage = errors.New("Invalid index key usage")
-	ErrKeyNotFound       = errors.New("Unable to find the key")
-	ErrBadJSONPath       = errors.New("Bad path: must start with $ and have more then 2 chars")
+	// ErrMalformedIndex returns when index key have invalid format
+	ErrMalformedIndex = errors.New("malformed index key")
+	// ErrInvalidIndexUsage returns when index key usage error
+	ErrInvalidIndexUsage = errors.New("invalid index key usage")
+	// ErrKeyNotFound returns when key not found
+	ErrKeyNotFound = errors.New("unable to find the key")
+	// ErrBadJSONPath returns when path have invalid syntax
+	ErrBadJSONPath = errors.New("bad path: must start with $ and have more then 2 chars")
 )
 
+// Lookup performs a lookup into a value, using a path of keys. The key should
+// match with a Field or a MapIndex. For slice you can use the syntax key[index]
+// to access a specific index. If one key owns to a slice and an index is not
+// specificied the rest of the path will be apllied to evaley value of the
+// slice, and the value will be merged into a slice.
 func Lookup(i interface{}, path string) (reflect.Value, error) {
 	if path == "" || path[0:1] != "$" {
 		return reflect.Value{}, ErrBadJSONPath
@@ -37,11 +49,6 @@ func Lookup(i interface{}, path string) (reflect.Value, error) {
 	return lookup(i, strings.Split(path[2:], SplitToken)...)
 }
 
-// Lookup performs a lookup into a value, using a path of keys. The key should
-// match with a Field or a MapIndex. For slice you can use the syntax key[index]
-// to access a specific index. If one key owns to a slice and an index is not
-// specificied the rest of the path will be apllied to evaley value of the
-// slice, and the value will be merged into a slice.
 func lookup(i interface{}, path ...string) (reflect.Value, error) {
 	value := reflect.ValueOf(i)
 	var parent reflect.Value

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.go

@@ -20,8 +20,8 @@ var bracketSplitter = regexp.MustCompile(`\[|\]`)
 
 // StructField contains struct field path its value and field
 type StructField struct {
-	Path  string
 	Value reflect.Value
+	Path  string
 	Field reflect.StructField
 }
 
@@ -245,8 +245,13 @@ func StructFields(src interface{}) ([]StructField, error) {
 
 		switch val.Kind() {
 		case reflect.Ptr:
-			//	if !val.IsValid()
-			if reflect.Indirect(val).Kind() == reflect.Struct {
+			if val.CanSet() && fld.Type.Elem().Kind() == reflect.Struct {
+				if val.IsNil() {
+					val.Set(reflect.New(fld.Type.Elem()))
+				}
+			}
+			switch reflect.Indirect(val).Kind() {
+			case reflect.Struct:
 				infields, err := StructFields(val.Interface())
 				if err != nil {
 					return nil, err
@@ -255,7 +260,7 @@ func StructFields(src interface{}) ([]StructField, error) {
 					infield.Path = fmt.Sprintf("%s.%s", fld.Name, infield.Path)
 					fields = append(fields, infield)
 				}
-			} else {
+			default:
 				fields = append(fields, StructField{Field: fld, Value: val, Path: fld.Name})
 			}
 		case reflect.Struct:
@@ -268,6 +273,7 @@ func StructFields(src interface{}) ([]StructField, error) {
 				fields = append(fields, infield)
 			}
 		default:
+
 			fields = append(fields, StructField{Field: fld, Value: val, Path: fld.Name})
 		}
 	}

util/reflect/struct_test.go

@@ -9,23 +9,56 @@ 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
+	}
 	type Config struct {
-		Wait     time.Duration
 		Time     time.Time
+		Nested   *NestedConfig
 		Metadata map[string]int
 		Broker   string
 		Addr     []string
+		Wait     time.Duration
 		Verbose  bool
-		Nested   *Config
 	}
-	cfg := &Config{Nested: &Config{}}
+	cfg := &Config{Nested: &NestedConfig{}}
 	fields, err := rutil.StructFields(cfg)
 	if err != nil {
 		t.Fatal(err)
 	}
-	if len(fields) != 13 {
-		t.Fatalf("invalid fields number: %v", fields)
+	if len(fields) != 7 {
+		for _, field := range fields {
+			t.Logf("field %#+v\n", field)
+		}
+		t.Fatalf("invalid fields number: %d != %d", 7, len(fields))
 	}
 }
 
@@ -124,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
@@ -172,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"`
 	}
@@ -191,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"`
 	}
@@ -254,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/register/util.go

@@ -1,7 +1,11 @@
 package register // import "go.unistack.org/micro/v3/util/register"
 
 import (
+	"context"
+	"time"
+
 	"go.unistack.org/micro/v3/register"
+	jitter "go.unistack.org/micro/v3/util/jitter"
 )
 
 func addNodes(old, neu []*register.Node) []*register.Node {
@@ -146,3 +150,30 @@ func Remove(old, del []*register.Service) []*register.Service {
 
 	return services
 }
+
+// WaitService using register wait for service to appear with min/max interval for check and optional timeout.
+// Timeout can be 0 to wait infinitive.
+func WaitService(ctx context.Context, reg register.Register, name string, min time.Duration, max time.Duration, timeout time.Duration, opts ...register.LookupOption) error {
+	if timeout > 0 {
+		var cancel context.CancelFunc
+		ctx, cancel = context.WithTimeout(ctx, timeout)
+		defer cancel()
+	}
+
+	ticker := jitter.NewTickerContext(ctx, min, max)
+	defer ticker.Stop()
+
+	for {
+		select {
+		case <-ctx.Done():
+			return ctx.Err()
+		case _, ok := <-ticker.C:
+			if _, err := reg.LookupService(ctx, name, opts...); err == nil {
+				return nil
+			}
+			if ok {
+				return register.ErrNotFound
+			}
+		}
+	}
+}

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 {