util/xpool: package pool

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

.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/v3/broker"
 import (
 	"context"
 	"errors"
+	"time"
 
 	"go.unistack.org/micro/v3/codec"
 	"go.unistack.org/micro/v3/metadata"
@@ -17,6 +18,8 @@ var (
 	ErrNotConnected = errors.New("broker not connected")
 	// ErrDisconnected returns when broker disconnected
 	ErrDisconnected = errors.New("broker disconnected")
+	// DefaultGracefulTimeout
+	DefaultGracefulTimeout = 5 * time.Second
 )
 
 // Broker is an interface used for asynchronous messaging.

broker/options.go

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

cluster/cluster.go

@@ -0,0 +1,41 @@
+package cluster
+
+import (
+	"context"
+
+	"go.unistack.org/micro/v3/metadata"
+)
+
+// Message sent to member in cluster
+type Message interface {
+	// Header returns message headers
+	Header() metadata.Metadata
+	// Body returns broker message may be []byte slice or some go struct or interface
+	Body() interface{}
+}
+
+type Node interface {
+	// Name returns node name
+	Name() string
+	// Address returns node address
+	Address() string
+	// Metadata returns node metadata
+	Metadata() metadata.Metadata
+}
+
+// Cluster interface used for cluster communication across nodes
+type Cluster interface {
+	// Join is used to take an existing members and performing state sync
+	Join(ctx context.Context, addr ...string) error
+	// Leave broadcast a leave message and stop listeners
+	Leave(ctx context.Context) error
+	// Ping is used to probe live status of the node
+	Ping(ctx context.Context, node Node, payload []byte) error
+	// Members returns the cluster members
+	Members() ([]Node, error)
+	// Broadcast send message for all members in cluster, if filter is not nil, nodes may be filtered
+	// by key/value pairs
+	Broadcast(ctx context.Context, msg Message, filter ...string) error
+	// Unicast send message to single member in cluster
+	Unicast(ctx context.Context, node Node, msg Message) error
+}

semconv/broker.go

@@ -0,0 +1,22 @@
+package semconv
+
+var (
+	// PublishMessageDurationSeconds specifies meter metric name
+	PublishMessageDurationSeconds = "publish_message_duration_seconds"
+	// PublishMessageLatencyMicroseconds specifies meter metric name
+	PublishMessageLatencyMicroseconds = "publish_message_latency_microseconds"
+	// PublishMessageTotal specifies meter metric name
+	PublishMessageTotal = "publish_message_total"
+	// PublishMessageInflight specifies meter metric name
+	PublishMessageInflight = "publish_message_inflight"
+	// SubscribeMessageDurationSeconds specifies meter metric name
+	SubscribeMessageDurationSeconds = "subscribe_message_duration_seconds"
+	// SubscribeMessageLatencyMicroseconds specifies meter metric name
+	SubscribeMessageLatencyMicroseconds = "subscribe_message_latency_microseconds"
+	// SubscribeMessageTotal specifies meter metric name
+	SubscribeMessageTotal = "subscribe_message_total"
+	// SubscribeMessageInflight specifies meter metric name
+	SubscribeMessageInflight = "subscribe_message_inflight"
+	// BrokerGroupLag specifies broker lag
+	BrokerGroupLag = "broker_group_lag"
+)

semconv/client.go

@@ -0,0 +1,12 @@
+package semconv
+
+var (
+	// ClientRequestDurationSeconds specifies meter metric name
+	ClientRequestDurationSeconds = "client_request_duration_seconds"
+	// ClientRequestLatencyMicroseconds specifies meter metric name
+	ClientRequestLatencyMicroseconds = "client_request_latency_microseconds"
+	// ClientRequestTotal specifies meter metric name
+	ClientRequestTotal = "client_request_total"
+	// ClientRequestInflight specifies meter metric name
+	ClientRequestInflight = "client_request_inflight"
+)

semconv/server.go

@@ -0,0 +1,12 @@
+package semconv
+
+var (
+	// ServerRequestDurationSeconds specifies meter metric name
+	ServerRequestDurationSeconds = "server_request_duration_seconds"
+	// ServerRequestLatencyMicroseconds specifies meter metric name
+	ServerRequestLatencyMicroseconds = "server_request_latency_microseconds"
+	// ServerRequestTotal specifies meter metric name
+	ServerRequestTotal = "server_request_total"
+	// ServerRequestInflight specifies meter metric name
+	ServerRequestInflight = "server_request_inflight"
+)

server/options.go

@@ -15,6 +15,7 @@ import (
 	"go.unistack.org/micro/v3/network/transport"
 	"go.unistack.org/micro/v3/options"
 	"go.unistack.org/micro/v3/register"
+	msync "go.unistack.org/micro/v3/sync"
 	"go.unistack.org/micro/v3/tracer"
 	"go.unistack.org/micro/v3/util/id"
 )
@@ -47,7 +48,7 @@ type Options struct {
 	// Listener may be passed if already created
 	Listener net.Listener
 	// Wait group
-	Wait *sync.WaitGroup
+	Wait *msync.WaitGroup
 	// TLSConfig specifies tls.Config for secure serving
 	TLSConfig *tls.Config
 	// Metadata holds the server metadata
@@ -282,7 +283,7 @@ func Wait(wg *sync.WaitGroup) Option {
 		if wg == nil {
 			wg = new(sync.WaitGroup)
 		}
-		o.Wait = wg
+		o.Wait = msync.WrapWaitGroup(wg)
 	}
 }
 
@@ -331,7 +332,6 @@ func GracefulTimeout(td time.Duration) Option {
 	}
 }
 
-
 // HandlerOptions struct
 type HandlerOptions struct {
 	// Context holds external options

store/options.go

@@ -144,6 +144,10 @@ type ReadOptions struct {
 	Context context.Context
 	// Namespace holds namespace
 	Namespace string
+	// Name holds mnemonic name
+	Name string
+	// Timeout specifies max timeout for operation
+	Timeout time.Duration
 }
 
 // NewReadOptions fills ReadOptions struct with opts slice
@@ -158,6 +162,20 @@ func NewReadOptions(opts ...ReadOption) ReadOptions {
 // ReadOption sets values in ReadOptions
 type ReadOption func(r *ReadOptions)
 
+// ReadTimeout pass timeout to ReadOptions
+func ReadTimeout(td time.Duration) ReadOption {
+	return func(o *ReadOptions) {
+		o.Timeout = td
+	}
+}
+
+// ReadName pass name to ReadOptions
+func ReadName(name string) ReadOption {
+	return func(o *ReadOptions) {
+		o.Name = name
+	}
+}
+
 // ReadContext pass context.Context to ReadOptions
 func ReadContext(ctx context.Context) ReadOption {
 	return func(o *ReadOptions) {
@@ -180,6 +198,10 @@ type WriteOptions struct {
 	Metadata metadata.Metadata
 	// Namespace holds namespace
 	Namespace string
+	// Name holds mnemonic name
+	Name string
+	// Timeout specifies max timeout for operation
+	Timeout time.Duration
 	// TTL specifies key TTL
 	TTL time.Duration
 }
@@ -224,12 +246,30 @@ func WriteNamespace(ns string) WriteOption {
 	}
 }
 
+// WriteName pass name to WriteOptions
+func WriteName(name string) WriteOption {
+	return func(o *WriteOptions) {
+		o.Name = name
+	}
+}
+
+// WriteTimeout pass timeout to WriteOptions
+func WriteTimeout(td time.Duration) WriteOption {
+	return func(o *WriteOptions) {
+		o.Timeout = td
+	}
+}
+
 // DeleteOptions configures an individual Delete operation
 type DeleteOptions struct {
 	// Context holds external options
 	Context context.Context
 	// Namespace holds namespace
 	Namespace string
+	// Name holds mnemonic name
+	Name string
+	// Timeout specifies max timeout for operation
+	Timeout time.Duration
 }
 
 // NewDeleteOptions fills DeleteOptions struct with opts slice
@@ -258,14 +298,32 @@ func DeleteNamespace(ns string) DeleteOption {
 	}
 }
 
+// DeleteName pass name to DeleteOptions
+func DeleteName(name string) DeleteOption {
+	return func(o *DeleteOptions) {
+		o.Name = name
+	}
+}
+
+// DeleteTimeout pass timeout to DeleteOptions
+func DeleteTimeout(td time.Duration) DeleteOption {
+	return func(o *DeleteOptions) {
+		o.Timeout = td
+	}
+}
+
 // ListOptions configures an individual List operation
 type ListOptions struct {
 	Context   context.Context
 	Prefix    string
 	Suffix    string
 	Namespace string
-	Limit     uint
+	// Name holds mnemonic name
-	Offset    uint
+	Name   string
+	Limit  uint
+	Offset uint
+	// Timeout specifies max timeout for operation
+	Timeout time.Duration
 }
 
 // NewListOptions fills ListOptions struct with opts slice
@@ -322,12 +380,23 @@ func ListNamespace(ns string) ListOption {
 	}
 }
 
+// ListTimeout pass timeout to ListOptions
+func ListTimeout(td time.Duration) ListOption {
+	return func(o *ListOptions) {
+		o.Timeout = td
+	}
+}
+
 // ExistsOptions holds options for Exists method
 type ExistsOptions struct {
 	// Context holds external options
 	Context context.Context
 	// Namespace contains namespace
 	Namespace string
+	// Name holds mnemonic name
+	Name string
+	// Timeout specifies max timeout for operation
+	Timeout time.Duration
 }
 
 // ExistsOption specifies Exists call options
@@ -358,6 +427,20 @@ func ExistsNamespace(ns string) ExistsOption {
 	}
 }
 
+// ExistsName pass name to exist options
+func ExistsName(name string) ExistsOption {
+	return func(o *ExistsOptions) {
+		o.Name = name
+	}
+}
+
+// ExistsTimeout timeout to ListOptions
+func ExistsTimeout(td time.Duration) ExistsOption {
+	return func(o *ExistsOptions) {
+		o.Timeout = td
+	}
+}
+
 /*
 // WrapStore adds a store Wrapper to a list of options passed into the store
 func WrapStore(w Wrapper) Option {

tracer/options.go

@@ -100,13 +100,13 @@ type EventOption func(o *EventOptions)
 
 func WithEventLabels(kv ...interface{}) EventOption {
 	return func(o *EventOptions) {
-		o.Labels = kv
+		o.Labels = append(o.Labels, kv...)
 	}
 }
 
 func WithSpanLabels(kv ...interface{}) SpanOption {
 	return func(o *SpanOptions) {
-		o.Labels = kv
+		o.Labels = append(o.Labels, kv...)
 	}
 }
 

tracer/tracer.go

@@ -3,8 +3,6 @@ package tracer // import "go.unistack.org/micro/v3/tracer"
 
 import (
 	"context"
-	"fmt"
-	"sort"
 
 	"go.unistack.org/micro/v3/logger"
 )
@@ -70,37 +68,3 @@ type Span interface {
 	// SpanID returns span id
 	SpanID() string
 }
-
-// sort labels alphabeticaly by label name
-type byKey []interface{}
-
-func (k byKey) Len() int           { return len(k) / 2 }
-func (k byKey) Less(i, j int) bool { return fmt.Sprintf("%s", k[i*2]) < fmt.Sprintf("%s", k[j*2]) }
-func (k byKey) Swap(i, j int) {
-	k[i*2], k[j*2] = k[j*2], k[i*2]
-	k[i*2+1], k[j*2+1] = k[j*2+1], k[i*2+1]
-}
-
-func UniqLabels(labels []interface{}) []interface{} {
-	if len(labels)%2 == 1 {
-		labels = labels[:len(labels)-1]
-	}
-
-	if len(labels) > 2 {
-		sort.Sort(byKey(labels))
-
-		idx := 0
-		for {
-			if labels[idx] == labels[idx+2] {
-				copy(labels[idx:], labels[idx+2:])
-				labels = labels[:len(labels)-2]
-			} else {
-				idx += 2
-			}
-			if idx+2 >= len(labels) {
-				break
-			}
-		}
-	}
-	return labels
-}

util/reflect/reflect.go

@@ -1,6 +1,7 @@
-package reflect // import "go.unistack.org/micro/v3/util/reflect"
+package reflect
 
 import (
+	"encoding/json"
 	"errors"
 	"fmt"
 	"reflect"
@@ -45,15 +46,23 @@ func SliceAppend(b bool) Option {
 
 // Merge merges map[string]interface{} to destination struct
 func Merge(dst interface{}, mp map[string]interface{}, opts ...Option) error {
-	var err error
-	var sval reflect.Value
-	var fname string
-
 	options := Options{}
 	for _, o := range opts {
 		o(&options)
 	}
 
+	if unmarshaler, ok := dst.(json.Unmarshaler); ok {
+		buf, err := json.Marshal(mp)
+		if err == nil {
+			err = unmarshaler.UnmarshalJSON(buf)
+		}
+		return err
+	}
+
+	var err error
+	var sval reflect.Value
+	var fname string
+
 	dviface := reflect.ValueOf(dst)
 	if dviface.Kind() == reflect.Ptr {
 		dviface = dviface.Elem()

util/sort/sort.go

@@ -0,0 +1,40 @@
+package sort
+
+import (
+	"fmt"
+	"sort"
+)
+
+// sort labels alphabeticaly by label name
+type byKey []interface{}
+
+func (k byKey) Len() int           { return len(k) / 2 }
+func (k byKey) Less(i, j int) bool { return fmt.Sprintf("%s", k[i*2]) < fmt.Sprintf("%s", k[j*2]) }
+func (k byKey) Swap(i, j int) {
+	k[i*2], k[j*2] = k[j*2], k[i*2]
+	k[i*2+1], k[j*2+1] = k[j*2+1], k[i*2+1]
+}
+
+func Uniq(labels []interface{}) []interface{} {
+	if len(labels)%2 == 1 {
+		labels = labels[:len(labels)-1]
+	}
+
+	if len(labels) > 2 {
+		sort.Sort(byKey(labels))
+
+		idx := 0
+		for {
+			if labels[idx] == labels[idx+2] {
+				copy(labels[idx:], labels[idx+2:])
+				labels = labels[:len(labels)-2]
+			} else {
+				idx += 2
+			}
+			if idx+2 >= len(labels) {
+				break
+			}
+		}
+	}
+	return labels
+}

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)
+}