Merge branch 'main' into dapr-state-store-clickhouse

middt · web-flow · commit 2292ac1a1ec0 · 2024-11-20T20:45:40.000+03:00
diff --git a/bindings/aws/dynamodb/dynamodb.go b/bindings/aws/dynamodb/dynamodb.go
@@ -122,5 +122,8 @@ func (d *DynamoDB) GetComponentMetadata() (metadataInfo metadata.MetadataMap) {
 }
 
 func (d *DynamoDB) Close() error {
-	return d.authProvider.Close()
+	if d.authProvider != nil {
+		return d.authProvider.Close()
+	}
+	return nil
 }
diff --git a/bindings/aws/kinesis/kinesis.go b/bindings/aws/kinesis/kinesis.go
@@ -266,7 +266,10 @@ func (a *AWSKinesis) Close() error {
 		close(a.closeCh)
 	}
 	a.wg.Wait()
-	return a.authProvider.Close()
+	if a.authProvider != nil {
+		return a.authProvider.Close()
+	}
+	return nil
 }
 
 func (a *AWSKinesis) ensureConsumer(ctx context.Context, streamARN *string) (*string, error) {
diff --git a/bindings/aws/s3/s3.go b/bindings/aws/s3/s3.go
@@ -153,7 +153,10 @@ func (s *AWSS3) Init(ctx context.Context, metadata bindings.Metadata) error {
 }
 
 func (s *AWSS3) Close() error {
-	return s.authProvider.Close()
+	if s.authProvider != nil {
+		return s.authProvider.Close()
+	}
+	return nil
 }
 
 func (s *AWSS3) Operations() []bindings.OperationKind {
diff --git a/bindings/aws/ses/ses.go b/bindings/aws/ses/ses.go
@@ -176,5 +176,8 @@ func (a *AWSSES) GetComponentMetadata() (metadataInfo contribMetadata.MetadataMa
 }
 
 func (a *AWSSES) Close() error {
-	return a.authProvider.Close()
+	if a.authProvider != nil {
+		return a.authProvider.Close()
+	}
+	return nil
 }
diff --git a/bindings/aws/sns/sns.go b/bindings/aws/sns/sns.go
@@ -128,5 +128,8 @@ func (a *AWSSNS) GetComponentMetadata() (metadataInfo metadata.MetadataMap) {
 }
 
 func (a *AWSSNS) Close() error {
-	return a.authProvider.Close()
+	if a.authProvider != nil {
+		return a.authProvider.Close()
+	}
+	return nil
 }
diff --git a/bindings/aws/sqs/sqs.go b/bindings/aws/sqs/sqs.go
@@ -173,7 +173,10 @@ func (a *AWSSQS) Close() error {
 		close(a.closeCh)
 	}
 	a.wg.Wait()
-	return a.authProvider.Close()
+	if a.authProvider != nil {
+		return a.authProvider.Close()
+	}
+	return nil
 }
 
 func (a *AWSSQS) parseSQSMetadata(meta bindings.Metadata) (*sqsMetadata, error) {
diff --git a/pubsub/aws/snssqs/snssqs.go b/pubsub/aws/snssqs/snssqs.go
@@ -875,7 +875,10 @@ func (s *snsSqs) Close() error {
 		s.subscriptionManager.Close()
 	}
 
-	return s.authProvider.Close()
+	if s.authProvider != nil {
+		return s.authProvider.Close()
+	}
+	return nil
 }
 
 func (s *snsSqs) Features() []pubsub.Feature {
diff --git a/secretstores/aws/parameterstore/parameterstore.go b/secretstores/aws/parameterstore/parameterstore.go
@@ -182,5 +182,8 @@ func (s *ssmSecretStore) GetComponentMetadata() (metadataInfo metadata.MetadataM
 }
 
 func (s *ssmSecretStore) Close() error {
-	return s.authProvider.Close()
+	if s.authProvider != nil {
+		return s.authProvider.Close()
+	}
+	return nil
 }
diff --git a/secretstores/aws/secretmanager/secretmanager.go b/secretstores/aws/secretmanager/secretmanager.go
@@ -170,5 +170,8 @@ func (s *smSecretStore) GetComponentMetadata() (metadataInfo metadata.MetadataMa
 }
 
 func (s *smSecretStore) Close() error {
-	return s.authProvider.Close()
+	if s.authProvider != nil {
+		return s.authProvider.Close()
+	}
+	return nil
 }
diff --git a/state/aws/dynamodb/dynamodb.go b/state/aws/dynamodb/dynamodb.go
@@ -275,7 +275,10 @@ func (d *StateStore) GetComponentMetadata() (metadataInfo metadata.MetadataMap)
 }
 
 func (d *StateStore) Close() error {
-	return d.authProvider.Close()
+	if d.authProvider != nil {
+		return d.authProvider.Close()
+	}
+	return nil
 }
 
 func (d *StateStore) getDynamoDBMetadata(meta state.Metadata) (*dynamoDBMetadata, error) {
diff --git a/state/mongodb/mongodb.go b/state/mongodb/mongodb.go
@@ -25,6 +25,8 @@ import (
 	"strings"
 	"time"
 
+	"github.com/dapr/components-contrib/contenttype"
+
 	"github.com/google/uuid"
 	"go.mongodb.org/mongo-driver/bson"
 	"go.mongodb.org/mongo-driver/bson/bsonrw"
@@ -528,7 +530,18 @@ func (m *MongoDB) doTransaction(sessCtx mongo.SessionContext, operations []state
 		var err error
 		switch req := o.(type) {
 		case state.SetRequest:
-			err = m.setInternal(sessCtx, &req)
+			{
+				isJSON := (len(req.Metadata) > 0 && req.Metadata[metadata.ContentType] == contenttype.JSONContentType)
+				if isJSON {
+					if bytes, ok := req.Value.([]byte); ok {
+						err = json.Unmarshal(bytes, &req.Value)
+						if err != nil {
+							break
+						}
+					}
+				}
+				err = m.setInternal(sessCtx, &req)
+			}
 		case state.DeleteRequest:
 			err = m.deleteInternal(sessCtx, &req)
 		}
diff --git a/tests/conformance/state/state.go b/tests/conformance/state/state.go
@@ -15,6 +15,7 @@ package state
 
 import (
 	"context"
+	"encoding/base64"
 	"encoding/json"
 	"fmt"
 	"slices"
@@ -784,6 +785,70 @@ func ConformanceTests(t *testing.T, props map[string]string, statestore state.St
 				assert.Equal(t, v, res.Data)
 			}
 		})
+
+		t.Run("transaction-serialization-grpc-json", func(t *testing.T) {
+			features := statestore.Features()
+			// this check for exclude redis 7
+			if state.FeatureQueryAPI.IsPresent(features) {
+				json := "{\"id\":1223,\"name\":\"test\"}"
+				keyTest1 := key + "-key-grpc"
+				valueTest := []byte(json)
+				keyTest2 := key + "-key-grpc-no-json"
+
+				metadataTest1 := map[string]string{
+					"contentType": "application/json",
+				}
+
+				operations := []state.TransactionalStateOperation{
+					state.SetRequest{
+						Key:      keyTest1,
+						Value:    valueTest,
+						Metadata: metadataTest1,
+					},
+					state.SetRequest{
+						Key:   keyTest2,
+						Value: valueTest,
+					},
+				}
+
+				expected := map[string][]byte{
+					keyTest1: []byte(json),
+					keyTest2: []byte(json),
+				}
+
+				expectedMetadata := map[string]map[string]string{
+					keyTest1: metadataTest1,
+				}
+
+				// Act
+				transactionStore, ok := statestore.(state.TransactionalStore)
+				assert.True(t, ok)
+				err := transactionStore.Multi(context.Background(), &state.TransactionalStateRequest{
+					Operations: operations,
+				})
+				require.NoError(t, err)
+
+				// Assert
+				for k, v := range expected {
+					res, err := statestore.Get(context.Background(), &state.GetRequest{
+						Key:      k,
+						Metadata: expectedMetadata[k],
+					})
+					expectedValue := res.Data
+
+					// In redisjson when set the value with contentType = application/Json store the value in base64
+					if strings.HasPrefix(string(expectedValue), "\"ey") {
+						valueBase64 := strings.Trim(string(expectedValue), "\"")
+						expectedValueDecoded, _ := base64.StdEncoding.DecodeString(valueBase64)
+						require.NoError(t, err)
+						assert.Equal(t, expectedValueDecoded, v)
+					} else {
+						require.NoError(t, err)
+						assert.Equal(t, expectedValue, v)
+					}
+				}
+			}
+		})
 	} else {
 		t.Run("component does not implement TransactionalStore interface", func(t *testing.T) {
 			_, ok := statestore.(state.TransactionalStore)

Original file line number	Diff line number	Diff line change
`@@ -122,5 +122,8 @@ func (d *DynamoDB) GetComponentMetadata() (metadataInfo metadata.MetadataMap) {`
`122`	`122`	`}`
`123`	`123`
`124`	`124`	`func (d *DynamoDB) Close() error {`
`125`		`- return d.authProvider.Close()`
	`125`	`+ if d.authProvider != nil {`
	`126`	`+ return d.authProvider.Close()`
	`127`	`+ }`
	`128`	`+ return nil`
`126`	`129`	`}`
Original file line number	Diff line number	Diff line change
`@@ -266,7 +266,10 @@ func (a *AWSKinesis) Close() error {`
`266`	`266`	`close(a.closeCh)`
`267`	`267`	`}`
`268`	`268`	`a.wg.Wait()`
`269`		`- return a.authProvider.Close()`
	`269`	`+ if a.authProvider != nil {`
	`270`	`+ return a.authProvider.Close()`
	`271`	`+ }`
	`272`	`+ return nil`
`270`	`273`	`}`
`271`	`274`
`272`	`275`	`func (a AWSKinesis) ensureConsumer(ctx context.Context, streamARN string) (*string, error) {`
Original file line number	Diff line number	Diff line change
`@@ -153,7 +153,10 @@ func (s *AWSS3) Init(ctx context.Context, metadata bindings.Metadata) error {`
`153`	`153`	`}`
`154`	`154`
`155`	`155`	`func (s *AWSS3) Close() error {`
`156`		`- return s.authProvider.Close()`
	`156`	`+ if s.authProvider != nil {`
	`157`	`+ return s.authProvider.Close()`
	`158`	`+ }`
	`159`	`+ return nil`
`157`	`160`	`}`
`158`	`161`
`159`	`162`	`func (s *AWSS3) Operations() []bindings.OperationKind {`
Original file line number	Diff line number	Diff line change
`@@ -176,5 +176,8 @@ func (a *AWSSES) GetComponentMetadata() (metadataInfo contribMetadata.MetadataMa`
`176`	`176`	`}`
`177`	`177`
`178`	`178`	`func (a *AWSSES) Close() error {`
`179`		`- return a.authProvider.Close()`
	`179`	`+ if a.authProvider != nil {`
	`180`	`+ return a.authProvider.Close()`
	`181`	`+ }`
	`182`	`+ return nil`
`180`	`183`	`}`
Original file line number	Diff line number	Diff line change
`@@ -128,5 +128,8 @@ func (a *AWSSNS) GetComponentMetadata() (metadataInfo metadata.MetadataMap) {`
`128`	`128`	`}`
`129`	`129`
`130`	`130`	`func (a *AWSSNS) Close() error {`
`131`		`- return a.authProvider.Close()`
	`131`	`+ if a.authProvider != nil {`
	`132`	`+ return a.authProvider.Close()`
	`133`	`+ }`
	`134`	`+ return nil`
`132`	`135`	`}`
Original file line number	Diff line number	Diff line change
`@@ -173,7 +173,10 @@ func (a *AWSSQS) Close() error {`
`173`	`173`	`close(a.closeCh)`
`174`	`174`	`}`
`175`	`175`	`a.wg.Wait()`
`176`		`- return a.authProvider.Close()`
	`176`	`+ if a.authProvider != nil {`
	`177`	`+ return a.authProvider.Close()`
	`178`	`+ }`
	`179`	`+ return nil`
`177`	`180`	`}`
`178`	`181`
`179`	`182`	`func (a AWSSQS) parseSQSMetadata(meta bindings.Metadata) (sqsMetadata, error) {`
Original file line number	Diff line number	Diff line change
`@@ -875,7 +875,10 @@ func (s *snsSqs) Close() error {`
`875`	`875`	`s.subscriptionManager.Close()`
`876`	`876`	`}`
`877`	`877`
`878`		`- return s.authProvider.Close()`
	`878`	`+ if s.authProvider != nil {`
	`879`	`+ return s.authProvider.Close()`
	`880`	`+ }`
	`881`	`+ return nil`
`879`	`882`	`}`
`880`	`883`
`881`	`884`	`func (s *snsSqs) Features() []pubsub.Feature {`
Original file line number	Diff line number	Diff line change
`@@ -182,5 +182,8 @@ func (s *ssmSecretStore) GetComponentMetadata() (metadataInfo metadata.MetadataM`
`182`	`182`	`}`
`183`	`183`
`184`	`184`	`func (s *ssmSecretStore) Close() error {`
`185`		`- return s.authProvider.Close()`
	`185`	`+ if s.authProvider != nil {`
	`186`	`+ return s.authProvider.Close()`
	`187`	`+ }`
	`188`	`+ return nil`
`186`	`189`	`}`
Original file line number	Diff line number	Diff line change
`@@ -170,5 +170,8 @@ func (s *smSecretStore) GetComponentMetadata() (metadataInfo metadata.MetadataMa`
`170`	`170`	`}`
`171`	`171`
`172`	`172`	`func (s *smSecretStore) Close() error {`
`173`		`- return s.authProvider.Close()`
	`173`	`+ if s.authProvider != nil {`
	`174`	`+ return s.authProvider.Close()`
	`175`	`+ }`
	`176`	`+ return nil`
`174`	`177`	`}`
Original file line number	Diff line number	Diff line change
`@@ -275,7 +275,10 @@ func (d *StateStore) GetComponentMetadata() (metadataInfo metadata.MetadataMap)`
`275`	`275`	`}`
`276`	`276`
`277`	`277`	`func (d *StateStore) Close() error {`
`278`		`- return d.authProvider.Close()`
	`278`	`+ if d.authProvider != nil {`
	`279`	`+ return d.authProvider.Close()`
	`280`	`+ }`
	`281`	`+ return nil`
`279`	`282`	`}`
`280`	`283`
`281`	`284`	`func (d StateStore) getDynamoDBMetadata(meta state.Metadata) (dynamoDBMetadata, error) {`