feat: optimize metadata delivery

internal/xstrings/string.go

@@ -0,0 +1,16 @@
+package xstrings
+
+func Distinct(values []string) []string {
+	m := make(map[string]struct{}, len(values))
+	newValues := make([]string, 0, len(values))
+	for _, value := range values {
+		_, ok := m[value]
+		if ok {
+			continue
+		}
+		newValues = append(newValues, value)
+		m[value] = struct{}{}
+	}
+
+	return newValues
+}

md/grpc_metadata.go

@@ -6,6 +6,8 @@ import (
 	"google.golang.org/grpc/metadata"
 )
 
+const zeroFlowFieldsKey = "zero-flow-fields"
+
 var _ Carrier = (*GrpcMetadataCarrier)(nil)
 
 type GrpcMetadataCarrier metadata.MD
@@ -30,11 +32,5 @@ func (h GrpcMetadataCarrier) Set(key string, value ...string) {
 }
 
 func (h GrpcMetadataCarrier) Keys() []string {
-	keys := make([]string, 0, len(h))
-	for k := range h {
-		k = strings.ToLower(k)
-		keys = append(keys, k)
-	}
-
-	return keys
+	return h[zeroFlowFieldsKey]
 }

md/metadata.go

@@ -5,7 +5,9 @@ import (
 	"net/url"
 	"strings"
 
+	"github.com/chenquan/zero-flow/internal/xstrings"
 	"google.golang.org/grpc/attributes"
+	"google.golang.org/grpc/metadata"
 )
 
 var _ Carrier = (*Metadata)(nil)
@@ -91,11 +93,17 @@ func (m Metadata) Clone() Metadata {
 
 func (m Metadata) Merge(metadata Metadata) {
 	metadata.Range(func(key string, values ...string) bool {
-		m.Append(strings.ToLower(key), copyOf(values)...)
+		m.Append(strings.ToLower(key), copyOf(xstrings.Distinct(values))...)
 		return true
 	})
 }
 
+func (m Metadata) Distinct() {
+	for k, v := range m {
+		m[k] = xstrings.Distinct(v)
+	}
+}
+
 func FromContext(ctx context.Context) (Metadata, bool) {
 	value := ctx.Value(metadataKey{})
 	if value == nil {
@@ -114,18 +122,29 @@ func NewContext(ctx context.Context, carrier Carrier) context.Context {
 	return context.WithValue(ctx, metadataKey{}, md)
 }
 
+func ToGrpcMetadata(m Metadata) metadata.MD {
+	md := metadata.MD{}
+	keys := m.Keys()
+	for _, key := range keys {
+		md.Set(key, m.Get(key)...)
+	}
+	m.Set(zeroFlowFieldsKey, keys...)
+
+	return md
+}
+
 func NewMetaDataFromContext(ctx context.Context, carrier Carrier) (context.Context, Metadata) {
 	metadata, ok := FromContext(ctx)
 	if !ok {
 		metadata = Metadata{}
 	} else {
 		metadata = metadata.Clone()
 	}
-
-	for _, key := range carrier.Keys() {
-		metadata.Append(strings.ToLower(key), carrier.Get(key)...)
+	keys := carrier.Keys()
+	for _, key := range keys {
+		metadata.Append(strings.ToLower(key), xstrings.Distinct(carrier.Get(key))...)
 	}
-
+	metadata.Distinct()
 	return context.WithValue(ctx, metadataKey{}, metadata), metadata
 }
 

zrpc/internal/clientinterceptors/mdinterceptor.go

@@ -31,10 +31,10 @@ func injectionMd(ctx context.Context, defaultMd md.Metadata) context.Context {
 		outgoingMd = metadata.MD{}
 	}
 
-	m.Range(func(key string, values ...string) bool {
+	grpcMetadata := md.ToGrpcMetadata(m)
+	for key, values := range grpcMetadata {
 		outgoingMd.Append(key, values...)
-		return true
-	})
+	}
 	ctx = metadata.NewOutgoingContext(ctx, outgoingMd)
 
 	m, ok = md.FromContext(ctx)

zrpc/internal/clientinterceptors/mdinterceptor_test.go

@@ -0,0 +1 @@
+package clientinterceptors

zrpc/rpc.go

@@ -35,6 +35,7 @@ func MustNewClient(c RpcClientConf, options ...zrpc.ClientOption) zrpc.Client {
 	}
 	metadata := md.Metadata(query)
 	metadata.Merge(selector.DefaultSelectorMd)
+	metadata.Distinct()
 	svcCfg := fmt.Sprintf(`{"loadBalancingPolicy":"%s"}`, p2c.Name)
 	options = append(options,
 		zrpc.WithUnaryClientInterceptor(clientinterceptors.UnaryMdInterceptor(metadata.Clone())),