Compare commits
6 Commits
Author | SHA1 | Date | |
---|---|---|---|
8d747c64a8 | |||
94beb5ed3b | |||
98981ba86c | |||
1013f50d0e | |||
0b190997b1 | |||
69a44eb190 |
@ -109,7 +109,7 @@ func (m *memoryBroker) Init(opts ...broker.Option) error {
|
||||
m.funcSubscribe = m.fnSubscribe
|
||||
m.funcBatchSubscribe = m.fnBatchSubscribe
|
||||
|
||||
m.opts.Hooks.EachNext(func(hook options.Hook) {
|
||||
m.opts.Hooks.EachPrev(func(hook options.Hook) {
|
||||
switch h := hook.(type) {
|
||||
case broker.HookPublish:
|
||||
m.funcPublish = h(m.funcPublish)
|
||||
|
@ -59,7 +59,7 @@ func (b *NoopBroker) Init(opts ...Option) error {
|
||||
b.funcSubscribe = b.fnSubscribe
|
||||
b.funcBatchSubscribe = b.fnBatchSubscribe
|
||||
|
||||
b.opts.Hooks.EachNext(func(hook options.Hook) {
|
||||
b.opts.Hooks.EachPrev(func(hook options.Hook) {
|
||||
switch h := hook.(type) {
|
||||
case HookPublish:
|
||||
b.funcPublish = h(b.funcPublish)
|
||||
|
@ -194,7 +194,7 @@ func (n *noopClient) Init(opts ...Option) error {
|
||||
n.funcPublish = n.fnPublish
|
||||
n.funcBatchPublish = n.fnBatchPublish
|
||||
|
||||
n.opts.Hooks.EachNext(func(hook options.Hook) {
|
||||
n.opts.Hooks.EachPrev(func(hook options.Hook) {
|
||||
switch h := hook.(type) {
|
||||
case HookCall:
|
||||
n.funcCall = h(n.funcCall)
|
||||
|
@ -37,7 +37,7 @@ func (c *defaultConfig) Init(opts ...Option) error {
|
||||
c.funcLoad = c.fnLoad
|
||||
c.funcSave = c.fnSave
|
||||
|
||||
c.opts.Hooks.EachNext(func(hook options.Hook) {
|
||||
c.opts.Hooks.EachPrev(func(hook options.Hook) {
|
||||
switch h := hook.(type) {
|
||||
case HookLoad:
|
||||
c.funcLoad = h(c.funcLoad)
|
||||
|
@ -68,15 +68,7 @@ func (md Metadata) Iterator() *Iterator {
|
||||
}
|
||||
|
||||
func (md Metadata) MustGet(key string) string {
|
||||
// fast path
|
||||
val, ok := md[key]
|
||||
if !ok {
|
||||
// slow path
|
||||
val, ok = md[textproto.CanonicalMIMEHeaderKey(key)]
|
||||
if !ok {
|
||||
val, ok = md[strings.ToLower(key)]
|
||||
}
|
||||
}
|
||||
val, ok := md.Get(key)
|
||||
if !ok {
|
||||
panic("missing metadata key")
|
||||
}
|
||||
@ -119,11 +111,18 @@ func (md Metadata) Del(keys ...string) {
|
||||
}
|
||||
}
|
||||
|
||||
// Copy makes a copy of the metadata
|
||||
func (md Metadata) CopyTo(dst Metadata) {
|
||||
for k, v := range md {
|
||||
dst[k] = v
|
||||
}
|
||||
}
|
||||
|
||||
// Copy makes a copy of the metadata
|
||||
func Copy(md Metadata, exclude ...string) Metadata {
|
||||
nmd := New(len(md))
|
||||
for key, val := range md {
|
||||
nmd.Set(key, val)
|
||||
for k, v := range md {
|
||||
nmd[k] = v
|
||||
}
|
||||
nmd.Del(exclude...)
|
||||
return nmd
|
||||
@ -147,7 +146,7 @@ func Merge(omd Metadata, mmd Metadata, overwrite bool) Metadata {
|
||||
case ok && !overwrite:
|
||||
continue
|
||||
case val != "":
|
||||
nmd.Set(key, val)
|
||||
nmd[key] = val
|
||||
case ok && val == "":
|
||||
nmd.Del(key)
|
||||
}
|
||||
@ -161,6 +160,8 @@ func Pairs(kv ...string) (Metadata, bool) {
|
||||
return nil, false
|
||||
}
|
||||
md := New(len(kv) / 2)
|
||||
md.Set(kv...)
|
||||
for idx := 0; idx < len(kv); idx += 2 {
|
||||
md[kv[idx]] = kv[idx+1]
|
||||
}
|
||||
return md, true
|
||||
}
|
||||
|
18
semconv/metadata.go
Normal file
18
semconv/metadata.go
Normal file
@ -0,0 +1,18 @@
|
||||
package semconv
|
||||
|
||||
var (
|
||||
// HeaderTopic is the header name that contains topic name
|
||||
HeaderTopic = "Micro-Topic"
|
||||
// HeaderContentType specifies content type of message
|
||||
HeaderContentType = "Content-Type"
|
||||
// HeaderEndpoint specifies endpoint in service
|
||||
HeaderEndpoint = "Micro-Endpoint"
|
||||
// HeaderService specifies service
|
||||
HeaderService = "Micro-Service"
|
||||
// HeaderTimeout specifies timeout of operation
|
||||
HeaderTimeout = "Micro-Timeout"
|
||||
// HeaderAuthorization specifies Authorization header
|
||||
HeaderAuthorization = "Authorization"
|
||||
// HeaderXRequestID specifies request id
|
||||
HeaderXRequestID = "X-Request-Id"
|
||||
)
|
@ -723,7 +723,7 @@ func (n *noopServer) createSubHandler(sb *subscriber, opts Options) broker.Handl
|
||||
return nil
|
||||
}
|
||||
|
||||
opts.Hooks.EachNext(func(hook options.Hook) {
|
||||
opts.Hooks.EachPrev(func(hook options.Hook) {
|
||||
if h, ok := hook.(HookSubHandler); ok {
|
||||
fn = h(fn)
|
||||
}
|
||||
|
@ -9,6 +9,7 @@ import (
|
||||
"go.unistack.org/micro/v3/client"
|
||||
"go.unistack.org/micro/v3/codec"
|
||||
"go.unistack.org/micro/v3/logger"
|
||||
"go.unistack.org/micro/v3/options"
|
||||
"go.unistack.org/micro/v3/server"
|
||||
)
|
||||
|
||||
@ -84,3 +85,40 @@ func TestNoopSub(t *testing.T) {
|
||||
}
|
||||
}()
|
||||
}
|
||||
|
||||
func TestHooks_Wrap(t *testing.T) {
|
||||
n := 5
|
||||
fn1 := func(next server.FuncSubHandler) server.FuncSubHandler {
|
||||
return func(ctx context.Context, msg server.Message) (err error) {
|
||||
n *= 2
|
||||
return next(ctx, msg)
|
||||
}
|
||||
}
|
||||
fn2 := func(next server.FuncSubHandler) server.FuncSubHandler {
|
||||
return func(ctx context.Context, msg server.Message) (err error) {
|
||||
n -= 10
|
||||
return next(ctx, msg)
|
||||
}
|
||||
}
|
||||
|
||||
hs := &options.Hooks{}
|
||||
hs.Append(server.HookSubHandler(fn1), server.HookSubHandler(fn2))
|
||||
|
||||
var fn = func(ctx context.Context, msg server.Message) error {
|
||||
return nil
|
||||
}
|
||||
|
||||
hs.EachPrev(func(hook options.Hook) {
|
||||
if h, ok := hook.(server.HookSubHandler); ok {
|
||||
fn = h(fn)
|
||||
}
|
||||
})
|
||||
|
||||
if err := fn(nil, nil); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
|
||||
if n != 0 {
|
||||
t.Fatalf("uncorrected hooks call")
|
||||
}
|
||||
}
|
||||
|
@ -123,7 +123,7 @@ func (m *memoryStore) Init(opts ...store.Option) error {
|
||||
m.funcList = m.fnList
|
||||
m.funcDelete = m.fnDelete
|
||||
|
||||
m.opts.Hooks.EachNext(func(hook options.Hook) {
|
||||
m.opts.Hooks.EachPrev(func(hook options.Hook) {
|
||||
switch h := hook.(type) {
|
||||
case store.HookRead:
|
||||
m.funcRead = h(m.funcRead)
|
||||
|
@ -54,7 +54,7 @@ func (n *noopStore) Init(opts ...Option) error {
|
||||
n.funcList = n.fnList
|
||||
n.funcDelete = n.fnDelete
|
||||
|
||||
n.opts.Hooks.EachNext(func(hook options.Hook) {
|
||||
n.opts.Hooks.EachPrev(func(hook options.Hook) {
|
||||
switch h := hook.(type) {
|
||||
case HookRead:
|
||||
n.funcRead = h(n.funcRead)
|
||||
|
@ -37,6 +37,15 @@ func NewContext(ctx context.Context, tracer Tracer) context.Context {
|
||||
|
||||
type spanKey struct{}
|
||||
|
||||
// SpanFromContext returns a span from context
|
||||
func SpanMustContext(ctx context.Context) Span {
|
||||
sp, ok := SpanFromContext(ctx)
|
||||
if !ok {
|
||||
panic("missing span")
|
||||
}
|
||||
return sp
|
||||
}
|
||||
|
||||
// SpanFromContext returns a span from context
|
||||
func SpanFromContext(ctx context.Context) (Span, bool) {
|
||||
if ctx == nil {
|
||||
|
@ -25,6 +25,7 @@ func (t *Tracer) Start(ctx context.Context, name string, opts ...tracer.SpanOpti
|
||||
name: name,
|
||||
ctx: ctx,
|
||||
tracer: t,
|
||||
labels: options.Labels,
|
||||
kind: options.Kind,
|
||||
startTime: time.Now(),
|
||||
}
|
||||
@ -37,6 +38,14 @@ func (t *Tracer) Start(ctx context.Context, name string, opts ...tracer.SpanOpti
|
||||
return tracer.NewSpanContext(ctx, span), span
|
||||
}
|
||||
|
||||
type memoryStringer struct {
|
||||
s string
|
||||
}
|
||||
|
||||
func (s memoryStringer) String() string {
|
||||
return s.s
|
||||
}
|
||||
|
||||
func (t *Tracer) Flush(_ context.Context) error {
|
||||
return nil
|
||||
}
|
||||
@ -52,14 +61,6 @@ func (t *Tracer) Name() string {
|
||||
return t.opts.Name
|
||||
}
|
||||
|
||||
type noopStringer struct {
|
||||
s string
|
||||
}
|
||||
|
||||
func (s noopStringer) String() string {
|
||||
return s.s
|
||||
}
|
||||
|
||||
type Span struct {
|
||||
ctx context.Context
|
||||
tracer tracer.Tracer
|
||||
@ -67,8 +68,8 @@ type Span struct {
|
||||
statusMsg string
|
||||
startTime time.Time
|
||||
finishTime time.Time
|
||||
traceID noopStringer
|
||||
spanID noopStringer
|
||||
traceID memoryStringer
|
||||
spanID memoryStringer
|
||||
events []*Event
|
||||
labels []interface{}
|
||||
logs []interface{}
|
||||
|
@ -2,6 +2,7 @@ package tracer
|
||||
|
||||
import (
|
||||
"context"
|
||||
"time"
|
||||
|
||||
"go.unistack.org/micro/v3/util/id"
|
||||
)
|
||||
@ -23,6 +24,7 @@ func (t *noopTracer) Start(ctx context.Context, name string, opts ...SpanOption)
|
||||
name: name,
|
||||
ctx: ctx,
|
||||
tracer: t,
|
||||
startTime: time.Now(),
|
||||
labels: options.Labels,
|
||||
kind: options.Kind,
|
||||
}
|
||||
@ -31,7 +33,6 @@ func (t *noopTracer) Start(ctx context.Context, name string, opts ...SpanOption)
|
||||
if span.ctx == nil {
|
||||
span.ctx = context.Background()
|
||||
}
|
||||
t.spans = append(t.spans, span)
|
||||
return NewSpanContext(ctx, span), span
|
||||
}
|
||||
|
||||
@ -58,21 +59,16 @@ func (t *noopTracer) Name() string {
|
||||
return t.opts.Name
|
||||
}
|
||||
|
||||
type noopEvent struct {
|
||||
name string
|
||||
labels []interface{}
|
||||
}
|
||||
|
||||
type noopSpan struct {
|
||||
ctx context.Context
|
||||
tracer Tracer
|
||||
name string
|
||||
statusMsg string
|
||||
startTime time.Time
|
||||
finishTime time.Time
|
||||
traceID noopStringer
|
||||
spanID noopStringer
|
||||
events []*noopEvent
|
||||
labels []interface{}
|
||||
logs []interface{}
|
||||
kind SpanKind
|
||||
status SpanStatus
|
||||
}
|
||||
@ -86,6 +82,7 @@ func (s *noopSpan) SpanID() string {
|
||||
}
|
||||
|
||||
func (s *noopSpan) Finish(_ ...SpanOption) {
|
||||
s.finishTime = time.Now()
|
||||
}
|
||||
|
||||
func (s *noopSpan) Context() context.Context {
|
||||
@ -97,8 +94,6 @@ func (s *noopSpan) Tracer() Tracer {
|
||||
}
|
||||
|
||||
func (s *noopSpan) AddEvent(name string, opts ...EventOption) {
|
||||
options := NewEventOptions(opts...)
|
||||
s.events = append(s.events, &noopEvent{name: name, labels: options.Labels})
|
||||
}
|
||||
|
||||
func (s *noopSpan) SetName(name string) {
|
||||
@ -106,7 +101,6 @@ func (s *noopSpan) SetName(name string) {
|
||||
}
|
||||
|
||||
func (s *noopSpan) AddLogs(kv ...interface{}) {
|
||||
s.logs = append(s.logs, kv...)
|
||||
}
|
||||
|
||||
func (s *noopSpan) AddLabels(kv ...interface{}) {
|
||||
|
Loading…
x
Reference in New Issue
Block a user