add message metadata support #59

Closed
vtolstov wants to merge 7 commits from messagemd into master
2 changed files with 14 additions and 16 deletions

View File

@ -15,11 +15,6 @@ jobs:
github.actor == 'vtolstov' || github.actor == 'vtolstov' ||
github.actor == 'dependabot[bot]' github.actor == 'dependabot[bot]'
steps: steps:
- name: metadata
id: metadata
uses: dependabot/fetch-metadata@v1.3.0
with:
github-token: "${{ secrets.TOKEN }}"
- name: approve - name: approve
run: gh pr review --approve "$PR_URL" run: gh pr review --approve "$PR_URL"
env: env:

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