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

bindings/rabbitmq/rabbitmq.go

@@ -31,6 +31,7 @@ import (
 	amqp "github.com/rabbitmq/amqp091-go"
 
 	"github.com/dapr/components-contrib/bindings"
+	common "github.com/dapr/components-contrib/common/component/rabbitmq"
 	"github.com/dapr/components-contrib/metadata"
 	"github.com/dapr/kit/logger"
 	kitmd "github.com/dapr/kit/metadata"
@@ -228,11 +229,6 @@ func (r *RabbitMQ) Invoke(ctx context.Context, req *bindings.InvokeRequest) (*bi
 		pub.Headers[k] = v
 	}
 
-	contentType, ok := metadata.TryGetContentType(req.Metadata)
-	if ok {
-		pub.ContentType = contentType
-	}
-
 	// The default time to live has been set in the queue
 	// We allow overriding on each call, by setting a value in request metadata
 	ttl, ok, err := metadata.TryGetTTL(req.Metadata)
@@ -252,6 +248,8 @@ func (r *RabbitMQ) Invoke(ctx context.Context, req *bindings.InvokeRequest) (*bi
 		pub.Priority = priority
 	}
 
+	common.ApplyMetadataToPublishing(req.Metadata, &pub)
+
 	err = ch.PublishWithContext(ctx, "", r.metadata.QueueName, false, false, pub)
 	if err != nil {
 		return nil, fmt.Errorf("failed to publish message: %w", err)

bindings/rabbitmq/rabbitmq_integration_test.go

@@ -447,3 +447,67 @@ func TestPublishWithHeaders(t *testing.T) {
 	// assert.Contains(t, msg.Header, "custom_header1")
 	// assert.Contains(t, msg.Header, "custom_header2")
 }
+
+func TestPublishMetadataProperties(t *testing.T) {
+	rabbitmqHost := getTestRabbitMQHost()
+	require.NotEmpty(t, rabbitmqHost, fmt.Sprintf("RabbitMQ host configuration must be set in environment variable '%s'", testRabbitMQHostEnvKey))
+
+	queueName := uuid.New().String()
+	durable := true
+	exclusive := false
+
+	metadata := bindings.Metadata{
+		Base: contribMetadata.Base{
+			Name: "testQueue",
+			Properties: map[string]string{
+				"queueName":        queueName,
+				"host":             rabbitmqHost,
+				"deleteWhenUnused": strconv.FormatBool(exclusive),
+				"durable":          strconv.FormatBool(durable),
+			},
+		},
+	}
+
+	logger := logger.NewLogger("test")
+	r := NewRabbitMQ(logger).(*RabbitMQ)
+	err := r.Init(t.Context(), metadata)
+	require.NoError(t, err)
+
+	conn, err := amqp.Dial(rabbitmqHost)
+	require.NoError(t, err)
+	defer conn.Close()
+
+	ch, err := conn.Channel()
+	require.NoError(t, err)
+	defer ch.Close()
+
+	const messageData = "test message"
+	const msgID = "msg-123"
+	const corrID = "corr-456"
+	const msgType = "testType"
+	const contentType = "application/json"
+
+	writeRequest := bindings.InvokeRequest{
+		Data: []byte(messageData),
+		Metadata: map[string]string{
+			"messageID":     msgID,
+			"correlationID": corrID,
+			"type":          msgType,
+			"contentType":   contentType,
+		},
+	}
+	_, err = r.Invoke(t.Context(), &writeRequest)
+	require.NoError(t, err)
+
+	// Retrieve the message.
+	msg, ok, err := getMessageWithRetries(ch, queueName, 2*time.Second)
+	require.NoError(t, err)
+	assert.True(t, ok)
+	assert.Equal(t, messageData, string(msg.Body))
+	assert.Equal(t, msgID, msg.MessageId)
+	assert.Equal(t, corrID, msg.CorrelationId)
+	assert.Equal(t, msgType, msg.Type)
+	assert.Equal(t, contentType, msg.ContentType)
+
+	require.NoError(t, r.Close())
+}

common/component/rabbitmq/rabbitmq.go

@@ -0,0 +1,70 @@
+package rabbitmq
+
+import (
+	"strings"
+
+	amqp "github.com/rabbitmq/amqp091-go"
+)
+
+// tryGetProperty checks for a property value using various key formats: original, camelCase, and case-insensitive
+func tryGetProperty(props map[string]string, keys ...string) (string, bool) {
+	// First try exact match for all provided keys
+	for _, key := range keys {
+		if val, ok := props[key]; ok && val != "" {
+			return val, true
+		}
+	}
+
+	// Then try case-insensitive match if no exact matches were found
+	for k, v := range props {
+		if v != "" {
+			lowerK := strings.ToLower(k)
+			for _, key := range keys {
+				if strings.ToLower(key) == lowerK {
+					return v, true
+				}
+			}
+		}
+	}
+
+	return "", false
+}
+
+func TryGetMessageID(props map[string]string) (string, bool) {
+	return tryGetProperty(props, "messageId", "messageID", "MessageId", "MessageID")
+}
+
+func TryGetCorrelationID(props map[string]string) (string, bool) {
+	return tryGetProperty(props, "correlationId", "correlationID", "CorrelationId", "CorrelationID")
+}
+
+func TryGetContentType(props map[string]string) (string, bool) {
+	return tryGetProperty(props, "contentType", "ContentType")
+}
+
+func TryGetType(props map[string]string) (string, bool) {
+	return tryGetProperty(props, "type", "Type")
+}
+
+// ApplyMetadataToPublishing applies common metadata fields to an AMQP publishing
+func ApplyMetadataToPublishing(metadata map[string]string, publishing *amqp.Publishing) {
+	contentType, ok := TryGetContentType(metadata)
+	if ok {
+		publishing.ContentType = contentType
+	}
+
+	messageID, ok := TryGetMessageID(metadata)
+	if ok {
+		publishing.MessageId = messageID
+	}
+
+	correlationID, ok := TryGetCorrelationID(metadata)
+	if ok {
+		publishing.CorrelationId = correlationID
+	}
+
+	aType, ok := TryGetType(metadata)
+	if ok {
+		publishing.Type = aType
+	}
+}

metadata/utils.go

@@ -114,14 +114,6 @@ func IsRawPayload(props map[string]string) (bool, error) {
 	return false, nil
 }
 
-func TryGetContentType(props map[string]string) (string, bool) {
-	if val, ok := props[ContentType]; 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

@@ -28,6 +28,7 @@ import (
 
 	amqp "github.com/rabbitmq/amqp091-go"
 
+	common "github.com/dapr/components-contrib/common/component/rabbitmq"
 	"github.com/dapr/components-contrib/metadata"
 	"github.com/dapr/components-contrib/pubsub"
 	"github.com/dapr/kit/logger"
@@ -259,6 +260,8 @@ func (r *rabbitMQ) publishSync(ctx context.Context, req *pubsub.PublishRequest)
 		p.Priority = priority
 	}
 
+	common.ApplyMetadataToPublishing(req.Metadata, &p)
+
 	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)

pubsub/rabbitmq/rabbitmq_test.go

@@ -461,10 +461,11 @@ func createAMQPMessage(body []byte) amqp.Delivery {
 }
 
 type rabbitMQInMemoryBroker struct {
-	buffer         chan amqp.Delivery
-	declaredQueues []string
-	connectCount   atomic.Int32
-	closeCount     atomic.Int32
+	buffer          chan amqp.Delivery
+	declaredQueues  []string
+	connectCount    atomic.Int32
+	closeCount      atomic.Int32
+	lastMsgMetadata *amqp.Publishing // Add this field to capture the last message metadata
 }
 
 func (r *rabbitMQInMemoryBroker) Qos(prefetchCount, prefetchSize int, global bool) error {
@@ -482,7 +483,17 @@ func (r *rabbitMQInMemoryBroker) PublishWithDeferredConfirmWithContext(ctx conte
 		return nil, errors.New(errorChannelConnection)
 	}
 
-	r.buffer <- createAMQPMessage(msg.Body)
+	// Store the last message metadata for inspection in tests
+	r.lastMsgMetadata = &msg
+
+	// Use a non-blocking send or a separate goroutine to prevent deadlock
+	// when there's no consumer reading from the buffer
+	select {
+	case r.buffer <- createAMQPMessage(msg.Body):
+		// Message sent successfully
+	default:
+		// Buffer is full or there's no consumer, but we don't want to block
+	}
 
 	return nil, nil
 }
@@ -525,3 +536,76 @@ func (r *rabbitMQInMemoryBroker) Close() error {
 func (r *rabbitMQInMemoryBroker) IsClosed() bool {
 	return r.connectCount.Load() <= r.closeCount.Load()
 }
+
+// TestPublishMetadataProperties tests that message metadata properties are correctly passed to the broker
+func TestPublishMetadataProperties(t *testing.T) {
+	broker := newBroker()
+	pubsubRabbitMQ := newRabbitMQTest(broker)
+	metadata := pubsub.Metadata{Base: mdata.Base{
+		Properties: map[string]string{
+			metadataHostnameKey:   "anyhost",
+			metadataConsumerIDKey: "consumer",
+		},
+	}}
+	err := pubsubRabbitMQ.Init(t.Context(), metadata)
+	require.NoError(t, err)
+
+	topic := "metadatatest"
+
+	// Create a consumer for the test to prevent channel deadlock
+	messageHandler := func(ctx context.Context, msg *pubsub.NewMessage) error {
+		return nil
+	}
+	err = pubsubRabbitMQ.Subscribe(t.Context(), pubsub.SubscribeRequest{Topic: topic}, messageHandler)
+	require.NoError(t, err)
+
+	// Test messageID
+	err = pubsubRabbitMQ.Publish(t.Context(), &pubsub.PublishRequest{
+		Topic: topic,
+		Data:  []byte("test message"),
+		Metadata: map[string]string{
+			"messageID": "msg-123",
+		},
+	})
+	require.NoError(t, err)
+	assert.Equal(t, "msg-123", broker.lastMsgMetadata.MessageId)
+
+	// Test correlationID
+	err = pubsubRabbitMQ.Publish(t.Context(), &pubsub.PublishRequest{
+		Topic: topic,
+		Data:  []byte("test message"),
+		Metadata: map[string]string{
+			"correlationID": "corr-456",
+		},
+	})
+	require.NoError(t, err)
+	assert.Equal(t, "corr-456", broker.lastMsgMetadata.CorrelationId)
+
+	// Test Type
+	err = pubsubRabbitMQ.Publish(t.Context(), &pubsub.PublishRequest{
+		Topic: topic,
+		Data:  []byte("test message"),
+		Metadata: map[string]string{
+			"type": "mytype",
+		},
+	})
+	require.NoError(t, err)
+	assert.Equal(t, "mytype", broker.lastMsgMetadata.Type)
+
+	// Test all properties together
+	err = pubsubRabbitMQ.Publish(t.Context(), &pubsub.PublishRequest{
+		Topic: topic,
+		Data:  []byte("test message"),
+		Metadata: map[string]string{
+			"messageID":     "msg-789",
+			"correlationID": "corr-789",
+			"type":          "complete-type",
+			"contentType":   "application/json",
+		},
+	})
+	require.NoError(t, err)
+	assert.Equal(t, "msg-789", broker.lastMsgMetadata.MessageId)
+	assert.Equal(t, "corr-789", broker.lastMsgMetadata.CorrelationId)
+	assert.Equal(t, "complete-type", broker.lastMsgMetadata.Type)
+	assert.Equal(t, "application/json", broker.lastMsgMetadata.ContentType)
+}