unistack-org/micro-client-grpc
4
0
Fork 1
Code Issues Pull Requests Actions Packages Projects Releases 35 Wiki Activity

fixup md #131

Merged
vtolstov merged 1 commits from pubmdfix into v3 2023-12-21 00:17:48 +03:00
Conversation 0 Commits 1 Files Changed 1 +9 -10
1 changed files with 9 additions and 10 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

19
grpc.go
View File

@ -718,6 +718,10 @@ func (g *grpcClient) publish(ctx context.Context, ps []client.Message, opts ...c
if v, ok := os.LookupEnv("MICRO_PROXY"); ok { if v, ok := os.LookupEnv("MICRO_PROXY"); ok {
exchange = v exchange = v
} }
// get the exchange
if len(options.Exchange) > 0 {
exchange = options.Exchange
}
msgs := make([]*broker.Message, 0, len(ps)) msgs := make([]*broker.Message, 0, len(ps))
@ -729,7 +733,11 @@ func (g *grpcClient) publish(ctx context.Context, ps []client.Message, opts ...c
for _, p := range ps { for _, p := range ps {
md := metadata.Copy(omd) md := metadata.Copy(omd)
md[metadata.HeaderContentType] = p.ContentType() md[metadata.HeaderContentType] = p.ContentType()
topic := p.Topic()
if len(exchange) > 0 {
topic = exchange
}
md.Set(metadata.HeaderTopic, topic)
iter := p.Metadata().Iterator() iter := p.Metadata().Iterator()
var k, v string var k, v string
for iter.Next(&k, &v) { for iter.Next(&k, &v) {
@ -753,15 +761,6 @@ func (g *grpcClient) publish(ctx context.Context, ps []client.Message, opts ...c
body = b body = b
} }
topic := p.Topic()
if len(exchange) > 0 {
topic = exchange
}
for k, v := range p.Metadata() {
md.Set(k, v)
}
md.Set(metadata.HeaderTopic, topic)
msgs = append(msgs, &broker.Message{Header: md, Body: body}) msgs = append(msgs, &broker.Message{Header: md, Body: body})
} }