Compare commits
22 Commits
Author | SHA1 | Date | |
---|---|---|---|
04d9d6675a | |||
c80068ccfc | |||
d0f9d44fe4 | |||
725ed992cc | |||
b8928d3da9 | |||
76090f7569 | |||
f8c68a81f7 | |||
5d997f7654 | |||
56d33ae823 | |||
c3cabc1fe5 | |||
47497b49b3 | |||
e89cfdc80d | |||
f954afb236 | |||
fbfacc82cd | |||
3d6df76dd5 | |||
2afcb51440 | |||
eb534a46d0 | |||
856aac4aa9 | |||
e416a5b38c | |||
2f8c0d9f9d | |||
4b261083d6 | |||
4c3c4859fc |
@ -1,24 +0,0 @@
|
||||
on:
|
||||
push:
|
||||
branches:
|
||||
- 'main'
|
||||
- 'master'
|
||||
- 'v3'
|
||||
schedule:
|
||||
#- cron: '* * * * *'
|
||||
- cron: '@hourly'
|
||||
|
||||
jobs:
|
||||
autoupdate:
|
||||
runs-on: ubuntu-latest
|
||||
steps:
|
||||
- name: setup-go
|
||||
uses: https://gitea.com/actions/setup-go@v3
|
||||
with:
|
||||
go-version: 1.21
|
||||
- name: checkout
|
||||
uses: https://gitea.com/actions/checkout@v3
|
||||
- name: get pkgdashcli
|
||||
run: GOPROXY=direct GONOSUMDB="git.unistack.org/*" GONOPROXY="git.unistack.org/*" GOBIN=/bin go install git.unistack.org/unistack-org/pkgdash/cmd/pkgdashcli@latest
|
||||
- name: pkgdashcli check
|
||||
run: /bin/pkgdashcli check
|
30
.gitea/workflows/coverage.yml
Normal file
30
.gitea/workflows/coverage.yml
Normal file
@ -0,0 +1,30 @@
|
||||
name: Go
|
||||
|
||||
on:
|
||||
push:
|
||||
branches: [ master, v3 ]
|
||||
pull_request:
|
||||
branches: [ master, v3 ]
|
||||
workflow_dispatch:
|
||||
|
||||
jobs:
|
||||
test:
|
||||
runs-on: ubuntu-latest
|
||||
steps:
|
||||
- name: checkout
|
||||
uses: actions/checkout@v4
|
||||
|
||||
- name: setup
|
||||
uses: actions/setup-go@v4
|
||||
with:
|
||||
go-version: stable
|
||||
|
||||
- name: coverage
|
||||
run: go test -v -coverprofile coverage.out ./...
|
||||
|
||||
- name: badge
|
||||
uses: ncruces/go-coverage-report@main
|
||||
with:
|
||||
coverage-file: coverage.out
|
||||
reuse-go: true
|
||||
amend: true
|
3
.gitignore
vendored
3
.gitignore
vendored
@ -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]
|
||||
|
@ -1,6 +1,5 @@
|
||||
run:
|
||||
concurrency: 4
|
||||
deadline: 5m
|
||||
issues-exit-code: 1
|
||||
tests: true
|
||||
|
||||
@ -13,15 +12,13 @@ linters-settings:
|
||||
linters:
|
||||
enable:
|
||||
- govet
|
||||
- deadcode
|
||||
- errcheck
|
||||
- govet
|
||||
- ineffassign
|
||||
- staticcheck
|
||||
- structcheck
|
||||
- typecheck
|
||||
- unused
|
||||
- varcheck
|
||||
- spancheck
|
||||
- bodyclose
|
||||
- gci
|
||||
- goconst
|
||||
@ -41,4 +38,5 @@ linters:
|
||||
- prealloc
|
||||
- unconvert
|
||||
- unparam
|
||||
- unused
|
||||
disable-all: false
|
||||
|
@ -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.
|
||||
|
@ -37,10 +37,10 @@ func TestMemoryBatchBroker(t *testing.T) {
|
||||
msgs := make([]Message, 0, count)
|
||||
for i := 0; i < count; i++ {
|
||||
message := &memoryMessage{
|
||||
header: map[string]string{
|
||||
metadata.HeaderTopic: topic,
|
||||
"foo": "bar",
|
||||
"id": fmt.Sprintf("%d", i),
|
||||
header: metadata.Metadata{
|
||||
metadata.HeaderTopic: []string{topic},
|
||||
"foo": []string{"bar"},
|
||||
"id": []string{fmt.Sprintf("%d", i)},
|
||||
},
|
||||
body: []byte(`"hello world"`),
|
||||
}
|
||||
@ -83,10 +83,10 @@ func TestMemoryBroker(t *testing.T) {
|
||||
msgs := make([]Message, 0, count)
|
||||
for i := 0; i < count; i++ {
|
||||
message := &memoryMessage{
|
||||
header: map[string]string{
|
||||
metadata.HeaderTopic: topic,
|
||||
"foo": "bar",
|
||||
"id": fmt.Sprintf("%d", i),
|
||||
header: metadata.Metadata{
|
||||
metadata.HeaderTopic: []string{topic},
|
||||
"foo": []string{"bar"},
|
||||
"id": []string{fmt.Sprintf("%d", i)},
|
||||
},
|
||||
body: []byte(`"hello world"`),
|
||||
}
|
||||
|
@ -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,36 +37,41 @@ 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{
|
||||
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
|
||||
type PublishOptions struct {
|
||||
// Context holds external options
|
||||
Context context.Context
|
||||
// BodyOnly flag says the message contains raw body bytes
|
||||
BodyOnly bool
|
||||
// Message metadata usually passed as message headers
|
||||
Metadata metadata.Metadata
|
||||
// Content-Type of message for marshal
|
||||
ContentType string
|
||||
// Topic destination
|
||||
Topic string
|
||||
// BodyOnly flag says the message contains raw body bytes
|
||||
BodyOnly bool
|
||||
}
|
||||
|
||||
// NewPublishOptions creates PublishOptions struct
|
||||
|
@ -19,8 +19,8 @@ var typeOfError = reflect.TypeOf((*error)(nil)).Elem()
|
||||
|
||||
// Is this an exported - upper case - name?
|
||||
func isExported(name string) bool {
|
||||
rune, _ := utf8.DecodeRuneInString(name)
|
||||
return unicode.IsUpper(rune)
|
||||
r, _ := utf8.DecodeRuneInString(name)
|
||||
return unicode.IsUpper(r)
|
||||
}
|
||||
|
||||
// Is this type exported or a builtin?
|
||||
|
@ -31,6 +31,9 @@ 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
|
||||
}
|
||||
|
||||
@ -105,3 +108,19 @@ 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")
|
||||
}
|
||||
}
|
||||
|
@ -75,7 +75,10 @@ func ParseDSN(dsn string) (*Config, error) {
|
||||
// Find last '/' that goes before dbname
|
||||
foundSlash := false
|
||||
for i := len(dsn) - 1; i >= 0; i-- {
|
||||
if dsn[i] == '/' {
|
||||
if dsn[i] != '/' {
|
||||
continue
|
||||
}
|
||||
|
||||
foundSlash = true
|
||||
var j, k int
|
||||
|
||||
@ -147,7 +150,6 @@ func ParseDSN(dsn string) (*Config, error) {
|
||||
|
||||
break
|
||||
}
|
||||
}
|
||||
|
||||
if !foundSlash && len(dsn) > 0 {
|
||||
return nil, ErrInvalidDSNNoSlash
|
||||
|
36
flow/flow_test.go
Normal file
36
flow/flow_test.go
Normal file
@ -0,0 +1,36 @@
|
||||
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)
|
||||
}
|
||||
})
|
||||
}
|
13
go.mod
13
go.mod
@ -5,17 +5,16 @@ go 1.20
|
||||
require (
|
||||
dario.cat/mergo v1.0.0
|
||||
github.com/DATA-DOG/go-sqlmock v1.5.0
|
||||
github.com/google/uuid v1.3.1
|
||||
github.com/google/uuid v1.6.0
|
||||
github.com/patrickmn/go-cache v2.1.0+incompatible
|
||||
github.com/silas/dag v0.0.0-20220518035006-a7e85ada93c5
|
||||
golang.org/x/sync v0.3.0
|
||||
golang.org/x/sys v0.12.0
|
||||
google.golang.org/grpc v1.58.2
|
||||
google.golang.org/protobuf v1.31.0
|
||||
golang.org/x/sync v0.6.0
|
||||
golang.org/x/sys v0.16.0
|
||||
google.golang.org/grpc v1.62.1
|
||||
google.golang.org/protobuf v1.32.0
|
||||
)
|
||||
|
||||
require (
|
||||
github.com/golang/protobuf v1.5.3 // indirect
|
||||
golang.org/x/net v0.15.0 // indirect
|
||||
google.golang.org/genproto/googleapis/rpc v0.0.0-20230711160842-782d3b101e98 // indirect
|
||||
google.golang.org/genproto/googleapis/rpc v0.0.0-20240123012728-ef4313101c80 // indirect
|
||||
)
|
||||
|
31
go.sum
31
go.sum
@ -6,29 +6,28 @@ github.com/golang/protobuf v1.5.0/go.mod h1:FsONVRAS9T7sI+LIUmWTfcYkHO4aIWwzhcaS
|
||||
github.com/golang/protobuf v1.5.3 h1:KhyjKVUg7Usr/dYsdSqoFveMYd5ko72D+zANwlG1mmg=
|
||||
github.com/golang/protobuf v1.5.3/go.mod h1:XVQd3VNwM+JqD3oG2Ue2ip4fOMUkwXdXDdiuN0vRsmY=
|
||||
github.com/google/go-cmp v0.5.5/go.mod h1:v8dTdLbMG2kIc/vJvl+f65V22dbkXbowE6jgT/gNBxE=
|
||||
github.com/google/go-cmp v0.5.9 h1:O2Tfq5qg4qc4AmwVlvv0oLiVAGB7enBSJ2x2DqQFi38=
|
||||
github.com/google/uuid v1.3.1 h1:KjJaJ9iWZ3jOFZIf1Lqf4laDRCasjl0BCmnEGxkdLb4=
|
||||
github.com/google/uuid v1.3.1/go.mod h1:TIyPZe4MgqvfeYDBFedMoGGpEw/LqOeaOT+nhxU+yHo=
|
||||
github.com/google/go-cmp v0.6.0 h1:ofyhxvXcZhMsU5ulbFiLKl/XBFqE1GSq7atu8tAmTRI=
|
||||
github.com/google/uuid v1.6.0 h1:NIvaJDMOsjHA8n1jAhLSgzrAzy1Hgr+hNrb57e+94F0=
|
||||
github.com/google/uuid v1.6.0/go.mod h1:TIyPZe4MgqvfeYDBFedMoGGpEw/LqOeaOT+nhxU+yHo=
|
||||
github.com/patrickmn/go-cache v2.1.0+incompatible h1:HRMgzkcYKYpi3C8ajMPV8OFXaaRUnok+kx1WdO15EQc=
|
||||
github.com/patrickmn/go-cache v2.1.0+incompatible/go.mod h1:3Qf8kWWT7OJRJbdiICTKqZju1ZixQ/KpMGzzAfe6+WQ=
|
||||
github.com/silas/dag v0.0.0-20220518035006-a7e85ada93c5 h1:G/FZtUu7a6NTWl3KUHMV9jkLAh/Rvtf03NWMHaEDl+E=
|
||||
github.com/silas/dag v0.0.0-20220518035006-a7e85ada93c5/go.mod h1:7RTUFBdIRC9nZ7/3RyRNH1bdqIShrDejd1YbLwgPS+I=
|
||||
golang.org/x/net v0.15.0 h1:ugBLEUaxABaB5AJqW9enI0ACdci2RUd4eP51NTBvuJ8=
|
||||
golang.org/x/net v0.15.0/go.mod h1:idbUs1IY1+zTqbi8yxTbhexhEEk5ur9LInksu6HrEpk=
|
||||
golang.org/x/sync v0.3.0 h1:ftCYgMx6zT/asHUrPw8BLLscYtGznsLAnjq5RH9P66E=
|
||||
golang.org/x/sync v0.3.0/go.mod h1:FU7BRWz2tNW+3quACPkgCx/L+uEAv1htQ0V83Z9Rj+Y=
|
||||
golang.org/x/sys v0.12.0 h1:CM0HF96J0hcLAwsHPJZjfdNzs0gftsLfgKt57wWHJ0o=
|
||||
golang.org/x/sys v0.12.0/go.mod h1:oPkhp1MJrh7nUepCBck5+mAzfO9JrbApNNgaTdGDITg=
|
||||
golang.org/x/text v0.13.0 h1:ablQoSUd0tRdKxZewP80B+BaqeKJuVhuRxj/dkrun3k=
|
||||
golang.org/x/net v0.20.0 h1:aCL9BSgETF1k+blQaYUBx9hJ9LOGP3gAVemcZlf1Kpo=
|
||||
golang.org/x/sync v0.6.0 h1:5BMeUDZ7vkXGfEr1x9B4bRcTH4lpkTkpdh0T/J+qjbQ=
|
||||
golang.org/x/sync v0.6.0/go.mod h1:Czt+wKu1gCyEFDUtn0jG5QVvpJ6rzVqr5aXyt9drQfk=
|
||||
golang.org/x/sys v0.16.0 h1:xWw16ngr6ZMtmxDyKyIgsE93KNKz5HKmMa3b8ALHidU=
|
||||
golang.org/x/sys v0.16.0/go.mod h1:/VUhepiaJMQUp4+oa/7Zr1D23ma6VTLIYjOOTFZPUcA=
|
||||
golang.org/x/text v0.14.0 h1:ScX5w1eTa3QqT8oi6+ziP7dTV1S2+ALU0bI+0zXKWiQ=
|
||||
golang.org/x/xerrors v0.0.0-20191204190536-9bdfabe68543/go.mod h1:I/5z698sn9Ka8TeJc9MKroUUfqBBauWjQqLJ2OPfmY0=
|
||||
google.golang.org/genproto/googleapis/rpc v0.0.0-20230711160842-782d3b101e98 h1:bVf09lpb+OJbByTj913DRJioFFAjf/ZGxEz7MajTp2U=
|
||||
google.golang.org/genproto/googleapis/rpc v0.0.0-20230711160842-782d3b101e98/go.mod h1:TUfxEVdsvPg18p6AslUXFoLdpED4oBnGwyqk3dV1XzM=
|
||||
google.golang.org/grpc v1.58.2 h1:SXUpjxeVF3FKrTYQI4f4KvbGD5u2xccdYdurwowix5I=
|
||||
google.golang.org/grpc v1.58.2/go.mod h1:tgX3ZQDlNJGU96V6yHh1T/JeoBQ2TXdr43YbYSsCJk0=
|
||||
google.golang.org/genproto/googleapis/rpc v0.0.0-20240123012728-ef4313101c80 h1:AjyfHzEPEFp/NpvfN5g+KDla3EMojjhRVZc1i7cj+oM=
|
||||
google.golang.org/genproto/googleapis/rpc v0.0.0-20240123012728-ef4313101c80/go.mod h1:PAREbraiVEVGVdTZsVWjSbbTtSyGbAgIIvni8a8CD5s=
|
||||
google.golang.org/grpc v1.62.1 h1:B4n+nfKzOICUXMgyrNd19h/I9oH0L1pizfk1d4zSgTk=
|
||||
google.golang.org/grpc v1.62.1/go.mod h1:IWTG0VlJLCh1SkC58F7np9ka9mx/WNkjl4PGJaiq+QE=
|
||||
google.golang.org/protobuf v1.26.0-rc.1/go.mod h1:jlhhOSvTdKEhbULTjvd4ARK9grFBp09yW+WbY/TyQbw=
|
||||
google.golang.org/protobuf v1.26.0/go.mod h1:9q0QmTI4eRPtz6boOQmLYwt+qCgq0jsYwAQnmE0givc=
|
||||
google.golang.org/protobuf v1.31.0 h1:g0LDEJHgrBl9N9r17Ru3sqWhkIx2NB67okBHPwC7hs8=
|
||||
google.golang.org/protobuf v1.31.0/go.mod h1:HV8QOd/L58Z+nl8r43ehVNZIU/HEI6OcFqwMG9pJV4I=
|
||||
google.golang.org/protobuf v1.32.0 h1:pPC6BG5ex8PDFnkbrGU3EixyhKcQ2aDuBS36lqK/C7I=
|
||||
google.golang.org/protobuf v1.32.0/go.mod h1:c6P6GXX6sHbq/GpV6MGZEdwhWPcYBgnhAHhKbcUYpos=
|
||||
gopkg.in/check.v1 v0.0.0-20161208181325-20d25e280405/go.mod h1:Co6ibVJAznAaIkqp8huTwlJQCZ016jof/cbN4VW5Yz0=
|
||||
gopkg.in/yaml.v3 v3.0.1 h1:fxVm/GzAzEWqLHuvctI91KS9hhNmmWOoWu0XTYJS7CA=
|
||||
gopkg.in/yaml.v3 v3.0.1/go.mod h1:K4uyk7z7BCEPqu6E+C64Yfv1cQ7kz7rIZviUmN+EgEM=
|
||||
|
@ -64,8 +64,8 @@ func Error(ctx context.Context, msg string, attrs ...interface{}) {
|
||||
DefaultLogger.Error(ctx, msg, attrs...)
|
||||
}
|
||||
|
||||
// Debugf writes formatted msg to default logger on debug level
|
||||
func Debugf(ctx context.Context, msg string, attrs ...interface{}) {
|
||||
// Debug writes formatted msg to default logger on debug level
|
||||
func Debug(ctx context.Context, msg string, attrs ...interface{}) {
|
||||
DefaultLogger.Debug(ctx, msg, attrs...)
|
||||
}
|
||||
|
||||
|
@ -6,6 +6,7 @@ import (
|
||||
"log/slog"
|
||||
"os"
|
||||
"reflect"
|
||||
"time"
|
||||
|
||||
"go.unistack.org/micro/v4/options"
|
||||
rutil "go.unistack.org/micro/v4/util/reflect"
|
||||
@ -17,18 +18,12 @@ type Options struct {
|
||||
Out io.Writer
|
||||
// Context holds exernal options
|
||||
Context context.Context
|
||||
// Attrs holds additional attributes
|
||||
Attrs []interface{}
|
||||
// Name holds the logger name
|
||||
Name string
|
||||
// The logging level the logger should log
|
||||
Level Level
|
||||
// CallerSkipCount number of frmaes to skip
|
||||
CallerSkipCount int
|
||||
// ContextAttrFuncs contains funcs that executed before log func on context
|
||||
ContextAttrFuncs []ContextAttrFunc
|
||||
// TimeFunc used to obtain current time
|
||||
TimeFunc func() time.Time
|
||||
// TimeKey is the key used for the time of the log call
|
||||
TimeKey string
|
||||
// Name holds the logger name
|
||||
Name string
|
||||
// LevelKey is the key used for the level of the log call
|
||||
LevelKey string
|
||||
// MessageKey is the key used for the message of the log call
|
||||
@ -39,6 +34,14 @@ type Options struct {
|
||||
SourceKey string
|
||||
// StacktraceKey is the key used for the stacktrace
|
||||
StacktraceKey string
|
||||
// Attrs holds additional attributes
|
||||
Attrs []interface{}
|
||||
// ContextAttrFuncs contains funcs that executed before log func on context
|
||||
ContextAttrFuncs []ContextAttrFunc
|
||||
// CallerSkipCount number of frmaes to skip
|
||||
CallerSkipCount int
|
||||
// The logging level the logger should log
|
||||
Level Level
|
||||
// AddStacktrace controls writing of stacktaces on error
|
||||
AddStacktrace bool
|
||||
// AddSource enabled writing source file and position in log
|
||||
@ -55,6 +58,7 @@ func NewOptions(opts ...options.Option) Options {
|
||||
Context: context.Background(),
|
||||
ContextAttrFuncs: DefaultContextAttrFuncs,
|
||||
AddSource: true,
|
||||
TimeFunc: time.Now,
|
||||
}
|
||||
|
||||
_ = WithMicroKeys()(&options)
|
||||
@ -210,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 {
|
||||
|
@ -8,7 +8,6 @@ import (
|
||||
"runtime"
|
||||
"strconv"
|
||||
"sync"
|
||||
"time"
|
||||
|
||||
"go.unistack.org/micro/v4/logger"
|
||||
"go.unistack.org/micro/v4/options"
|
||||
@ -62,8 +61,8 @@ func (s *slogLogger) renameAttr(_ []string, a slog.Attr) slog.Attr {
|
||||
}
|
||||
|
||||
type slogLogger struct {
|
||||
slog *slog.Logger
|
||||
leveler *slog.LevelVar
|
||||
handler slog.Handler
|
||||
opts logger.Options
|
||||
mu sync.RWMutex
|
||||
}
|
||||
@ -71,6 +70,7 @@ type slogLogger struct {
|
||||
func (s *slogLogger) Clone(opts ...options.Option) logger.Logger {
|
||||
s.mu.RLock()
|
||||
options := s.opts
|
||||
s.mu.RUnlock()
|
||||
|
||||
for _, o := range opts {
|
||||
_ = o(&options)
|
||||
@ -87,10 +87,7 @@ func (s *slogLogger) Clone(opts ...options.Option) logger.Logger {
|
||||
AddSource: l.opts.AddSource,
|
||||
}
|
||||
l.leveler.Set(loggerToSlogLevel(l.opts.Level))
|
||||
handler := slog.NewJSONHandler(options.Out, handleOpt)
|
||||
l.slog = slog.New(handler).With(options.Attrs...)
|
||||
|
||||
s.mu.RUnlock()
|
||||
l.handler = slog.New(slog.NewJSONHandler(options.Out, handleOpt)).With(options.Attrs...).Handler()
|
||||
|
||||
return l
|
||||
}
|
||||
@ -109,9 +106,13 @@ func (s *slogLogger) Options() logger.Options {
|
||||
|
||||
func (s *slogLogger) Attrs(attrs ...interface{}) logger.Logger {
|
||||
s.mu.RLock()
|
||||
l := &slogLogger{opts: s.opts}
|
||||
level := s.leveler.Level()
|
||||
options := s.opts
|
||||
s.mu.RUnlock()
|
||||
|
||||
l := &slogLogger{opts: options}
|
||||
l.leveler = new(slog.LevelVar)
|
||||
l.leveler.Set(s.leveler.Level())
|
||||
l.leveler.Set(level)
|
||||
|
||||
handleOpt := &slog.HandlerOptions{
|
||||
ReplaceAttr: l.renameAttr,
|
||||
@ -119,10 +120,7 @@ func (s *slogLogger) Attrs(attrs ...interface{}) logger.Logger {
|
||||
AddSource: l.opts.AddSource,
|
||||
}
|
||||
|
||||
handler := slog.NewJSONHandler(s.opts.Out, handleOpt)
|
||||
l.slog = slog.New(handler).With(attrs...)
|
||||
|
||||
s.mu.RUnlock()
|
||||
l.handler = slog.New(slog.NewJSONHandler(s.opts.Out, handleOpt)).With(attrs...).Handler()
|
||||
|
||||
return l
|
||||
}
|
||||
@ -147,8 +145,7 @@ func (s *slogLogger) Init(opts ...options.Option) error {
|
||||
AddSource: s.opts.AddSource,
|
||||
}
|
||||
s.leveler.Set(loggerToSlogLevel(s.opts.Level))
|
||||
handler := slog.NewJSONHandler(s.opts.Out, handleOpt)
|
||||
s.slog = slog.New(handler).With(s.opts.Attrs...)
|
||||
s.handler = slog.New(slog.NewJSONHandler(s.opts.Out, handleOpt)).With(s.opts.Attrs...).Handler()
|
||||
s.mu.Unlock()
|
||||
|
||||
return nil
|
||||
@ -160,13 +157,13 @@ func (s *slogLogger) Log(ctx context.Context, lvl logger.Level, msg string, attr
|
||||
}
|
||||
var pcs [1]uintptr
|
||||
runtime.Callers(s.opts.CallerSkipCount, pcs[:]) // skip [Callers, Infof]
|
||||
r := slog.NewRecord(time.Now(), loggerToSlogLevel(lvl), msg, pcs[0])
|
||||
r := slog.NewRecord(s.opts.TimeFunc(), loggerToSlogLevel(lvl), msg, pcs[0])
|
||||
for _, fn := range s.opts.ContextAttrFuncs {
|
||||
attrs = append(attrs, fn(ctx)...)
|
||||
}
|
||||
for _, a := range attrs {
|
||||
if ve, ok := a.(error); ok && ve != nil {
|
||||
attrs = append(attrs, slog.String(s.opts.ErrorKey, ve.Error()))
|
||||
for idx, attr := range attrs {
|
||||
if ve, ok := attr.(error); ok && ve != nil {
|
||||
attrs[idx] = slog.String(s.opts.ErrorKey, ve.Error())
|
||||
break
|
||||
}
|
||||
}
|
||||
@ -189,7 +186,7 @@ func (s *slogLogger) Log(ctx context.Context, lvl logger.Level, msg string, attr
|
||||
}
|
||||
return true
|
||||
})
|
||||
_ = s.slog.Handler().Handle(ctx, r)
|
||||
_ = s.handler.Handle(ctx, r)
|
||||
}
|
||||
|
||||
func (s *slogLogger) Info(ctx context.Context, msg string, attrs ...interface{}) {
|
||||
@ -198,12 +195,18 @@ func (s *slogLogger) Info(ctx context.Context, msg string, attrs ...interface{})
|
||||
}
|
||||
var pcs [1]uintptr
|
||||
runtime.Callers(s.opts.CallerSkipCount, pcs[:]) // skip [Callers, Infof]
|
||||
r := slog.NewRecord(time.Now(), slog.LevelInfo, msg, pcs[0])
|
||||
r := slog.NewRecord(s.opts.TimeFunc(), slog.LevelInfo, msg, pcs[0])
|
||||
for _, fn := range s.opts.ContextAttrFuncs {
|
||||
attrs = append(attrs, fn(ctx)...)
|
||||
}
|
||||
for idx, attr := range attrs {
|
||||
if ve, ok := attr.(error); ok && ve != nil {
|
||||
attrs[idx] = slog.String(s.opts.ErrorKey, ve.Error())
|
||||
break
|
||||
}
|
||||
}
|
||||
r.Add(attrs...)
|
||||
_ = s.slog.Handler().Handle(ctx, r)
|
||||
_ = s.handler.Handle(ctx, r)
|
||||
}
|
||||
|
||||
func (s *slogLogger) Debug(ctx context.Context, msg string, attrs ...interface{}) {
|
||||
@ -212,12 +215,18 @@ func (s *slogLogger) Debug(ctx context.Context, msg string, attrs ...interface{}
|
||||
}
|
||||
var pcs [1]uintptr
|
||||
runtime.Callers(s.opts.CallerSkipCount, pcs[:]) // skip [Callers, Infof]
|
||||
r := slog.NewRecord(time.Now(), slog.LevelDebug, msg, pcs[0])
|
||||
r := slog.NewRecord(s.opts.TimeFunc(), slog.LevelDebug, msg, pcs[0])
|
||||
for _, fn := range s.opts.ContextAttrFuncs {
|
||||
attrs = append(attrs, fn(ctx)...)
|
||||
}
|
||||
for idx, attr := range attrs {
|
||||
if ve, ok := attr.(error); ok && ve != nil {
|
||||
attrs[idx] = slog.String(s.opts.ErrorKey, ve.Error())
|
||||
break
|
||||
}
|
||||
}
|
||||
r.Add(attrs...)
|
||||
_ = s.slog.Handler().Handle(ctx, r)
|
||||
_ = s.handler.Handle(ctx, r)
|
||||
}
|
||||
|
||||
func (s *slogLogger) Trace(ctx context.Context, msg string, attrs ...interface{}) {
|
||||
@ -226,12 +235,18 @@ func (s *slogLogger) Trace(ctx context.Context, msg string, attrs ...interface{}
|
||||
}
|
||||
var pcs [1]uintptr
|
||||
runtime.Callers(s.opts.CallerSkipCount, pcs[:]) // skip [Callers, Infof]
|
||||
r := slog.NewRecord(time.Now(), slog.LevelDebug-1, msg, pcs[0])
|
||||
r := slog.NewRecord(s.opts.TimeFunc(), slog.LevelDebug-1, msg, pcs[0])
|
||||
for _, fn := range s.opts.ContextAttrFuncs {
|
||||
attrs = append(attrs, fn(ctx)...)
|
||||
}
|
||||
for idx, attr := range attrs {
|
||||
if ve, ok := attr.(error); ok && ve != nil {
|
||||
attrs[idx] = slog.String(s.opts.ErrorKey, ve.Error())
|
||||
break
|
||||
}
|
||||
}
|
||||
r.Add(attrs...)
|
||||
_ = s.slog.Handler().Handle(ctx, r)
|
||||
_ = s.handler.Handle(ctx, r)
|
||||
}
|
||||
|
||||
func (s *slogLogger) Error(ctx context.Context, msg string, attrs ...interface{}) {
|
||||
@ -240,13 +255,13 @@ func (s *slogLogger) Error(ctx context.Context, msg string, attrs ...interface{}
|
||||
}
|
||||
var pcs [1]uintptr
|
||||
runtime.Callers(s.opts.CallerSkipCount, pcs[:]) // skip [Callers, Infof]
|
||||
r := slog.NewRecord(time.Now(), slog.LevelError, msg, pcs[0])
|
||||
r := slog.NewRecord(s.opts.TimeFunc(), slog.LevelError, msg, pcs[0])
|
||||
for _, fn := range s.opts.ContextAttrFuncs {
|
||||
attrs = append(attrs, fn(ctx)...)
|
||||
}
|
||||
for _, a := range attrs {
|
||||
if ve, ok := a.(error); ok && ve != nil {
|
||||
attrs = append(attrs, slog.String(s.opts.ErrorKey, ve.Error()))
|
||||
for idx, attr := range attrs {
|
||||
if ve, ok := attr.(error); ok && ve != nil {
|
||||
attrs[idx] = slog.String(s.opts.ErrorKey, ve.Error())
|
||||
break
|
||||
}
|
||||
}
|
||||
@ -269,7 +284,7 @@ func (s *slogLogger) Error(ctx context.Context, msg string, attrs ...interface{}
|
||||
}
|
||||
return true
|
||||
})
|
||||
_ = s.slog.Handler().Handle(ctx, r)
|
||||
_ = s.handler.Handle(ctx, r)
|
||||
}
|
||||
|
||||
func (s *slogLogger) Fatal(ctx context.Context, msg string, attrs ...interface{}) {
|
||||
@ -278,12 +293,18 @@ func (s *slogLogger) Fatal(ctx context.Context, msg string, attrs ...interface{}
|
||||
}
|
||||
var pcs [1]uintptr
|
||||
runtime.Callers(s.opts.CallerSkipCount, pcs[:]) // skip [Callers, Infof]
|
||||
r := slog.NewRecord(time.Now(), slog.LevelError+1, msg, pcs[0])
|
||||
r := slog.NewRecord(s.opts.TimeFunc(), slog.LevelError+1, msg, pcs[0])
|
||||
for _, fn := range s.opts.ContextAttrFuncs {
|
||||
attrs = append(attrs, fn(ctx)...)
|
||||
}
|
||||
for idx, attr := range attrs {
|
||||
if ve, ok := attr.(error); ok && ve != nil {
|
||||
attrs[idx] = slog.String(s.opts.ErrorKey, ve.Error())
|
||||
break
|
||||
}
|
||||
}
|
||||
r.Add(attrs...)
|
||||
_ = s.slog.Handler().Handle(ctx, r)
|
||||
_ = s.handler.Handle(ctx, r)
|
||||
os.Exit(1)
|
||||
}
|
||||
|
||||
@ -293,12 +314,18 @@ func (s *slogLogger) Warn(ctx context.Context, msg string, attrs ...interface{})
|
||||
}
|
||||
var pcs [1]uintptr
|
||||
runtime.Callers(s.opts.CallerSkipCount, pcs[:]) // skip [Callers, Infof]
|
||||
r := slog.NewRecord(time.Now(), slog.LevelWarn, msg, pcs[0])
|
||||
r := slog.NewRecord(s.opts.TimeFunc(), slog.LevelWarn, msg, pcs[0])
|
||||
for _, fn := range s.opts.ContextAttrFuncs {
|
||||
attrs = append(attrs, fn(ctx)...)
|
||||
}
|
||||
for idx, attr := range attrs {
|
||||
if ve, ok := attr.(error); ok && ve != nil {
|
||||
attrs[idx] = slog.String(s.opts.ErrorKey, ve.Error())
|
||||
break
|
||||
}
|
||||
}
|
||||
r.Add(attrs...)
|
||||
_ = s.slog.Handler().Handle(ctx, r)
|
||||
_ = s.handler.Handle(ctx, r)
|
||||
}
|
||||
|
||||
func (s *slogLogger) Name() string {
|
||||
|
@ -56,9 +56,9 @@ type Wrapper struct {
|
||||
s fmt.State
|
||||
pointers map[uintptr]int
|
||||
opts *Options
|
||||
takeMap map[int]bool
|
||||
depth int
|
||||
ignoreNextType bool
|
||||
takeMap map[int]bool
|
||||
protoWrapperType bool
|
||||
sqlWrapperType bool
|
||||
}
|
||||
|
@ -17,11 +17,11 @@ func FromIncomingContext(ctx context.Context) (Metadata, bool) {
|
||||
if ctx == nil {
|
||||
return nil, false
|
||||
}
|
||||
md, ok := ctx.Value(mdIncomingKey{}).(*rawMetadata)
|
||||
if !ok || md.md == nil {
|
||||
md, ok := ctx.Value(mdIncomingKey{}).(Metadata)
|
||||
if !ok || md == nil {
|
||||
return nil, false
|
||||
}
|
||||
return md.md, ok
|
||||
return md, ok
|
||||
}
|
||||
|
||||
// FromOutgoingContext returns metadata from outgoing ctx
|
||||
@ -30,11 +30,11 @@ func FromOutgoingContext(ctx context.Context) (Metadata, bool) {
|
||||
if ctx == nil {
|
||||
return nil, false
|
||||
}
|
||||
md, ok := ctx.Value(mdOutgoingKey{}).(*rawMetadata)
|
||||
if !ok || md.md == nil {
|
||||
md, ok := ctx.Value(mdOutgoingKey{}).(Metadata)
|
||||
if !ok || md == nil {
|
||||
return nil, false
|
||||
}
|
||||
return md.md, ok
|
||||
return md, ok
|
||||
}
|
||||
|
||||
// FromContext returns metadata from the given context
|
||||
@ -43,11 +43,11 @@ func FromContext(ctx context.Context) (Metadata, bool) {
|
||||
if ctx == nil {
|
||||
return nil, false
|
||||
}
|
||||
md, ok := ctx.Value(mdKey{}).(*rawMetadata)
|
||||
if !ok || md.md == nil {
|
||||
md, ok := ctx.Value(mdKey{}).(Metadata)
|
||||
if !ok || md == nil {
|
||||
return nil, false
|
||||
}
|
||||
return md.md, ok
|
||||
return md, ok
|
||||
}
|
||||
|
||||
// NewContext creates a new context with the given metadata
|
||||
@ -55,45 +55,16 @@ func NewContext(ctx context.Context, md Metadata) context.Context {
|
||||
if ctx == nil {
|
||||
ctx = context.Background()
|
||||
}
|
||||
ctx = context.WithValue(ctx, mdKey{}, &rawMetadata{md})
|
||||
ctx = context.WithValue(ctx, mdIncomingKey{}, &rawMetadata{})
|
||||
ctx = context.WithValue(ctx, mdOutgoingKey{}, &rawMetadata{})
|
||||
ctx = context.WithValue(ctx, mdKey{}, md)
|
||||
return ctx
|
||||
}
|
||||
|
||||
// SetOutgoingContext modify outgoing context with given metadata
|
||||
func SetOutgoingContext(ctx context.Context, md Metadata) bool {
|
||||
if ctx == nil {
|
||||
return false
|
||||
}
|
||||
if omd, ok := ctx.Value(mdOutgoingKey{}).(*rawMetadata); ok {
|
||||
omd.md = md
|
||||
return true
|
||||
}
|
||||
return false
|
||||
}
|
||||
|
||||
// SetIncomingContext modify incoming context with given metadata
|
||||
func SetIncomingContext(ctx context.Context, md Metadata) bool {
|
||||
if ctx == nil {
|
||||
return false
|
||||
}
|
||||
if omd, ok := ctx.Value(mdIncomingKey{}).(*rawMetadata); ok {
|
||||
omd.md = md
|
||||
return true
|
||||
}
|
||||
return false
|
||||
}
|
||||
|
||||
// NewIncomingContext creates a new context with incoming metadata attached
|
||||
func NewIncomingContext(ctx context.Context, md Metadata) context.Context {
|
||||
if ctx == nil {
|
||||
ctx = context.Background()
|
||||
}
|
||||
ctx = context.WithValue(ctx, mdIncomingKey{}, &rawMetadata{md})
|
||||
if v, ok := ctx.Value(mdOutgoingKey{}).(*rawMetadata); !ok || v == nil {
|
||||
ctx = context.WithValue(ctx, mdOutgoingKey{}, &rawMetadata{})
|
||||
}
|
||||
ctx = context.WithValue(ctx, mdIncomingKey{}, md)
|
||||
return ctx
|
||||
}
|
||||
|
||||
@ -102,41 +73,28 @@ func NewOutgoingContext(ctx context.Context, md Metadata) context.Context {
|
||||
if ctx == nil {
|
||||
ctx = context.Background()
|
||||
}
|
||||
ctx = context.WithValue(ctx, mdOutgoingKey{}, &rawMetadata{md})
|
||||
if v, ok := ctx.Value(mdIncomingKey{}).(*rawMetadata); !ok || v == nil {
|
||||
ctx = context.WithValue(ctx, mdIncomingKey{}, &rawMetadata{})
|
||||
}
|
||||
ctx = context.WithValue(ctx, mdOutgoingKey{}, md)
|
||||
return ctx
|
||||
}
|
||||
|
||||
// AppendOutgoingContext apends new md to context
|
||||
func AppendOutgoingContext(ctx context.Context, kv ...string) context.Context {
|
||||
md, ok := Pairs(kv...)
|
||||
if !ok {
|
||||
return ctx
|
||||
}
|
||||
md := Pairs(kv...)
|
||||
omd, ok := FromOutgoingContext(ctx)
|
||||
if !ok {
|
||||
return NewOutgoingContext(ctx, md)
|
||||
}
|
||||
for k, v := range md {
|
||||
omd.Set(k, v)
|
||||
}
|
||||
return NewOutgoingContext(ctx, omd)
|
||||
nmd := Merge(omd, md, true)
|
||||
return NewOutgoingContext(ctx, nmd)
|
||||
}
|
||||
|
||||
// AppendIncomingContext apends new md to context
|
||||
func AppendIncomingContext(ctx context.Context, kv ...string) context.Context {
|
||||
md, ok := Pairs(kv...)
|
||||
if !ok {
|
||||
return ctx
|
||||
}
|
||||
md := Pairs(kv...)
|
||||
omd, ok := FromIncomingContext(ctx)
|
||||
if !ok {
|
||||
return NewIncomingContext(ctx, md)
|
||||
}
|
||||
for k, v := range md {
|
||||
omd.Set(k, v)
|
||||
}
|
||||
return NewIncomingContext(ctx, omd)
|
||||
nmd := Merge(omd, md, true)
|
||||
return NewIncomingContext(ctx, nmd)
|
||||
}
|
||||
|
@ -24,7 +24,7 @@ func TestNewNilContext(t *testing.T) {
|
||||
}
|
||||
|
||||
func TestFromContext(t *testing.T) {
|
||||
ctx := context.WithValue(context.TODO(), mdKey{}, &rawMetadata{New(0)})
|
||||
ctx := context.WithValue(context.TODO(), mdKey{}, New(0))
|
||||
|
||||
c, ok := FromContext(ctx)
|
||||
if c == nil || !ok {
|
||||
@ -42,7 +42,7 @@ func TestNewContext(t *testing.T) {
|
||||
}
|
||||
|
||||
func TestFromIncomingContext(t *testing.T) {
|
||||
ctx := context.WithValue(context.TODO(), mdIncomingKey{}, &rawMetadata{New(0)})
|
||||
ctx := context.WithValue(context.TODO(), mdIncomingKey{}, New(0))
|
||||
|
||||
c, ok := FromIncomingContext(ctx)
|
||||
if c == nil || !ok {
|
||||
@ -51,7 +51,7 @@ func TestFromIncomingContext(t *testing.T) {
|
||||
}
|
||||
|
||||
func TestFromOutgoingContext(t *testing.T) {
|
||||
ctx := context.WithValue(context.TODO(), mdOutgoingKey{}, &rawMetadata{New(0)})
|
||||
ctx := context.WithValue(context.TODO(), mdOutgoingKey{}, New(0))
|
||||
|
||||
c, ok := FromOutgoingContext(ctx)
|
||||
if c == nil || !ok {
|
||||
@ -59,36 +59,6 @@ func TestFromOutgoingContext(t *testing.T) {
|
||||
}
|
||||
}
|
||||
|
||||
func TestSetIncomingContext(t *testing.T) {
|
||||
md := New(1)
|
||||
md.Set("key", "val")
|
||||
ctx := context.WithValue(context.TODO(), mdIncomingKey{}, &rawMetadata{})
|
||||
if !SetIncomingContext(ctx, md) {
|
||||
t.Fatal("SetIncomingContext not works")
|
||||
}
|
||||
md, ok := FromIncomingContext(ctx)
|
||||
if md == nil || !ok {
|
||||
t.Fatal("SetIncomingContext not works")
|
||||
} else if v, ok := md.Get("key"); !ok || v != "val" {
|
||||
t.Fatal("SetIncomingContext not works")
|
||||
}
|
||||
}
|
||||
|
||||
func TestSetOutgoingContext(t *testing.T) {
|
||||
md := New(1)
|
||||
md.Set("key", "val")
|
||||
ctx := context.WithValue(context.TODO(), mdOutgoingKey{}, &rawMetadata{})
|
||||
if !SetOutgoingContext(ctx, md) {
|
||||
t.Fatal("SetOutgoingContext not works")
|
||||
}
|
||||
md, ok := FromOutgoingContext(ctx)
|
||||
if md == nil || !ok {
|
||||
t.Fatal("SetOutgoingContext not works")
|
||||
} else if v, ok := md.Get("key"); !ok || v != "val" {
|
||||
t.Fatal("SetOutgoingContext not works")
|
||||
}
|
||||
}
|
||||
|
||||
func TestNewIncomingContext(t *testing.T) {
|
||||
md := New(1)
|
||||
md.Set("key", "val")
|
||||
|
@ -1,9 +1,9 @@
|
||||
// Package metadata is a way of defining message headers
|
||||
package metadata // import "go.unistack.org/micro/v4/metadata"
|
||||
package metadata
|
||||
|
||||
import (
|
||||
"net/textproto"
|
||||
"sort"
|
||||
"strings"
|
||||
)
|
||||
|
||||
var (
|
||||
@ -24,47 +24,7 @@ var (
|
||||
)
|
||||
|
||||
// Metadata is our way of representing request headers internally.
|
||||
// They're used at the RPC level and translate back and forth
|
||||
// from Transport headers.
|
||||
type Metadata map[string]string
|
||||
|
||||
type rawMetadata struct {
|
||||
md Metadata
|
||||
}
|
||||
|
||||
// defaultMetadataSize used when need to init new Metadata
|
||||
var defaultMetadataSize = 2
|
||||
|
||||
// Iterator used to iterate over metadata with order
|
||||
type Iterator struct {
|
||||
md Metadata
|
||||
keys []string
|
||||
cur int
|
||||
cnt int
|
||||
}
|
||||
|
||||
// Next advance iterator to next element
|
||||
func (iter *Iterator) Next(k, v *string) bool {
|
||||
if iter.cur+1 > iter.cnt {
|
||||
return false
|
||||
}
|
||||
|
||||
*k = iter.keys[iter.cur]
|
||||
*v = iter.md[*k]
|
||||
iter.cur++
|
||||
return true
|
||||
}
|
||||
|
||||
// Iterator returns the itarator for metadata in sorted order
|
||||
func (md Metadata) Iterator() *Iterator {
|
||||
iter := &Iterator{md: md, cnt: len(md)}
|
||||
iter.keys = make([]string, 0, iter.cnt)
|
||||
for k := range md {
|
||||
iter.keys = append(iter.keys, k)
|
||||
}
|
||||
sort.Strings(iter.keys)
|
||||
return iter
|
||||
}
|
||||
type Metadata map[string][]string
|
||||
|
||||
// Get returns value from metadata by key
|
||||
func (md Metadata) Get(key string) (string, bool) {
|
||||
@ -74,7 +34,7 @@ func (md Metadata) Get(key string) (string, bool) {
|
||||
// slow path
|
||||
val, ok = md[textproto.CanonicalMIMEHeaderKey(key)]
|
||||
}
|
||||
return val, ok
|
||||
return strings.Join(val, ","), ok
|
||||
}
|
||||
|
||||
// Set is used to store value in metadata
|
||||
@ -83,10 +43,19 @@ func (md Metadata) Set(kv ...string) {
|
||||
kv = kv[:len(kv)-1]
|
||||
}
|
||||
for idx := 0; idx < len(kv); idx += 2 {
|
||||
md[textproto.CanonicalMIMEHeaderKey(kv[idx])] = kv[idx+1]
|
||||
md[textproto.CanonicalMIMEHeaderKey(kv[idx])] = []string{kv[idx+1]}
|
||||
}
|
||||
}
|
||||
|
||||
// Append is used to append value in metadata
|
||||
func (md Metadata) Append(k string, v ...string) {
|
||||
if len(v) == 0 {
|
||||
return
|
||||
}
|
||||
k = textproto.CanonicalMIMEHeaderKey(k)
|
||||
md[k] = append(md[k], v...)
|
||||
}
|
||||
|
||||
// Del is used to remove value from metadata
|
||||
func (md Metadata) Del(keys ...string) {
|
||||
for _, key := range keys {
|
||||
@ -98,46 +67,52 @@ func (md Metadata) Del(keys ...string) {
|
||||
}
|
||||
|
||||
// Copy makes a copy of the metadata
|
||||
func Copy(md Metadata) Metadata {
|
||||
nmd := New(len(md))
|
||||
for key, val := range md {
|
||||
nmd.Set(key, val)
|
||||
func Copy(md Metadata, exclude ...string) Metadata {
|
||||
nmd := make(Metadata, len(md))
|
||||
for k, v := range md {
|
||||
nmd[k] = v
|
||||
}
|
||||
nmd.Del(exclude...)
|
||||
return nmd
|
||||
}
|
||||
|
||||
// New return new sized metadata
|
||||
func New(size int) Metadata {
|
||||
if size == 0 {
|
||||
size = defaultMetadataSize
|
||||
size = 2
|
||||
}
|
||||
return make(Metadata, size)
|
||||
}
|
||||
|
||||
// Merge merges metadata to existing metadata, overwriting if specified
|
||||
func Merge(omd Metadata, mmd Metadata, overwrite bool) Metadata {
|
||||
var ok bool
|
||||
nmd := Copy(omd)
|
||||
for key, val := range mmd {
|
||||
_, ok = nmd[key]
|
||||
nval, ok := nmd[key]
|
||||
switch {
|
||||
case ok && overwrite:
|
||||
nmd[key] = nval
|
||||
continue
|
||||
case ok && !overwrite:
|
||||
continue
|
||||
case val != "":
|
||||
nmd.Set(key, val)
|
||||
case ok && val == "":
|
||||
nmd.Del(key)
|
||||
case !ok:
|
||||
for _, v := range val {
|
||||
if v != "" {
|
||||
nval = append(nval, v)
|
||||
}
|
||||
}
|
||||
nmd[key] = nval
|
||||
}
|
||||
}
|
||||
return nmd
|
||||
}
|
||||
|
||||
// Pairs from which metadata created
|
||||
func Pairs(kv ...string) (Metadata, bool) {
|
||||
func Pairs(kv ...string) Metadata {
|
||||
if len(kv)%2 == 1 {
|
||||
return nil, false
|
||||
kv = kv[:len(kv)-1]
|
||||
}
|
||||
md := New(len(kv) / 2)
|
||||
md := make(Metadata, len(kv)/2)
|
||||
md.Set(kv...)
|
||||
return md, true
|
||||
return md
|
||||
}
|
||||
|
@ -33,30 +33,52 @@ func TestAppend(t *testing.T) {
|
||||
}
|
||||
|
||||
func TestPairs(t *testing.T) {
|
||||
md, ok := Pairs("key1", "val1", "key2", "val2")
|
||||
if !ok {
|
||||
t.Fatal("odd number of kv")
|
||||
}
|
||||
if _, ok = md.Get("key1"); !ok {
|
||||
md := Pairs("key1", "val1", "key2", "val2")
|
||||
|
||||
if _, ok := md.Get("key1"); !ok {
|
||||
t.Fatal("key1 not found")
|
||||
}
|
||||
}
|
||||
|
||||
func testCtx(ctx context.Context) {
|
||||
md := New(2)
|
||||
func testIncomingCtx(ctx context.Context) {
|
||||
if md, ok := FromIncomingContext(ctx); ok && md != nil {
|
||||
md.Set("Key1", "Val1_new")
|
||||
md.Set("Key3", "Val3")
|
||||
SetOutgoingContext(ctx, md)
|
||||
}
|
||||
}
|
||||
|
||||
func TestPassing(t *testing.T) {
|
||||
func testOutgoingCtx(ctx context.Context) {
|
||||
if md, ok := FromOutgoingContext(ctx); ok && md != nil {
|
||||
md.Set("Key1", "Val1_new")
|
||||
md.Set("Key3", "Val3")
|
||||
}
|
||||
}
|
||||
|
||||
func TestIncoming(t *testing.T) {
|
||||
ctx := context.TODO()
|
||||
md1 := New(2)
|
||||
md1.Set("Key1", "Val1")
|
||||
md1.Set("Key2", "Val2")
|
||||
|
||||
ctx = NewIncomingContext(ctx, md1)
|
||||
testCtx(ctx)
|
||||
testIncomingCtx(ctx)
|
||||
md, ok := FromIncomingContext(ctx)
|
||||
if !ok {
|
||||
t.Fatalf("missing metadata from incoming context")
|
||||
}
|
||||
if v, ok := md.Get("Key1"); !ok || v != "Val1_new" {
|
||||
t.Fatalf("invalid metadata value %#+v", md)
|
||||
}
|
||||
}
|
||||
|
||||
func TestOutgoing(t *testing.T) {
|
||||
ctx := context.TODO()
|
||||
md1 := New(2)
|
||||
md1.Set("Key1", "Val1")
|
||||
md1.Set("Key2", "Val2")
|
||||
|
||||
ctx = NewOutgoingContext(ctx, md1)
|
||||
testOutgoingCtx(ctx)
|
||||
md, ok := FromOutgoingContext(ctx)
|
||||
if !ok {
|
||||
t.Fatalf("missing metadata from outgoing context")
|
||||
@ -68,10 +90,10 @@ func TestPassing(t *testing.T) {
|
||||
|
||||
func TestMerge(t *testing.T) {
|
||||
omd := Metadata{
|
||||
"key1": "val1",
|
||||
"key1": []string{"val1"},
|
||||
}
|
||||
mmd := Metadata{
|
||||
"key2": "val2",
|
||||
"key2": []string{"val2"},
|
||||
}
|
||||
|
||||
nmd := Merge(omd, mmd, true)
|
||||
@ -80,21 +102,6 @@ func TestMerge(t *testing.T) {
|
||||
}
|
||||
}
|
||||
|
||||
func TestIterator(t *testing.T) {
|
||||
md := Metadata{
|
||||
"1Last": "last",
|
||||
"2First": "first",
|
||||
"3Second": "second",
|
||||
}
|
||||
|
||||
iter := md.Iterator()
|
||||
var k, v string
|
||||
|
||||
for iter.Next(&k, &v) {
|
||||
// fmt.Printf("k: %s, v: %s\n", k, v)
|
||||
}
|
||||
}
|
||||
|
||||
func TestMedataCanonicalKey(t *testing.T) {
|
||||
md := New(1)
|
||||
md.Set("x-request-id", "12345")
|
||||
@ -134,10 +141,7 @@ func TestMetadataSet(t *testing.T) {
|
||||
}
|
||||
|
||||
func TestMetadataDelete(t *testing.T) {
|
||||
md := Metadata{
|
||||
"Foo": "bar",
|
||||
"Baz": "empty",
|
||||
}
|
||||
md := Pairs("Foo", "bar", "Baz", "empty")
|
||||
|
||||
md.Del("Baz")
|
||||
_, ok := md.Get("Baz")
|
||||
@ -156,24 +160,19 @@ func TestNilContext(t *testing.T) {
|
||||
}
|
||||
|
||||
func TestMetadataCopy(t *testing.T) {
|
||||
md := Metadata{
|
||||
"Foo": "bar",
|
||||
"Bar": "baz",
|
||||
}
|
||||
md := Pairs("Foo", "bar", "Bar", "baz")
|
||||
|
||||
cp := Copy(md)
|
||||
|
||||
for k, v := range md {
|
||||
if cv := cp[k]; cv != v {
|
||||
if cv := cp[k]; len(cv) != len(v) {
|
||||
t.Fatalf("Got %s:%s for %s:%s", k, cv, k, v)
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
func TestMetadataContext(t *testing.T) {
|
||||
md := Metadata{
|
||||
"Foo": "bar",
|
||||
}
|
||||
md := Pairs("Foo", "bar")
|
||||
|
||||
ctx := NewContext(context.TODO(), md)
|
||||
|
||||
@ -182,7 +181,7 @@ func TestMetadataContext(t *testing.T) {
|
||||
t.Errorf("Unexpected error retrieving metadata, got %t", ok)
|
||||
}
|
||||
|
||||
if emd["Foo"] != md["Foo"] {
|
||||
if len(emd["Foo"]) != len(md["Foo"]) {
|
||||
t.Errorf("Expected key: %s val: %s, got key: %s val: %s", "Foo", md["Foo"], "Foo", emd["Foo"])
|
||||
}
|
||||
|
||||
@ -190,3 +189,14 @@ func TestMetadataContext(t *testing.T) {
|
||||
t.Errorf("Expected metadata length 1 got %d", i)
|
||||
}
|
||||
}
|
||||
|
||||
func TestCopy(t *testing.T) {
|
||||
md := New(2)
|
||||
md.Set("key1", "val1", "key2", "val2")
|
||||
nmd := Copy(md, "key2")
|
||||
if len(nmd) != 1 {
|
||||
t.Fatal("Copy exclude not works")
|
||||
} else if nmd["Key1"][0] != "val1" {
|
||||
t.Fatal("Copy exclude not works")
|
||||
}
|
||||
}
|
||||
|
36
micro.go
36
micro.go
@ -65,6 +65,8 @@ func As(b any, target any) bool {
|
||||
break
|
||||
case targetType.Implements(routerType):
|
||||
break
|
||||
case targetType.Implements(tracerType):
|
||||
break
|
||||
default:
|
||||
return false
|
||||
}
|
||||
@ -76,19 +78,21 @@ func As(b any, target any) bool {
|
||||
return false
|
||||
}
|
||||
|
||||
var brokerType = reflect.TypeOf((*broker.Broker)(nil)).Elem()
|
||||
var loggerType = reflect.TypeOf((*logger.Logger)(nil)).Elem()
|
||||
var clientType = reflect.TypeOf((*client.Client)(nil)).Elem()
|
||||
var serverType = reflect.TypeOf((*server.Server)(nil)).Elem()
|
||||
var codecType = reflect.TypeOf((*codec.Codec)(nil)).Elem()
|
||||
var flowType = reflect.TypeOf((*flow.Flow)(nil)).Elem()
|
||||
var fsmType = reflect.TypeOf((*fsm.FSM)(nil)).Elem()
|
||||
var meterType = reflect.TypeOf((*meter.Meter)(nil)).Elem()
|
||||
var registerType = reflect.TypeOf((*register.Register)(nil)).Elem()
|
||||
var resolverType = reflect.TypeOf((*resolver.Resolver)(nil)).Elem()
|
||||
var routerType = reflect.TypeOf((*router.Router)(nil)).Elem()
|
||||
var selectorType = reflect.TypeOf((*selector.Selector)(nil)).Elem()
|
||||
var storeType = reflect.TypeOf((*store.Store)(nil)).Elem()
|
||||
var syncType = reflect.TypeOf((*sync.Sync)(nil)).Elem()
|
||||
var tracerType = reflect.TypeOf((*tracer.Tracer)(nil)).Elem()
|
||||
var serviceType = reflect.TypeOf((*Service)(nil)).Elem()
|
||||
var (
|
||||
brokerType = reflect.TypeOf((*broker.Broker)(nil)).Elem()
|
||||
loggerType = reflect.TypeOf((*logger.Logger)(nil)).Elem()
|
||||
clientType = reflect.TypeOf((*client.Client)(nil)).Elem()
|
||||
serverType = reflect.TypeOf((*server.Server)(nil)).Elem()
|
||||
codecType = reflect.TypeOf((*codec.Codec)(nil)).Elem()
|
||||
flowType = reflect.TypeOf((*flow.Flow)(nil)).Elem()
|
||||
fsmType = reflect.TypeOf((*fsm.FSM)(nil)).Elem()
|
||||
meterType = reflect.TypeOf((*meter.Meter)(nil)).Elem()
|
||||
registerType = reflect.TypeOf((*register.Register)(nil)).Elem()
|
||||
resolverType = reflect.TypeOf((*resolver.Resolver)(nil)).Elem()
|
||||
routerType = reflect.TypeOf((*router.Router)(nil)).Elem()
|
||||
selectorType = reflect.TypeOf((*selector.Selector)(nil)).Elem()
|
||||
storeType = reflect.TypeOf((*store.Store)(nil)).Elem()
|
||||
syncType = reflect.TypeOf((*sync.Sync)(nil)).Elem()
|
||||
tracerType = reflect.TypeOf((*tracer.Tracer)(nil)).Elem()
|
||||
serviceType = reflect.TypeOf((*Service)(nil)).Elem()
|
||||
)
|
||||
|
@ -158,17 +158,28 @@ func Metadata(md ...any) Option {
|
||||
case metadata.Metadata:
|
||||
result = metadata.Copy(vt)
|
||||
case map[string]string:
|
||||
result = make(metadata.Metadata, len(vt))
|
||||
for k, v := range vt {
|
||||
result.Set(k, v)
|
||||
}
|
||||
case map[string][]string:
|
||||
result = metadata.Copy(vt)
|
||||
default:
|
||||
result = metadata.New(0)
|
||||
}
|
||||
} else {
|
||||
result = metadata.New(len(md) / 2)
|
||||
for idx := 0; idx < len(md)/2; idx += 2 {
|
||||
k, kok := md[idx].(string)
|
||||
v, vok := md[idx+1].(string)
|
||||
if kok && vok {
|
||||
result.Set(k, v)
|
||||
for idx := 0; idx <= len(md)/2; idx += 2 {
|
||||
k, ok := md[idx].(string)
|
||||
switch vt := md[idx+1].(type) {
|
||||
case string:
|
||||
if ok {
|
||||
result.Set(k, vt)
|
||||
}
|
||||
case []string:
|
||||
if ok {
|
||||
result.Append(k, vt...)
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
|
@ -1,7 +1,6 @@
|
||||
package options_test
|
||||
|
||||
import (
|
||||
"fmt"
|
||||
"testing"
|
||||
|
||||
"go.unistack.org/micro/v4/codec"
|
||||
@ -100,40 +99,29 @@ func TestMetadataAny(t *testing.T) {
|
||||
}{
|
||||
{
|
||||
"strings_even",
|
||||
[]any{"key1", "val1", "key2", "val2"},
|
||||
metadata.Metadata{
|
||||
"Key1": "val1",
|
||||
"Key2": "val2",
|
||||
},
|
||||
[]any{"Strkey1", []string{"val1"}, "Strkey2", []string{"val2"}},
|
||||
metadata.Pairs("Strkey1", "val1", "Strkey2", "val2"),
|
||||
},
|
||||
{
|
||||
"strings_odd",
|
||||
[]any{"key1", "val1", "key2"},
|
||||
metadata.Metadata{
|
||||
"Key1": "val1",
|
||||
},
|
||||
metadata.Pairs("Key1", "val1"),
|
||||
},
|
||||
{
|
||||
Name: "map",
|
||||
Data: map[string]string{
|
||||
"key1": "val1",
|
||||
"key2": "val2",
|
||||
Data: map[string][]string{
|
||||
"Mapkey1": {"val1"},
|
||||
"Mapkey2": {"val2"},
|
||||
},
|
||||
Expected: metadata.Metadata{
|
||||
"Key1": "val1",
|
||||
"Key2": "val2",
|
||||
"Mapkey1": []string{"val1"},
|
||||
"Mapkey2": []string{"val2"},
|
||||
},
|
||||
},
|
||||
{
|
||||
"metadata.Metadata",
|
||||
metadata.Metadata{
|
||||
"key1": "val1",
|
||||
"key2": "val2",
|
||||
},
|
||||
metadata.Metadata{
|
||||
"Key1": "val1",
|
||||
"Key2": "val2",
|
||||
},
|
||||
metadata.Pairs("key1", "val1", "key2", "val2"),
|
||||
metadata.Pairs("Key1", "val1", "Key2", "val2"),
|
||||
},
|
||||
}
|
||||
|
||||
@ -144,7 +132,7 @@ func TestMetadataAny(t *testing.T) {
|
||||
switch valData := tt.Data.(type) {
|
||||
case []any:
|
||||
opts = append(opts, options.Metadata(valData...))
|
||||
case map[string]string, metadata.Metadata:
|
||||
case map[string]string, map[string][]string, metadata.Metadata:
|
||||
opts = append(opts, options.Metadata(valData))
|
||||
}
|
||||
|
||||
@ -153,7 +141,7 @@ func TestMetadataAny(t *testing.T) {
|
||||
t.Fatal(err)
|
||||
}
|
||||
if !reflect.Equal(tt.Expected, src.Metadata) {
|
||||
t.Fatal(fmt.Sprintf("expected: %v, actual: %v", tt.Expected, src.Metadata))
|
||||
t.Fatalf("expected: %v, actual: %v", tt.Expected, src.Metadata)
|
||||
}
|
||||
}
|
||||
})
|
||||
|
@ -2,10 +2,12 @@ package memory
|
||||
|
||||
import (
|
||||
"context"
|
||||
"go.unistack.org/micro/v4/register"
|
||||
"sync"
|
||||
"time"
|
||||
|
||||
"go.unistack.org/micro/v4/metadata"
|
||||
"go.unistack.org/micro/v4/register"
|
||||
|
||||
"go.unistack.org/micro/v4/logger"
|
||||
"go.unistack.org/micro/v4/util/id"
|
||||
)
|
||||
@ -24,16 +26,16 @@ type node struct {
|
||||
type record struct {
|
||||
Name string
|
||||
Version string
|
||||
Metadata map[string]string
|
||||
Metadata metadata.Metadata
|
||||
Nodes map[string]*node
|
||||
Endpoints []*register.Endpoint
|
||||
}
|
||||
|
||||
type memory struct {
|
||||
sync.RWMutex
|
||||
records map[string]services
|
||||
watchers map[string]*watcher
|
||||
opts register.Options
|
||||
sync.RWMutex
|
||||
}
|
||||
|
||||
// services is a KV map with service name as the key and a map of records as the value
|
||||
@ -100,10 +102,20 @@ func (m *memory) sendEvent(r *register.Result) {
|
||||
}
|
||||
|
||||
func (m *memory) Connect(ctx context.Context) error {
|
||||
select {
|
||||
case <-ctx.Done():
|
||||
return ctx.Err()
|
||||
default:
|
||||
}
|
||||
return nil
|
||||
}
|
||||
|
||||
func (m *memory) Disconnect(ctx context.Context) error {
|
||||
select {
|
||||
case <-ctx.Done():
|
||||
return ctx.Err()
|
||||
default:
|
||||
}
|
||||
return nil
|
||||
}
|
||||
|
||||
@ -124,6 +136,11 @@ func (m *memory) Options() register.Options {
|
||||
}
|
||||
|
||||
func (m *memory) Register(ctx context.Context, s *register.Service, opts ...register.RegisterOption) error {
|
||||
select {
|
||||
case <-ctx.Done():
|
||||
return ctx.Err()
|
||||
default:
|
||||
}
|
||||
m.Lock()
|
||||
defer m.Unlock()
|
||||
|
||||
@ -137,11 +154,11 @@ func (m *memory) Register(ctx context.Context, s *register.Service, opts ...regi
|
||||
|
||||
// domain is set in metadata so it can be passed to watchers
|
||||
if s.Metadata == nil {
|
||||
s.Metadata = map[string]string{"domain": options.Domain}
|
||||
} else {
|
||||
s.Metadata["domain"] = options.Domain
|
||||
s.Metadata = metadata.New(0)
|
||||
}
|
||||
|
||||
s.Metadata.Set("domain", options.Domain)
|
||||
|
||||
// ensure the service name exists
|
||||
r := serviceToRecord(s, options.TTL)
|
||||
if _, ok := srvs[s.Name]; !ok {
|
||||
@ -165,15 +182,8 @@ func (m *memory) Register(ctx context.Context, s *register.Service, opts ...regi
|
||||
continue
|
||||
}
|
||||
|
||||
metadata := make(map[string]string, len(n.Metadata))
|
||||
|
||||
// make copy of metadata
|
||||
for k, v := range n.Metadata {
|
||||
metadata[k] = v
|
||||
}
|
||||
|
||||
// set the domain
|
||||
metadata["domain"] = options.Domain
|
||||
metadata := metadata.Copy(n.Metadata)
|
||||
metadata.Set("domain", options.Domain)
|
||||
|
||||
// add the node
|
||||
srvs[s.Name][s.Version].Nodes[n.ID] = &node{
|
||||
@ -217,10 +227,9 @@ func (m *memory) Deregister(ctx context.Context, s *register.Service, opts ...re
|
||||
|
||||
// domain is set in metadata so it can be passed to watchers
|
||||
if s.Metadata == nil {
|
||||
s.Metadata = map[string]string{"domain": options.Domain}
|
||||
} else {
|
||||
s.Metadata["domain"] = options.Domain
|
||||
s.Metadata = metadata.New(0)
|
||||
}
|
||||
s.Metadata.Set("domain", options.Domain)
|
||||
|
||||
// if the domain doesn't exist, there is nothing to deregister
|
||||
services, ok := m.records[options.Domain]
|
||||
@ -425,16 +434,24 @@ func (m *watcher) Next() (*register.Result, error) {
|
||||
continue
|
||||
}
|
||||
|
||||
if m.wo.Domain == register.WildcardDomain {
|
||||
return r, nil
|
||||
}
|
||||
|
||||
if r.Service.Metadata == nil {
|
||||
continue
|
||||
}
|
||||
|
||||
// extract domain from service metadata
|
||||
var domain string
|
||||
if r.Service.Metadata != nil && len(r.Service.Metadata["domain"]) > 0 {
|
||||
domain = r.Service.Metadata["domain"]
|
||||
if v, ok := r.Service.Metadata.Get("domain"); ok && v != "" {
|
||||
domain = v
|
||||
} else {
|
||||
domain = register.DefaultDomain
|
||||
}
|
||||
|
||||
// only send the event if watching the wildcard or this specific domain
|
||||
if m.wo.Domain == register.WildcardDomain || m.wo.Domain == domain {
|
||||
if m.wo.Domain == domain {
|
||||
return r, nil
|
||||
}
|
||||
case <-m.exit:
|
||||
@ -453,10 +470,7 @@ func (m *watcher) Stop() {
|
||||
}
|
||||
|
||||
func serviceToRecord(s *register.Service, ttl time.Duration) *record {
|
||||
metadata := make(map[string]string, len(s.Metadata))
|
||||
for k, v := range s.Metadata {
|
||||
metadata[k] = v
|
||||
}
|
||||
metadata := metadata.Copy(s.Metadata)
|
||||
|
||||
nodes := make(map[string]*node, len(s.Nodes))
|
||||
for _, n := range s.Nodes {
|
||||
@ -468,9 +482,7 @@ func serviceToRecord(s *register.Service, ttl time.Duration) *record {
|
||||
}
|
||||
|
||||
endpoints := make([]*register.Endpoint, len(s.Endpoints))
|
||||
for i, e := range s.Endpoints { // TODO: vtolstov use copy
|
||||
endpoints[i] = e
|
||||
}
|
||||
copy(endpoints, s.Endpoints)
|
||||
|
||||
return &record{
|
||||
Name: s.Name,
|
||||
@ -482,20 +494,11 @@ func serviceToRecord(s *register.Service, ttl time.Duration) *record {
|
||||
}
|
||||
|
||||
func recordToService(r *record, domain string) *register.Service {
|
||||
metadata := make(map[string]string, len(r.Metadata))
|
||||
for k, v := range r.Metadata {
|
||||
metadata[k] = v
|
||||
}
|
||||
|
||||
// set the domain in metadata so it can be determined when a wildcard query is performed
|
||||
metadata["domain"] = domain
|
||||
|
||||
endpoints := make([]*register.Endpoint, len(r.Endpoints))
|
||||
for i, e := range r.Endpoints {
|
||||
md := make(map[string]string, len(e.Metadata))
|
||||
for k, v := range e.Metadata {
|
||||
md[k] = v
|
||||
}
|
||||
md := metadata.Copy(e.Metadata)
|
||||
// set the domain in metadata so it can be determined when a wildcard query is performed
|
||||
md.Set("domain", domain)
|
||||
|
||||
endpoints[i] = ®ister.Endpoint{
|
||||
Name: e.Name,
|
||||
@ -508,15 +511,10 @@ func recordToService(r *record, domain string) *register.Service {
|
||||
nodes := make([]*register.Node, len(r.Nodes))
|
||||
i := 0
|
||||
for _, n := range r.Nodes {
|
||||
md := make(map[string]string, len(n.Metadata))
|
||||
for k, v := range n.Metadata {
|
||||
md[k] = v
|
||||
}
|
||||
|
||||
nodes[i] = ®ister.Node{
|
||||
ID: n.ID,
|
||||
Address: n.Address,
|
||||
Metadata: md,
|
||||
Metadata: metadata.Copy(n.Metadata),
|
||||
}
|
||||
i++
|
||||
}
|
||||
@ -524,7 +522,7 @@ func recordToService(r *record, domain string) *register.Service {
|
||||
return ®ister.Service{
|
||||
Name: r.Name,
|
||||
Version: r.Version,
|
||||
Metadata: metadata,
|
||||
Metadata: metadata.Copy(r.Metadata),
|
||||
Endpoints: endpoints,
|
||||
Nodes: nodes,
|
||||
}
|
||||
|
@ -3,12 +3,13 @@ package memory
|
||||
import (
|
||||
"context"
|
||||
"fmt"
|
||||
"go.unistack.org/micro/v4"
|
||||
"go.unistack.org/micro/v4/register"
|
||||
"reflect"
|
||||
"sync"
|
||||
"testing"
|
||||
"time"
|
||||
|
||||
"go.unistack.org/micro/v4"
|
||||
"go.unistack.org/micro/v4/register"
|
||||
)
|
||||
|
||||
var testData = map[string][]*register.Service{
|
||||
@ -209,9 +210,9 @@ func TestMemoryRegistryTTLConcurrent(t *testing.T) {
|
||||
}
|
||||
}
|
||||
|
||||
//if len(os.Getenv("IN_TRAVIS_CI")) == 0 {
|
||||
// if len(os.Getenv("IN_TRAVIS_CI")) == 0 {
|
||||
// t.Logf("test will wait %v, then check TTL timeouts", waitTime)
|
||||
//}
|
||||
// }
|
||||
|
||||
errChan := make(chan error, concurrency)
|
||||
syncChan := make(chan struct{})
|
||||
@ -252,6 +253,13 @@ func TestMemoryWildcard(t *testing.T) {
|
||||
m := NewRegister()
|
||||
ctx := context.TODO()
|
||||
|
||||
if err := m.Init(); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
|
||||
if err := m.Connect(ctx); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
testSrv := ®ister.Service{Name: "foo", Version: "1.0.0"}
|
||||
|
||||
if err := m.Register(ctx, testSrv, register.RegisterDomain("one")); err != nil {
|
||||
@ -291,8 +299,12 @@ func TestWatcher(t *testing.T) {
|
||||
|
||||
ctx := context.TODO()
|
||||
m := NewRegister()
|
||||
m.Init()
|
||||
m.Connect(ctx)
|
||||
if err := m.Init(); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
if err := m.Connect(ctx); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
wc, err := m.Watch(ctx)
|
||||
if err != nil {
|
||||
t.Fatalf("cant watch: %v", err)
|
||||
|
@ -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"
|
||||
)
|
||||
|
12
semconv/cache.go
Normal file
12
semconv/cache.go
Normal file
@ -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"
|
||||
)
|
@ -112,8 +112,8 @@ func (n *noopServer) Register() error {
|
||||
}
|
||||
n.RUnlock()
|
||||
|
||||
service.Nodes[0].Metadata["protocol"] = "noop"
|
||||
service.Nodes[0].Metadata["transport"] = service.Nodes[0].Metadata["protocol"]
|
||||
service.Nodes[0].Metadata.Set("protocol", "noop")
|
||||
service.Nodes[0].Metadata.Set("transport", "noop")
|
||||
service.Endpoints = endpoints
|
||||
|
||||
n.RLock()
|
||||
|
@ -13,6 +13,7 @@ import (
|
||||
"go.unistack.org/micro/v4/meter"
|
||||
"go.unistack.org/micro/v4/options"
|
||||
"go.unistack.org/micro/v4/register"
|
||||
msync "go.unistack.org/micro/v4/sync"
|
||||
"go.unistack.org/micro/v4/tracer"
|
||||
"go.unistack.org/micro/v4/util/id"
|
||||
)
|
||||
@ -32,7 +33,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
|
||||
@ -146,8 +147,11 @@ func Wait(wg *sync.WaitGroup) options.Option {
|
||||
if wg == nil {
|
||||
wg = new(sync.WaitGroup)
|
||||
}
|
||||
|
||||
wrap := msync.WrapWaitGroup(wg)
|
||||
|
||||
return func(src interface{}) error {
|
||||
return options.Set(src, wg, ".Wait")
|
||||
return options.Set(src, wrap, ".Wait")
|
||||
}
|
||||
}
|
||||
|
||||
|
@ -77,8 +77,8 @@ func NewRegisterService(s Server) (*register.Service, error) {
|
||||
}
|
||||
node.Metadata = metadata.Copy(opts.Metadata)
|
||||
|
||||
node.Metadata["server"] = s.String()
|
||||
node.Metadata["register"] = opts.Register.String()
|
||||
node.Metadata.Set("server", s.String())
|
||||
node.Metadata.Set("register", opts.Register.String())
|
||||
|
||||
return ®ister.Service{
|
||||
Name: opts.Name,
|
||||
|
69
sync/waitgroup.go
Normal file
69
sync/waitgroup.go
Normal file
@ -0,0 +1,69 @@
|
||||
package sync
|
||||
|
||||
import (
|
||||
"context"
|
||||
"sync"
|
||||
)
|
||||
|
||||
type WaitGroup struct {
|
||||
wg *sync.WaitGroup
|
||||
c int
|
||||
mu sync.Mutex
|
||||
}
|
||||
|
||||
func WrapWaitGroup(wg *sync.WaitGroup) *WaitGroup {
|
||||
g := &WaitGroup{
|
||||
wg: wg,
|
||||
}
|
||||
return g
|
||||
}
|
||||
|
||||
func NewWaitGroup() *WaitGroup {
|
||||
var wg sync.WaitGroup
|
||||
return WrapWaitGroup(&wg)
|
||||
}
|
||||
|
||||
func (g *WaitGroup) Add(n int) {
|
||||
g.mu.Lock()
|
||||
g.c += n
|
||||
g.wg.Add(n)
|
||||
g.mu.Unlock()
|
||||
}
|
||||
|
||||
func (g *WaitGroup) Done() {
|
||||
g.mu.Lock()
|
||||
g.c += -1
|
||||
g.wg.Add(-1)
|
||||
g.mu.Unlock()
|
||||
}
|
||||
|
||||
func (g *WaitGroup) Wait() {
|
||||
g.wg.Wait()
|
||||
}
|
||||
|
||||
func (g *WaitGroup) WaitContext(ctx context.Context) {
|
||||
done := make(chan struct{})
|
||||
go func() {
|
||||
g.wg.Wait()
|
||||
close(done)
|
||||
}()
|
||||
|
||||
select {
|
||||
case <-ctx.Done():
|
||||
g.mu.Lock()
|
||||
g.wg.Add(-g.c)
|
||||
<-done
|
||||
g.wg.Add(g.c)
|
||||
g.mu.Unlock()
|
||||
return
|
||||
case <-done:
|
||||
return
|
||||
}
|
||||
}
|
||||
|
||||
func (g *WaitGroup) Waiters() int {
|
||||
g.mu.Lock()
|
||||
c := g.c
|
||||
g.mu.Unlock()
|
||||
return c
|
||||
}
|
37
sync/waitgroup_test.go
Normal file
37
sync/waitgroup_test.go
Normal file
@ -0,0 +1,37 @@
|
||||
package sync
|
||||
|
||||
import (
|
||||
"context"
|
||||
"testing"
|
||||
"time"
|
||||
)
|
||||
|
||||
func TestWaitGroupContext(t *testing.T) {
|
||||
wg := NewWaitGroup()
|
||||
_ = t
|
||||
wg.Add(1)
|
||||
ctx, cancel := context.WithTimeout(context.TODO(), 1*time.Second)
|
||||
defer cancel()
|
||||
wg.WaitContext(ctx)
|
||||
}
|
||||
|
||||
func TestWaitGroupReuse(t *testing.T) {
|
||||
wg := NewWaitGroup()
|
||||
defer func() {
|
||||
if wg.Waiters() != 0 {
|
||||
t.Fatal("lost goroutines")
|
||||
}
|
||||
}()
|
||||
|
||||
wg.Add(1)
|
||||
defer wg.Done()
|
||||
ctx, cancel := context.WithTimeout(context.TODO(), 1*time.Second)
|
||||
defer cancel()
|
||||
wg.WaitContext(ctx)
|
||||
|
||||
wg.Add(1)
|
||||
defer wg.Done()
|
||||
ctx, cancel = context.WithTimeout(context.TODO(), 1*time.Second)
|
||||
defer cancel()
|
||||
wg.WaitContext(ctx)
|
||||
}
|
@ -172,6 +172,7 @@ func NewEventOptions(opts ...options.Option) EventOptions {
|
||||
func NewOptions(opts ...options.Option) Options {
|
||||
options := Options{
|
||||
Logger: logger.DefaultLogger,
|
||||
Context: context.Background(),
|
||||
}
|
||||
for _, o := range opts {
|
||||
o(&options)
|
||||
|
@ -1,10 +1,8 @@
|
||||
// Package tracer provides an interface for distributed tracing
|
||||
package tracer // import "go.unistack.org/micro/v4/tracer"
|
||||
package tracer
|
||||
|
||||
import (
|
||||
"context"
|
||||
"fmt"
|
||||
"sort"
|
||||
|
||||
"go.unistack.org/micro/v4/logger"
|
||||
"go.unistack.org/micro/v4/options"
|
||||
@ -71,37 +69,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
|
||||
}
|
||||
|
@ -1,6 +1,7 @@
|
||||
package reflect // import "go.unistack.org/micro/v4/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()
|
||||
@ -532,6 +541,9 @@ func Equal(src interface{}, dst interface{}, excptFields ...string) bool {
|
||||
}
|
||||
s := srcVal.MapIndex(key)
|
||||
d := dstVal.MapIndex(key)
|
||||
if !s.IsValid() || !d.IsValid() {
|
||||
return false
|
||||
}
|
||||
if !Equal(s.Interface(), d.Interface(), excptFields...) {
|
||||
return false
|
||||
}
|
||||
|
@ -109,13 +109,12 @@ func Merge(olist []*register.Service, nlist []*register.Service) []*register.Ser
|
||||
seen = true
|
||||
srv = append(srv, sp)
|
||||
break
|
||||
} else {
|
||||
}
|
||||
sp := ®ister.Service{}
|
||||
// make copy
|
||||
*sp = *o
|
||||
srv = append(srv, sp)
|
||||
}
|
||||
}
|
||||
if !seen {
|
||||
srv = append(srv, Copy([]*register.Service{n})...)
|
||||
}
|
||||
|
40
util/sort/sort.go
Normal file
40
util/sort/sort.go
Normal file
@ -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
|
||||
}
|
@ -12,7 +12,7 @@ type EC2Metadata struct {
|
||||
InstanceType string `json:"instance-type"`
|
||||
LocalHostname string `json:"local-hostname"`
|
||||
LocalIPv4 string `json:"local-ipv4"`
|
||||
kernelID int `json:"kernel-id"`
|
||||
KernelID int `json:"kernel-id"`
|
||||
Placement string `json:"placement"`
|
||||
AvailabilityZone string `json:"availability-zone"`
|
||||
ProductCodes string `json:"product-codes"`
|
||||
|
@ -67,9 +67,9 @@ func (fi *fileInfo) Name() string {
|
||||
|
||||
func (fi *fileInfo) Mode() os.FileMode {
|
||||
if strings.HasSuffix(fi.name, "/") {
|
||||
return os.FileMode(0755) | os.ModeDir
|
||||
return os.FileMode(0o755) | os.ModeDir
|
||||
}
|
||||
return os.FileMode(0644)
|
||||
return os.FileMode(0o644)
|
||||
}
|
||||
|
||||
func (fi *fileInfo) IsDir() bool {
|
||||
@ -112,15 +112,14 @@ func (f *file) Readdir(count int) ([]os.FileInfo, error) {
|
||||
func (f *file) Seek(offset int64, whence int) (int64, error) {
|
||||
// log.Printf("seek %d %d %s\n", offset, whence, f.name)
|
||||
switch whence {
|
||||
case os.SEEK_SET:
|
||||
case io.SeekStart:
|
||||
f.offset = offset
|
||||
case os.SEEK_CUR:
|
||||
case io.SeekCurrent:
|
||||
f.offset += offset
|
||||
case os.SEEK_END:
|
||||
case io.SeekEnd:
|
||||
f.offset = int64(len(f.data)) + offset
|
||||
}
|
||||
return f.offset, nil
|
||||
|
||||
}
|
||||
|
||||
func (f *file) Stat() (os.FileInfo, error) {
|
||||
|
@ -2,7 +2,7 @@ package structfs
|
||||
|
||||
import (
|
||||
"encoding/json"
|
||||
"io/ioutil"
|
||||
"io"
|
||||
"net/http"
|
||||
"reflect"
|
||||
"testing"
|
||||
@ -82,7 +82,7 @@ func get(path string) ([]byte, error) {
|
||||
return nil, err
|
||||
}
|
||||
defer res.Body.Close()
|
||||
return ioutil.ReadAll(res.Body)
|
||||
return io.ReadAll(res.Body)
|
||||
}
|
||||
|
||||
func TestAll(t *testing.T) {
|
||||
|
@ -35,7 +35,7 @@ func TestUnmarshalJSON(t *testing.T) {
|
||||
err = json.Unmarshal([]byte(`{"ttl":"1y"}`), v)
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
} else if v.TTL != 31536000000000000 {
|
||||
} else if v.TTL != 31622400000000000 {
|
||||
t.Fatalf("invalid duration %v != 31536000000000000", v.TTL)
|
||||
}
|
||||
}
|
||||
@ -55,7 +55,7 @@ func TestParseDuration(t *testing.T) {
|
||||
if err != nil {
|
||||
t.Fatalf("ParseDuration error: %v", err)
|
||||
}
|
||||
if td.String() != "8760h0m0s" {
|
||||
if td.String() != "8784h0m0s" {
|
||||
t.Fatalf("ParseDuration 1y != 8760h0m0s : %s", td.String())
|
||||
}
|
||||
}
|
||||
|
25
util/xpool/pool.go
Normal file
25
util/xpool/pool.go
Normal file
@ -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)
|
||||
}
|
27
util/xpool/pool_test.go
Normal file
27
util/xpool/pool_test.go
Normal file
@ -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)
|
||||
}
|
Loading…
x
Reference in New Issue
Block a user