Compare commits
No commits in common. "master" and "v4.0.19" have entirely different histories.
3
.gitignore
vendored
3
.gitignore
vendored
@ -1,8 +1,6 @@
|
|||||||
# Develop tools
|
# Develop tools
|
||||||
/.vscode/
|
/.vscode/
|
||||||
/.idea/
|
/.idea/
|
||||||
.idea
|
|
||||||
.vscode
|
|
||||||
|
|
||||||
# Binaries for programs and plugins
|
# Binaries for programs and plugins
|
||||||
*.exe
|
*.exe
|
||||||
@ -15,7 +13,6 @@
|
|||||||
_obj
|
_obj
|
||||||
_test
|
_test
|
||||||
_build
|
_build
|
||||||
.DS_Store
|
|
||||||
|
|
||||||
# Architecture specific extensions/prefixes
|
# Architecture specific extensions/prefixes
|
||||||
*.[568vq]
|
*.[568vq]
|
||||||
|
@ -4,7 +4,6 @@ package broker // import "go.unistack.org/micro/v4/broker"
|
|||||||
import (
|
import (
|
||||||
"context"
|
"context"
|
||||||
"errors"
|
"errors"
|
||||||
"time"
|
|
||||||
|
|
||||||
"go.unistack.org/micro/v4/metadata"
|
"go.unistack.org/micro/v4/metadata"
|
||||||
"go.unistack.org/micro/v4/options"
|
"go.unistack.org/micro/v4/options"
|
||||||
@ -20,8 +19,6 @@ var (
|
|||||||
ErrDisconnected = errors.New("broker disconnected")
|
ErrDisconnected = errors.New("broker disconnected")
|
||||||
// ErrInvalidMessage returns when message has nvalid format
|
// ErrInvalidMessage returns when message has nvalid format
|
||||||
ErrInvalidMessage = errors.New("broker message has invalid format")
|
ErrInvalidMessage = errors.New("broker message has invalid format")
|
||||||
// DefaultGracefulTimeout
|
|
||||||
DefaultGracefulTimeout = 5 * time.Second
|
|
||||||
)
|
)
|
||||||
|
|
||||||
// Broker is an interface used for asynchronous messaging.
|
// Broker is an interface used for asynchronous messaging.
|
||||||
|
@ -11,7 +11,6 @@ import (
|
|||||||
"go.unistack.org/micro/v4/meter"
|
"go.unistack.org/micro/v4/meter"
|
||||||
"go.unistack.org/micro/v4/options"
|
"go.unistack.org/micro/v4/options"
|
||||||
"go.unistack.org/micro/v4/register"
|
"go.unistack.org/micro/v4/register"
|
||||||
"go.unistack.org/micro/v4/sync"
|
|
||||||
"go.unistack.org/micro/v4/tracer"
|
"go.unistack.org/micro/v4/tracer"
|
||||||
)
|
)
|
||||||
|
|
||||||
@ -37,27 +36,22 @@ type Options struct {
|
|||||||
Name string
|
Name string
|
||||||
// Address holds the broker address
|
// Address holds the broker address
|
||||||
Address []string
|
Address []string
|
||||||
|
|
||||||
Wait *sync.WaitGroup
|
|
||||||
|
|
||||||
GracefulTimeout time.Duration
|
|
||||||
}
|
}
|
||||||
|
|
||||||
// NewOptions create new Options
|
// NewOptions create new Options
|
||||||
func NewOptions(opts ...options.Option) Options {
|
func NewOptions(opts ...options.Option) Options {
|
||||||
newOpts := 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,
|
||||||
Codecs: make(map[string]codec.Codec),
|
Codecs: make(map[string]codec.Codec),
|
||||||
Tracer: tracer.DefaultTracer,
|
Tracer: tracer.DefaultTracer,
|
||||||
GracefulTimeout: DefaultGracefulTimeout,
|
|
||||||
}
|
}
|
||||||
for _, o := range opts {
|
for _, o := range opts {
|
||||||
o(&newOpts)
|
o(&options)
|
||||||
}
|
}
|
||||||
return newOpts
|
return options
|
||||||
}
|
}
|
||||||
|
|
||||||
// PublishOptions struct
|
// PublishOptions struct
|
||||||
|
@ -31,9 +31,6 @@ func (c *cfg) Validate() error {
|
|||||||
if c.IntValue != 10 {
|
if c.IntValue != 10 {
|
||||||
return fmt.Errorf("invalid IntValue %d != %d", 10, c.IntValue)
|
return fmt.Errorf("invalid IntValue %d != %d", 10, c.IntValue)
|
||||||
}
|
}
|
||||||
if c.MapValue["key1"] != true {
|
|
||||||
return fmt.Errorf("invalid MapValue %t != %t", true, c.MapValue["key1"])
|
|
||||||
}
|
|
||||||
return nil
|
return nil
|
||||||
}
|
}
|
||||||
|
|
||||||
@ -108,19 +105,3 @@ func TestValidate(t *testing.T) {
|
|||||||
t.Fatal(err)
|
t.Fatal(err)
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
func TestString(t *testing.T) {
|
|
||||||
cfg := config.NewConfig()
|
|
||||||
res := cfg.String()
|
|
||||||
if res != "default" {
|
|
||||||
t.Fatalf("string value invalid: %s", res)
|
|
||||||
}
|
|
||||||
}
|
|
||||||
|
|
||||||
func TestName(t *testing.T) {
|
|
||||||
cfg := config.NewConfig()
|
|
||||||
res := cfg.Name()
|
|
||||||
if res != "" {
|
|
||||||
t.Fatal("name value not empty")
|
|
||||||
}
|
|
||||||
}
|
|
||||||
|
@ -1,36 +0,0 @@
|
|||||||
package flow
|
|
||||||
|
|
||||||
import (
|
|
||||||
"reflect"
|
|
||||||
"testing"
|
|
||||||
)
|
|
||||||
|
|
||||||
func FuzzMarshall(f *testing.F) {
|
|
||||||
f.Fuzz(func(t *testing.T, ref []byte) {
|
|
||||||
rm := RawMessage(ref)
|
|
||||||
|
|
||||||
b, err := rm.MarshalJSON()
|
|
||||||
if err != nil {
|
|
||||||
t.Errorf("Error MarshalJSON: %s", err)
|
|
||||||
}
|
|
||||||
|
|
||||||
if !reflect.DeepEqual(ref, b) {
|
|
||||||
t.Errorf("Error. Expected '%s', was '%s'", ref, b)
|
|
||||||
}
|
|
||||||
})
|
|
||||||
}
|
|
||||||
|
|
||||||
func FuzzUnmarshall(f *testing.F) {
|
|
||||||
f.Fuzz(func(t *testing.T, ref string) {
|
|
||||||
b := []byte(ref)
|
|
||||||
rm := RawMessage(b)
|
|
||||||
|
|
||||||
if err := rm.UnmarshalJSON(b); err != nil {
|
|
||||||
t.Errorf("Error UnmarshalJSON: %s", err)
|
|
||||||
}
|
|
||||||
|
|
||||||
if ref != string(rm) {
|
|
||||||
t.Errorf("Error. Expected '%s', was '%s'", ref, rm)
|
|
||||||
}
|
|
||||||
})
|
|
||||||
}
|
|
@ -214,20 +214,6 @@ 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
|
// WithAddStacktrace controls writing stacktrace on error
|
||||||
func WithAddStacktrace(v bool) options.Option {
|
func WithAddStacktrace(v bool) options.Option {
|
||||||
return func(src interface{}) error {
|
return func(src interface{}) error {
|
||||||
|
@ -17,6 +17,4 @@ var (
|
|||||||
SubscribeMessageTotal = "subscribe_message_total"
|
SubscribeMessageTotal = "subscribe_message_total"
|
||||||
// SubscribeMessageInflight specifies meter metric name
|
// SubscribeMessageInflight specifies meter metric name
|
||||||
SubscribeMessageInflight = "subscribe_message_inflight"
|
SubscribeMessageInflight = "subscribe_message_inflight"
|
||||||
// BrokerGroupLag specifies broker lag
|
|
||||||
BrokerGroupLag = "broker_lag"
|
|
||||||
)
|
)
|
||||||
|
@ -1,12 +0,0 @@
|
|||||||
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"
|
|
||||||
)
|
|
@ -1,8 +1,10 @@
|
|||||||
// Package tracer provides an interface for distributed tracing
|
// Package tracer provides an interface for distributed tracing
|
||||||
package tracer
|
package tracer // import "go.unistack.org/micro/v4/tracer"
|
||||||
|
|
||||||
import (
|
import (
|
||||||
"context"
|
"context"
|
||||||
|
"fmt"
|
||||||
|
"sort"
|
||||||
|
|
||||||
"go.unistack.org/micro/v4/logger"
|
"go.unistack.org/micro/v4/logger"
|
||||||
"go.unistack.org/micro/v4/options"
|
"go.unistack.org/micro/v4/options"
|
||||||
@ -69,3 +71,37 @@ type Span interface {
|
|||||||
// SpanID returns span id
|
// SpanID returns span id
|
||||||
SpanID() string
|
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
|
||||||
|
}
|
||||||
|
@ -1,7 +1,6 @@
|
|||||||
package reflect
|
package reflect // import "go.unistack.org/micro/v4/util/reflect"
|
||||||
|
|
||||||
import (
|
import (
|
||||||
"encoding/json"
|
|
||||||
"errors"
|
"errors"
|
||||||
"fmt"
|
"fmt"
|
||||||
"reflect"
|
"reflect"
|
||||||
@ -46,23 +45,15 @@ func SliceAppend(b bool) Option {
|
|||||||
|
|
||||||
// Merge merges map[string]interface{} to destination struct
|
// Merge merges map[string]interface{} to destination struct
|
||||||
func Merge(dst interface{}, mp map[string]interface{}, opts ...Option) error {
|
func Merge(dst interface{}, mp map[string]interface{}, opts ...Option) error {
|
||||||
|
var err error
|
||||||
|
var sval reflect.Value
|
||||||
|
var fname string
|
||||||
|
|
||||||
options := Options{}
|
options := Options{}
|
||||||
for _, o := range opts {
|
for _, o := range opts {
|
||||||
o(&options)
|
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)
|
dviface := reflect.ValueOf(dst)
|
||||||
if dviface.Kind() == reflect.Ptr {
|
if dviface.Kind() == reflect.Ptr {
|
||||||
dviface = dviface.Elem()
|
dviface = dviface.Elem()
|
||||||
|
@ -1,40 +0,0 @@
|
|||||||
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
|
|
||||||
}
|
|
@ -1,25 +0,0 @@
|
|||||||
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)
|
|
||||||
}
|
|
@ -1,27 +0,0 @@
|
|||||||
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)
|
|
||||||
}
|
|
Loading…
Reference in New Issue
Block a user