set the correct Type and Content-Type headers on out going messages

gbus/bus.go

@@ -655,12 +655,13 @@ func (b *DefaultBus) sendImpl(sctx context.Context, tx *sql.Tx, toService, reply
 	}
 
 	msg := amqp.Publishing{
-		Body:            buffer,
-		ReplyTo:         replyTo,
-		MessageId:       message.ID,
-		CorrelationId:   message.CorrelationID,
-		ContentEncoding: b.Serializer.Name(),
-		Headers:         headers,
+		Type:          message.PayloadFQN,
+		Body:          buffer,
+		ReplyTo:       replyTo,
+		MessageId:     message.ID,
+		CorrelationId: message.CorrelationID,
+		ContentType:   b.Serializer.Name(),
+		Headers:       headers,
 	}
 	span.LogFields(message.GetTraceLog()...)
 

gbus/policy/generic.go

@@ -0,0 +1,11 @@
+package policy
+
+import "github.com/streadway/amqp"
+
+type Generic struct {
+	Funk func(publishing *amqp.Publishing)
+}
+
+func (g *Generic) Apply(publishing *amqp.Publishing) {
+	g.Funk(publishing)
+}

gbus/serialization/avro.go

@@ -62,7 +62,7 @@ func NewAvroSerializer(schemaRegistryUrls ...string) *Avro {
 
 //Name implements Serializer.Name
 func (as *Avro) Name() string {
-	return "avro"
+	return "avro/binary"
 }
 
 //Encode encodes an object into a byte array

gbus/serialization/proto.go

@@ -33,7 +33,7 @@ func NewProtoSerializer(logger logrus.FieldLogger) gbus.Serializer {
 
 //Name implements Serializer.Name
 func (as *Proto) Name() string {
-	return "proto"
+	return "application/x-protobuf"
 }
 
 //Encode encodes an object into a byte array

tests/bus_test.go

@@ -10,6 +10,7 @@ import (
 	"time"
 
 	"github.com/wework/grabbit/gbus/metrics"
+	"github.com/wework/grabbit/gbus/policy"
 
 	"github.com/opentracing/opentracing-go"
 	olog "github.com/opentracing/opentracing-go/log"
@@ -619,6 +620,27 @@ func TestFailHandlerInvokeOfMessageWithEmptyBody(t *testing.T) {
 	}, t)
 }
 
+func TestTypeAndContentTypeHeadersSet(t *testing.T) {
+	cmd := Command1{}
+
+	bus := createNamedBusForTest(testSvc1)
+
+	policy := &policy.Generic{
+		Funk: func(publishing *amqp.Publishing) {
+			if publishing.Type != cmd.SchemaName() {
+				t.Errorf("publishing.Type != cmd.SchemaName()")
+			}
+			dfb := bus.(*gbus.DefaultBus)
+			if publishing.ContentType != dfb.Serializer.Name() {
+				t.Errorf("expected %s as content-type but actual value was %s", dfb.Serializer.Name(), publishing.ContentType)
+			}
+		}}
+
+	bus.Start()
+	defer bus.Shutdown()
+	bus.Send(context.Background(), testSvc1, gbus.NewBusMessage(cmd), policy)
+}
+
 func TestHealthCheck(t *testing.T) {
 	svc1 := createNamedBusForTest(testSvc1)
 	err := svc1.Start()

tests/protoSerialization_test.go

@@ -24,8 +24,8 @@ func TestProtoSerialization(t *testing.T) {
 	serializer := serialization.NewProtoSerializer(logger)
 
 	name := serializer.Name()
-	if name != "proto" {
-		t.Fatalf("incorrect serializer name. expected:proto actual:%s", name)
+	if name != "application/x-protobuf" {
+		t.Fatalf("incorrect serializer name. expected:application/x-protobuf actual:%s", name)
 	}
 	cmd := getProtoCommand()
 	schemaName := cmd.SchemaName()