unistack-org/micro
4
0
Fork 2
Code Issues 18 Pull Requests Actions Packages Projects Releases 331 Wiki Activity

Set topic header

Browse Source
This commit is contained in:
Asim Aslam
2019-01-10 20:35:10 +00:00
parent 59d82b0abe
commit 40ff5b749b
1 changed files with 12 additions and 1 deletions
Download Patch File Download Diff File Expand all files Collapse all files

13
client/rpc_client.go
View File

@@ -9,6 +9,7 @@ import (
"sync/atomic" "sync/atomic"
"github.com/google/uuid"
"github.com/micro/go-micro/broker" "github.com/micro/go-micro/broker"
"github.com/micro/go-micro/codec" "github.com/micro/go-micro/codec"
"github.com/micro/go-micro/errors" "github.com/micro/go-micro/errors"
@@ -444,7 +445,11 @@ func (r *rpcClient) Publish(ctx context.Context, msg Message, opts ...PublishOpt
if !ok { if !ok {
md = make(map[string]string) md = make(map[string]string)
} }
id := uuid.New().String()
md["Content-Type"] = msg.ContentType() md["Content-Type"] = msg.ContentType()
md["X-Micro-Topic"] = msg.Topic()
md["X-Micro-Id"] = id
// encode message body // encode message body
cf, err := r.newCodec(msg.ContentType()) cf, err := r.newCodec(msg.ContentType())
@@ -452,7 +457,13 @@ func (r *rpcClient) Publish(ctx context.Context, msg Message, opts ...PublishOpt
return errors.InternalServerError("go.micro.client", err.Error()) return errors.InternalServerError("go.micro.client", err.Error())
} }
b := &buffer{bytes.NewBuffer(nil)} b := &buffer{bytes.NewBuffer(nil)}
if err := cf(b).Write(&codec.Message{Type: codec.Publication}, msg.Payload()); err != nil { if err := cf(b).Write(&codec.Message{
Type: codec.Publication,
Header: map[string]string{
"X-Micro-Id": id,
"X-Micro-Topic": msg.Topic(),
},
}, msg.Payload()); err != nil {
return errors.InternalServerError("go.micro.client", err.Error()) return errors.InternalServerError("go.micro.client", err.Error())
} }
r.once.Do(func() { r.once.Do(func() {