set the correct Type and Content-Type headers on out going messages (#160)

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

* refactoring

Author: Guy Baron

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

@@ -15,6 +15,8 @@ import (
 
 var _ gbus.Serializer = &Proto{}
 
+const ProtoContentType = "application/x-protobuf"
+
 //Proto a serializer for GBus uses protobuf
 type Proto struct {
 	lock              *sync.Mutex
@@ -33,7 +35,7 @@ func NewProtoSerializer(logger logrus.FieldLogger) gbus.Serializer {
 
 //Name implements Serializer.Name
 func (as *Proto) Name() string {
-	return "proto"
+	return ProtoContentType
 }
 
 //Encode encodes an object into a byte array

tests/bus_test.go

@@ -12,6 +12,7 @@ import (
 	"github.com/wework/grabbit/gbus/serialization"
 
 	"github.com/wework/grabbit/gbus/metrics"
+	"github.com/wework/grabbit/gbus/policy"
 
 	"github.com/opentracing/opentracing-go"
 	olog "github.com/opentracing/opentracing-go/log"
@@ -621,6 +622,27 @@ func TestOnlyRawMessageHandlersInvoked(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 TestSendEmptyBody(t *testing.T) {
 	/*
 			test sending of message with len(payload) == 0 .

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 != serialization.ProtoContentType {
+		t.Fatalf("incorrect serializer name. expected:application/x-protobuf actual:%s", name)
 	}
 	cmd := getProtoCommand()
 	schemaName := cmd.SchemaName()