Repository: cloudevents/sdk-go (go)

Issue:
spec v1, pubsub transport: couldn't make it work, errs
I used this example as base:

https://github.com/cloudevents/sdk-go/tree/master/cmd/samples/pubsub

Using PubSub Transport, everything worked fine when used v03.

When trying to use v1, just changing `cloudevents.VersionV03` for `cloudevents.VersionV1` in:

https://github.com/cloudevents/sdk-go/blob/master/cmd/samples/pubsub/sender/main.go#L45

Throws this error on a loop:

```
{"level":"error","ts":1573823257.217474,"logger":"fallback","caller":"pubsub/transport.go:241","msg":"failed to decode message","error":"unknown encoding: Unknown Encoding","stacktrace":"github.com/cloudevents/sdk-go/pkg/cloudevents/transport/pubsub.(*Transport).startSubscriber.func1\n\t/Users/lopezator/go/pkg/mod/github.com/cloudevents/sdk-go@v0.10.0/pkg/cloudevents/transport/pubsub/transport.go:241\ngithub.com/cloudevents/sdk-go/pkg/cloudevents/transport/pubsub/internal.(*Connection).Receive.func1\n\t/Users/lopezator/go/pkg/mod/github.com/cloudevents/sdk-go@v0.10.0/pkg/cloudevents/transport/pubsub/internal/connection.go:187\ncloud.google.com/go/pubsub.(*Subscription).receive.func3\n\t/Users/lopezator/go/pkg/mod/cloud.google.com/go@v0.40.0/pubsub/subscription.go:633"}
```

What could I do?

Maybe this also could be a good opportunity to update the examples ans start using v1 on them. I could take over this if you give me any hint.





Pull request description:
pubsub: fix v1 transport, default encoding, refactors
- fix v1 transport, set default encoding
- apply refactors from http tranport
- update pubsub transport examples to use v1

Closes #255

Patch:
diff --git a/cmd/samples/pubsub/multisender/main.go b/cmd/samples/pubsub/multisender/main.go
index f2378cf..e4dbd1a 100644
--- a/cmd/samples/pubsub/multisender/main.go
+++ b/cmd/samples/pubsub/multisender/main.go
@@ -58,7 +58,7 @@ func main() {
 
 	for i, topic := range strings.Split(env.TopicIDs, ",") {
 		ctx := cecontext.WithTopic(context.Background(), topic)
-		event := cloudevents.NewEvent(cloudevents.VersionV03)
+		event := cloudevents.NewEvent(cloudevents.VersionV01)
 		event.SetType("com.cloudevents.sample.sent")
 		event.SetSource("github.com/cloudevents/sdk-go/cmd/samples/pubsub/multisender/")
 		_ = event.SetData(&Example{
diff --git a/cmd/samples/pubsub/sender/main.go b/cmd/samples/pubsub/sender/main.go
index ef77c25..b448bb6 100644
--- a/cmd/samples/pubsub/sender/main.go
+++ b/cmd/samples/pubsub/sender/main.go
@@ -42,7 +42,7 @@ func main() {
 		os.Exit(1)
 	}
 
-	event := cloudevents.NewEvent(cloudevents.VersionV03)
+	event := cloudevents.NewEvent(cloudevents.VersionV1)
 	event.SetType("com.cloudevents.sample.sent")
 	event.SetSource("github.com/cloudevents/sdk-go/cmd/samples/pubsub/sender/")
 	_ = event.SetData(&Example{
diff --git a/pkg/cloudevents/transport/pubsub/codec.go b/pkg/cloudevents/transport/pubsub/codec.go
index 951b991..4401ef7 100644
--- a/pkg/cloudevents/transport/pubsub/codec.go
+++ b/pkg/cloudevents/transport/pubsub/codec.go
@@ -2,9 +2,12 @@ package pubsub
 
 import (
 	"context"
+	"errors"
 	"fmt"
+	"sync"
 
 	"github.com/cloudevents/sdk-go/pkg/cloudevents"
+	cecontext "github.com/cloudevents/sdk-go/pkg/cloudevents/context"
 	"github.com/cloudevents/sdk-go/pkg/cloudevents/transport"
 )
 
@@ -16,6 +19,9 @@ type Codec struct {
 
 	v03 *CodecV03
 	v1  *CodecV1
+
+	_v03 sync.Once
+	_v1  sync.Once
 }
 
 const (
@@ -24,46 +30,83 @@ const (
 
 var _ transport.Codec = (*Codec)(nil)
 
+func (c *Codec) loadCodec(encoding Encoding) (transport.Codec, error) {
+	switch encoding {
+	case Default:
+		fallthrough
+	case BinaryV1, StructuredV1:
+		c._v1.Do(func() {
+			c.v1 = &CodecV1{DefaultEncoding: c.Encoding}
+		})
+		return c.v1, nil
+	case BinaryV03, StructuredV03:
+		c._v03.Do(func() {
+			c.v03 = &CodecV03{DefaultEncoding: c.Encoding}
+		})
+		return c.v03, nil
+	}
+
+	return nil, fmt.Errorf("unknown encoding: %s", encoding)
+}
+
 func (c *Codec) Encode(ctx context.Context, e cloudevents.Event) (transport.Message, error) {
-	switch c.Encoding {
-	case Default, BinaryV03, StructuredV03:
-		if c.v03 == nil {
-			c.v03 = &CodecV03{Encoding: c.Encoding}
-		}
-		return c.v03.Encode(ctx, e)
-	case StructuredV1:
-		if c.v1 == nil {
-			c.v1 = &CodecV1{Encoding: c.Encoding}
-		}
-		return c.v1.Encode(ctx, e)
-	default:
-		return nil, transport.NewErrMessageEncodingUnknown("wrapper", TransportName)
+	encoding := c.Encoding
+	if encoding == Default && c.DefaultEncodingSelectionFn != nil {
+		encoding = c.DefaultEncodingSelectionFn(ctx, e)
+	}
+	codec, err := c.loadCodec(encoding)
+	if err != nil {
+		return nil, err
 	}
+	ctx = cecontext.WithEncoding(ctx, encoding.Name())
+	return codec.Encode(ctx, e)
 }
 
 func (c *Codec) Decode(ctx context.Context, msg transport.Message) (*cloudevents.Event, error) {
-	switch encoding := c.inspectEncoding(ctx, msg); encoding {
+	codec, err := c.loadCodec(c.inspectEncoding(ctx, msg))
+	if err != nil {
+		return nil, err
+	}
+	event, err := codec.Decode(ctx, msg)
+	if err != nil {
+		return nil, err
+	}
+	return c.convertEvent(event)
+}
+
+// Give the context back as the user expects
+func (c *Codec) convertEvent(event *cloudevents.Event) (*cloudevents.Event, error) {
+	if event == nil {
+		return nil, errors.New("event is nil, can not convert")
+	}
+
+	switch c.Encoding {
+	case Default:
+		return event, nil
 	case BinaryV03, StructuredV03:
-		if c.v03 == nil {
-			c.v03 = &CodecV03{Encoding: encoding}
-		}
-		return c.v03.Decode(ctx, msg)
+		ca := event.Context.AsV03()
+		event.Context = ca
+		return event, nil
 	case BinaryV1, StructuredV1:
-		if c.v1 == nil {
-			c.v1 = &CodecV1{Encoding: encoding}
-		}
-		return c.v1.Decode(ctx, msg)
+		ca := event.Context.AsV1()
+		event.Context = ca
+		return event, nil
 	default:
-		return nil, fmt.Errorf("unknown encoding: %s", encoding)
+		return nil, fmt.Errorf("unknown encoding: %s", c.Encoding)
 	}
 }
 
 func (c *Codec) inspectEncoding(ctx context.Context, msg transport.Message) Encoding {
-	if c.v03 == nil {
-		c.v03 = &CodecV03{Encoding: c.Encoding}
+	// Try v1.0.
+	_, _ = c.loadCodec(BinaryV1)
+	encoding := c.v1.inspectEncoding(ctx, msg)
+	if encoding != Unknown {
+		return encoding
 	}
+
 	// Try v0.3.
-	encoding := c.v03.inspectEncoding(ctx, msg)
+	_, _ = c.loadCodec(BinaryV03)
+	encoding = c.v03.inspectEncoding(ctx, msg)
 	if encoding != Unknown {
 		return encoding
 	}
diff --git a/pkg/cloudevents/transport/pubsub/codec_v03.go b/pkg/cloudevents/transport/pubsub/codec_v03.go
index da90cdc..d96ab3a 100644
--- a/pkg/cloudevents/transport/pubsub/codec_v03.go
+++ b/pkg/cloudevents/transport/pubsub/codec_v03.go
@@ -7,6 +7,7 @@ import (
 	"strings"
 
 	"github.com/cloudevents/sdk-go/pkg/cloudevents"
+	cecontext "github.com/cloudevents/sdk-go/pkg/cloudevents/context"
 	"github.com/cloudevents/sdk-go/pkg/cloudevents/transport"
 	"github.com/cloudevents/sdk-go/pkg/cloudevents/types"
 )
@@ -18,19 +19,31 @@ const (
 type CodecV03 struct {
 	CodecStructured
 
-	Encoding Encoding
+	DefaultEncoding Encoding
 }
 
 var _ transport.Codec = (*CodecV03)(nil)
 
 func (v CodecV03) Encode(ctx context.Context, e cloudevents.Event) (transport.Message, error) {
-	switch v.Encoding {
-	case Default, StructuredV03:
+	encoding := v.DefaultEncoding
+	strEnc := cecontext.EncodingFrom(ctx)
+	if strEnc != "" {
+		switch strEnc {
+		case Binary:
+			encoding = BinaryV03
+		case Structured:
+			encoding = StructuredV03
+		}
+	}
+	switch encoding {
+	case Default:
+		fallthrough
+	case StructuredV03:
 		return v.encodeStructured(ctx, e)
 	case BinaryV03:
 		return v.encodeBinary(ctx, e)
 	default:
-		return nil, fmt.Errorf("unknown encoding: %d", v.Encoding)
+		return nil, fmt.Errorf("unknown encoding: %d", encoding)
 	}
 }
 
diff --git a/pkg/cloudevents/transport/pubsub/codec_v1.go b/pkg/cloudevents/transport/pubsub/codec_v1.go
index e7b3ef3..85396de 100644
--- a/pkg/cloudevents/transport/pubsub/codec_v1.go
+++ b/pkg/cloudevents/transport/pubsub/codec_v1.go
@@ -7,6 +7,7 @@ import (
 	"strings"
 
 	"github.com/cloudevents/sdk-go/pkg/cloudevents"
+	cecontext "github.com/cloudevents/sdk-go/pkg/cloudevents/context"
 	"github.com/cloudevents/sdk-go/pkg/cloudevents/transport"
 	"github.com/cloudevents/sdk-go/pkg/cloudevents/types"
 )
@@ -14,14 +15,26 @@ import (
 type CodecV1 struct {
 	CodecStructured
 
-	Encoding Encoding
+	DefaultEncoding Encoding
 }
 
 var _ transport.Codec = (*CodecV1)(nil)
 
 func (v CodecV1) Encode(ctx context.Context, e cloudevents.Event) (transport.Message, error) {
-	switch v.Encoding {
-	case Default, StructuredV1:
+	encoding := v.DefaultEncoding
+	strEnc := cecontext.EncodingFrom(ctx)
+	if strEnc != "" {
+		switch strEnc {
+		case Binary:
+			encoding = BinaryV1
+		case Structured:
+			encoding = StructuredV1
+		}
+	}
+	switch encoding {
+	case Default:
+		fallthrough
+	case StructuredV1:
 		return v.encodeStructured(ctx, e)
 	case BinaryV1:
 		return v.encodeBinary(ctx, e)
diff --git a/pkg/cloudevents/transport/pubsub/encoding.go b/pkg/cloudevents/transport/pubsub/encoding.go
index 2bdf564..33717f7 100644
--- a/pkg/cloudevents/transport/pubsub/encoding.go
+++ b/pkg/cloudevents/transport/pubsub/encoding.go
@@ -22,8 +22,17 @@ const (
 	StructuredV03
 	// StructuredV1 is Structured CloudEvents spec v1.0.
 	StructuredV1
+
 	// Unknown is unknown.
 	Unknown
+
+	// Binary is used for Context Based Encoding Selections to use the
+	// DefaultBinaryEncodingSelectionStrategy
+	Binary = "binary"
+
+	// Structured is used for Context Based Encoding Selections to use the
+	// DefaultStructuredEncodingSelectionStrategy
+	Structured = "structured"
 )
 
 // DefaultBinaryEncodingSelectionStrategy implements a selection process for
@@ -89,3 +98,17 @@ func (e Encoding) Version() string {
 		return "Unknown"
 	}
 }
+
+// Name creates a string to represent the the codec name.
+func (e Encoding) Name() string {
+	switch e {
+	case Default:
+		return Binary
+	case BinaryV03, BinaryV1:
+		return Binary
+	case StructuredV03, StructuredV1:
+		return Structured
+	default:
+		return Binary
+	}
+}


Test patch:
diff --git a/pkg/cloudevents/transport/pubsub/codec_test.go b/pkg/cloudevents/transport/pubsub/codec_test.go
index f2360b7..ca58a38 100644
--- a/pkg/cloudevents/transport/pubsub/codec_test.go
+++ b/pkg/cloudevents/transport/pubsub/codec_test.go
@@ -23,14 +23,14 @@ func TestCodecEncode(t *testing.T) {
 	source := &types.URLRef{URL: *sourceUrl}
 
 	testCases := map[string]struct {
-		codec   pubsub.Codec
-		event   cloudevents.Event
+		codec   *pubsub.Codec
+		event   *cloudevents.Event
 		want    *pubsub.Message
 		wantErr error
 	}{
 		"simple v03 structured": {
-			codec: pubsub.Codec{Encoding: pubsub.StructuredV03},
-			event: cloudevents.Event{
+			codec: &pubsub.Codec{Encoding: pubsub.StructuredV03},
+			event: &cloudevents.Event{
 				Context: cloudevents.EventContextV03{
 					Type:    "com.example.test",
 					Source:  *source,
@@ -55,8 +55,8 @@ func TestCodecEncode(t *testing.T) {
 			},
 		},
 		"simple v03 binary": {
-			codec: pubsub.Codec{Encoding: pubsub.BinaryV03},
-			event: cloudevents.Event{
+			codec: &pubsub.Codec{Encoding: pubsub.BinaryV03},
+			event: &cloudevents.Event{
 				Context: &cloudevents.EventContextV03{
 					Type:    "com.example.test",
 					Source:  *source,
@@ -79,7 +79,7 @@ func TestCodecEncode(t *testing.T) {
 	for n, tc := range testCases {
 		t.Run(n, func(t *testing.T) {
 
-			got, err := tc.codec.Encode(context.TODO(), tc.event)
+			got, err := tc.codec.Encode(context.TODO(), *tc.event)
 
 			if tc.wantErr != nil || err != nil {
 				if diff := cmp.Diff(tc.wantErr, err); diff != "" {
@@ -109,13 +109,13 @@ func TestCodecDecode(t *testing.T) {
 	source := &types.URLRef{URL: *sourceUrl}
 
 	testCases := map[string]struct {
-		codec   pubsub.Codec
+		codec   *pubsub.Codec
 		msg     *pubsub.Message
 		want    *cloudevents.Event
 		wantErr error
 	}{
 		"simple v3 structured": {
-			codec: pubsub.Codec{Encoding: pubsub.StructuredV03},
+			codec: &pubsub.Codec{Encoding: pubsub.StructuredV03},
 			msg: &pubsub.Message{
 				Attributes: map[string]string{
 					"Content-Type": cloudevents.ApplicationCloudEventsJSON,
@@ -176,14 +176,14 @@ func TestCodecAsMiddleware(t *testing.T) {
 	for _, encoding := range []pubsub.Encoding{pubsub.StructuredV03} {
 
 		testCases := map[string]struct {
-			codec   pubsub.Codec
-			event   cloudevents.Event
-			want    cloudevents.Event
+			codec   *pubsub.Codec
+			event   *cloudevents.Event
+			want    *cloudevents.Event
 			wantErr error
 		}{
 			"simple data": {
-				codec: pubsub.Codec{Encoding: encoding},
-				event: cloudevents.Event{
+				codec: &pubsub.Codec{Encoding: encoding},
+				event: &cloudevents.Event{
 					Context: cloudevents.EventContextV03{
 						Type:   "com.example.test",
 						Source: *source,
@@ -194,7 +194,7 @@ func TestCodecAsMiddleware(t *testing.T) {
 						"b": "banana",
 					},
 				},
-				want: cloudevents.Event{
+				want: &cloudevents.Event{
 					Context: &cloudevents.EventContextV03{
 						SpecVersion: cloudevents.CloudEventsVersionV03,
 						Type:        "com.example.test",
@@ -209,8 +209,8 @@ func TestCodecAsMiddleware(t *testing.T) {
 				},
 			},
 			"struct data": {
-				codec: pubsub.Codec{Encoding: encoding},
-				event: cloudevents.Event{
+				codec: &pubsub.Codec{Encoding: encoding},
+				event: &cloudevents.Event{
 					Context: cloudevents.EventContextV03{
 						Type:   "com.example.test",
 						Source: *source,
@@ -221,7 +221,7 @@ func TestCodecAsMiddleware(t *testing.T) {
 						AString: "testing",
 					},
 				},
-				want: cloudevents.Event{
+				want: &cloudevents.Event{
 					Context: &cloudevents.EventContextV03{
 						SpecVersion: cloudevents.CloudEventsVersionV03,
 						Type:        "com.example.test",
@@ -240,7 +240,7 @@ func TestCodecAsMiddleware(t *testing.T) {
 			n = fmt.Sprintf("%s, %s", encoding, n)
 			t.Run(n, func(t *testing.T) {
 
-				msg1, err := tc.codec.Encode(context.TODO(), tc.event)
+				msg1, err := tc.codec.Encode(context.TODO(), *tc.event)
 				if err != nil {
 					if diff := cmp.Diff(tc.wantErr, err); diff != "" {
 						t.Errorf("unexpected error (-want, +got) = %v", diff)
@@ -288,12 +288,11 @@ func TestCodecAsMiddleware(t *testing.T) {
 				ctxv3 := got.Context.AsV03()
 				got.Context = ctxv3
 
-				if diff := cmp.Diff(tc.want, *got); diff != "" {
+				if diff := cmp.Diff(tc.want, got); diff != "" {
 					t.Errorf("unexpected event (-want, +got) = %v", diff)
 				}
 			})
 		}
-
 	}
 }
 
diff --git a/pkg/cloudevents/transport/pubsub/codec_v03_test.go b/pkg/cloudevents/transport/pubsub/codec_v03_test.go
index d1e4ed4..2ade7d4 100644
--- a/pkg/cloudevents/transport/pubsub/codec_v03_test.go
+++ b/pkg/cloudevents/transport/pubsub/codec_v03_test.go
@@ -92,7 +92,7 @@ func TestCodecV03_Encode(t *testing.T) {
 			},
 		},
 		"simple v0.3 structured": {
-			codec: pubsub.CodecV03{Encoding: pubsub.StructuredV03},
+			codec: pubsub.CodecV03{DefaultEncoding: pubsub.StructuredV03},
 			event: cloudevents.Event{
 				Context: cloudevents.EventContextV03{
 					Type:   "com.example.test",
@@ -116,7 +116,7 @@ func TestCodecV03_Encode(t *testing.T) {
 			},
 		},
 		"full v0.3 structured": {
-			codec: pubsub.CodecV03{Encoding: pubsub.StructuredV03},
+			codec: pubsub.CodecV03{DefaultEncoding: pubsub.StructuredV03},
 			event: cloudevents.Event{
 				Context: cloudevents.EventContextV03{
 					ID:              "ABC-123",
diff --git a/pkg/cloudevents/transport/pubsub/codec_v1_test.go b/pkg/cloudevents/transport/pubsub/codec_v1_test.go
index 8b75d25..0b4dfb7 100644
--- a/pkg/cloudevents/transport/pubsub/codec_v1_test.go
+++ b/pkg/cloudevents/transport/pubsub/codec_v1_test.go
@@ -92,7 +92,7 @@ func TestCodecV1_Encode(t *testing.T) {
 			},
 		},
 		"simple v1.0 structured": {
-			codec: pubsub.CodecV1{Encoding: pubsub.StructuredV1},
+			codec: pubsub.CodecV1{DefaultEncoding: pubsub.StructuredV1},
 			event: cloudevents.Event{
 				Context: cloudevents.EventContextV1{
 					Type:   "com.example.test",
@@ -116,7 +116,7 @@ func TestCodecV1_Encode(t *testing.T) {
 			},
 		},
 		"full v1.0 structured": {
-			codec: pubsub.CodecV1{Encoding: pubsub.StructuredV1},
+			codec: pubsub.CodecV1{DefaultEncoding: pubsub.StructuredV1},
 			event: cloudevents.Event{
 				Context: cloudevents.EventContextV1{
 					ID:              "ABC-123",