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

codec/codec.go

@@ -3,8 +3,6 @@ package codec
 
 import (
 	"errors"
-
-	"gopkg.in/yaml.v3"
 )
 
 var (
@@ -68,10 +66,10 @@ func (m *RawMessage) MarshalYAML() ([]byte, error) {
 }
 
 // UnmarshalYAML sets *m to a copy of data.
-func (m *RawMessage) UnmarshalYAML(n *yaml.Node) error {
+func (m *RawMessage) UnmarshalYAML(data []byte) error {
 	if m == nil {
 		return errors.New("RawMessage UnmarshalYAML on nil pointer")
 	}
-	*m = append((*m)[0:0], []byte(n.Value)...)
+	*m = append((*m)[0:0], data...)
 	return nil
 }

codec/frame.go

@@ -1,7 +1,5 @@
 package codec
 
-import "gopkg.in/yaml.v3"
-
 // Frame gives us the ability to define raw data to send over the pipes
 type Frame struct {
 	Data []byte
@@ -28,8 +26,8 @@ func (m *Frame) MarshalYAML() ([]byte, error) {
 }
 
 // UnmarshalYAML set frame data
-func (m *Frame) UnmarshalYAML(n *yaml.Node) error {
-	m.Data = []byte(n.Value)
+func (m *Frame) UnmarshalYAML(data []byte) error {
+	m.Data = append((m.Data)[0:0], data...)
 	return nil
 }
 

go.mod

@@ -6,7 +6,7 @@ require (
 	dario.cat/mergo v1.0.1
 	github.com/DATA-DOG/go-sqlmock v1.5.2
 	github.com/KimMachineGun/automemlimit v0.7.0
-	github.com/ash3in/uuidv8 v1.2.0
+	github.com/goccy/go-yaml v1.17.1
 	github.com/google/uuid v1.6.0
 	github.com/matoous/go-nanoid v1.5.1
 	github.com/patrickmn/go-cache v2.1.0+incompatible
@@ -19,7 +19,6 @@ require (
 	golang.org/x/sync v0.10.0
 	google.golang.org/grpc v1.69.4
 	google.golang.org/protobuf v1.36.3
-	gopkg.in/yaml.v3 v3.0.1
 )
 
 require (
@@ -31,4 +30,5 @@ require (
 	golang.org/x/sys v0.29.0 // indirect
 	google.golang.org/genproto/googleapis/rpc v0.0.0-20241216192217-9240e9c98484 // indirect
 	gopkg.in/check.v1 v1.0.0-20201130134442-10cb98267c6c // indirect
+	gopkg.in/yaml.v3 v3.0.1 // indirect
 )

go.sum

@@ -4,12 +4,12 @@ github.com/DATA-DOG/go-sqlmock v1.5.2 h1:OcvFkGmslmlZibjAjaHm3L//6LiuBgolP7Oputl
 github.com/DATA-DOG/go-sqlmock v1.5.2/go.mod h1:88MAG/4G7SMwSE3CeA0ZKzrT5CiOU3OJ+JlNzwDqpNU=
 github.com/KimMachineGun/automemlimit v0.7.0 h1:7G06p/dMSf7G8E6oq+f2uOPuVncFyIlDI/pBWK49u88=
 github.com/KimMachineGun/automemlimit v0.7.0/go.mod h1:QZxpHaGOQoYvFhv/r4u3U0JTC2ZcOwbSr11UZF46UBM=
-github.com/ash3in/uuidv8 v1.2.0 h1:2oogGdtCPwaVtyvPPGin4TfZLtOGE5F+W++E880G6SI=
-github.com/ash3in/uuidv8 v1.2.0/go.mod h1:BnU0wJBxnzdEKmVg4xckBkD+VZuecTFTUP3M0dWgyY4=
 github.com/davecgh/go-spew v1.1.2-0.20180830191138-d8f796af33cc h1:U9qPSI2PIWSS1VwoXQT9A3Wy9MM3WgvqSxFWenqJduM=
 github.com/davecgh/go-spew v1.1.2-0.20180830191138-d8f796af33cc/go.mod h1:J7Y8YcW2NihsgmVo/mv3lAwl/skON4iLHjSsI+c5H38=
 github.com/frankban/quicktest v1.14.6 h1:7Xjx+VpznH+oBnejlPUj8oUpdxnVs4f8XU8WnHkI4W8=
 github.com/frankban/quicktest v1.14.6/go.mod h1:4ptaffx2x8+WTWXmUCuVU6aPUX1/Mz7zb5vbUoiM6w0=
+github.com/goccy/go-yaml v1.17.1 h1:LI34wktB2xEE3ONG/2Ar54+/HJVBriAGJ55PHls4YuY=
+github.com/goccy/go-yaml v1.17.1/go.mod h1:XBurs7gK8ATbW4ZPGKgcbrY1Br56PdM69F7LkFRi1kA=
 github.com/golang/protobuf v1.5.4 h1:i7eJL8qZTpSEXOPTxNKhASYpMn+8e5Q6AdndVa1dWek=
 github.com/golang/protobuf v1.5.4/go.mod h1:lnTiLA8Wa4RWRcIUkrtSVa5nRhsEGBg48fD6rSs7xps=
 github.com/google/go-cmp v0.6.0 h1:ofyhxvXcZhMsU5ulbFiLKl/XBFqE1GSq7atu8tAmTRI=

util/id/id.go

@@ -2,12 +2,8 @@ package id
 
 import (
 	"crypto/rand"
-	"encoding/binary"
 	"errors"
-	"fmt"
-	"time"
 
-	uuidv8 "github.com/ash3in/uuidv8"
 	"github.com/google/uuid"
 	nanoid "github.com/matoous/go-nanoid"
 )
@@ -25,6 +21,7 @@ type Type int
 const (
 	TypeUnspecified Type = iota
 	TypeNanoid
+	TypeUUIDv7
 	TypeUUIDv8
 )
 
@@ -58,14 +55,14 @@ func (g *Generator) New() (string, error) {
 		}
 
 		return nanoid.Generate(g.opts.NanoidAlphabet, g.opts.NanoidSize)
-	case TypeUUIDv8:
-		timestamp := uint64(time.Now().UnixNano())
-		clockSeq := make([]byte, 2)
-		if _, err := rand.Read(clockSeq); err != nil {
-			return "", fmt.Errorf("failed to generate random clock sequence: %w", err)
+	case TypeUUIDv7:
+		uid, err := uuid.NewV7()
+		if err != nil {
+			return "", err
 		}
-		clockSeqValue := binary.BigEndian.Uint16(clockSeq) & 0x0FFF // Mask to 12 bits
-		return uuidv8.NewWithParams(timestamp, clockSeqValue, g.opts.UUIDNode[:], uuidv8.TimestampBits48)
+		return uid.String(), nil
+	case TypeUUIDv8:
+		return "", errors.New("unsupported uuid version v8")
 	}
 	return "", errors.New("invalid option, Type unspecified")
 }
@@ -82,16 +79,15 @@ func New(opts ...Option) (string, error) {
 		if options.NanoidSize <= 0 {
 			return "", errors.New("invalid option, NanoidSize must be positive integer")
 		}
-
 		return nanoid.Generate(options.NanoidAlphabet, options.NanoidSize)
-	case TypeUUIDv8:
-		timestamp := uint64(time.Now().UnixNano())
-		clockSeq := make([]byte, 2)
-		if _, err := rand.Read(clockSeq); err != nil {
-			return "", fmt.Errorf("failed to generate random clock sequence: %w", err)
+	case TypeUUIDv7:
+		uid, err := uuid.NewV7()
+		if err != nil {
+			return "", err
 		}
-		clockSeqValue := binary.BigEndian.Uint16(clockSeq) & 0x0FFF // Mask to 12 bits
-		return uuidv8.NewWithParams(timestamp, clockSeqValue, options.UUIDNode[:], uuidv8.TimestampBits48)
+		return uid.String(), nil
+	case TypeUUIDv8:
+		return "", errors.New("unsupported uuid version v8")
 	}
 
 	return "", errors.New("invalid option, Type unspecified")
@@ -145,7 +141,7 @@ func WithUUIDNode(node [6]byte) Option {
 // NewOptions returns new Options struct filled by opts
 func NewOptions(opts ...Option) Options {
 	options := Options{
-		Type:           TypeUUIDv8,
+		Type:           TypeUUIDv7,
 		NanoidAlphabet: DefaultNanoidAlphabet,
 		NanoidSize:     DefaultNanoidSize,
 		UUIDNode:       generatedNode,

util/time/duration.go

@@ -6,7 +6,7 @@ import (
 	"strconv"
 	"time"
 
-	"gopkg.in/yaml.v3"
+	"github.com/goccy/go-yaml"
 )
 
 type Duration int64
@@ -58,9 +58,9 @@ func (d Duration) MarshalYAML() (interface{}, error) {
 	return time.Duration(d).String(), nil
 }
 
-func (d *Duration) UnmarshalYAML(n *yaml.Node) error {
+func (d *Duration) UnmarshalYAML(data []byte) error {
 	var v interface{}
-	if err := yaml.Unmarshal([]byte(n.Value), &v); err != nil {
+	if err := yaml.Unmarshal(data, &v); err != nil {
 		return err
 	}
 	switch value := v.(type) {

util/time/duration_test.go

@@ -6,7 +6,7 @@ import (
 	"testing"
 	"time"
 
-	"gopkg.in/yaml.v3"
+	"github.com/goccy/go-yaml"
 )
 
 func TestMarshalYAML(t *testing.T) {