abossard
diff --git a/‎bindings/rabbitmq/rabbitmq.go
Lines changed: 4 additions & 21 deletions b/‎bindings/rabbitmq/rabbitmq.go
Lines changed: 4 additions & 21 deletions
diff --git a/‎bindings/rabbitmq/rabbitmq_integration_test.go
Lines changed: 64 additions & 0 deletions b/‎bindings/rabbitmq/rabbitmq_integration_test.go
Lines changed: 64 additions & 0 deletions
diff --git a/‎common/component/rabbitmq/rabbitmq.go
Lines changed: 69 additions & 0 deletions b/‎common/component/rabbitmq/rabbitmq.go
Lines changed: 69 additions & 0 deletions
diff --git a/‎conversation/deepseek/metadata.yaml
Lines changed: 1 addition & 1 deletion b/‎conversation/deepseek/metadata.yaml
Lines changed: 1 addition & 1 deletion
diff --git a/‎go.mod
Lines changed: 4 additions & 3 deletions b/‎go.mod
Lines changed: 4 additions & 3 deletions
@@ -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,20 +248,7 @@ 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
-	}
+	common.ApplyMetadataToPublishing(req.Metadata, &pub)
 
 	err = ch.PublishWithContext(ctx, "", r.metadata.QueueName, false, false, pub)
 	if err != nil {
@@ -488,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
 				}
 			}
 
 
@@ -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())
+}
@@ -0,0 +1,69 @@
+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 != "" {
+			for _, key := range keys {
+				if strings.EqualFold(key, k) {
+					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
+	}
+}
@@ -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"
 
@@ -126,9 +126,9 @@ require (
 	go.uber.org/multierr v1.11.0
 	go.uber.org/ratelimit v0.3.0
 	golang.org/x/crypto v0.35.0
-	golang.org/x/mod v0.23.0
+	golang.org/x/mod v0.17.0
 	golang.org/x/net v0.36.0
-	golang.org/x/oauth2 v0.27.0
+	golang.org/x/oauth2 v0.21.0
 	google.golang.org/api v0.180.0
 	google.golang.org/grpc v1.66.0
 	google.golang.org/protobuf v1.34.2
@@ -164,6 +164,7 @@ require (
 	github.com/AzureAD/microsoft-authentication-library-for-go v1.2.2 // indirect
 	github.com/Code-Hex/go-generics-cache v1.3.1 // indirect
 	github.com/DataDog/zstd v1.5.2 // indirect
+	github.com/Microsoft/go-winio v0.6.2 // indirect
 	github.com/OneOfOne/xxhash v1.2.8 // indirect
 	github.com/RoaringBitmap/roaring v1.1.0 // indirect
 	github.com/Workiva/go-datastructures v1.0.53 // indirect
@@ -404,7 +405,7 @@ require (
 	golang.org/x/term v0.29.0 // indirect
 	golang.org/x/text v0.22.0 // indirect
 	golang.org/x/time v0.6.0 // indirect
-	golang.org/x/tools v0.30.0 // indirect
+	golang.org/x/tools v0.21.1-0.20240508182429-e35e4ccd0d2d // indirect
 	google.golang.org/genproto v0.0.0-20240401170217-c3f982113cda // indirect
 	google.golang.org/genproto/googleapis/api v0.0.0-20240701130421-f6361c86f094 // indirect
 	google.golang.org/genproto/googleapis/rpc v0.0.0-20240701130421-f6361c86f094 // indirect