broker: pass broker content type if message options not pass it

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

README.md

@@ -1,5 +1,5 @@
 # Micro
-![Coverage](https://img.shields.io/badge/Coverage-33.6%25-yellow)
+![Coverage](https://img.shields.io/badge/Coverage-33.7%25-yellow)
 [![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/go.unistack.org/micro/v4?tab=overview)
 [![Status](https://git.unistack.org/unistack-org/micro/actions/workflows/job_tests.yml/badge.svg?branch=v4)](https://git.unistack.org/unistack-org/micro/actions?query=workflow%3Abuild+branch%3Av4+event%3Apush)

broker/memory/memory.go

@@ -159,6 +159,9 @@ func (b *Broker) Init(opts ...broker.Option) error {
 
 func (b *Broker) NewMessage(ctx context.Context, hdr metadata.Metadata, body interface{}, opts ...broker.PublishOption) (broker.Message, error) {
 	options := broker.NewPublishOptions(opts...)
+	if options.ContentType == "" {
+		options.ContentType = b.opts.ContentType
+	}
 	m := &memoryMessage{ctx: ctx, hdr: hdr, opts: options}
 	c, err := b.newCodec(m.opts.ContentType)
 	if err == nil {

broker/noop.go

@@ -128,6 +128,9 @@ func (m *noopMessage) Unmarshal(dst interface{}, opts ...codec.Option) error {
 
 func (b *NoopBroker) NewMessage(ctx context.Context, hdr metadata.Metadata, body interface{}, opts ...PublishOption) (Message, error) {
 	options := NewPublishOptions(opts...)
+	if options.ContentType == "" {
+		options.ContentType = b.opts.ContentType
+	}
 	m := &noopMessage{ctx: ctx, hdr: hdr, opts: options}
 	c, err := b.newCodec(m.opts.ContentType)
 	if err == nil {

broker/options.go

@@ -45,6 +45,9 @@ type Options struct {
 
 	// GracefulTimeout contains time to wait to finish in flight requests
 	GracefulTimeout time.Duration
+
+	// ContentType will be used if no content-type set when creating message
+	ContentType string
 }
 
 // NewOptions create new Options
@@ -57,14 +60,19 @@ func NewOptions(opts ...Option) Options {
 		Codecs:          make(map[string]codec.Codec),
 		Tracer:          tracer.DefaultTracer,
 		GracefulTimeout: DefaultGracefulTimeout,
+		ContentType:     DefaultContentType,
 	}
 
 	for _, o := range opts {
 		o(&options)
 	}
+
 	return options
 }
 
+// DefaultContentType is the default content-type if not specified
+var DefaultContentType = ""
+
 // Context sets the context option
 func Context(ctx context.Context) Option {
 	return func(o *Options) {
@@ -72,6 +80,13 @@ func Context(ctx context.Context) Option {
 	}
 }
 
+// ContentType used by default if not specified
+func ContentType(ct string) Option {
+	return func(o *Options) {
+		o.ContentType = ct
+	}
+}
+
 // PublishOptions struct
 type PublishOptions struct {
 	// ContentType for message body