broker/context_test.go

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

client/backoff.go

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

client/backoff_test.go

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

client/client.go

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

client/client_call_options_test.go

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

client/context_test.go

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

client/noop.go

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

client/retry_test.go

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

codec/context_test.go

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

config/context_test.go

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

context_test.go

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

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

logger/context_test.go

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

metadata/context_test.go

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

register/context_test.go

@@ -0,0 +1,36 @@
+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")
+	}
+}
+

router/context_test.go

@@ -0,0 +1,36 @@
+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

@@ -282,13 +282,14 @@ func (n *noopServer) Deregister() error {
 				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)
 					}

server/noop_test.go

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

server/server.go

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

store/context_test.go

@@ -0,0 +1,36 @@
+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")
+	}
+}
+

tracer/context_test.go

@@ -0,0 +1,25 @@
+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")
+	}
+}
+