Repository: cloudevents/sdk-go
Pull request #289 (2020-01-23 18:02:24)

Fixing PubSub Codec to only support v1 string extensions 
- v1 should support string extensions. In the pubsub codec, we had some legacy code from 0.3, which supported maps, and other types as well.
- This caused a panic when using it from a receive adapter code, trying to decode a "int64" extension: 
panic: cannot convert 1.57947362937361e+15 to int32: out of range

Signed-off-by: Nacho Cano <nachoacano@gmail.com>

nachocano: /assign @n3wscott 
n3wscott: LGTM

Patch:
diff --git a/pkg/cloudevents/transport/pubsub/codec_v1.go b/pkg/cloudevents/transport/pubsub/codec_v1.go
index 85396de..2d09101 100644
--- a/pkg/cloudevents/transport/pubsub/codec_v1.go
+++ b/pkg/cloudevents/transport/pubsub/codec_v1.go
@@ -2,7 +2,6 @@ package pubsub
 
 import (
 	"context"
-	"encoding/json"
 	"fmt"
 	"strings"
 
@@ -118,27 +117,13 @@ func (v CodecV1) toAttributes(e cloudevents.Event) (map[string]string, error) {
 	}
 
 	for k, v := range e.Extensions() {
-		if mapVal, ok := v.(map[string]interface{}); ok {
-			for subkey, subval := range mapVal {
-				encoded, err := json.Marshal(subval)
-				if err != nil {
-					return nil, err
-				}
-				a[prefix+k+"-"+subkey] = string(encoded)
-			}
-			continue
-		}
-		if s, ok := v.(string); ok {
-			a[prefix+k] = s
-			continue
-		}
-		encoded, err := json.Marshal(v)
+		k = strings.ToLower(k)
+		cstr, err := types.Format(v)
 		if err != nil {
-			return nil, err
+			return a, err
 		}
-		a[prefix+k] = string(encoded)
+		a[prefix+k] = cstr
 	}
-
 	return a, nil
 }
 
@@ -244,32 +229,7 @@ func (v CodecV1) fromAttributes(a map[string]string, event *cloudevents.Event) e
 	for k, v := range a {
 		if len(k) > len(prefix) && strings.EqualFold(k[:len(prefix)], prefix) {
 			ak := strings.ToLower(k[len(prefix):])
-			if i := strings.Index(ak, "-"); i > 0 {
-				// attrib-key
-				attrib := ak[:i]
-				key := ak[(i + 1):]
-				if xv, ok := extensions[attrib]; ok {
-					if m, ok := xv.(map[string]interface{}); ok {
-						m[key] = v
-						continue
-					}
-					// TODO: revisit how we want to bubble errors up.
-					return fmt.Errorf("failed to process map type extension")
-				} else {
-					m := make(map[string]interface{})
-					m[key] = v
-					extensions[attrib] = m
-				}
-			} else {
-				// key
-				var tmp interface{}
-				if err := json.Unmarshal([]byte(v), &tmp); err == nil {
-					extensions[ak] = tmp
-				} else {
-					// If we can't unmarshal the data, treat it as a string.
-					extensions[ak] = v
-				}
-			}
+			extensions[ak] = v
 		}
 	}
 	event.Context = ec


Test patch:
diff --git a/pkg/cloudevents/transport/pubsub/codec_test.go b/pkg/cloudevents/transport/pubsub/codec_test.go
index ca58a38..12af951 100644
--- a/pkg/cloudevents/transport/pubsub/codec_test.go
+++ b/pkg/cloudevents/transport/pubsub/codec_test.go
@@ -14,6 +14,10 @@ import (
 	"github.com/google/go-cmp/cmp"
 )
 
+const (
+	prefix = "ce-"
+)
+
 func strptr(s string) *string {
 	return &s
 }
diff --git a/pkg/cloudevents/transport/pubsub/codec_v1_test.go b/pkg/cloudevents/transport/pubsub/codec_v1_test.go
index 0b4dfb7..255b620 100644
--- a/pkg/cloudevents/transport/pubsub/codec_v1_test.go
+++ b/pkg/cloudevents/transport/pubsub/codec_v1_test.go
@@ -62,7 +62,8 @@ func TestCodecV1_Encode(t *testing.T) {
 					DataContentType: cloudevents.StringOfApplicationJSON(),
 					Source:          *source,
 					Extensions: map[string]interface{}{
-						"test": "extended",
+						"test":       "extended",
+						"generation": "1579743478182200",
 					},
 				}.AsV1(),
 				Data: map[string]interface{}{
@@ -84,6 +85,7 @@ func TestCodecV1_Encode(t *testing.T) {
 						"time":       now,
 						"type":       "com.example.test",
 						"test":       "extended",
+						"generation": "1579743478182200",
 						"dataschema": "http://example.com/schema",
 						"source":     "http://example.com/source",
 					}
@@ -91,6 +93,46 @@ func TestCodecV1_Encode(t *testing.T) {
 				}(),
 			},
 		},
+		"full v1.0 binary": {
+			codec: pubsub.CodecV1{
+				DefaultEncoding: pubsub.BinaryV1,
+			},
+			event: cloudevents.Event{
+				Context: cloudevents.EventContextV1{
+					ID:         "ABC-123",
+					Time:       &now,
+					Type:       "com.example.test",
+					DataSchema: schema,
+					Source:     *source,
+					Extensions: map[string]interface{}{
+						"test":       "extended",
+						"generation": "1579743478182200",
+					},
+				}.AsV1(),
+				Data: map[string]interface{}{
+					"hello": "world",
+				},
+			},
+			want: &pubsub.Message{
+				Attributes: map[string]string{
+					prefix + "specversion":     "1.0",
+					prefix + "id":              "ABC-123",
+					prefix + "datacontenttype": cloudevents.ApplicationJSON,
+					prefix + "time":            now.String(),
+					prefix + "type":            "com.example.test",
+					prefix + "dataschema":      "http://example.com/schema",
+					prefix + "source":          "http://example.com/source",
+					prefix + "test":            "extended",
+					prefix + "generation":      "1579743478182200",
+				},
+				Data: func() []byte {
+					data := map[string]interface{}{
+						"hello": "world",
+					}
+					return toBytes(data)
+				}(),
+			},
+		},
 		"simple v1.0 structured": {
 			codec: pubsub.CodecV1{DefaultEncoding: pubsub.StructuredV1},
 			event: cloudevents.Event{
@@ -238,6 +280,7 @@ func TestCodecV1_Decode(t *testing.T) {
 					"time":       now,
 					"type":       "com.example.test",
 					"test":       "extended",
+					"generation": "1579743478182200",
 					"dataschema": "http://example.com/schema",
 					"source":     "http://example.com/source",
 				}),
@@ -252,7 +295,44 @@ func TestCodecV1_Decode(t *testing.T) {
 					DataContentType: cloudevents.StringOfApplicationJSON(),
 					Source:          *source,
 					Extensions: map[string]interface{}{
-						"test": "extended",
+						"test":       "extended",
+						"generation": "1579743478182200",
+					},
+				},
+				Data: toBytes(map[string]interface{}{
+					"hello": "world",
+				}),
+				DataEncoded: true,
+			},
+		},
+		"full v1.0 binary": {
+			codec: pubsub.CodecV1{},
+			msg: &pubsub.Message{
+				Attributes: map[string]string{
+					prefix + "specversion": "1.0",
+					prefix + "id":          "ABC-123",
+					prefix + "time":        now.String(),
+					prefix + "type":        "com.example.test",
+					prefix + "dataschema":  "http://example.com/schema",
+					prefix + "source":      "http://example.com/source",
+					prefix + "test":        "extended",
+					prefix + "generation":  "1579743478182200",
+				},
+				Data: toBytes(map[string]interface{}{
+					"hello": "world",
+				}),
+			},
+			want: &cloudevents.Event{
+				Context: &cloudevents.EventContextV1{
+					SpecVersion: cloudevents.CloudEventsVersionV1,
+					ID:          "ABC-123",
+					Time:        &now,
+					Type:        "com.example.test",
+					DataSchema:  schema,
+					Source:      *source,
+					Extensions: map[string]interface{}{
+						"test":       "extended",
+						"generation": "1579743478182200",
 					},
 				},
 				Data: toBytes(map[string]interface{}{