unistack-org/micro
4
0
Fork 3
Code Issues 19 Pull Requests 2 Actions 2 Packages Projects Releases 262 Wiki Activity

rework metadata #318

Merged
vtolstov merged 1 commits from metav4 into master 2024-03-09 17:24:03 +03:00
Conversation 0 Commits 1 Files Changed 12 +152 -279
12 changed files with 152 additions and 279 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files
Showing only changes of commit 2f8c0d9f9d - Show all commits

16
broker/memory_test.go
View File

@ -37,10 +37,10 @@ func TestMemoryBatchBroker(t *testing.T) {
msgs := make([]Message, 0, count) msgs := make([]Message, 0, count)
for i := 0; i < count; i++ { for i := 0; i < count; i++ {
message := &memoryMessage{ message := &memoryMessage{
header: map[string]string{ header: metadata.Metadata{
metadata.HeaderTopic: topic, metadata.HeaderTopic: []string{topic},
"foo": "bar", "foo": []string{"bar"},
"id": fmt.Sprintf("%d", i), "id": []string{fmt.Sprintf("%d", i)},
}, },
body: []byte(`"hello world"`), body: []byte(`"hello world"`),
} }
@ -83,10 +83,10 @@ func TestMemoryBroker(t *testing.T) {
msgs := make([]Message, 0, count) msgs := make([]Message, 0, count)
for i := 0; i < count; i++ { for i := 0; i < count; i++ {
message := &memoryMessage{ message := &memoryMessage{
header: map[string]string{ header: metadata.Metadata{
metadata.HeaderTopic: topic, metadata.HeaderTopic: []string{topic},
"foo": "bar", "foo": []string{"bar"},
"id": fmt.Sprintf("%d", i), "id": []string{fmt.Sprintf("%d", i)},
}, },
body: []byte(`"hello world"`), body: []byte(`"hello world"`),
} }

78
metadata/context.go
View File

@ -17,11 +17,11 @@ func FromIncomingContext(ctx context.Context) (Metadata, bool) {
if ctx == nil { if ctx == nil {
return nil, false return nil, false
} }
md, ok := ctx.Value(mdIncomingKey{}).(*rawMetadata) md, ok := ctx.Value(mdIncomingKey{}).(Metadata)
if !ok || md.md == nil { if !ok || md == nil {
return nil, false return nil, false
} }
return md.md, ok return md, ok
} }
// FromOutgoingContext returns metadata from outgoing ctx // FromOutgoingContext returns metadata from outgoing ctx
@ -30,11 +30,11 @@ func FromOutgoingContext(ctx context.Context) (Metadata, bool) {
if ctx == nil { if ctx == nil {
return nil, false return nil, false
} }
md, ok := ctx.Value(mdOutgoingKey{}).(*rawMetadata) md, ok := ctx.Value(mdOutgoingKey{}).(Metadata)
if !ok || md.md == nil { if !ok || md == nil {
return nil, false return nil, false
} }
return md.md, ok return md, ok
} }
// FromContext returns metadata from the given context // FromContext returns metadata from the given context
@ -43,11 +43,11 @@ func FromContext(ctx context.Context) (Metadata, bool) {
if ctx == nil { if ctx == nil {
return nil, false return nil, false
} }
md, ok := ctx.Value(mdKey{}).(*rawMetadata) md, ok := ctx.Value(mdKey{}).(Metadata)
if !ok || md.md == nil { if !ok || md == nil {
return nil, false return nil, false
} }
return md.md, ok return md, ok
} }
// NewContext creates a new context with the given metadata // 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 { if ctx == nil {
ctx = context.Background() ctx = context.Background()
} }
ctx = context.WithValue(ctx, mdKey{}, &rawMetadata{md}) ctx = context.WithValue(ctx, mdKey{}, md)
ctx = context.WithValue(ctx, mdIncomingKey{}, &rawMetadata{})
ctx = context.WithValue(ctx, mdOutgoingKey{}, &rawMetadata{})
return ctx 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 // NewIncomingContext creates a new context with incoming metadata attached
func NewIncomingContext(ctx context.Context, md Metadata) context.Context { func NewIncomingContext(ctx context.Context, md Metadata) context.Context {
if ctx == nil { if ctx == nil {
ctx = context.Background() ctx = context.Background()
} }
ctx = context.WithValue(ctx, mdIncomingKey{}, &rawMetadata{md}) ctx = context.WithValue(ctx, mdIncomingKey{}, md)
if v, ok := ctx.Value(mdOutgoingKey{}).(*rawMetadata); !ok || v == nil {
ctx = context.WithValue(ctx, mdOutgoingKey{}, &rawMetadata{})
}
return ctx return ctx
} }
@ -102,41 +73,28 @@ func NewOutgoingContext(ctx context.Context, md Metadata) context.Context {
if ctx == nil { if ctx == nil {
ctx = context.Background() ctx = context.Background()
} }
ctx = context.WithValue(ctx, mdOutgoingKey{}, &rawMetadata{md}) ctx = context.WithValue(ctx, mdOutgoingKey{}, md)
if v, ok := ctx.Value(mdIncomingKey{}).(*rawMetadata); !ok || v == nil {
ctx = context.WithValue(ctx, mdIncomingKey{}, &rawMetadata{})
}
return ctx return ctx
} }
// AppendOutgoingContext apends new md to context // AppendOutgoingContext apends new md to context
func AppendOutgoingContext(ctx context.Context, kv ...string) context.Context { func AppendOutgoingContext(ctx context.Context, kv ...string) context.Context {
md, ok := Pairs(kv...) md := Pairs(kv...)
if !ok {
return ctx
}
omd, ok := FromOutgoingContext(ctx) omd, ok := FromOutgoingContext(ctx)
if !ok { if !ok {
return NewOutgoingContext(ctx, md) return NewOutgoingContext(ctx, md)
} }
for k, v := range md { nmd := Merge(omd, md, true)
omd.Set(k, v) return NewOutgoingContext(ctx, nmd)
}
return NewOutgoingContext(ctx, omd)
} }
// AppendIncomingContext apends new md to context // AppendIncomingContext apends new md to context
func AppendIncomingContext(ctx context.Context, kv ...string) context.Context { func AppendIncomingContext(ctx context.Context, kv ...string) context.Context {
md, ok := Pairs(kv...) md := Pairs(kv...)
if !ok {
return ctx
}
omd, ok := FromIncomingContext(ctx) omd, ok := FromIncomingContext(ctx)
if !ok { if !ok {
return NewIncomingContext(ctx, md) return NewIncomingContext(ctx, md)
} }
for k, v := range md { nmd := Merge(omd, md, true)
omd.Set(k, v) return NewIncomingContext(ctx, nmd)
}
return NewIncomingContext(ctx, omd)
} }

36
metadata/context_test.go
View File

@ -24,7 +24,7 @@ func TestNewNilContext(t *testing.T) {
} }
func TestFromContext(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) c, ok := FromContext(ctx)
if c == nil || !ok { if c == nil || !ok {
@ -42,7 +42,7 @@ func TestNewContext(t *testing.T) {
} }
func TestFromIncomingContext(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) c, ok := FromIncomingContext(ctx)
if c == nil || !ok { if c == nil || !ok {
@ -51,7 +51,7 @@ func TestFromIncomingContext(t *testing.T) {
} }
func TestFromOutgoingContext(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) c, ok := FromOutgoingContext(ctx)
if c == nil || !ok { 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) { func TestNewIncomingContext(t *testing.T) {
md := New(1) md := New(1)
md.Set("key", "val") md.Set("key", "val")

92
metadata/metadata.go
View File

@ -1,9 +1,9 @@
// Package metadata is a way of defining message headers // Package metadata is a way of defining message headers
package metadata // import "go.unistack.org/micro/v4/metadata" package metadata
import ( import (
"net/textproto" "net/textproto"
"sort" "strings"
) )
var ( var (
@ -24,47 +24,7 @@ var (
) )
// Metadata is our way of representing request headers internally. // Metadata is our way of representing request headers internally.
// They're used at the RPC level and translate back and forth type Metadata map[string][]string
// 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
}
// Get returns value from metadata by key // Get returns value from metadata by key
func (md Metadata) Get(key string) (string, bool) { func (md Metadata) Get(key string) (string, bool) {
@ -74,7 +34,7 @@ func (md Metadata) Get(key string) (string, bool) {
// slow path // slow path
val, ok = md[textproto.CanonicalMIMEHeaderKey(key)] val, ok = md[textproto.CanonicalMIMEHeaderKey(key)]
} }
return val, ok return strings.Join(val, ","), ok
} }
// Set is used to store value in metadata // Set is used to store value in metadata
@ -83,10 +43,19 @@ func (md Metadata) Set(kv ...string) {
kv = kv[:len(kv)-1] kv = kv[:len(kv)-1]
} }
for idx := 0; idx < len(kv); idx += 2 { 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 // Del is used to remove value from metadata
func (md Metadata) Del(keys ...string) { func (md Metadata) Del(keys ...string) {
for _, key := range keys { for _, key := range keys {
@ -99,9 +68,9 @@ func (md Metadata) Del(keys ...string) {
// Copy makes a copy of the metadata // Copy makes a copy of the metadata
func Copy(md Metadata) Metadata { func Copy(md Metadata) Metadata {
nmd := New(len(md)) nmd := make(Metadata, len(md))
for key, val := range md { for k, v := range md {
nmd.Set(key, val) nmd[k] = v
} }
return nmd return nmd
} }
@ -109,35 +78,40 @@ func Copy(md Metadata) Metadata {
// New return new sized metadata // New return new sized metadata
func New(size int) Metadata { func New(size int) Metadata {
if size == 0 { if size == 0 {
size = defaultMetadataSize size = 2
} }
return make(Metadata, size) return make(Metadata, size)
} }
// Merge merges metadata to existing metadata, overwriting if specified // Merge merges metadata to existing metadata, overwriting if specified
func Merge(omd Metadata, mmd Metadata, overwrite bool) Metadata { func Merge(omd Metadata, mmd Metadata, overwrite bool) Metadata {
var ok bool
nmd := Copy(omd) nmd := Copy(omd)
for key, val := range mmd { for key, val := range mmd {
_, ok = nmd[key] nval, ok := nmd[key]
switch { switch {
case ok && overwrite:
nmd[key] = nval
continue
case ok && !overwrite: case ok && !overwrite:
continue continue
case val != "": case !ok:
nmd.Set(key, val) for _, v := range val {
case ok && val == "": if v != "" {
nmd.Del(key) nval = append(nval, v)
}
}
nmd[key] = nval
} }
} }
return nmd return nmd
} }
// Pairs from which metadata created // Pairs from which metadata created
func Pairs(kv ...string) (Metadata, bool) { func Pairs(kv ...string) Metadata {
if len(kv)%2 == 1 { 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...) md.Set(kv...)
return md, true return md
} }

62
metadata/metadata_test.go
View File

@ -33,20 +33,25 @@ func TestAppend(t *testing.T) {
} }
func TestPairs(t *testing.T) { func TestPairs(t *testing.T) {
md, ok := Pairs("key1", "val1", "key2", "val2") md := Pairs("key1", "val1", "key2", "val2")
if !ok {
t.Fatal("odd number of kv") if _, ok := md.Get("key1"); !ok {
}
if _, ok = md.Get("key1"); !ok {
t.Fatal("key1 not found") t.Fatal("key1 not found")
} }
} }
func testCtx(ctx context.Context) { func testIncomingCtx(ctx context.Context) {
md := New(2) if md, ok := FromIncomingContext(ctx); ok && md != nil {
md.Set("Key1", "Val1_new") md.Set("Key1", "Val1_new")
md.Set("Key3", "Val3") md.Set("Key3", "Val3")
SetOutgoingContext(ctx, md) }
}
func testOutgoingCtx(ctx context.Context) {
if md, ok := FromOutgoingContext(ctx); ok && md != nil {
md.Set("Key1", "Val1_new")
md.Set("Key3", "Val3")
}
} }
func TestPassing(t *testing.T) { func TestPassing(t *testing.T) {
@ -55,8 +60,8 @@ func TestPassing(t *testing.T) {
md1.Set("Key1", "Val1") md1.Set("Key1", "Val1")
md1.Set("Key2", "Val2") md1.Set("Key2", "Val2")
ctx = NewIncomingContext(ctx, md1) ctx = NewOutgoingContext(ctx, md1)
testCtx(ctx) testOutgoingCtx(ctx)
md, ok := FromOutgoingContext(ctx) md, ok := FromOutgoingContext(ctx)
if !ok { if !ok {
t.Fatalf("missing metadata from outgoing context") t.Fatalf("missing metadata from outgoing context")
@ -68,10 +73,10 @@ func TestPassing(t *testing.T) {
func TestMerge(t *testing.T) { func TestMerge(t *testing.T) {
omd := Metadata{ omd := Metadata{
"key1": "val1", "key1": []string{"val1"},
} }
mmd := Metadata{ mmd := Metadata{
"key2": "val2", "key2": []string{"val2"},
} }
nmd := Merge(omd, mmd, true) nmd := Merge(omd, mmd, true)
@ -80,21 +85,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) { func TestMedataCanonicalKey(t *testing.T) {
md := New(1) md := New(1)
md.Set("x-request-id", "12345") md.Set("x-request-id", "12345")
@ -134,10 +124,7 @@ func TestMetadataSet(t *testing.T) {
} }
func TestMetadataDelete(t *testing.T) { func TestMetadataDelete(t *testing.T) {
md := Metadata{ md := Pairs("Foo", "bar", "Baz", "empty")
"Foo": "bar",
"Baz": "empty",
}
md.Del("Baz") md.Del("Baz")
_, ok := md.Get("Baz") _, ok := md.Get("Baz")
@ -156,24 +143,19 @@ func TestNilContext(t *testing.T) {
} }
func TestMetadataCopy(t *testing.T) { func TestMetadataCopy(t *testing.T) {
md := Metadata{ md := Pairs("Foo", "bar", "Bar", "baz")
"Foo": "bar",
"Bar": "baz",
}
cp := Copy(md) cp := Copy(md)
for k, v := range 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) t.Fatalf("Got %s:%s for %s:%s", k, cv, k, v)
} }
} }
} }
func TestMetadataContext(t *testing.T) { func TestMetadataContext(t *testing.T) {
md := Metadata{ md := Pairs("Foo", "bar")
"Foo": "bar",
}
ctx := NewContext(context.TODO(), md) ctx := NewContext(context.TODO(), md)
@ -182,7 +164,7 @@ func TestMetadataContext(t *testing.T) {
t.Errorf("Unexpected error retrieving metadata, got %t", ok) 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"]) t.Errorf("Expected key: %s val: %s, got key: %s val: %s", "Foo", md["Foo"], "Foo", emd["Foo"])
} }

21
options/options.go
View File

@ -158,17 +158,28 @@ func Metadata(md ...any) Option {
case metadata.Metadata: case metadata.Metadata:
result = metadata.Copy(vt) result = metadata.Copy(vt)
case map[string]string: 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) result = metadata.Copy(vt)
default: default:
result = metadata.New(0) result = metadata.New(0)
} }
} else { } else {
result = metadata.New(len(md) / 2) result = metadata.New(len(md) / 2)
for idx := 0; idx < len(md)/2; idx += 2 { for idx := 0; idx <= len(md)/2; idx += 2 {
k, kok := md[idx].(string) k, ok := md[idx].(string)
v, vok := md[idx+1].(string) switch vt := md[idx+1].(type) {
if kok && vok { case string:
result.Set(k, v) if ok {
result.Set(k, vt)
}
case []string:
if ok {
result.Append(k, vt...)
}
} }
} }
} }

36
options/options_test.go
View File

@ -100,40 +100,29 @@ func TestMetadataAny(t *testing.T) {
}{ }{
{ {
"strings_even", "strings_even",
[]any{"key1", "val1", "key2", "val2"}, []any{"Strkey1", []string{"val1"}, "Strkey2", []string{"val2"}},
metadata.Metadata{ metadata.Pairs("Strkey1", "val1", "Strkey2", "val2"),
"Key1": "val1",
"Key2": "val2",
},
}, },
{ {
"strings_odd", "strings_odd",
[]any{"key1", "val1", "key2"}, []any{"key1", "val1", "key2"},
metadata.Metadata{ metadata.Pairs("Key1", "val1"),
"Key1": "val1",
},
}, },
{ {
Name: "map", Name: "map",
Data: map[string]string{ Data: map[string][]string{
"key1": "val1", "Mapkey1": {"val1"},
"key2": "val2", "Mapkey2": {"val2"},
}, },
Expected: metadata.Metadata{ Expected: metadata.Metadata{
"Key1": "val1", "Mapkey1": []string{"val1"},
"Key2": "val2", "Mapkey2": []string{"val2"},
}, },
}, },
{ {
"metadata.Metadata", "metadata.Metadata",
metadata.Metadata{ metadata.Pairs("key1", "val1", "key2", "val2"),
"key1": "val1", metadata.Pairs("Key1", "val1", "Key2", "val2"),
"key2": "val2",
},
metadata.Metadata{
"Key1": "val1",
"Key2": "val2",
},
}, },
} }
@ -143,8 +132,9 @@ func TestMetadataAny(t *testing.T) {
var opts []options.Option var opts []options.Option
switch valData := tt.Data.(type) { switch valData := tt.Data.(type) {
case []any: case []any:
fmt.Printf("%s any %#+v\n", tt.Name, valData)
opts = append(opts, options.Metadata(valData...)) 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)) opts = append(opts, options.Metadata(valData))
} }
@ -153,7 +143,7 @@ func TestMetadataAny(t *testing.T) {
t.Fatal(err) t.Fatal(err)
} }
if !reflect.Equal(tt.Expected, src.Metadata) { 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)
} }
} }
}) })

71
register/memory/memory.go
View File

@ -2,10 +2,12 @@ package memory
import ( import (
"context" "context"
"go.unistack.org/micro/v4/register"
"sync" "sync"
"time" "time"
"go.unistack.org/micro/v4/metadata"
"go.unistack.org/micro/v4/register"
"go.unistack.org/micro/v4/logger" "go.unistack.org/micro/v4/logger"
"go.unistack.org/micro/v4/util/id" "go.unistack.org/micro/v4/util/id"
) )
@ -24,7 +26,7 @@ type node struct {
type record struct { type record struct {
Name string Name string
Version string Version string
Metadata map[string]string Metadata metadata.Metadata
Nodes map[string]*node Nodes map[string]*node
Endpoints []*register.Endpoint Endpoints []*register.Endpoint
} }
@ -137,11 +139,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 // domain is set in metadata so it can be passed to watchers
if s.Metadata == nil { if s.Metadata == nil {
s.Metadata = map[string]string{"domain": options.Domain} s.Metadata = metadata.New(0)
} else {
s.Metadata["domain"] = options.Domain
} }
s.Metadata.Set("domain", options.Domain)
// ensure the service name exists // ensure the service name exists
r := serviceToRecord(s, options.TTL) r := serviceToRecord(s, options.TTL)
if _, ok := srvs[s.Name]; !ok { if _, ok := srvs[s.Name]; !ok {
@ -165,15 +167,8 @@ func (m *memory) Register(ctx context.Context, s *register.Service, opts ...regi
continue continue
} }
metadata := make(map[string]string, len(n.Metadata)) metadata := metadata.Copy(n.Metadata)
metadata.Set("domain", options.Domain)
// make copy of metadata
for k, v := range n.Metadata {
metadata[k] = v
}
// set the domain
metadata["domain"] = options.Domain
// add the node // add the node
srvs[s.Name][s.Version].Nodes[n.ID] = &node{ srvs[s.Name][s.Version].Nodes[n.ID] = &node{
@ -217,10 +212,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 // domain is set in metadata so it can be passed to watchers
if s.Metadata == nil { if s.Metadata == nil {
s.Metadata = map[string]string{"domain": options.Domain} s.Metadata = metadata.New(0)
} else {
s.Metadata["domain"] = options.Domain
} }
s.Metadata.Set("domain", options.Domain)
// if the domain doesn't exist, there is nothing to deregister // if the domain doesn't exist, there is nothing to deregister
services, ok := m.records[options.Domain] services, ok := m.records[options.Domain]
@ -425,16 +419,24 @@ func (m *watcher) Next() (*register.Result, error) {
continue continue
} }
if m.wo.Domain == register.WildcardDomain {
return r, nil
}
if r.Service.Metadata == nil {
continue
}
// extract domain from service metadata // extract domain from service metadata
var domain string var domain string
if r.Service.Metadata != nil && len(r.Service.Metadata["domain"]) > 0 { if v, ok := r.Service.Metadata.Get("domain"); ok && v != "" {
domain = r.Service.Metadata["domain"] domain = v
} else { } else {
domain = register.DefaultDomain domain = register.DefaultDomain
} }
// only send the event if watching the wildcard or this specific domain // 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 return r, nil
} }
case <-m.exit: case <-m.exit:
@ -453,10 +455,7 @@ func (m *watcher) Stop() {
} }
func serviceToRecord(s *register.Service, ttl time.Duration) *record { func serviceToRecord(s *register.Service, ttl time.Duration) *record {
metadata := make(map[string]string, len(s.Metadata)) metadata := metadata.Copy(s.Metadata)
for k, v := range s.Metadata {
metadata[k] = v
}
nodes := make(map[string]*node, len(s.Nodes)) nodes := make(map[string]*node, len(s.Nodes))
for _, n := range s.Nodes { for _, n := range s.Nodes {
@ -482,20 +481,11 @@ func serviceToRecord(s *register.Service, ttl time.Duration) *record {
} }
func recordToService(r *record, domain string) *register.Service { 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)) endpoints := make([]*register.Endpoint, len(r.Endpoints))
for i, e := range r.Endpoints { for i, e := range r.Endpoints {
md := make(map[string]string, len(e.Metadata)) md := metadata.Copy(e.Metadata)
for k, v := range e.Metadata { // set the domain in metadata so it can be determined when a wildcard query is performed
md[k] = v md.Set("domain", domain)
}
endpoints[i] = &register.Endpoint{ endpoints[i] = &register.Endpoint{
Name: e.Name, Name: e.Name,
@ -508,15 +498,10 @@ func recordToService(r *record, domain string) *register.Service {
nodes := make([]*register.Node, len(r.Nodes)) nodes := make([]*register.Node, len(r.Nodes))
i := 0 i := 0
for _, n := range r.Nodes { for _, n := range r.Nodes {
md := make(map[string]string, len(n.Metadata))
for k, v := range n.Metadata {
md[k] = v
}
nodes[i] = &register.Node{ nodes[i] = &register.Node{
ID: n.ID, ID: n.ID,
Address: n.Address, Address: n.Address,
Metadata: md, Metadata: metadata.Copy(n.Metadata),
} }
i++ i++
} }
@ -524,7 +509,7 @@ func recordToService(r *record, domain string) *register.Service {
return &register.Service{ return &register.Service{
Name: r.Name, Name: r.Name,
Version: r.Version, Version: r.Version,
Metadata: metadata, Metadata: metadata.Copy(r.Metadata),
Endpoints: endpoints, Endpoints: endpoints,
Nodes: nodes, Nodes: nodes,
} }

4
server/noop.go
View File

@ -112,8 +112,8 @@ func (n *noopServer) Register() error {
} }
n.RUnlock() n.RUnlock()
service.Nodes[0].Metadata["protocol"] = "noop" service.Nodes[0].Metadata.Set("protocol", "noop")
service.Nodes[0].Metadata["transport"] = service.Nodes[0].Metadata["protocol"] service.Nodes[0].Metadata.Set("transport", "noop")
service.Endpoints = endpoints service.Endpoints = endpoints
n.RLock() n.RLock()

4
server/registry.go
View File

@ -77,8 +77,8 @@ func NewRegisterService(s Server) (*register.Service, error) {
} }
node.Metadata = metadata.Copy(opts.Metadata) node.Metadata = metadata.Copy(opts.Metadata)
node.Metadata["server"] = s.String() node.Metadata.Set("server", s.String())
node.Metadata["register"] = opts.Register.String() node.Metadata.Set("register", opts.Register.String())
return &register.Service{ return &register.Service{
Name: opts.Name, Name: opts.Name,

3
util/reflect/reflect.go
View File

@ -532,6 +532,9 @@ func Equal(src interface{}, dst interface{}, excptFields ...string) bool {
} }
s := srcVal.MapIndex(key) s := srcVal.MapIndex(key)
d := dstVal.MapIndex(key) d := dstVal.MapIndex(key)
if !s.IsValid() || !d.IsValid() {
return false
}
if !Equal(s.Interface(), d.Interface(), excptFields...) { if !Equal(s.Interface(), d.Interface(), excptFields...) {
return false return false
} }

4
util/time/duration_test.go
View File

@ -35,7 +35,7 @@ func TestUnmarshalJSON(t *testing.T) {
err = json.Unmarshal([]byte(`{"ttl":"1y"}`), v) err = json.Unmarshal([]byte(`{"ttl":"1y"}`), v)
if err != nil { if err != nil {
t.Fatal(err) t.Fatal(err)
} else if v.TTL != 31536000000000000 { } else if v.TTL != 31622400000000000 {
t.Fatalf("invalid duration %v != 31536000000000000", v.TTL) t.Fatalf("invalid duration %v != 31536000000000000", v.TTL)
} }
} }
@ -55,7 +55,7 @@ func TestParseDuration(t *testing.T) {
if err != nil { if err != nil {
t.Fatalf("ParseDuration error: %v", err) t.Fatalf("ParseDuration error: %v", err)
} }
if td.String() != "8760h0m0s" { if td.String() != "8784h0m0s" {
t.Fatalf("ParseDuration 1y != 8760h0m0s : %s", td.String()) t.Fatalf("ParseDuration 1y != 8760h0m0s : %s", td.String())
} }
} }