diff --git a/.gitignore b/.gitignore index 500d68ca..c2fff381 100644 --- a/.gitignore +++ b/.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] diff --git a/broker/broker.go b/broker/broker.go index 53099b7a..7e8dd9ae 100644 --- a/broker/broker.go +++ b/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. diff --git a/broker/options.go b/broker/options.go index ddc12e32..aa3a7272 100644 --- a/broker/options.go +++ b/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 diff --git a/logger/options.go b/logger/options.go index 8755c9b7..d3748749 100644 --- a/logger/options.go +++ b/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 { diff --git a/semconv/broker.go b/semconv/broker.go index 9b61b523..8c51a8f2 100644 --- a/semconv/broker.go +++ b/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" ) diff --git a/semconv/cache.go b/semconv/cache.go new file mode 100644 index 00000000..f12739c1 --- /dev/null +++ b/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" +) diff --git a/util/xpool/pool.go b/util/xpool/pool.go new file mode 100644 index 00000000..1ffe4293 --- /dev/null +++ b/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) +} diff --git a/util/xpool/pool_test.go b/util/xpool/pool_test.go new file mode 100644 index 00000000..8e7a9b81 --- /dev/null +++ b/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) +}