client/noop: fixup md #285
@ -3,6 +3,7 @@ package client
|
|||||||
import (
|
import (
|
||||||
"context"
|
"context"
|
||||||
"fmt"
|
"fmt"
|
||||||
|
"os"
|
||||||
"time"
|
"time"
|
||||||
|
|
||||||
"go.unistack.org/micro/v3/broker"
|
"go.unistack.org/micro/v3/broker"
|
||||||
@ -485,30 +486,35 @@ func (n *noopClient) publish(ctx context.Context, ps []Message, opts ...PublishO
|
|||||||
|
|
||||||
msgs := make([]*broker.Message, 0, len(ps))
|
msgs := make([]*broker.Message, 0, len(ps))
|
||||||
|
|
||||||
for _, p := range ps {
|
// get proxy
|
||||||
|
exchange := ""
|
||||||
|
if v, ok := os.LookupEnv("MICRO_PROXY"); ok {
|
||||||
|
exchange = v
|
||||||
|
}
|
||||||
|
// get the exchange
|
||||||
|
if len(options.Exchange) > 0 {
|
||||||
|
exchange = options.Exchange
|
||||||
|
}
|
||||||
|
|
||||||
omd, ok := metadata.FromOutgoingContext(ctx)
|
omd, ok := metadata.FromOutgoingContext(ctx)
|
||||||
if !ok {
|
if !ok {
|
||||||
omd = metadata.New(0)
|
omd = metadata.New(0)
|
||||||
}
|
}
|
||||||
|
|
||||||
|
for _, p := range ps {
|
||||||
md := metadata.Copy(omd)
|
md := metadata.Copy(omd)
|
||||||
|
md[metadata.HeaderContentType] = p.ContentType()
|
||||||
|
topic := p.Topic()
|
||||||
|
if len(exchange) > 0 {
|
||||||
|
topic = exchange
|
||||||
|
}
|
||||||
|
md[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) {
|
||||||
md.Set(k, v)
|
md.Set(k, v)
|
||||||
}
|
}
|
||||||
|
|
||||||
md[metadata.HeaderContentType] = p.ContentType()
|
|
||||||
|
|
||||||
topic := p.Topic()
|
|
||||||
|
|
||||||
// get the exchange
|
|
||||||
if len(options.Exchange) > 0 {
|
|
||||||
topic = options.Exchange
|
|
||||||
}
|
|
||||||
|
|
||||||
md[metadata.HeaderTopic] = topic
|
|
||||||
|
|
||||||
var body []byte
|
var body []byte
|
||||||
|
|
||||||
// passed in raw data
|
// passed in raw data
|
||||||
|
Loading…
Reference in New Issue
Block a user