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
|
||||
/.vscode/
|
||||
/.idea/
|
||||
.idea
|
||||
.vscode
|
||||
|
||||
# Binaries for programs and plugins
|
||||
*.exe
|
||||
@ -15,7 +13,6 @@
|
||||
_obj
|
||||
_test
|
||||
_build
|
||||
.DS_Store
|
||||
|
||||
# Architecture specific extensions/prefixes
|
||||
*.[568vq]
|
||||
|
@ -4,7 +4,6 @@ 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"
|
||||
@ -20,8 +19,6 @@ 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.
|
||||
|
@ -11,7 +11,6 @@ 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"
|
||||
)
|
||||
|
||||
@ -37,27 +36,22 @@ 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 {
|
||||
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,
|
||||
options := Options{
|
||||
Register: register.DefaultRegister,
|
||||
Logger: logger.DefaultLogger,
|
||||
Context: context.Background(),
|
||||
Meter: meter.DefaultMeter,
|
||||
Codecs: make(map[string]codec.Codec),
|
||||
Tracer: tracer.DefaultTracer,
|
||||
}
|
||||
for _, o := range opts {
|
||||
o(&newOpts)
|
||||
o(&options)
|
||||
}
|
||||
return newOpts
|
||||
return options
|
||||
}
|
||||
|
||||
// PublishOptions struct
|
||||
|
@ -31,9 +31,6 @@ func (c *cfg) Validate() error {
|
||||
if c.IntValue != 10 {
|
||||
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
|
||||
}
|
||||
|
||||
@ -108,19 +105,3 @@ func TestValidate(t *testing.T) {
|
||||
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
|
||||
func WithAddStacktrace(v bool) options.Option {
|
||||
return func(src interface{}) error {
|
||||
|
@ -17,6 +17,4 @@ var (
|
||||
SubscribeMessageTotal = "subscribe_message_total"
|
||||
// SubscribeMessageInflight specifies meter metric name
|
||||
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
|
||||
package tracer // import "go.unistack.org/micro/v4/tracer"
|
||||
|
||||
import (
|
||||
"context"
|
||||
"fmt"
|
||||
"sort"
|
||||
|
||||
"go.unistack.org/micro/v4/logger"
|
||||
"go.unistack.org/micro/v4/options"
|
||||
@ -69,3 +71,37 @@ 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
|
||||
}
|
||||
|
@ -1,7 +1,6 @@
|
||||
package reflect
|
||||
package reflect // import "go.unistack.org/micro/v4/util/reflect"
|
||||
|
||||
import (
|
||||
"encoding/json"
|
||||
"errors"
|
||||
"fmt"
|
||||
"reflect"
|
||||
@ -46,23 +45,15 @@ 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()
|
||||
|
@ -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