Add application/json support to outbox

pkg/runtime/pubsub/outbox.go

@@ -17,6 +17,7 @@
 	"context"
 	"encoding/json"
 	"fmt"
+	"strings"
 	"sync"
 	"time"
 
@@ -151,6 +152,13 @@
 			bt, ok := sr.Value.([]byte)
 			if ok {
 				ceData = bt
+			} else if sr.ContentType != nil && strings.EqualFold(*sr.ContentType, "application/json") {
+				b, sErr := json.Marshal(sr.Value)
+				if sErr != nil {
+					return nil, sErr
+				}
+
+				ceData = b
 			} else {
 				ceData = []byte(fmt.Sprintf("%v", sr.Value))
 			}

pkg/runtime/pubsub/outbox_test.go

@@ -259,7 +259,7 @@ func TestPublishInternal(t *testing.T) {
 		assert.NoError(t, err)
 	})
 
-	t.Run("valid operation, custom datacontenttype", func(t *testing.T) {
+	t.Run("valid operation, no datacontenttype", func(t *testing.T) {
 		o := newTestOutbox().(*outboxImpl)
 		o.publishFn = func(ctx context.Context, pr *contribPubsub.PublishRequest) error {
 			var cloudEvent map[string]interface{}
@@ -270,7 +270,7 @@ func TestPublishInternal(t *testing.T) {
 			assert.Equal(t, "a", pr.PubsubName)
 			assert.Equal(t, "testapp1outbox", pr.Topic)
 			assert.Equal(t, "testapp", cloudEvent["source"])
-			assert.Equal(t, "application/json", cloudEvent["datacontenttype"])
+			assert.Equal(t, "text/plain", cloudEvent["datacontenttype"])
 			assert.Equal(t, "a", cloudEvent["pubsubname"])
 
 			return nil
@@ -302,7 +302,7 @@ func TestPublishInternal(t *testing.T) {
 			},
 		})
 
-		contentType := "application/json"
+		contentType := ""
 		_, err := o.PublishInternal(context.TODO(), "test", []state.TransactionalStateOperation{
 			state.SetRequest{
 				Key:         "key",
@@ -314,6 +314,77 @@ func TestPublishInternal(t *testing.T) {
 		assert.NoError(t, err)
 	})
 
+	type customData struct {
+		Name string `json:"name"`
+	}
+
+	t.Run("valid operation, application/json datacontenttype", func(t *testing.T) {
+		o := newTestOutbox().(*outboxImpl)
+		o.publishFn = func(ctx context.Context, pr *contribPubsub.PublishRequest) error {
+			var cloudEvent map[string]interface{}
+			err := json.Unmarshal(pr.Data, &cloudEvent)
+			assert.NoError(t, err)
+
+			data := cloudEvent["data"]
+			j := customData{}
+
+			err = json.Unmarshal([]byte(data.(string)), &j)
+			assert.NoError(t, err)
+
+			assert.Equal(t, "test", j.Name)
+			assert.Equal(t, "a", pr.PubsubName)
+			assert.Equal(t, "testapp1outbox", pr.Topic)
+			assert.Equal(t, "testapp", cloudEvent["source"])
+			assert.Equal(t, "application/json", cloudEvent["datacontenttype"])
+			assert.Equal(t, "a", cloudEvent["pubsubname"])
+
+			return nil
+		}
+
+		o.AddOrUpdateOutbox(v1alpha1.Component{
+			ObjectMeta: metav1.ObjectMeta{
+				Name: "test",
+			},
+			Spec: v1alpha1.ComponentSpec{
+				Metadata: []common.NameValuePair{
+					{
+						Name: outboxPublishPubsubKey,
+						Value: common.DynamicValue{
+							JSON: v1.JSON{
+								Raw: []byte("a"),
+							},
+						},
+					},
+					{
+						Name: outboxPublishTopicKey,
+						Value: common.DynamicValue{
+							JSON: v1.JSON{
+								Raw: []byte("1"),
+							},
+						},
+					},
+				},
+			},
+		})
+
+		j := customData{
+			Name: "test",
+		}
+		b, err := json.Marshal(&j)
+		require.NoError(t, err)
+
+		contentType := "application/json"
+		_, err = o.PublishInternal(context.TODO(), "test", []state.TransactionalStateOperation{
+			state.SetRequest{
+				Key:         "key",
+				Value:       string(b),
+				ContentType: &contentType,
+			},
+		}, "testapp", "", "")
+
+		assert.NoError(t, err)
+	})
+
 	t.Run("missing state store", func(t *testing.T) {
 		o := newTestOutbox().(*outboxImpl)