unistack-org/micro-client-http
3
0
Fork 2
Code Issues 2 Pull Requests Actions Packages Projects Releases 63 Wiki Activity

parent bce6f8c3e75dfe7844a0458a2a01fb437a9ccc77

Browse Source 
author Vasiliy Tolstov <v.tolstov@unistack.org> 1646489425 +0300
committer Vasiliy Tolstov <v.tolstov@unistack.org> 1646494543 +0300

add message metadata support

Signed-off-by: Vasiliy Tolstov <v.tolstov@unistack.org>
This commit is contained in:
Василий Толстов 2022-03-05 17:10:25 +03:00
parent bce6f8c3e7
commit fe764ae5a3
2 changed files with 15 additions and 12 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

2
.github/workflows/automerge.yml vendored
View File

@ -1,4 +1,4 @@
name: "prautomerge"
name: "automerge"
on:
pull_request_target:

25
http.go
View File

@ -604,6 +604,8 @@ func (h *httpClient) Publish(ctx context.Context, p client.Message, opts ...clie
}
func (h *httpClient) publish(ctx context.Context, ps []client.Message, opts ...client.PublishOption) error {
var body []byte
options := client.NewPublishOptions(opts...)
// get proxy
@ -622,24 +624,22 @@ func (h *httpClient) publish(ctx context.Context, ps []client.Message, opts ...c
for _, p := range ps {
md := metadata.Copy(omd)
md[metadata.HeaderContentType] = p.ContentType()
md[metadata.HeaderTopic] = p.Topic()
cf, err := h.newCodec(p.ContentType())
if err != nil {
return errors.InternalServerError("go.micro.client", err.Error())
}
var body []byte
// passed in raw data
if d, ok := p.Payload().(*codec.Frame); ok {
body = d.Data
} else {
b := bytes.NewBuffer(nil)
if err := cf.Write(b, &codec.Message{Type: codec.Event}, p.Payload()); err != nil {
// use codec for payload
cf, err := h.newCodec(p.ContentType())
if err != nil {
return errors.InternalServerError("go.micro.client", err.Error())
}
body = b.Bytes()
// set the body
b, err := cf.Marshal(p.Payload())
if err != nil {
return errors.InternalServerError("go.micro.client", err.Error())
}
body = b
}
topic := p.Topic()
@ -647,6 +647,9 @@ func (h *httpClient) publish(ctx context.Context, ps []client.Message, opts ...c
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})
}