Add support for messageId, correlationId, and type in RabbitMQ bindings

Signed-off-by: Andre Bossard <[email protected]>

bindings/rabbitmq/rabbitmq.go

@@ -252,6 +252,21 @@ func (r *RabbitMQ) Invoke(ctx context.Context, req *bindings.InvokeRequest) (*bi
 		pub.Priority = priority
 	}
 
+	messageId, ok := metadata.TryGetMessageId(req.Metadata)
+	if ok {
+		pub.MessageId = messageId
+	}
+
+	correlationId, ok := metadata.TryGetCorrelationId(req.Metadata)
+	if ok {
+		pub.CorrelationId = correlationId
+	}
+
+	aType, ok := metadata.TryGetType(req.Metadata)
+	if ok {
+		pub.Type = aType
+	}
+
 	err = ch.PublishWithContext(ctx, "", r.metadata.QueueName, false, false, pub)
 	if err != nil {
 		return nil, fmt.Errorf("failed to publish message: %w", err)

metadata/utils.go

@@ -113,6 +113,21 @@ func IsRawPayload(props map[string]string) (bool, error) {
 
 	return false, nil
 }
+func TryGetMessageId(props map[string]string) (string, bool) {
+	if val, ok := props["messageId"]; ok && val != "" {
+		return val, true
+	}
+
+	return "", false
+}
+
+func TryGetCorrelationId(props map[string]string) (string, bool) {
+	if val, ok := props["correlationId"]; ok && val != "" {
+		return val, true
+	}
+
+	return "", false
+}
 
 func TryGetContentType(props map[string]string) (string, bool) {
 	if val, ok := props[ContentType]; ok && val != "" {
@@ -122,6 +137,14 @@ func TryGetContentType(props map[string]string) (string, bool) {
 	return "", false
 }
 
+func TryGetType(props map[string]string) (string, bool) {
+	if val, ok := props["type"]; ok && val != "" {
+		return val, true
+	}
+
+	return "", false
+}
+
 func TryGetQueryIndexName(props map[string]string) (string, bool) {
 	if val, ok := props[QueryIndexName]; ok && val != "" {
 		return val, true

pubsub/rabbitmq/rabbitmq.go

@@ -259,6 +259,26 @@ func (r *rabbitMQ) publishSync(ctx context.Context, req *pubsub.PublishRequest)
 		p.Priority = priority
 	}
 
+	contentType, ok := metadata.TryGetContentType(req.Metadata)
+	if ok {
+		p.ContentType = contentType
+	}
+
+	messageId, ok := metadata.TryGetMessageId(req.Metadata)
+	if ok {
+		p.MessageId = messageId
+	}
+
+	correlationId, ok := metadata.TryGetCorrelationId(req.Metadata)
+	if ok {
+		p.CorrelationId = correlationId
+	}
+
+	aType, ok := metadata.TryGetType(req.Metadata)
+	if ok {
+		p.Type = aType
+	}
+
 	confirm, err := r.channel.PublishWithDeferredConfirmWithContext(ctx, req.Topic, routingKey, false, false, p)
 	if err != nil {
 		r.logger.Errorf("%s publishing to %s failed in channel.Publish: %v", logMessagePrefix, req.Topic, err)