metadata: minor fixup for NewXXXContext functions

Signed-off-by: Vasiliy Tolstov <v.tolstov@unistack.org>

.github/renovate.json

@@ -2,6 +2,7 @@
   "extends": [
     "config:base"
   ],
+  "postUpdateOptions": ["gomodTidy"],
   "packageRules": [
     {
       "matchUpdateTypes": ["minor", "patch", "pin", "digest"],

go.mod

@@ -6,12 +6,8 @@ require (
 	github.com/dgrijalva/jwt-go v3.2.0+incompatible
 	github.com/ef-ds/deque v1.0.4
 	github.com/google/uuid v1.2.0
-	github.com/heimdalr/dag v1.0.1 // indirect
 	github.com/imdario/mergo v0.3.11
-	github.com/kr/text v0.2.0 // indirect
-	github.com/niemeyer/pretty v0.0.0-20200227124842-a10e7caefd8e // indirect
 	github.com/patrickmn/go-cache v2.1.0+incompatible
-	github.com/silas/dag v0.0.0-20210121180416-41cf55125c34 // indirect
-	golang.org/x/net v0.0.0-20210119194325-5f4716e94777
-	gopkg.in/check.v1 v1.0.0-20200227125254-8fa46927fb4f // indirect
+	github.com/silas/dag v0.0.0-20210121180416-41cf55125c34
+	golang.org/x/net v0.0.0-20210226172049-e18ecbb05110
 )

go.sum

@@ -1,35 +1,21 @@
-github.com/creack/pty v1.1.9/go.mod h1:oKZEueFk5CKHvIhNR5MUki03XCEU+Q6VDXinZuGJ33E=
 github.com/dgrijalva/jwt-go v3.2.0+incompatible h1:7qlOGliEKZXTDg6OTjfoBKDXWrumCAMpl/TFQ4/5kLM=
 github.com/dgrijalva/jwt-go v3.2.0+incompatible/go.mod h1:E3ru+11k8xSBh+hMPgOLZmtrrCbhqsmaPHjLKYnJCaQ=
 github.com/ef-ds/deque v1.0.4 h1:iFAZNmveMT9WERAkqLJ+oaABF9AcVQ5AjXem/hroniI=
 github.com/ef-ds/deque v1.0.4/go.mod h1:gXDnTC3yqvBcHbq2lcExjtAcVrOnJCbMcZXmuj8Z4tg=
-github.com/go-test/deep v1.0.7 h1:/VSMRlnY/JSyqxQUzQLKVMAskpY/NZKFA5j2P+0pP2M=
-github.com/go-test/deep v1.0.7/go.mod h1:QV8Hv/iy04NyLBxAdO9njL0iVPN1S4d/A3NVv1V36o8=
-github.com/google/uuid v1.1.2/go.mod h1:TIyPZe4MgqvfeYDBFedMoGGpEw/LqOeaOT+nhxU+yHo=
 github.com/google/uuid v1.2.0 h1:qJYtXnJRWmpe7m/3XlyhrsLrEURqHRM2kxzoxXqyUDs=
 github.com/google/uuid v1.2.0/go.mod h1:TIyPZe4MgqvfeYDBFedMoGGpEw/LqOeaOT+nhxU+yHo=
-github.com/heimdalr/dag v1.0.1 h1:iR2K3DSUFDYx0GeV7iXBnZkedWS1xePSGrylQ197uxg=
-github.com/heimdalr/dag v1.0.1/go.mod h1:t+ZkR+sjKL4xhlE1B9rwpvwfo+x+2R0363efS+Oghns=
 github.com/imdario/mergo v0.3.11 h1:3tnifQM4i+fbajXKBHXWEH+KvNHqojZ778UH75j3bGA=
 github.com/imdario/mergo v0.3.11/go.mod h1:jmQim1M+e3UYxmgPu/WyfjB3N3VflVyUjjjwH0dnCYA=
-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/niemeyer/pretty v0.0.0-20200227124842-a10e7caefd8e h1:fD57ERR4JtEqsWbfPhv4DMiApHyliiK5xCTNVSPiaAs=
-github.com/niemeyer/pretty v0.0.0-20200227124842-a10e7caefd8e/go.mod h1:zD1mROLANZcx1PVRCS0qkT7pwLkGfwJo4zjcN/Tysno=
 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/silas/dag v0.0.0-20210121180416-41cf55125c34 h1:vBfVmA5mZhsQa2jr1FOL9nfA37N/jnbBmi5XUfviVTI=
 github.com/silas/dag v0.0.0-20210121180416-41cf55125c34/go.mod h1:7RTUFBdIRC9nZ7/3RyRNH1bdqIShrDejd1YbLwgPS+I=
-golang.org/x/net v0.0.0-20210119194325-5f4716e94777 h1:003p0dJM77cxMSyCPFphvZf/Y5/NXf5fzg6ufd1/Oew=
-golang.org/x/net v0.0.0-20210119194325-5f4716e94777/go.mod h1:m0MpNAwzfU5UDzcl9v0D8zg8gWTRqZa9RBIspLL5mdg=
+golang.org/x/net v0.0.0-20210226172049-e18ecbb05110 h1:qWPm9rbaAMKs8Bq/9LRpbMqxWRVUAQwMI9fVrssnTfw=
+golang.org/x/net v0.0.0-20210226172049-e18ecbb05110/go.mod h1:m0MpNAwzfU5UDzcl9v0D8zg8gWTRqZa9RBIspLL5mdg=
 golang.org/x/sys v0.0.0-20201119102817-f84b799fce68/go.mod h1:h1NjWce9XRLGQEsW7wpKNCjG9DtNlClVuFLEZdDNbEs=
 golang.org/x/term v0.0.0-20201126162022-7de9c90e9dd1/go.mod h1:bj7SfCRtBDWHUb9snDiAeCFNEtKQo2Wmx5Cou7ajbmo=
 golang.org/x/text v0.3.3/go.mod h1:5Zoc/QRtKVWzQhOtBMvqHzDpF6irO9z98xDceosuGiQ=
 golang.org/x/tools v0.0.0-20180917221912-90fa682c2a6e/go.mod h1:n7NCudcB/nEzxVGmLbDWY5pfWTLqBcC2KZ6jyYvM4mQ=
 gopkg.in/check.v1 v0.0.0-20161208181325-20d25e280405/go.mod h1:Co6ibVJAznAaIkqp8huTwlJQCZ016jof/cbN4VW5Yz0=
-gopkg.in/check.v1 v1.0.0-20200227125254-8fa46927fb4f h1:BLraFXnmrev5lT+xlilqcH8XK9/i0At2xKjWk4p6zsU=
-gopkg.in/check.v1 v1.0.0-20200227125254-8fa46927fb4f/go.mod h1:Co6ibVJAznAaIkqp8huTwlJQCZ016jof/cbN4VW5Yz0=
 gopkg.in/yaml.v2 v2.3.0 h1:clyUAQHOM3G0M3f5vQj7LuJrETvjVot3Z5el9nffUtU=
 gopkg.in/yaml.v2 v2.3.0/go.mod h1:hI93XBmqTisBFMUTm0b8Fm+jr3Dg1NNxqwp+5A1VGuI=

metadata/context.go

@@ -93,7 +93,9 @@ func NewIncomingContext(ctx context.Context, md Metadata) context.Context {
 		ctx = context.Background()
 	}
 	ctx = context.WithValue(ctx, mdIncomingKey{}, &rawMetadata{md})
-	ctx = context.WithValue(ctx, mdOutgoingKey{}, &rawMetadata{})
+	if v, ok := ctx.Value(mdOutgoingKey{}).(*rawMetadata); !ok || v == nil {
+		ctx = context.WithValue(ctx, mdOutgoingKey{}, &rawMetadata{})
+	}
 	return ctx
 }
 
@@ -103,6 +105,40 @@ func NewOutgoingContext(ctx context.Context, md Metadata) context.Context {
 		ctx = context.Background()
 	}
 	ctx = context.WithValue(ctx, mdOutgoingKey{}, &rawMetadata{md})
-	ctx = context.WithValue(ctx, mdIncomingKey{}, &rawMetadata{})
+	if v, ok := ctx.Value(mdIncomingKey{}).(*rawMetadata); !ok || v == nil {
+		ctx = context.WithValue(ctx, mdIncomingKey{}, &rawMetadata{})
+	}
 	return ctx
 }
+
+// AppendOutgoingContext apends new md to context
+func AppendOutgoingContext(ctx context.Context, kv ...string) context.Context {
+	md, ok := Pairs(kv...)
+	if !ok {
+		return ctx
+	}
+	omd, ok := FromOutgoingContext(ctx)
+	if !ok {
+		return NewOutgoingContext(ctx, md)
+	}
+	for k, v := range md {
+		omd.Set(k, v)
+	}
+	return NewOutgoingContext(ctx, omd)
+}
+
+// AppendIncomingContext apends new md to context
+func AppendIncomingContext(ctx context.Context, kv ...string) context.Context {
+	md, ok := Pairs(kv...)
+	if !ok {
+		return ctx
+	}
+	omd, ok := FromIncomingContext(ctx)
+	if !ok {
+		return NewIncomingContext(ctx, md)
+	}
+	for k, v := range md {
+		omd.Set(k, v)
+	}
+	return NewIncomingContext(ctx, omd)
+}

metadata/metadata.go

@@ -111,3 +111,19 @@ func Merge(omd Metadata, mmd Metadata, overwrite bool) Metadata {
 	}
 	return nmd
 }
+
+func Pairs(kv ...string) (Metadata, bool) {
+	if len(kv)%2 == 1 {
+		return nil, false
+	}
+	md := New(len(kv) / 2)
+	var k string
+	for i, v := range kv {
+		if i%2 == 0 {
+			k = v
+			continue
+		}
+		md.Set(k, v)
+	}
+	return md, true
+}

metadata/metadata_test.go

@@ -5,6 +5,28 @@ import (
 	"testing"
 )
 
+func TestAppend(t *testing.T) {
+	ctx := context.Background()
+	ctx = AppendIncomingContext(ctx, "key1", "val1", "key2", "val2")
+	md, ok := FromIncomingContext(ctx)
+	if !ok {
+		t.Fatal("metadata empty")
+	}
+	if _, ok := md.Get("key1"); !ok {
+		t.Fatal("key1 not found")
+	}
+}
+
+func TestPairs(t *testing.T) {
+	md, ok := Pairs("key1", "val1", "key2", "val2")
+	if !ok {
+		t.Fatal("odd number of kv")
+	}
+	if _, ok = md.Get("key1"); !ok {
+		t.Fatal("key1 not found")
+	}
+}
+
 func testCtx(ctx context.Context) {
 	md := New(2)
 	md.Set("Key1", "Val1_new")

server/errors.go

@@ -0,0 +1,59 @@
+package server
+
+import "github.com/unistack-org/micro/v3/errors"
+
+type Error struct {
+	id string
+}
+
+func NewError(id string) *Error {
+	return &Error{id}
+}
+
+func (e *Error) BadRequest(format string, a ...interface{}) error {
+	return errors.BadRequest(e.id, format, a...)
+}
+
+func (e *Error) Unauthorized(format string, a ...interface{}) error {
+	return errors.Unauthorized(e.id, format, a...)
+}
+
+func (e *Error) Forbidden(format string, a ...interface{}) error {
+	return errors.Forbidden(e.id, format, a...)
+}
+
+func (e *Error) NotFound(format string, a ...interface{}) error {
+	return errors.NotFound(e.id, format, a...)
+}
+
+func (e *Error) MethodNotAllowed(format string, a ...interface{}) error {
+	return errors.MethodNotAllowed(e.id, format, a...)
+}
+
+func (e *Error) Timeout(format string, a ...interface{}) error {
+	return errors.Timeout(e.id, format, a...)
+}
+
+func (e *Error) Conflict(format string, a ...interface{}) error {
+	return errors.Conflict(e.id, format, a...)
+}
+
+func (e *Error) InternalServerError(format string, a ...interface{}) error {
+	return errors.InternalServerError(e.id, format, a...)
+}
+
+func (e *Error) NotImplemented(format string, a ...interface{}) error {
+	return errors.NotImplemented(e.id, format, a...)
+}
+
+func (e *Error) BadGateway(format string, a ...interface{}) error {
+	return errors.BadGateway(e.id, format, a...)
+}
+
+func (e *Error) ServiceUnavailable(format string, a ...interface{}) error {
+	return errors.ServiceUnavailable(e.id, format, a...)
+}
+
+func (e *Error) GatewayTimeout(format string, a ...interface{}) error {
+	return errors.GatewayTimeout(e.id, format, a...)
+}

server/errors_test.go

@@ -0,0 +1,19 @@
+package server
+
+import (
+	"testing"
+
+	"github.com/unistack-org/micro/v3/errors"
+)
+
+func TestError(t *testing.T) {
+	e := NewError("svc1")
+	err := e.BadRequest("%s", "test")
+	merr, ok := err.(*errors.Error)
+	if !ok {
+		t.Fatal("error not *errors.Error")
+	}
+	if merr.Id != "svc1" {
+		t.Fatal("id != svc1")
+	}
+}

server/options.go

@@ -332,6 +332,7 @@ type SubscriberOptions struct {
 	AutoAck  bool
 	Queue    string
 	Internal bool
+	BodyOnly bool
 	Context  context.Context
 }
 
@@ -389,6 +390,20 @@ func SubscriberQueue(n string) SubscriberOption {
 	}
 }
 
+// SubscriberGroup sets the shared group name distributed messages across subscribers
+func SubscriberGroup(n string) SubscriberOption {
+	return func(o *SubscriberOptions) {
+		o.Queue = n
+	}
+}
+
+// SubscriberBodyOnly says broker that message contains raw data with absence of micro broker.Message format
+func SubscriberBodyOnly(b bool) SubscriberOption {
+	return func(o *SubscriberOptions) {
+		o.BodyOnly = b
+	}
+}
+
 // SubscriberContext set context options to allow broker SubscriberOption passed
 func SubscriberContext(ctx context.Context) SubscriberOption {
 	return func(o *SubscriberOptions) {