client: allow to set metadata for message

Signed-off-by: Vasiliy Tolstov <v.tolstov@unistack.org>
This commit is contained in:
Василий Толстов 2021-12-16 15:03:42 +03:00
parent a36f99e30b
commit e143e2b547
5 changed files with 21 additions and 3 deletions

View File

@ -49,6 +49,7 @@ type Message interface {
Topic() string Topic() string
Payload() interface{} Payload() interface{}
ContentType() string ContentType() string
Metadata() metadata.Metadata
} }
// Request is the interface for a synchronous request used by Call or Stream // Request is the interface for a synchronous request used by Call or Stream

View File

@ -12,7 +12,7 @@ import (
type LookupFunc func(context.Context, Request, CallOptions) ([]string, error) type LookupFunc func(context.Context, Request, CallOptions) ([]string, error)
// LookupRoute for a request using the router and then choose one using the selector // LookupRoute for a request using the router and then choose one using the selector
func LookupRoute(ctx context.Context, req Request, opts CallOptions) ([]string, error) { func LookupRoute(_ context.Context, req Request, opts CallOptions) ([]string, error) {
// check to see if an address was provided as a call option // check to see if an address was provided as a call option
if len(opts.Address) > 0 { if len(opts.Address) > 0 {
return opts.Address, nil return opts.Address, nil

View File

@ -139,6 +139,10 @@ func (n *noopMessage) ContentType() string {
return n.opts.ContentType return n.opts.ContentType
} }
func (n *noopMessage) Metadata() metadata.Metadata {
return n.opts.Metadata
}
func (n *noopClient) newCodec(contentType string) (codec.Codec, error) { func (n *noopClient) newCodec(contentType string) (codec.Codec, error) {
if cf, ok := n.opts.Codecs[contentType]; ok { if cf, ok := n.opts.Codecs[contentType]; ok {
return cf, nil return cf, nil

View File

@ -8,6 +8,7 @@ import (
"go.unistack.org/micro/v3/broker" "go.unistack.org/micro/v3/broker"
"go.unistack.org/micro/v3/codec" "go.unistack.org/micro/v3/codec"
"go.unistack.org/micro/v3/logger" "go.unistack.org/micro/v3/logger"
"go.unistack.org/micro/v3/metadata"
"go.unistack.org/micro/v3/meter" "go.unistack.org/micro/v3/meter"
"go.unistack.org/micro/v3/network/transport" "go.unistack.org/micro/v3/network/transport"
"go.unistack.org/micro/v3/register" "go.unistack.org/micro/v3/register"
@ -128,7 +129,7 @@ type PublishOptions struct {
// NewMessageOptions creates message options struct // NewMessageOptions creates message options struct
func NewMessageOptions(opts ...MessageOption) MessageOptions { func NewMessageOptions(opts ...MessageOption) MessageOptions {
options := MessageOptions{} options := MessageOptions{Metadata: metadata.New(1)}
for _, o := range opts { for _, o := range opts {
o(&options) o(&options)
} }
@ -137,7 +138,10 @@ func NewMessageOptions(opts ...MessageOption) MessageOptions {
// MessageOptions holds client message options // MessageOptions holds client message options
type MessageOptions struct { type MessageOptions struct {
// Metadata additional metadata
Metadata metadata.Metadata
// ContentType specify content-type of message // ContentType specify content-type of message
// deprecated
ContentType string ContentType string
} }
@ -517,6 +521,7 @@ func WithSelectOptions(sops ...selector.SelectOption) CallOption {
// Deprecated // Deprecated
func WithMessageContentType(ct string) MessageOption { func WithMessageContentType(ct string) MessageOption {
return func(o *MessageOptions) { return func(o *MessageOptions) {
o.Metadata.Set(metadata.HeaderContentType, ct)
o.ContentType = ct o.ContentType = ct
} }
} }
@ -524,10 +529,18 @@ func WithMessageContentType(ct string) MessageOption {
// MessageContentType sets the message content type // MessageContentType sets the message content type
func MessageContentType(ct string) MessageOption { func MessageContentType(ct string) MessageOption {
return func(o *MessageOptions) { return func(o *MessageOptions) {
o.Metadata.Set(metadata.HeaderContentType, ct)
o.ContentType = ct o.ContentType = ct
} }
} }
// MessageMetadata sets the message metadata
func MessageMetadata(k, v string) MessageOption {
return func(o *MessageOptions) {
o.Metadata.Set(k, v)
}
}
// StreamingRequest specifies that request is streaming // StreamingRequest specifies that request is streaming
func StreamingRequest(b bool) RequestOption { func StreamingRequest(b bool) RequestOption {
return func(o *RequestOptions) { return func(o *RequestOptions) {

View File

@ -20,7 +20,7 @@ func RetryNever(ctx context.Context, req Request, retryCount int, err error) (bo
} }
// RetryOnError retries a request on a 500 or timeout error // RetryOnError retries a request on a 500 or timeout error
func RetryOnError(ctx context.Context, req Request, retryCount int, err error) (bool, error) { func RetryOnError(_ context.Context, _ Request, _ int, err error) (bool, error) {
if err == nil { if err == nil {
return false, nil return false, nil
} }