Supporting more properties for rabbitmq (#3806)

Signed-off-by: Andre Bossard <[email protected]>
Signed-off-by: Fabian Martinez <[email protected]>
Co-authored-by: Andre Bossard <[email protected]>
Co-authored-by: Fabian Martinez <[email protected]>
Co-authored-by: Claude <[email protected]>
Co-authored-by: Dapr Bot <[email protected]>
Co-authored-by: Yaron Schneider <[email protected]>

Author: 5 people

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)
@@ -473,9 +471,9 @@ func (r *RabbitMQ) handleMessage(ctx context.Context, handler bindings.Handler,
 			// Passthrough any custom metadata to the handler.
 			for k, v := range d.Headers {
 				if s, ok := v.(string); ok {
-					// Escape the key and value to ensure they are valid URL query parameters.
+					// Escape the key to ensure they are valid URL query parameters.
 					// This is necessary for them to be sent as HTTP Metadata.
-					metadata[url.QueryEscape(k)] = url.QueryEscape(s)
+					metadata[url.QueryEscape(k)] = s
 				}
 			}
 

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,52 @@
+package rabbitmq
+
+import (
+	"strings"
+
+	amqp "github.com/rabbitmq/amqp091-go"
+)
+
+const (
+	MetadataKeyMessageID     = "messageID"
+	MetadataKeyCorrelationID = "correlationID"
+	MetadataKeyContentType   = "contentType"
+	MetadataKeyType          = "type"
+	MetadataKeyPriority      = "priority"
+	MetadataKeyTTL           = "ttl"
+)
+
+// TryGetProperty finds a property value using case-insensitive matching
+func TryGetProperty(props map[string]string, key string) (string, bool) {
+	// First try exact match
+	if val, ok := props[key]; ok && val != "" {
+		return val, true
+	}
+
+	// Then try case-insensitive match
+	for k, v := range props {
+		if v != "" && strings.EqualFold(key, k) {
+			return v, true
+		}
+	}
+
+	return "", false
+}
+
+// ApplyMetadataToPublishing applies common metadata fields to an AMQP publishing
+func ApplyMetadataToPublishing(metadata map[string]string, publishing *amqp.Publishing) {
+	if contentType, ok := TryGetProperty(metadata, MetadataKeyContentType); ok {
+		publishing.ContentType = contentType
+	}
+
+	if messageID, ok := TryGetProperty(metadata, MetadataKeyMessageID); ok {
+		publishing.MessageId = messageID
+	}
+
+	if correlationID, ok := TryGetProperty(metadata, MetadataKeyCorrelationID); ok {
+		publishing.CorrelationId = correlationID
+	}
+
+	if aType, ok := TryGetProperty(metadata, MetadataKeyType); ok {
+		publishing.Type = aType
+	}
+}

common/component/rabbitmq/rabbitmq_test.go

@@ -0,0 +1,68 @@
+package rabbitmq
+
+import (
+	"testing"
+
+	"github.com/stretchr/testify/assert"
+)
+
+func TestTryGetProperty(t *testing.T) {
+	tests := []struct {
+		name     string
+		props    map[string]string
+		key      string
+		expected string
+		found    bool
+	}{
+		{
+			name:     "exact match",
+			props:    map[string]string{"messageID": "test-id"},
+			key:      "messageID",
+			expected: "test-id",
+			found:    true,
+		},
+		{
+			name:     "case insensitive match",
+			props:    map[string]string{"messageid": "test-id"},
+			key:      "messageID",
+			expected: "test-id",
+			found:    true,
+		},
+		{
+			name:     "uppercase match",
+			props:    map[string]string{"MESSAGEID": "test-id"},
+			key:      "messageID",
+			expected: "test-id",
+			found:    true,
+		},
+		{
+			name:     "not found",
+			props:    map[string]string{"otherKey": "value"},
+			key:      "messageID",
+			expected: "",
+			found:    false,
+		},
+		{
+			name:     "empty value",
+			props:    map[string]string{"messageID": ""},
+			key:      "messageID",
+			expected: "",
+			found:    false,
+		},
+		{
+			name:     "whitespace value",
+			props:    map[string]string{"messageID": "   "},
+			key:      "messageID",
+			expected: "   ",
+			found:    true,
+		},
+	}
+
+	for _, tt := range tests {
+		t.Run(tt.name, func(t *testing.T) {
+			value, found := TryGetProperty(tt.props, tt.key)
+			assert.Equal(t, tt.expected, value)
+			assert.Equal(t, tt.found, found)
+		})
+	}
+}

conversation/deepseek/metadata.yaml

@@ -7,7 +7,7 @@ status: alpha
 title: "Deepseek"
 urls:
   - title: Reference
-    url: https://docs.dapr.io/reference/components-reference/supported-conversation/deepseek/
+    url: https://docs.dapr.io/reference/components-reference/supported-conversation/setup-deepseek/
 authenticationProfiles:
   - title: "API Key"
     description: "Authenticate using an API key"

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

metadata/utils_test.go

@@ -175,34 +175,6 @@ func TestIsRawPayload(t *testing.T) {
 	})
 }
 
-func TestTryGetContentType(t *testing.T) {
-	t.Run("Metadata without content type", func(t *testing.T) {
-		val, ok := TryGetContentType(map[string]string{})
-
-		assert.Equal(t, "", val)
-		assert.False(t, ok)
-	})
-
-	t.Run("Metadata with empty content type", func(t *testing.T) {
-		val, ok := TryGetContentType(map[string]string{
-			"contentType": "",
-		})
-
-		assert.Equal(t, "", val)
-		assert.False(t, ok)
-	})
-
-	t.Run("Metadata with corrent content type", func(t *testing.T) {
-		const contentType = "application/cloudevent+json"
-		val, ok := TryGetContentType(map[string]string{
-			"contentType": contentType,
-		})
-
-		assert.Equal(t, contentType, val)
-		assert.True(t, ok)
-	})
-}
-
 func TestMetadataStructToStringMap(t *testing.T) {
 	t.Run("Test metadata struct to metadata info conversion", func(t *testing.T) {
 		type NestedStruct struct {