Refactor and fix contenttype logic for pubsub. (#602)

Co-authored-by: Yaron Schneider <yaronsc@microsoft.com>

contenttype/utils.go

@@ -0,0 +1,39 @@
+// ------------------------------------------------------------
+// Copyright (c) Microsoft Corporation.
+// Licensed under the MIT License.
+// ------------------------------------------------------------
+
+package contenttype
+
+import "strings"
+
+const (
+	// CloudEventContentType is the content type for cloud event.
+	CloudEventContentType = "application/cloudevents+json"
+	// JSONContentType is the content type for JSON.
+	JSONContentType = "application/json"
+)
+
+// IsCloudEventContentType checks for content type.
+func IsCloudEventContentType(contentType string) bool {
+	return isContentType(contentType, CloudEventContentType)
+}
+
+// IsJSONContentType checks for content type.
+func IsJSONContentType(contentType string) bool {
+	return isContentType(contentType, JSONContentType)
+}
+
+func isContentType(contentType string, expected string) bool {
+	lowerContentType := strings.ToLower(contentType)
+	if lowerContentType == expected {
+		return true
+	}
+
+	semiColonPos := strings.Index(lowerContentType, ";")
+	if semiColonPos >= 0 {
+		return lowerContentType[0:semiColonPos] == expected
+	}
+
+	return false
+}

pubsub/envelope.go

@@ -9,6 +9,7 @@ import (
 	"fmt"
 	"time"
 
+	contrib_contenttype "github.com/dapr/components-contrib/contenttype"
 	contrib_metadata "github.com/dapr/components-contrib/metadata"
 	"github.com/google/uuid"
 	jsoniter "github.com/json-iterator/go"
@@ -19,9 +20,6 @@ const (
 	DefaultCloudEventType = "com.dapr.event.sent"
 	// CloudEventsSpecVersion is the specversion used by Dapr for the cloud events implementation
 	CloudEventsSpecVersion = "1.0"
-	// ContentType is the Cloud Events HTTP content type
-	ContentType     = "application/cloudevents+json"
-	JSONContentType = "application/json"
 	// DefaultCloudEventSource is the default event source
 	DefaultCloudEventSource = "Dapr"
 	// DefaultCloudEventDataContentType is the default content-type for the data attribute
@@ -57,11 +55,13 @@ func NewCloudEventsEnvelope(id, source, eventType, subject string, topic string,
 
 	var ceData interface{}
 	var err error
-	if dataContentType == JSONContentType {
+	if contrib_contenttype.IsJSONContentType(dataContentType) {
 		err = jsoniter.Unmarshal(data, &ceData)
+	} else {
+		ceData = string(data)
 	}
 
-	if err != nil || dataContentType != JSONContentType {
+	if err != nil {
 		ceData = string(data)
 	}
 

pubsub/envelope_test.go

@@ -56,15 +56,16 @@ func TestCreateFromJSON(t *testing.T) {
 		envelope := NewCloudEventsEnvelope("a", "source", "", "", "", "mypubsub", "application/json", data, "1")
 		t.Logf("data: %v", envelope[DataField])
 		assert.Equal(t, "application/json", envelope[DataContentTypeField])
+		assert.Equal(t, map[string]interface{}{"Val1": "test", "Val2": float64(1)}, envelope[DataField])
+	})
 
-		obj2 := struct {
-			Val1 string
-			Val2 int
-		}{}
-		err := json.Unmarshal(data, &obj2)
-		assert.NoError(t, err)
-		assert.Equal(t, obj1.Val1, obj2.Val1)
-		assert.Equal(t, obj1.Val2, obj2.Val2)
+	t.Run("has JSON string with rich contenttype", func(t *testing.T) {
+		obj1 := "message"
+		data, _ := json.Marshal(obj1)
+		envelope := NewCloudEventsEnvelope("a", "source", "", "", "", "mypubsub", "application/JSON; charset=utf-8", data, "1")
+		t.Logf("data: %v", envelope[DataField])
+		assert.Equal(t, "application/JSON; charset=utf-8", envelope[DataContentTypeField])
+		assert.Equal(t, "message", envelope[DataField])
 	})
 }