metadata: add default headers

Signed-off-by: Vasiliy Tolstov <v.tolstov@unistack.org>
This commit is contained in:
Василий Толстов 2021-07-23 12:03:18 +03:00
parent d357fb1e0d
commit 675e121049
6 changed files with 33 additions and 13 deletions

View File

@ -112,6 +112,13 @@ type Message struct {
Body RawMessage Body RawMessage
} }
// NewMessage create broker message with topic filled
func NewMessage(topic string) *Message {
m := &Message{Header: metadata.New(2)}
m.Header.Set(metadata.HeaderTopic, topic)
return m
}
// Subscriber is a convenience return type for the Subscribe method // Subscriber is a convenience return type for the Subscribe method
type Subscriber interface { type Subscriber interface {
// Options returns subscriber options // Options returns subscriber options

View File

@ -6,6 +6,7 @@ import (
"github.com/google/uuid" "github.com/google/uuid"
"github.com/unistack-org/micro/v3/logger" "github.com/unistack-org/micro/v3/logger"
"github.com/unistack-org/micro/v3/metadata"
maddr "github.com/unistack-org/micro/v3/util/addr" maddr "github.com/unistack-org/micro/v3/util/addr"
mnet "github.com/unistack-org/micro/v3/util/net" mnet "github.com/unistack-org/micro/v3/util/net"
"github.com/unistack-org/micro/v3/util/rand" "github.com/unistack-org/micro/v3/util/rand"
@ -113,14 +114,14 @@ func (m *memoryBroker) BatchPublish(ctx context.Context, msgs []*Message, opts .
if m.opts.Codec == nil { if m.opts.Codec == nil {
m.RLock() m.RLock()
for _, msg := range msgs { for _, msg := range msgs {
topic, _ := msg.Header.Get("Micro-Topic") topic, _ := msg.Header.Get(metadata.HeaderTopic)
vs = append(vs, msgWrapper{topic: topic, body: m}) vs = append(vs, msgWrapper{topic: topic, body: m})
} }
m.RUnlock() m.RUnlock()
} else { } else {
m.RLock() m.RLock()
for _, msg := range msgs { for _, msg := range msgs {
topic, _ := msg.Header.Get("Micro-Topic") topic, _ := msg.Header.Get(metadata.HeaderTopic)
buf, err := m.opts.Codec.Marshal(msg) buf, err := m.opts.Codec.Marshal(msg)
if err != nil { if err != nil {
m.RUnlock() m.RUnlock()

View File

@ -4,6 +4,8 @@ import (
"context" "context"
"fmt" "fmt"
"testing" "testing"
"github.com/unistack-org/micro/v3/metadata"
) )
func TestMemoryBatchBroker(t *testing.T) { func TestMemoryBatchBroker(t *testing.T) {
@ -30,9 +32,9 @@ func TestMemoryBatchBroker(t *testing.T) {
for i := 0; i < count; i++ { for i := 0; i < count; i++ {
message := &Message{ message := &Message{
Header: map[string]string{ Header: map[string]string{
"Micro-Topic": topic, metadata.HeaderTopic: topic,
"foo": "bar", "foo": "bar",
"id": fmt.Sprintf("%d", i), "id": fmt.Sprintf("%d", i),
}, },
Body: []byte(`"hello world"`), Body: []byte(`"hello world"`),
} }
@ -75,9 +77,9 @@ func TestMemoryBroker(t *testing.T) {
for i := 0; i < count; i++ { for i := 0; i < count; i++ {
message := &Message{ message := &Message{
Header: map[string]string{ Header: map[string]string{
"Micro-Topic": topic, metadata.HeaderTopic: topic,
"foo": "bar", "foo": "bar",
"id": fmt.Sprintf("%d", i), "id": fmt.Sprintf("%d", i),
}, },
Body: []byte(`"hello world"`), Body: []byte(`"hello world"`),
} }

View File

@ -190,8 +190,8 @@ func (n *noopClient) Publish(ctx context.Context, p Message, opts ...PublishOpti
if !ok { if !ok {
md = metadata.New(0) md = metadata.New(0)
} }
md["Content-Type"] = p.ContentType() md[metadata.HeaderContentType] = p.ContentType()
md["Micro-Topic"] = p.Topic() md[metadata.HeaderTopic] = p.Topic()
// passed in raw data // passed in raw data
if d, ok := p.Payload().(*codec.Frame); ok { if d, ok := p.Payload().(*codec.Frame); ok {

View File

@ -6,8 +6,18 @@ import (
"sort" "sort"
) )
// HeaderPrefix for all headers passed var (
var HeaderPrefix = "Micro-" // HeaderTopic is the header name that contains topic name
HeaderTopic = "Micro-Topic"
// HeaderContentType specifies content type of message
HeaderContentType = "Content-Type"
// HeaderEndpoint specifies endpoint in service
HeaderEndpoint = "Micro-Endpoint"
// HeaderService specifies service
HeaderService = "Micro-Service"
// HeaderTimeout specifies timeout of operation
HeaderTimeout = "Micro-Timeout"
)
// Metadata is our way of representing request headers internally. // Metadata is our way of representing request headers internally.
// They're used at the RPC level and translate back and forth // They're used at the RPC level and translate back and forth

View File

@ -78,7 +78,7 @@ func (t *tunBroker) BatchPublish(ctx context.Context, msgs []*broker.Message, op
var err error var err error
for _, msg := range msgs { for _, msg := range msgs {
topic, _ := msg.Header.Get("Micro-Topic") topic, _ := msg.Header.Get(metadata.HeaderTopic)
c, ok := topicMap[topic] c, ok := topicMap[topic]
if !ok { if !ok {
c, err := t.tunnel.Dial(ctx, topic, tunnel.DialMode(tunnel.Multicast)) c, err := t.tunnel.Dial(ctx, topic, tunnel.DialMode(tunnel.Multicast))