Merge branch 'master' into test_coverage

.gitignore

@@ -1,6 +1,8 @@
 # Develop tools
 /.vscode/
 /.idea/
+.idea
+.vscode
 
 # Binaries for programs and plugins
 *.exe
@@ -13,6 +15,7 @@
 _obj
 _test
 _build
+.DS_Store
 
 # Architecture specific extensions/prefixes
 *.[568vq]

broker/broker.go

@@ -4,6 +4,7 @@ package broker // import "go.unistack.org/micro/v4/broker"
 import (
 	"context"
 	"errors"
+	"time"
 
 	"go.unistack.org/micro/v4/metadata"
 	"go.unistack.org/micro/v4/options"
@@ -19,6 +20,8 @@ var (
 	ErrDisconnected = errors.New("broker disconnected")
 	// ErrInvalidMessage returns when message has nvalid format
 	ErrInvalidMessage = errors.New("broker message has invalid format")
+	// DefaultGracefulTimeout
+	DefaultGracefulTimeout = 5 * time.Second
 )
 
 // Broker is an interface used for asynchronous messaging.

broker/options.go

@@ -11,6 +11,7 @@ import (
 	"go.unistack.org/micro/v4/meter"
 	"go.unistack.org/micro/v4/options"
 	"go.unistack.org/micro/v4/register"
+	"go.unistack.org/micro/v4/sync"
 	"go.unistack.org/micro/v4/tracer"
 )
 
@@ -36,22 +37,27 @@ type Options struct {
 	Name string
 	// Address holds the broker address
 	Address []string
+
+	Wait *sync.WaitGroup
+
+	GracefulTimeout time.Duration
 }
 
 // NewOptions create new Options
 func NewOptions(opts ...options.Option) Options {
-	options := Options{
-		Register: register.DefaultRegister,
-		Logger:   logger.DefaultLogger,
-		Context:  context.Background(),
-		Meter:    meter.DefaultMeter,
-		Codecs:   make(map[string]codec.Codec),
-		Tracer:   tracer.DefaultTracer,
+	newOpts := Options{
+		Register:        register.DefaultRegister,
+		Logger:          logger.DefaultLogger,
+		Context:         context.Background(),
+		Meter:           meter.DefaultMeter,
+		Codecs:          make(map[string]codec.Codec),
+		Tracer:          tracer.DefaultTracer,
+		GracefulTimeout: DefaultGracefulTimeout,
 	}
 	for _, o := range opts {
-		o(&options)
+		o(&newOpts)
 	}
-	return options
+	return newOpts
 }
 
 // PublishOptions struct

logger/options.go

@@ -214,6 +214,20 @@ func WithMicroKeys() options.Option {
 	}
 }
 
+// WithAddCallerSkipCount add skip count for copy logger
+func WithAddCallerSkipCount(n int) options.Option {
+	return func(src interface{}) error {
+		c, err := options.Get(src, ".CallerSkipCount")
+		if err != nil {
+			return err
+		}
+		if err = options.Set(src, c.(int)+n, ".CallerSkipCount"); err != nil {
+			return err
+		}
+		return nil
+	}
+}
+
 // WithAddStacktrace controls writing stacktrace on error
 func WithAddStacktrace(v bool) options.Option {
 	return func(src interface{}) error {

semconv/broker.go

@@ -17,4 +17,6 @@ var (
 	SubscribeMessageTotal = "subscribe_message_total"
 	// SubscribeMessageInflight specifies meter metric name
 	SubscribeMessageInflight = "subscribe_message_inflight"
+	// BrokerGroupLag specifies broker lag
+	BrokerGroupLag = "broker_lag"
 )

semconv/cache.go

@@ -0,0 +1,12 @@
+package semconv
+
+var (
+	// CacheRequestDurationSeconds specifies meter metric name
+	CacheRequestDurationSeconds = "cache_request_duration_seconds"
+	// ClientRequestLatencyMicroseconds specifies meter metric name
+	CacheRequestLatencyMicroseconds = "cache_request_latency_microseconds"
+	// CacheRequestTotal specifies meter metric name
+	CacheRequestTotal = "cache_request_total"
+	// CacheRequestInflight specifies meter metric name
+	CacheRequestInflight = "cache_request_inflight"
+)

util/xpool/pool.go

@@ -0,0 +1,25 @@
+package pool
+
+import "sync"
+
+type Pool[T any] struct {
+	p *sync.Pool
+}
+
+func NewPool[T any](fn func() T) Pool[T] {
+	return Pool[T]{
+		p: &sync.Pool{
+			New: func() interface{} {
+				return fn()
+			},
+		},
+	}
+}
+
+func (p Pool[T]) Get() T {
+	return p.p.Get().(T)
+}
+
+func (p Pool[T]) Put(t T) {
+	p.p.Put(t)
+}

util/xpool/pool_test.go

@@ -0,0 +1,27 @@
+package pool
+
+import (
+	"bytes"
+	"strings"
+	"testing"
+)
+
+func TestBytes(t *testing.T) {
+	p := NewPool(func() *bytes.Buffer { return bytes.NewBuffer(nil) })
+	b := p.Get()
+	b.Write([]byte(`test`))
+	if b.String() != "test" {
+		t.Fatal("pool not works")
+	}
+	p.Put(b)
+}
+
+func TestStrings(t *testing.T) {
+	p := NewPool(func() *strings.Builder { return &strings.Builder{} })
+	b := p.Get()
+	b.Write([]byte(`test`))
+	if b.String() != "test" {
+		t.Fatal("pool not works")
+	}
+	p.Put(b)
+}