Repository: cloudevents/sdk-go
Pull request #155 (2019-07-10 15:06:13)

Adding context to codec methods.
Future work requires that context be available in the methods related to codecs. 

This also adds tracing to these methods in the context of the call stack.

Signed-off-by: Scott Nichols <nicholss@google.com>

n3wscott: No code change. Just adding context methods.

Patch:
diff --git a/pkg/cloudevents/datacodec/codec.go b/pkg/cloudevents/datacodec/codec.go
index 5ce030d..41425c2 100644
--- a/pkg/cloudevents/datacodec/codec.go
+++ b/pkg/cloudevents/datacodec/codec.go
@@ -12,11 +12,11 @@ import (
 // Decoder is the expected function signature for decoding `in` to `out`. What
 // `in` is could be decoder dependent. For example, `in` could be bytes, or a
 // base64 string.
-type Decoder func(in, out interface{}) error
+type Decoder func(ctx context.Context, in, out interface{}) error
 
 // Encoder is the expected function signature for encoding `in` to bytes.
 // Returns an error if the encoder has an issue encoding `in`.
-type Encoder func(in interface{}) ([]byte, error)
+type Encoder func(ctx context.Context, in interface{}) ([]byte, error)
 
 var decoder map[string]Decoder
 var encoder map[string]Encoder
@@ -53,10 +53,9 @@ func AddEncoder(contentType string, fn Encoder) {
 // Decode looks up and invokes the decoder registered for the given content
 // type. An error is returned if no decoder is registered for the given
 // content type.
-func Decode(contentType string, in, out interface{}) error {
-	// TODO: wire in context.
-	_, r := observability.NewReporter(context.Background(), reportDecode)
-	err := obsDecode(contentType, in, out)
+func Decode(ctx context.Context, contentType string, in, out interface{}) error {
+	_, r := observability.NewReporter(ctx, reportDecode)
+	err := obsDecode(ctx, contentType, in, out)
 	if err != nil {
 		r.Error()
 	} else {
@@ -65,9 +64,9 @@ func Decode(contentType string, in, out interface{}) error {
 	return err
 }
 
-func obsDecode(contentType string, in, out interface{}) error {
+func obsDecode(ctx context.Context, contentType string, in, out interface{}) error {
 	if fn, ok := decoder[contentType]; ok {
-		return fn(in, out)
+		return fn(ctx, in, out)
 	}
 	return fmt.Errorf("[decode] unsupported content type: %q", contentType)
 }
@@ -75,10 +74,9 @@ func obsDecode(contentType string, in, out interface{}) error {
 // Encode looks up and invokes the encoder registered for the given content
 // type. An error is returned if no encoder is registered for the given
 // content type.
-func Encode(contentType string, in interface{}) ([]byte, error) {
-	// TODO: wire in context.
-	_, r := observability.NewReporter(context.Background(), reportEncode)
-	b, err := obsEncode(contentType, in)
+func Encode(ctx context.Context, contentType string, in interface{}) ([]byte, error) {
+	_, r := observability.NewReporter(ctx, reportEncode)
+	b, err := obsEncode(ctx, contentType, in)
 	if err != nil {
 		r.Error()
 	} else {
@@ -87,9 +85,9 @@ func Encode(contentType string, in interface{}) ([]byte, error) {
 	return b, err
 }
 
-func obsEncode(contentType string, in interface{}) ([]byte, error) {
+func obsEncode(ctx context.Context, contentType string, in interface{}) ([]byte, error) {
 	if fn, ok := encoder[contentType]; ok {
-		return fn(in)
+		return fn(ctx, in)
 	}
 	return nil, fmt.Errorf("[encode] unsupported content type: %q", contentType)
 }
diff --git a/pkg/cloudevents/datacodec/json/data.go b/pkg/cloudevents/datacodec/json/data.go
index 0f92a96..926c344 100644
--- a/pkg/cloudevents/datacodec/json/data.go
+++ b/pkg/cloudevents/datacodec/json/data.go
@@ -13,10 +13,9 @@ import (
 // Decode takes `in` as []byte, or base64 string, normalizes in to unquoted and
 // base64 decoded []byte if required, and then attempts to use json.Unmarshal
 // to convert those bytes to `out`. Returns and error if this process fails.
-func Decode(in, out interface{}) error {
-	// TODO: wire in context.
-	_, r := observability.NewReporter(context.Background(), reportDecode)
-	err := obsDecode(in, out)
+func Decode(ctx context.Context, in, out interface{}) error {
+	_, r := observability.NewReporter(ctx, reportDecode)
+	err := obsDecode(ctx, in, out)
 	if err != nil {
 		r.Error()
 	} else {
@@ -25,7 +24,7 @@ func Decode(in, out interface{}) error {
 	return err
 }
 
-func obsDecode(in, out interface{}) error {
+func obsDecode(ctx context.Context, in, out interface{}) error {
 	if in == nil {
 		return nil
 	}
@@ -63,10 +62,9 @@ func obsDecode(in, out interface{}) error {
 // Encode attempts to json.Marshal `in` into bytes. Encode will inspect `in`
 // and returns `in` unmodified if it is detected that `in` is already a []byte;
 // Or json.Marshal errors.
-func Encode(in interface{}) ([]byte, error) {
-	// TODO: wire in context.
-	_, r := observability.NewReporter(context.Background(), reportEncode)
-	b, err := obsEncode(in)
+func Encode(ctx context.Context, in interface{}) ([]byte, error) {
+	_, r := observability.NewReporter(ctx, reportEncode)
+	b, err := obsEncode(ctx, in)
 	if err != nil {
 		r.Error()
 	} else {
@@ -75,7 +73,7 @@ func Encode(in interface{}) ([]byte, error) {
 	return b, err
 }
 
-func obsEncode(in interface{}) ([]byte, error) {
+func obsEncode(ctx context.Context, in interface{}) ([]byte, error) {
 	if in == nil {
 		return nil, nil
 	}
diff --git a/pkg/cloudevents/datacodec/xml/data.go b/pkg/cloudevents/datacodec/xml/data.go
index ae529c2..6339e44 100644
--- a/pkg/cloudevents/datacodec/xml/data.go
+++ b/pkg/cloudevents/datacodec/xml/data.go
@@ -13,10 +13,9 @@ import (
 // Decode takes `in` as []byte, or base64 string, normalizes in to unquoted and
 // base64 decoded []byte if required, and then attempts to use xml.Unmarshal
 // to convert those bytes to `out`. Returns and error if this process fails.
-func Decode(in, out interface{}) error {
-	// TODO: wire in context.
-	_, r := observability.NewReporter(context.Background(), reportDecode)
-	err := obsDecode(in, out)
+func Decode(ctx context.Context, in, out interface{}) error {
+	_, r := observability.NewReporter(ctx, reportDecode)
+	err := obsDecode(ctx, in, out)
 	if err != nil {
 		r.Error()
 	} else {
@@ -25,7 +24,7 @@ func Decode(in, out interface{}) error {
 	return err
 }
 
-func obsDecode(in, out interface{}) error {
+func obsDecode(ctx context.Context, in, out interface{}) error {
 	if in == nil {
 		return nil
 	}
@@ -68,10 +67,9 @@ func obsDecode(in, out interface{}) error {
 // Encode attempts to xml.Marshal `in` into bytes. Encode will inspect `in`
 // and returns `in` unmodified if it is detected that `in` is already a []byte;
 // Or xml.Marshal errors.
-func Encode(in interface{}) ([]byte, error) {
-	// TODO: wire in context.
-	_, r := observability.NewReporter(context.Background(), reportEncode)
-	b, err := obsEncode(in)
+func Encode(ctx context.Context, in interface{}) ([]byte, error) {
+	_, r := observability.NewReporter(ctx, reportEncode)
+	b, err := obsEncode(ctx, in)
 	if err != nil {
 		r.Error()
 	} else {
@@ -80,7 +78,7 @@ func Encode(in interface{}) ([]byte, error) {
 	return b, err
 }
 
-func obsEncode(in interface{}) ([]byte, error) {
+func obsEncode(ctx context.Context, in interface{}) ([]byte, error) {
 	if b, ok := in.([]byte); ok {
 		// check to see if it is a pre-encoded byte string.
 		if len(b) > 0 && b[0] == byte('"') {
diff --git a/pkg/cloudevents/event_data.go b/pkg/cloudevents/event_data.go
index 9631057..9e4cb9d 100644
--- a/pkg/cloudevents/event_data.go
+++ b/pkg/cloudevents/event_data.go
@@ -1,6 +1,7 @@
 package cloudevents
 
 import (
+	"context"
 	"encoding/base64"
 	"errors"
 	"fmt"
@@ -13,7 +14,7 @@ import (
 
 // SetData implements EventWriter.SetData
 func (e *Event) SetData(obj interface{}) error {
-	data, err := datacodec.Encode(e.DataMediaType(), obj)
+	data, err := datacodec.Encode(context.Background(), e.DataMediaType(), obj)
 	if err != nil {
 		return err
 	}
@@ -94,5 +95,5 @@ func (e Event) DataAs(data interface{}) error { // TODO: Clean this function up
 	if err != nil {
 		return err
 	}
-	return datacodec.Decode(mediaType, obj, data)
+	return datacodec.Decode(context.Background(), mediaType, obj, data)
 }
diff --git a/pkg/cloudevents/transport/amqp/codec.go b/pkg/cloudevents/transport/amqp/codec.go
index 9e8d88e..4498318 100644
--- a/pkg/cloudevents/transport/amqp/codec.go
+++ b/pkg/cloudevents/transport/amqp/codec.go
@@ -1,6 +1,7 @@
 package amqp
 
 import (
+	"context"
 	"encoding/json"
 	"fmt"
 	"net/textproto"
@@ -24,51 +25,51 @@ const (
 
 var _ transport.Codec = (*Codec)(nil)
 
-func (c *Codec) Encode(e cloudevents.Event) (transport.Message, error) {
+func (c *Codec) Encode(ctx context.Context, e cloudevents.Event) (transport.Message, error) {
 	switch c.Encoding {
 	case Default:
 		fallthrough
 	case BinaryV02, BinaryV03:
-		return c.encodeBinary(e)
+		return c.encodeBinary(ctx, e)
 	case StructuredV02:
 		if c.v02 == nil {
 			c.v02 = &CodecV02{Encoding: c.Encoding}
 		}
-		return c.v02.Encode(e)
+		return c.v02.Encode(ctx, e)
 	case StructuredV03:
 		if c.v03 == nil {
 			c.v03 = &CodecV03{Encoding: c.Encoding}
 		}
-		return c.v03.Encode(e)
+		return c.v03.Encode(ctx, e)
 	default:
 		return nil, fmt.Errorf("unknown encoding: %d", c.Encoding)
 	}
 }
 
-func (c *Codec) Decode(msg transport.Message) (*cloudevents.Event, error) {
-	switch encoding := c.inspectEncoding(msg); encoding {
+func (c *Codec) Decode(ctx context.Context, msg transport.Message) (*cloudevents.Event, error) {
+	switch encoding := c.inspectEncoding(ctx, msg); encoding {
 	case BinaryV02:
 		event := cloudevents.New(cloudevents.CloudEventsVersionV02)
-		return c.decodeBinary(msg, &event)
+		return c.decodeBinary(ctx, msg, &event)
 	case BinaryV03:
 		event := cloudevents.New(cloudevents.CloudEventsVersionV03)
-		return c.decodeBinary(msg, &event)
+		return c.decodeBinary(ctx, msg, &event)
 	case StructuredV02:
 		if c.v02 == nil {
 			c.v02 = &CodecV02{Encoding: encoding}
 		}
-		return c.v02.Decode(msg)
+		return c.v02.Decode(ctx, msg)
 	case StructuredV03:
 		if c.v03 == nil {
 			c.v03 = &CodecV03{Encoding: encoding}
 		}
-		return c.v03.Decode(msg)
+		return c.v03.Decode(ctx, msg)
 	default:
 		return nil, transport.NewErrMessageEncodingUnknown("wrapper", TransportName)
 	}
 }
 
-func (c Codec) encodeBinary(e cloudevents.Event) (transport.Message, error) {
+func (c Codec) encodeBinary(ctx context.Context, e cloudevents.Event) (transport.Message, error) {
 	headers, err := c.toHeaders(e)
 	if err != nil {
 		return nil, err
@@ -137,7 +138,7 @@ func (c Codec) toHeaders(e cloudevents.Event) (map[string]interface{}, error) {
 	return h, nil
 }
 
-func (c Codec) decodeBinary(msg transport.Message, event *cloudevents.Event) (*cloudevents.Event, error) {
+func (c Codec) decodeBinary(ctx context.Context, msg transport.Message, event *cloudevents.Event) (*cloudevents.Event, error) {
 	m, ok := msg.(*Message)
 	if !ok {
 		return nil, fmt.Errorf("failed to convert transport.Message to amqp.Message")
@@ -272,12 +273,12 @@ func (c Codec) fromHeaders(h map[string]interface{}, event *cloudevents.Event) e
 	return nil
 }
 
-func (c *Codec) inspectEncoding(msg transport.Message) Encoding {
+func (c *Codec) inspectEncoding(ctx context.Context, msg transport.Message) Encoding {
 	if c.v02 == nil {
 		c.v02 = &CodecV02{Encoding: c.Encoding}
 	}
 	// Try v0.2.
-	encoding := c.v02.inspectEncoding(msg)
+	encoding := c.v02.inspectEncoding(ctx, msg)
 	if encoding != Unknown {
 		return encoding
 	}
@@ -286,7 +287,7 @@ func (c *Codec) inspectEncoding(msg transport.Message) Encoding {
 		c.v03 = &CodecV03{Encoding: c.Encoding}
 	}
 	// Try v0.3.
-	encoding = c.v03.inspectEncoding(msg)
+	encoding = c.v03.inspectEncoding(ctx, msg)
 	if encoding != Unknown {
 		return encoding
 	}
diff --git a/pkg/cloudevents/transport/amqp/codec_structured.go b/pkg/cloudevents/transport/amqp/codec_structured.go
index 754c540..c1395f6 100644
--- a/pkg/cloudevents/transport/amqp/codec_structured.go
+++ b/pkg/cloudevents/transport/amqp/codec_structured.go
@@ -1,6 +1,7 @@
 package amqp
 
 import (
+	"context"
 	"encoding/json"
 	"fmt"
 
@@ -14,7 +15,7 @@ type CodecStructured struct {
 	Encoding Encoding
 }
 
-func (v CodecStructured) encodeStructured(e cloudevents.Event) (transport.Message, error) {
+func (v CodecStructured) encodeStructured(ctx context.Context, e cloudevents.Event) (transport.Message, error) {
 	body, err := json.Marshal(e)
 	if err != nil {
 		return nil, err
@@ -28,7 +29,7 @@ func (v CodecStructured) encodeStructured(e cloudevents.Event) (transport.Messag
 	return msg, nil
 }
 
-func (v CodecStructured) decodeStructured(version string, msg transport.Message) (*cloudevents.Event, error) {
+func (v CodecStructured) decodeStructured(ctx context.Context, version string, msg transport.Message) (*cloudevents.Event, error) {
 	m, ok := msg.(*Message)
 	if !ok {
 		return nil, fmt.Errorf("failed to convert transport.Message to amqp.Message")
diff --git a/pkg/cloudevents/transport/amqp/codec_v02.go b/pkg/cloudevents/transport/amqp/codec_v02.go
index e0ae9e3..5bd864a 100644
--- a/pkg/cloudevents/transport/amqp/codec_v02.go
+++ b/pkg/cloudevents/transport/amqp/codec_v02.go
@@ -1,6 +1,7 @@
 package amqp
 
 import (
+	"context"
 	"fmt"
 
 	"github.com/cloudevents/sdk-go/pkg/cloudevents"
@@ -15,28 +16,28 @@ type CodecV02 struct {
 
 var _ transport.Codec = (*CodecV02)(nil)
 
-func (v CodecV02) Encode(e cloudevents.Event) (transport.Message, error) {
+func (v CodecV02) Encode(ctx context.Context, e cloudevents.Event) (transport.Message, error) {
 	switch v.Encoding {
 	case Default:
 		fallthrough
 	case StructuredV02:
-		return v.encodeStructured(e)
+		return v.encodeStructured(ctx, e)
 	default:
 		return nil, fmt.Errorf("unknown encoding: %d", v.Encoding)
 	}
 }
 
-func (v CodecV02) Decode(msg transport.Message) (*cloudevents.Event, error) {
+func (v CodecV02) Decode(ctx context.Context, msg transport.Message) (*cloudevents.Event, error) {
 	// only structured is supported as of v0.2
-	switch v.inspectEncoding(msg) {
+	switch v.inspectEncoding(ctx, msg) {
 	case StructuredV02:
-		return v.decodeStructured(cloudevents.CloudEventsVersionV02, msg)
+		return v.decodeStructured(ctx, cloudevents.CloudEventsVersionV02, msg)
 	default:
 		return nil, transport.NewErrMessageEncodingUnknown("v02", TransportName)
 	}
 }
 
-func (v CodecV02) inspectEncoding(msg transport.Message) Encoding {
+func (v CodecV02) inspectEncoding(ctx context.Context, msg transport.Message) Encoding {
 	version := msg.CloudEventsVersion()
 	if version != cloudevents.CloudEventsVersionV02 {
 		return Unknown
diff --git a/pkg/cloudevents/transport/amqp/codec_v03.go b/pkg/cloudevents/transport/amqp/codec_v03.go
index 49bc860..605c38a 100644
--- a/pkg/cloudevents/transport/amqp/codec_v03.go
+++ b/pkg/cloudevents/transport/amqp/codec_v03.go
@@ -1,6 +1,7 @@
 package amqp
 
 import (
+	"context"
 	"fmt"
 
 	"github.com/cloudevents/sdk-go/pkg/cloudevents"
@@ -15,28 +16,28 @@ type CodecV03 struct {
 
 var _ transport.Codec = (*CodecV03)(nil)
 
-func (v CodecV03) Encode(e cloudevents.Event) (transport.Message, error) {
+func (v CodecV03) Encode(ctx context.Context, e cloudevents.Event) (transport.Message, error) {
 	switch v.Encoding {
 	case Default:
 		fallthrough
 	case StructuredV03:
-		return v.encodeStructured(e)
+		return v.encodeStructured(ctx, e)
 	default:
 		return nil, fmt.Errorf("unknown encoding: %d", v.Encoding)
 	}
 }
 
-func (v CodecV03) Decode(msg transport.Message) (*cloudevents.Event, error) {
+func (v CodecV03) Decode(ctx context.Context, msg transport.Message) (*cloudevents.Event, error) {
 	// only structured is supported as of v0.3
-	switch v.inspectEncoding(msg) {
+	switch v.inspectEncoding(ctx, msg) {
 	case StructuredV03:
-		return v.decodeStructured(cloudevents.CloudEventsVersionV03, msg)
+		return v.decodeStructured(ctx, cloudevents.CloudEventsVersionV03, msg)
 	default:
 		return nil, transport.NewErrMessageEncodingUnknown("v03", TransportName)
 	}
 }
 
-func (v CodecV03) inspectEncoding(msg transport.Message) Encoding {
+func (v CodecV03) inspectEncoding(ctx context.Context, msg transport.Message) Encoding {
 	version := msg.CloudEventsVersion()
 	if version != cloudevents.CloudEventsVersionV03 {
 		return Unknown
diff --git a/pkg/cloudevents/transport/amqp/transport.go b/pkg/cloudevents/transport/amqp/transport.go
index ce1a731..ffc5a38 100644
--- a/pkg/cloudevents/transport/amqp/transport.go
+++ b/pkg/cloudevents/transport/amqp/transport.go
@@ -112,7 +112,7 @@ func (t *Transport) Send(ctx context.Context, event cloudevents.Event) (*cloudev
 		return nil, fmt.Errorf("unknown encoding set on transport: %d", t.Encoding)
 	}
 
-	msg, err := t.codec.Encode(event)
+	msg, err := t.codec.Encode(ctx, event)
 	if err != nil {
 		return nil, err
 	}
@@ -178,7 +178,7 @@ func (t *Transport) StartReceiver(ctx context.Context) error {
 				ContentType:           msg.Properties.ContentType,
 				ApplicationProperties: msg.ApplicationProperties,
 			}
-			event, err := t.codec.Decode(m)
+			event, err := t.codec.Decode(ctx, m)
 			if err != nil {
 				logger.Errorw("failed to decode message", zap.Error(err)) // TODO: create an error channel to pass this up
 			} else {
diff --git a/pkg/cloudevents/transport/codec.go b/pkg/cloudevents/transport/codec.go
index d0648a2..091064c 100644
--- a/pkg/cloudevents/transport/codec.go
+++ b/pkg/cloudevents/transport/codec.go
@@ -1,6 +1,7 @@
 package transport
 
 import (
+	"context"
 	"fmt"
 
 	"github.com/cloudevents/sdk-go/pkg/cloudevents"
@@ -9,8 +10,8 @@ import (
 // Codec is the interface for transport codecs to convert between transport
 // specific payloads and the Message interface.
 type Codec interface {
-	Encode(cloudevents.Event) (Message, error)
-	Decode(Message) (*cloudevents.Event, error)
+	Encode(context.Context, cloudevents.Event) (Message, error)
+	Decode(context.Context, Message) (*cloudevents.Event, error)
 }
 
 // ErrMessageEncodingUnknown is an error produced when the encoding for an incoming
diff --git a/pkg/cloudevents/transport/http/codec.go b/pkg/cloudevents/transport/http/codec.go
index 9337ca4..6774111 100644
--- a/pkg/cloudevents/transport/http/codec.go
+++ b/pkg/cloudevents/transport/http/codec.go
@@ -1,6 +1,7 @@
 package http
 
 import (
+	"context"
 	"fmt"
 
 	"github.com/cloudevents/sdk-go/pkg/cloudevents"
@@ -26,11 +27,11 @@ type Codec struct {
 var _ transport.Codec = (*Codec)(nil)
 
 // Encode encodes the provided event into a transport message.
-func (c *Codec) Encode(e cloudevents.Event) (transport.Message, error) {
+func (c *Codec) Encode(ctx context.Context, e cloudevents.Event) (transport.Message, error) {
 	encoding := c.Encoding
 
 	if encoding == Default && c.DefaultEncodingSelectionFn != nil {
-		encoding = c.DefaultEncodingSelectionFn(e)
+		encoding = c.DefaultEncodingSelectionFn(ctx, e)
 	}
 
 	switch encoding {
@@ -42,34 +43,34 @@ func (c *Codec) Encode(e cloudevents.Event) (transport.Message, error) {
 		if c.v01 == nil {
 			c.v01 = &CodecV01{Encoding: encoding}
 		}
-		return c.v01.Encode(e)
+		return c.v01.Encode(ctx, e)
 	case BinaryV02:
 		fallthrough
 	case StructuredV02:
 		if c.v02 == nil {
 			c.v02 = &CodecV02{Encoding: encoding}
 		}
-		return c.v02.Encode(e)
+		return c.v02.Encode(ctx, e)
 	case BinaryV03:
 		fallthrough
 	case StructuredV03:
 		if c.v03 == nil {
 			c.v03 = &CodecV03{Encoding: encoding}
 		}
-		return c.v03.Encode(e)
+		return c.v03.Encode(ctx, e)
 	default:
 		return nil, fmt.Errorf("unknown encoding: %s", encoding)
 	}
 }
 
 // Decode converts a provided transport message into an Event, or error.
-func (c *Codec) Decode(msg transport.Message) (*cloudevents.Event, error) {
-	switch c.inspectEncoding(msg) {
+func (c *Codec) Decode(ctx context.Context, msg transport.Message) (*cloudevents.Event, error) {
+	switch c.inspectEncoding(ctx, msg) {
 	case BinaryV01, StructuredV01:
 		if c.v01 == nil {
 			c.v01 = &CodecV01{Encoding: c.Encoding}
 		}
-		if event, err := c.v01.Decode(msg); err != nil {
+		if event, err := c.v01.Decode(ctx, msg); err != nil {
 			return nil, err
 		} else {
 			return c.convertEvent(event), nil
@@ -79,7 +80,7 @@ func (c *Codec) Decode(msg transport.Message) (*cloudevents.Event, error) {
 		if c.v02 == nil {
 			c.v02 = &CodecV02{Encoding: c.Encoding}
 		}
-		if event, err := c.v02.Decode(msg); err != nil {
+		if event, err := c.v02.Decode(ctx, msg); err != nil {
 			return nil, err
 		} else {
 			return c.convertEvent(event), nil
@@ -89,7 +90,7 @@ func (c *Codec) Decode(msg transport.Message) (*cloudevents.Event, error) {
 		if c.v03 == nil {
 			c.v03 = &CodecV03{Encoding: c.Encoding}
 		}
-		if event, err := c.v03.Decode(msg); err != nil {
+		if event, err := c.v03.Decode(ctx, msg); err != nil {
 			return nil, err
 		} else {
 			return c.convertEvent(event), nil
@@ -113,8 +114,8 @@ func (c *Codec) convertEvent(event *cloudevents.Event) *cloudevents.Event {
 		if c.v01 == nil {
 			c.v01 = &CodecV01{Encoding: c.Encoding}
 		}
-		ctx := event.Context.AsV01()
-		event.Context = ctx
+		ca := event.Context.AsV01()
+		event.Context = ca
 		return event
 	case BinaryV02:
 		fallthrough
@@ -122,8 +123,8 @@ func (c *Codec) convertEvent(event *cloudevents.Event) *cloudevents.Event {
 		if c.v02 == nil {
 			c.v02 = &CodecV02{Encoding: c.Encoding}
 		}
-		ctx := event.Context.AsV02()
-		event.Context = ctx
+		ca := event.Context.AsV02()
+		event.Context = ca
 		return event
 	case BinaryV03:
 		fallthrough
@@ -133,21 +134,21 @@ func (c *Codec) convertEvent(event *cloudevents.Event) *cloudevents.Event {
 		if c.v03 == nil {
 			c.v03 = &CodecV03{Encoding: c.Encoding}
 		}
-		ctx := event.Context.AsV03()
-		event.Context = ctx
+		ca := event.Context.AsV03()
+		event.Context = ca
 		return event
 	default:
 		return nil
 	}
 }
 
-func (c *Codec) inspectEncoding(msg transport.Message) Encoding {
+func (c *Codec) inspectEncoding(ctx context.Context, msg transport.Message) Encoding {
 	// TODO: there should be a better way to make the version codecs on demand.
 	if c.v01 == nil {
 		c.v01 = &CodecV01{Encoding: c.Encoding}
 	}
 	// Try v0.1 first.
-	encoding := c.v01.inspectEncoding(msg)
+	encoding := c.v01.inspectEncoding(ctx, msg)
 	if encoding != Unknown {
 		return encoding
 	}
@@ -156,7 +157,7 @@ func (c *Codec) inspectEncoding(msg transport.Message) Encoding {
 		c.v02 = &CodecV02{Encoding: c.Encoding}
 	}
 	// Try v0.2.
-	encoding = c.v02.inspectEncoding(msg)
+	encoding = c.v02.inspectEncoding(ctx, msg)
 	if encoding != Unknown {
 		return encoding
 	}
@@ -165,7 +166,7 @@ func (c *Codec) inspectEncoding(msg transport.Message) Encoding {
 		c.v03 = &CodecV03{Encoding: c.Encoding}
 	}
 	// Try v0.3.
-	encoding = c.v03.inspectEncoding(msg)
+	encoding = c.v03.inspectEncoding(ctx, msg)
 	if encoding != Unknown {
 		return encoding
 	}
diff --git a/pkg/cloudevents/transport/http/codec_structured.go b/pkg/cloudevents/transport/http/codec_structured.go
index 96d79fd..098cb5a 100644
--- a/pkg/cloudevents/transport/http/codec_structured.go
+++ b/pkg/cloudevents/transport/http/codec_structured.go
@@ -1,6 +1,7 @@
 package http
 
 import (
+	"context"
 	"encoding/json"
 	"fmt"
 	"net/http"
@@ -15,7 +16,7 @@ type CodecStructured struct {
 	Encoding Encoding
 }
 
-func (v CodecStructured) encodeStructured(e cloudevents.Event) (transport.Message, error) {
+func (v CodecStructured) encodeStructured(ctx context.Context, e cloudevents.Event) (transport.Message, error) {
 	header := http.Header{}
 	header.Set("Content-Type", cloudevents.ApplicationCloudEventsJSON)
 
@@ -32,7 +33,7 @@ func (v CodecStructured) encodeStructured(e cloudevents.Event) (transport.Messag
 	return msg, nil
 }
 
-func (v CodecStructured) decodeStructured(version string, msg transport.Message) (*cloudevents.Event, error) {
+func (v CodecStructured) decodeStructured(ctx context.Context, version string, msg transport.Message) (*cloudevents.Event, error) {
 	m, ok := msg.(*Message)
 	if !ok {
 		return nil, fmt.Errorf("failed to convert transport.Message to http.Message")
diff --git a/pkg/cloudevents/transport/http/codec_v01.go b/pkg/cloudevents/transport/http/codec_v01.go
index b640e3e..e414c09 100644
--- a/pkg/cloudevents/transport/http/codec_v01.go
+++ b/pkg/cloudevents/transport/http/codec_v01.go
@@ -25,10 +25,10 @@ type CodecV01 struct {
 var _ transport.Codec = (*CodecV01)(nil)
 
 // Encode implements Codec.Encode
-func (v CodecV01) Encode(e cloudevents.Event) (transport.Message, error) {
+func (v CodecV01) Encode(ctx context.Context, e cloudevents.Event) (transport.Message, error) {
 	// TODO: wire context
 	_, r := observability.NewReporter(context.Background(), CodecObserved{o: reportEncode, c: v.Encoding.Codec()})
-	m, err := v.obsEncode(e)
+	m, err := v.obsEncode(ctx, e)
 	if err != nil {
 		r.Error()
 	} else {
@@ -37,24 +37,24 @@ func (v CodecV01) Encode(e cloudevents.Event) (transport.Message, error) {
 	return m, err
 }
 
-func (v CodecV01) obsEncode(e cloudevents.Event) (transport.Message, error) {
+func (v CodecV01) obsEncode(ctx context.Context, e cloudevents.Event) (transport.Message, error) {
 	switch v.Encoding {
 	case Default:
 		fallthrough
 	case BinaryV01:
-		return v.encodeBinary(e)
+		return v.encodeBinary(ctx, e)
 	case StructuredV01:
-		return v.encodeStructured(e)
+		return v.encodeStructured(ctx, e)
 	default:
 		return nil, fmt.Errorf("unknown encoding: %d", v.Encoding)
 	}
 }
 
 // Decode implements Codec.Decode
-func (v CodecV01) Decode(msg transport.Message) (*cloudevents.Event, error) {
+func (v CodecV01) Decode(ctx context.Context, msg transport.Message) (*cloudevents.Event, error) {
 	// TODO: wire context
-	_, r := observability.NewReporter(context.Background(), CodecObserved{o: reportDecode, c: v.inspectEncoding(msg).Codec()}) // TODO: inspectEncoding is not free.
-	e, err := v.obsDecode(msg)
+	_, r := observability.NewReporter(ctx, CodecObserved{o: reportDecode, c: v.inspectEncoding(ctx, msg).Codec()}) // TODO: inspectEncoding is not free.
+	e, err := v.obsDecode(ctx, msg)
 	if err != nil {
 		r.Error()
 	} else {
@@ -63,18 +63,18 @@ func (v CodecV01) Decode(msg transport.Message) (*cloudevents.Event, error) {
 	return e, err
 }
 
-func (v CodecV01) obsDecode(msg transport.Message) (*cloudevents.Event, error) {
-	switch v.inspectEncoding(msg) {
+func (v CodecV01) obsDecode(ctx context.Context, msg transport.Message) (*cloudevents.Event, error) {
+	switch v.inspectEncoding(ctx, msg) {
 	case BinaryV01:
-		return v.decodeBinary(msg)
+		return v.decodeBinary(ctx, msg)
 	case StructuredV01:
-		return v.decodeStructured(cloudevents.CloudEventsVersionV01, msg)
+		return v.decodeStructured(ctx, cloudevents.CloudEventsVersionV01, msg)
 	default:
 		return nil, transport.NewErrMessageEncodingUnknown("v01", TransportName)
 	}
 }
 
-func (v CodecV01) encodeBinary(e cloudevents.Event) (transport.Message, error) {
+func (v CodecV01) encodeBinary(ctx context.Context, e cloudevents.Event) (transport.Message, error) {
 	header, err := v.toHeaders(e.Context.AsV01())
 	if err != nil {
 		return nil, err
@@ -131,12 +131,12 @@ func (v CodecV01) toHeaders(ec *cloudevents.EventContextV01) (http.Header, error
 	return h, nil
 }
 
-func (v CodecV01) decodeBinary(msg transport.Message) (*cloudevents.Event, error) {
+func (v CodecV01) decodeBinary(ctx context.Context, msg transport.Message) (*cloudevents.Event, error) {
 	m, ok := msg.(*Message)
 	if !ok {
 		return nil, fmt.Errorf("failed to convert transport.Message to http.Message")
 	}
-	ctx, err := v.fromHeaders(m.Header)
+	ca, err := v.fromHeaders(m.Header)
 	if err != nil {
 		return nil, err
 	}
@@ -145,7 +145,7 @@ func (v CodecV01) decodeBinary(msg transport.Message) (*cloudevents.Event, error
 		body = m.Body
 	}
 	return &cloudevents.Event{
-		Context:     &ctx,
+		Context:     &ca,
 		Data:        body,
 		DataEncoded: body != nil,
 	}, nil
@@ -204,7 +204,7 @@ func (v CodecV01) fromHeaders(h http.Header) (cloudevents.EventContextV01, error
 	return ec, nil
 }
 
-func (v CodecV01) inspectEncoding(msg transport.Message) Encoding {
+func (v CodecV01) inspectEncoding(ctx context.Context, msg transport.Message) Encoding {
 	version := msg.CloudEventsVersion()
 	if version != cloudevents.CloudEventsVersionV01 {
 		return Unknown
diff --git a/pkg/cloudevents/transport/http/codec_v02.go b/pkg/cloudevents/transport/http/codec_v02.go
index 0cfd872..939e602 100644
--- a/pkg/cloudevents/transport/http/codec_v02.go
+++ b/pkg/cloudevents/transport/http/codec_v02.go
@@ -25,10 +25,10 @@ type CodecV02 struct {
 var _ transport.Codec = (*CodecV02)(nil)
 
 // Encode implements Codec.Encode
-func (v CodecV02) Encode(e cloudevents.Event) (transport.Message, error) {
+func (v CodecV02) Encode(ctx context.Context, e cloudevents.Event) (transport.Message, error) {
 	// TODO: wire context
-	_, r := observability.NewReporter(context.Background(), CodecObserved{o: reportEncode, c: v.Encoding.Codec()})
-	m, err := v.obsEncode(e)
+	_, r := observability.NewReporter(ctx, CodecObserved{o: reportEncode, c: v.Encoding.Codec()})
+	m, err := v.obsEncode(ctx, e)
 	if err != nil {
 		r.Error()
 	} else {
@@ -37,24 +37,24 @@ func (v CodecV02) Encode(e cloudevents.Event) (transport.Message, error) {
 	return m, err
 }
 
-func (v CodecV02) obsEncode(e cloudevents.Event) (transport.Message, error) {
+func (v CodecV02) obsEncode(ctx context.Context, e cloudevents.Event) (transport.Message, error) {
 	switch v.Encoding {
 	case Default:
 		fallthrough
 	case BinaryV02:
-		return v.encodeBinary(e)
+		return v.encodeBinary(ctx, e)
 	case StructuredV02:
-		return v.encodeStructured(e)
+		return v.encodeStructured(ctx, e)
 	default:
 		return nil, fmt.Errorf("unknown encoding: %d", v.Encoding)
 	}
 }
 
 // Decode implements Codec.Decode
-func (v CodecV02) Decode(msg transport.Message) (*cloudevents.Event, error) {
+func (v CodecV02) Decode(ctx context.Context, msg transport.Message) (*cloudevents.Event, error) {
 	// TODO: wire context
-	_, r := observability.NewReporter(context.Background(), CodecObserved{o: reportDecode, c: v.inspectEncoding(msg).Codec()}) // TODO: inspectEncoding is not free.
-	e, err := v.obsDecode(msg)
+	_, r := observability.NewReporter(ctx, CodecObserved{o: reportDecode, c: v.inspectEncoding(ctx, msg).Codec()}) // TODO: inspectEncoding is not free.
+	e, err := v.obsDecode(ctx, msg)
 	if err != nil {
 		r.Error()
 	} else {
@@ -63,18 +63,18 @@ func (v CodecV02) Decode(msg transport.Message) (*cloudevents.Event, error) {
 	return e, err
 }
 
-func (v CodecV02) obsDecode(msg transport.Message) (*cloudevents.Event, error) {
-	switch v.inspectEncoding(msg) {
+func (v CodecV02) obsDecode(ctx context.Context, msg transport.Message) (*cloudevents.Event, error) {
+	switch v.inspectEncoding(ctx, msg) {
 	case BinaryV02:
-		return v.decodeBinary(msg)
+		return v.decodeBinary(ctx, msg)
 	case StructuredV02:
-		return v.decodeStructured(cloudevents.CloudEventsVersionV02, msg)
+		return v.decodeStructured(ctx, cloudevents.CloudEventsVersionV02, msg)
 	default:
 		return nil, transport.NewErrMessageEncodingUnknown("v02", TransportName)
 	}
 }
 
-func (v CodecV02) encodeBinary(e cloudevents.Event) (transport.Message, error) {
+func (v CodecV02) encodeBinary(ctx context.Context, e cloudevents.Event) (transport.Message, error) {
 	header, err := v.toHeaders(e.Context.AsV02())
 	if err != nil {
 		return nil, err
@@ -135,12 +135,12 @@ func (v CodecV02) toHeaders(ec *cloudevents.EventContextV02) (http.Header, error
 	return h, nil
 }
 
-func (v CodecV02) decodeBinary(msg transport.Message) (*cloudevents.Event, error) {
+func (v CodecV02) decodeBinary(ctx context.Context, msg transport.Message) (*cloudevents.Event, error) {
 	m, ok := msg.(*Message)
 	if !ok {
 		return nil, fmt.Errorf("failed to convert transport.Message to http.Message")
 	}
-	ctx, err := v.fromHeaders(m.Header)
+	ca, err := v.fromHeaders(m.Header)
 	if err != nil {
 		return nil, err
 	}
@@ -149,7 +149,7 @@ func (v CodecV02) decodeBinary(msg transport.Message) (*cloudevents.Event, error
 		body = m.Body
 	}
 	return &cloudevents.Event{
-		Context:     &ctx,
+		Context:     &ca,
 		Data:        body,
 		DataEncoded: body != nil,
 	}, nil
@@ -235,7 +235,7 @@ func (v CodecV02) fromHeaders(h http.Header) (cloudevents.EventContextV02, error
 	return ec, nil
 }
 
-func (v CodecV02) inspectEncoding(msg transport.Message) Encoding {
+func (v CodecV02) inspectEncoding(ctx context.Context, msg transport.Message) Encoding {
 	version := msg.CloudEventsVersion()
 	if version != cloudevents.CloudEventsVersionV02 {
 		return Unknown
diff --git a/pkg/cloudevents/transport/http/codec_v03.go b/pkg/cloudevents/transport/http/codec_v03.go
index 706f81b..4364119 100644
--- a/pkg/cloudevents/transport/http/codec_v03.go
+++ b/pkg/cloudevents/transport/http/codec_v03.go
@@ -25,10 +25,10 @@ type CodecV03 struct {
 var _ transport.Codec = (*CodecV03)(nil)
 
 // Encode implements Codec.Encode
-func (v CodecV03) Encode(e cloudevents.Event) (transport.Message, error) {
+func (v CodecV03) Encode(ctx context.Context, e cloudevents.Event) (transport.Message, error) {
 	// TODO: wire context
-	_, r := observability.NewReporter(context.Background(), CodecObserved{o: reportEncode, c: v.Encoding.Codec()})
-	m, err := v.obsEncode(e)
+	_, r := observability.NewReporter(ctx, CodecObserved{o: reportEncode, c: v.Encoding.Codec()})
+	m, err := v.obsEncode(ctx, e)
 	if err != nil {
 		r.Error()
 	} else {
@@ -37,14 +37,14 @@ func (v CodecV03) Encode(e cloudevents.Event) (transport.Message, error) {
 	return m, err
 }
 
-func (v CodecV03) obsEncode(e cloudevents.Event) (transport.Message, error) {
+func (v CodecV03) obsEncode(ctx context.Context, e cloudevents.Event) (transport.Message, error) {
 	switch v.Encoding {
 	case Default:
 		fallthrough
 	case BinaryV03:
-		return v.encodeBinary(e)
+		return v.encodeBinary(ctx, e)
 	case StructuredV03:
-		return v.encodeStructured(e)
+		return v.encodeStructured(ctx, e)
 	case BatchedV03:
 		return nil, fmt.Errorf("not implemented")
 	default:
@@ -53,10 +53,10 @@ func (v CodecV03) obsEncode(e cloudevents.Event) (transport.Message, error) {
 }
 
 // Decode implements Codec.Decode
-func (v CodecV03) Decode(msg transport.Message) (*cloudevents.Event, error) {
+func (v CodecV03) Decode(ctx context.Context, msg transport.Message) (*cloudevents.Event, error) {
 	// TODO: wire context
-	_, r := observability.NewReporter(context.Background(), CodecObserved{o: reportDecode, c: v.inspectEncoding(msg).Codec()}) // TODO: inspectEncoding is not free.
-	e, err := v.obsDecode(msg)
+	_, r := observability.NewReporter(ctx, CodecObserved{o: reportDecode, c: v.inspectEncoding(ctx, msg).Codec()}) // TODO: inspectEncoding is not free.
+	e, err := v.obsDecode(ctx, msg)
 	if err != nil {
 		r.Error()
 	} else {
@@ -65,12 +65,12 @@ func (v CodecV03) Decode(msg transport.Message) (*cloudevents.Event, error) {
 	return e, err
 }
 
-func (v CodecV03) obsDecode(msg transport.Message) (*cloudevents.Event, error) {
-	switch v.inspectEncoding(msg) {
+func (v CodecV03) obsDecode(ctx context.Context, msg transport.Message) (*cloudevents.Event, error) {
+	switch v.inspectEncoding(ctx, msg) {
 	case BinaryV03:
-		return v.decodeBinary(msg)
+		return v.decodeBinary(ctx, msg)
 	case StructuredV03:
-		return v.decodeStructured(cloudevents.CloudEventsVersionV03, msg)
+		return v.decodeStructured(ctx, cloudevents.CloudEventsVersionV03, msg)
 	case BatchedV03:
 		return nil, fmt.Errorf("not implemented")
 	default:
@@ -78,7 +78,7 @@ func (v CodecV03) obsDecode(msg transport.Message) (*cloudevents.Event, error) {
 	}
 }
 
-func (v CodecV03) encodeBinary(e cloudevents.Event) (transport.Message, error) {
+func (v CodecV03) encodeBinary(ctx context.Context, e cloudevents.Event) (transport.Message, error) {
 	header, err := v.toHeaders(e.Context.AsV03())
 	if err != nil {
 		return nil, err
@@ -147,12 +147,12 @@ func (v CodecV03) toHeaders(ec *cloudevents.EventContextV03) (http.Header, error
 	return h, nil
 }
 
-func (v CodecV03) decodeBinary(msg transport.Message) (*cloudevents.Event, error) {
+func (v CodecV03) decodeBinary(ctx context.Context, msg transport.Message) (*cloudevents.Event, error) {
 	m, ok := msg.(*Message)
 	if !ok {
 		return nil, fmt.Errorf("failed to convert transport.Message to http.Message")
 	}
-	ctx, err := v.fromHeaders(m.Header)
+	ca, err := v.fromHeaders(m.Header)
 	if err != nil {
 		return nil, err
 	}
@@ -161,7 +161,7 @@ func (v CodecV03) decodeBinary(msg transport.Message) (*cloudevents.Event, error
 		body = m.Body
 	}
 	return &cloudevents.Event{
-		Context:     &ctx,
+		Context:     &ca,
 		Data:        body,
 		DataEncoded: body != nil,
 	}, nil
@@ -259,7 +259,7 @@ func (v CodecV03) fromHeaders(h http.Header) (cloudevents.EventContextV03, error
 	return ec, nil
 }
 
-func (v CodecV03) inspectEncoding(msg transport.Message) Encoding {
+func (v CodecV03) inspectEncoding(ctx context.Context, msg transport.Message) Encoding {
 	version := msg.CloudEventsVersion()
 	if version != cloudevents.CloudEventsVersionV03 {
 		return Unknown
diff --git a/pkg/cloudevents/transport/http/encoding.go b/pkg/cloudevents/transport/http/encoding.go
index 8b4a360..58c03b6 100644
--- a/pkg/cloudevents/transport/http/encoding.go
+++ b/pkg/cloudevents/transport/http/encoding.go
@@ -1,11 +1,15 @@
 package http
 
-import "github.com/cloudevents/sdk-go/pkg/cloudevents"
+import (
+	"context"
+
+	"github.com/cloudevents/sdk-go/pkg/cloudevents"
+)
 
 // Encoding to use for HTTP transport.
 type Encoding int32
 
-type EncodingSelector func(e cloudevents.Event) Encoding
+type EncodingSelector func(context.Context, cloudevents.Event) Encoding
 
 const (
 	// Default
@@ -30,7 +34,7 @@ const (
 
 // DefaultBinaryEncodingSelectionStrategy implements a selection process for
 // which binary encoding to use based on spec version of the event.
-func DefaultBinaryEncodingSelectionStrategy(e cloudevents.Event) Encoding {
+func DefaultBinaryEncodingSelectionStrategy(ctx context.Context, e cloudevents.Event) Encoding {
 	switch e.SpecVersion() {
 	case cloudevents.CloudEventsVersionV01:
 		return BinaryV01
@@ -45,7 +49,7 @@ func DefaultBinaryEncodingSelectionStrategy(e cloudevents.Event) Encoding {
 
 // DefaultStructuredEncodingSelectionStrategy implements a selection process
 // for which structured encoding to use based on spec version of the event.
-func DefaultStructuredEncodingSelectionStrategy(e cloudevents.Event) Encoding {
+func DefaultStructuredEncodingSelectionStrategy(ctx context.Context, e cloudevents.Event) Encoding {
 	switch e.SpecVersion() {
 	case cloudevents.CloudEventsVersionV01:
 		return StructuredV01
diff --git a/pkg/cloudevents/transport/http/transport.go b/pkg/cloudevents/transport/http/transport.go
index 74330a4..752abbe 100644
--- a/pkg/cloudevents/transport/http/transport.go
+++ b/pkg/cloudevents/transport/http/transport.go
@@ -193,7 +193,7 @@ func (t *Transport) obsSend(ctx context.Context, event cloudevents.Event) (*clou
 		return nil, fmt.Errorf("unknown encoding set on transport: %d", t.Encoding)
 	}
 
-	msg, err := t.codec.Encode(event)
+	msg, err := t.codec.Encode(ctx, event)
 	if err != nil {
 		return nil, err
 	}
@@ -243,7 +243,7 @@ func (t *Transport) MessageToEvent(ctx context.Context, msg *Message) (*cloudeve
 			err = transport.NewErrTransportMessageConversion("http", fmt.Sprintf("unknown encoding set on transport: %d", t.Encoding), true)
 			logger.Error("failed to load codec", zap.Error(err))
 		} else {
-			event, err = t.codec.Decode(msg)
+			event, err = t.codec.Decode(ctx, msg)
 		}
 	} else {
 		err = transport.NewErrTransportMessageConversion("http", "cloudevents version unknown", false)
@@ -490,7 +490,7 @@ func (t *Transport) obsInvokeReceiver(ctx context.Context, event cloudevents.Eve
 
 		if eventResp.Event != nil {
 			if t.loadCodec(ctx) {
-				if m, err := t.codec.Encode(*eventResp.Event); err != nil {
+				if m, err := t.codec.Encode(ctx, *eventResp.Event); err != nil {
 					logger.Errorw("failed to encode response from receiver fn", zap.Error(err))
 				} else if msg, ok := m.(*Message); ok {
 					resp.Message = *msg
diff --git a/pkg/cloudevents/transport/nats/codec.go b/pkg/cloudevents/transport/nats/codec.go
index 6c0a659..dab140e 100644
--- a/pkg/cloudevents/transport/nats/codec.go
+++ b/pkg/cloudevents/transport/nats/codec.go
@@ -1,6 +1,7 @@
 package nats
 
 import (
+	"context"
 	"fmt"
 
 	"github.com/cloudevents/sdk-go/pkg/cloudevents"
@@ -16,7 +17,7 @@ type Codec struct {
 
 var _ transport.Codec = (*Codec)(nil)
 
-func (c *Codec) Encode(e cloudevents.Event) (transport.Message, error) {
+func (c *Codec) Encode(ctx context.Context, e cloudevents.Event) (transport.Message, error) {
 	switch c.Encoding {
 	case Default:
 		fallthrough
@@ -24,32 +25,56 @@ func (c *Codec) Encode(e cloudevents.Event) (transport.Message, error) {
 		if c.v02 == nil {
 			c.v02 = &CodecV02{Encoding: c.Encoding}
 		}
-		return c.v02.Encode(e)
+		return c.v02.Encode(ctx, e)
 	case StructuredV03:
 		if c.v03 == nil {
 			c.v03 = &CodecV03{Encoding: c.Encoding}
 		}
-		return c.v03.Encode(e)
+		return c.v03.Encode(ctx, e)
 	default:
 		return nil, fmt.Errorf("unknown encoding: %d", c.Encoding)
 	}
 }
 
-func (c *Codec) Decode(msg transport.Message) (*cloudevents.Event, error) {
-	switch c.Encoding {
+func (c *Codec) Decode(ctx context.Context, msg transport.Message) (*cloudevents.Event, error) {
+	switch c.inspectEncoding(ctx, msg) {
 	case Default:
 		fallthrough
 	case StructuredV02:
 		if c.v02 == nil {
 			c.v02 = &CodecV02{Encoding: c.Encoding}
 		}
-		return c.v02.Decode(msg)
+		return c.v02.Decode(ctx, msg)
 	case StructuredV03:
 		if c.v03 == nil {
 			c.v03 = &CodecV03{Encoding: c.Encoding}
 		}
-		return c.v03.Decode(msg)
+		return c.v03.Decode(ctx, msg)
 	default:
 		return nil, transport.NewErrMessageEncodingUnknown("wrapper", TransportName)
 	}
 }
+
+func (c *Codec) inspectEncoding(ctx context.Context, msg transport.Message) Encoding {
+	// TODO: there should be a better way to make the version codecs on demand.
+	if c.v02 == nil {
+		c.v02 = &CodecV02{Encoding: c.Encoding}
+	}
+	// Try v0.2.
+	encoding := c.v02.inspectEncoding(ctx, msg)
+	if encoding != Unknown {
+		return encoding
+	}
+
+	if c.v03 == nil {
+		c.v03 = &CodecV03{Encoding: c.Encoding}
+	}
+	// Try v0.3.
+	encoding = c.v03.inspectEncoding(ctx, msg)
+	if encoding != Unknown {
+		return encoding
+	}
+
+	// We do not understand the message encoding.
+	return Unknown
+}
diff --git a/pkg/cloudevents/transport/nats/codec_structured.go b/pkg/cloudevents/transport/nats/codec_structured.go
index 8753747..d20e4d8 100644
--- a/pkg/cloudevents/transport/nats/codec_structured.go
+++ b/pkg/cloudevents/transport/nats/codec_structured.go
@@ -1,6 +1,7 @@
 package nats
 
 import (
+	"context"
 	"encoding/json"
 	"fmt"
 
@@ -14,7 +15,7 @@ type CodecStructured struct {
 	Encoding Encoding
 }
 
-func (v CodecStructured) encodeStructured(e cloudevents.Event) (transport.Message, error) {
+func (v CodecStructured) encodeStructured(ctx context.Context, e cloudevents.Event) (transport.Message, error) {
 	body, err := json.Marshal(e)
 	if err != nil {
 		return nil, err
@@ -27,7 +28,7 @@ func (v CodecStructured) encodeStructured(e cloudevents.Event) (transport.Messag
 	return msg, nil
 }
 
-func (v CodecStructured) decodeStructured(version string, msg transport.Message) (*cloudevents.Event, error) {
+func (v CodecStructured) decodeStructured(ctx context.Context, version string, msg transport.Message) (*cloudevents.Event, error) {
 	m, ok := msg.(*Message)
 	if !ok {
 		return nil, fmt.Errorf("failed to convert transport.Message to nats.Message")
diff --git a/pkg/cloudevents/transport/nats/codec_v02.go b/pkg/cloudevents/transport/nats/codec_v02.go
index 4fd2fcb..920c19a 100644
--- a/pkg/cloudevents/transport/nats/codec_v02.go
+++ b/pkg/cloudevents/transport/nats/codec_v02.go
@@ -1,6 +1,7 @@
 package nats
 
 import (
+	"context"
 	"fmt"
 
 	"github.com/cloudevents/sdk-go/pkg/cloudevents"
@@ -15,28 +16,28 @@ type CodecV02 struct {
 
 var _ transport.Codec = (*CodecV02)(nil)
 
-func (v CodecV02) Encode(e cloudevents.Event) (transport.Message, error) {
+func (v CodecV02) Encode(ctx context.Context, e cloudevents.Event) (transport.Message, error) {
 	switch v.Encoding {
 	case Default:
 		fallthrough
 	case StructuredV02:
-		return v.encodeStructured(e)
+		return v.encodeStructured(ctx, e)
 	default:
 		return nil, fmt.Errorf("unknown encoding: %d", v.Encoding)
 	}
 }
 
-func (v CodecV02) Decode(msg transport.Message) (*cloudevents.Event, error) {
+func (v CodecV02) Decode(ctx context.Context, msg transport.Message) (*cloudevents.Event, error) {
 	// only structured is supported as of v0.2
-	switch v.inspectEncoding(msg) {
+	switch v.inspectEncoding(ctx, msg) {
 	case StructuredV02:
-		return v.decodeStructured(cloudevents.CloudEventsVersionV02, msg)
+		return v.decodeStructured(ctx, cloudevents.CloudEventsVersionV02, msg)
 	default:
 		return nil, transport.NewErrMessageEncodingUnknown("v02", TransportName)
 	}
 }
 
-func (v CodecV02) inspectEncoding(msg transport.Message) Encoding {
+func (v CodecV02) inspectEncoding(ctx context.Context, msg transport.Message) Encoding {
 	version := msg.CloudEventsVersion()
 	if version != cloudevents.CloudEventsVersionV02 {
 		return Unknown
diff --git a/pkg/cloudevents/transport/nats/codec_v03.go b/pkg/cloudevents/transport/nats/codec_v03.go
index 2ec23f4..dee8388 100644
--- a/pkg/cloudevents/transport/nats/codec_v03.go
+++ b/pkg/cloudevents/transport/nats/codec_v03.go
@@ -1,6 +1,7 @@
 package nats
 
 import (
+	"context"
 	"fmt"
 
 	"github.com/cloudevents/sdk-go/pkg/cloudevents"
@@ -15,28 +16,28 @@ type CodecV03 struct {
 
 var _ transport.Codec = (*CodecV03)(nil)
 
-func (v CodecV03) Encode(e cloudevents.Event) (transport.Message, error) {
+func (v CodecV03) Encode(ctx context.Context, e cloudevents.Event) (transport.Message, error) {
 	switch v.Encoding {
 	case Default:
 		fallthrough
 	case StructuredV03:
-		return v.encodeStructured(e)
+		return v.encodeStructured(ctx, e)
 	default:
 		return nil, fmt.Errorf("unknown encoding: %d", v.Encoding)
 	}
 }
 
-func (v CodecV03) Decode(msg transport.Message) (*cloudevents.Event, error) {
+func (v CodecV03) Decode(ctx context.Context, msg transport.Message) (*cloudevents.Event, error) {
 	// only structured is supported as of v0.3
-	switch v.inspectEncoding(msg) {
+	switch v.inspectEncoding(ctx, msg) {
 	case StructuredV03:
-		return v.decodeStructured(cloudevents.CloudEventsVersionV03, msg)
+		return v.decodeStructured(ctx, cloudevents.CloudEventsVersionV03, msg)
 	default:
 		return nil, transport.NewErrMessageEncodingUnknown("v03", TransportName)
 	}
 }
 
-func (v CodecV03) inspectEncoding(msg transport.Message) Encoding {
+func (v CodecV03) inspectEncoding(ctx context.Context, msg transport.Message) Encoding {
 	version := msg.CloudEventsVersion()
 	if version != cloudevents.CloudEventsVersionV03 {
 		return Unknown
diff --git a/pkg/cloudevents/transport/nats/transport.go b/pkg/cloudevents/transport/nats/transport.go
index f0a9880..7964c37 100644
--- a/pkg/cloudevents/transport/nats/transport.go
+++ b/pkg/cloudevents/transport/nats/transport.go
@@ -83,7 +83,7 @@ func (t *Transport) Send(ctx context.Context, event cloudevents.Event) (*cloudev
 		return nil, fmt.Errorf("unknown encoding set on transport: %d", t.Encoding)
 	}
 
-	msg, err := t.codec.Encode(event)
+	msg, err := t.codec.Encode(ctx, event)
 	if err != nil {
 		return nil, err
 	}
@@ -137,7 +137,7 @@ func (t *Transport) StartReceiver(ctx context.Context) error {
 		msg := &Message{
 			Body: m.Data,
 		}
-		event, err := t.codec.Decode(msg)
+		event, err := t.codec.Decode(ctx, msg)
 		// If codec returns and error, try with the converter if it is set.
 		if err != nil && t.HasConverter() {
 			event, err = t.Converter.Convert(ctx, msg, err)
diff --git a/pkg/cloudevents/transport/pubsub/codec.go b/pkg/cloudevents/transport/pubsub/codec.go
index fb81d6a..98b3d77 100644
--- a/pkg/cloudevents/transport/pubsub/codec.go
+++ b/pkg/cloudevents/transport/pubsub/codec.go
@@ -1,6 +1,7 @@
 package pubsub
 
 import (
+	"context"
 	"encoding/json"
 	"fmt"
 	"strings"
@@ -25,38 +26,38 @@ const (
 
 var _ transport.Codec = (*Codec)(nil)
 
-func (c *Codec) Encode(e cloudevents.Event) (transport.Message, error) {
+func (c *Codec) Encode(ctx context.Context, e cloudevents.Event) (transport.Message, error) {
 	switch c.Encoding {
 	case Default:
 		fallthrough
 	case BinaryV03:
-		return c.encodeBinary(e)
+		return c.encodeBinary(ctx, e)
 	case StructuredV03:
 		if c.v03 == nil {
 			c.v03 = &CodecV03{Encoding: c.Encoding}
 		}
-		return c.v03.Encode(e)
+		return c.v03.Encode(ctx, e)
 	default:
 		return nil, transport.NewErrMessageEncodingUnknown("wrapper", TransportName)
 	}
 }
 
-func (c *Codec) Decode(msg transport.Message) (*cloudevents.Event, error) {
-	switch encoding := c.inspectEncoding(msg); encoding {
+func (c *Codec) Decode(ctx context.Context, msg transport.Message) (*cloudevents.Event, error) {
+	switch encoding := c.inspectEncoding(ctx, msg); encoding {
 	case BinaryV03:
 		event := cloudevents.New(cloudevents.CloudEventsVersionV03)
-		return c.decodeBinary(msg, &event)
+		return c.decodeBinary(ctx, msg, &event)
 	case StructuredV03:
 		if c.v03 == nil {
 			c.v03 = &CodecV03{Encoding: encoding}
 		}
-		return c.v03.Decode(msg)
+		return c.v03.Decode(ctx, msg)
 	default:
 		return nil, fmt.Errorf("unknown encoding: %s", encoding)
 	}
 }
 
-func (c Codec) encodeBinary(e cloudevents.Event) (transport.Message, error) {
+func (c Codec) encodeBinary(ctx context.Context, e cloudevents.Event) (transport.Message, error) {
 	attributes, err := c.toAttributes(e)
 	if err != nil {
 		return nil, err
@@ -127,7 +128,7 @@ func (c Codec) toAttributes(e cloudevents.Event) (map[string]string, error) {
 	return a, nil
 }
 
-func (c Codec) decodeBinary(msg transport.Message, event *cloudevents.Event) (*cloudevents.Event, error) {
+func (c Codec) decodeBinary(ctx context.Context, msg transport.Message, event *cloudevents.Event) (*cloudevents.Event, error) {
 	m, ok := msg.(*Message)
 	if !ok {
 		return nil, fmt.Errorf("failed to convert transport.Message to pubsub.Message")
@@ -266,12 +267,12 @@ func (c Codec) fromAttributes(a map[string]string, event *cloudevents.Event) err
 	return nil
 }
 
-func (c *Codec) inspectEncoding(msg transport.Message) Encoding {
+func (c *Codec) inspectEncoding(ctx context.Context, msg transport.Message) Encoding {
 	if c.v03 == nil {
 		c.v03 = &CodecV03{Encoding: c.Encoding}
 	}
 	// Try v0.3.
-	encoding := c.v03.inspectEncoding(msg)
+	encoding := c.v03.inspectEncoding(ctx, msg)
 	if encoding != Unknown {
 		return encoding
 	}
diff --git a/pkg/cloudevents/transport/pubsub/codec_structured.go b/pkg/cloudevents/transport/pubsub/codec_structured.go
index 2e42856..9ecafe6 100644
--- a/pkg/cloudevents/transport/pubsub/codec_structured.go
+++ b/pkg/cloudevents/transport/pubsub/codec_structured.go
@@ -1,6 +1,7 @@
 package pubsub
 
 import (
+	"context"
 	"encoding/json"
 	"fmt"
 
@@ -14,7 +15,7 @@ type CodecStructured struct {
 	Encoding Encoding
 }
 
-func (v CodecStructured) encodeStructured(e cloudevents.Event) (transport.Message, error) {
+func (v CodecStructured) encodeStructured(ctx context.Context, e cloudevents.Event) (transport.Message, error) {
 	data, err := json.Marshal(e)
 	if err != nil {
 		return nil, err
@@ -28,7 +29,7 @@ func (v CodecStructured) encodeStructured(e cloudevents.Event) (transport.Messag
 	return msg, nil
 }
 
-func (v CodecStructured) decodeStructured(version string, msg transport.Message) (*cloudevents.Event, error) {
+func (v CodecStructured) decodeStructured(ctx context.Context, version string, msg transport.Message) (*cloudevents.Event, error) {
 	m, ok := msg.(*Message)
 	if !ok {
 		return nil, fmt.Errorf("failed to convert transport.Message to pubsub.Message")
diff --git a/pkg/cloudevents/transport/pubsub/codec_v03.go b/pkg/cloudevents/transport/pubsub/codec_v03.go
index b6d85fe..78f506d 100644
--- a/pkg/cloudevents/transport/pubsub/codec_v03.go
+++ b/pkg/cloudevents/transport/pubsub/codec_v03.go
@@ -1,6 +1,7 @@
 package pubsub
 
 import (
+	"context"
 	"fmt"
 
 	"github.com/cloudevents/sdk-go/pkg/cloudevents"
@@ -19,28 +20,28 @@ type CodecV03 struct {
 
 var _ transport.Codec = (*CodecV03)(nil)
 
-func (v CodecV03) Encode(e cloudevents.Event) (transport.Message, error) {
+func (v CodecV03) Encode(ctx context.Context, e cloudevents.Event) (transport.Message, error) {
 	switch v.Encoding {
 	case Default:
 		fallthrough
 	case StructuredV03:
-		return v.encodeStructured(e)
+		return v.encodeStructured(ctx, e)
 	default:
 		return nil, fmt.Errorf("unknown encoding: %d", v.Encoding)
 	}
 }
 
-func (v CodecV03) Decode(msg transport.Message) (*cloudevents.Event, error) {
+func (v CodecV03) Decode(ctx context.Context, msg transport.Message) (*cloudevents.Event, error) {
 	// only structured is supported as of v0.3
-	switch v.inspectEncoding(msg) {
+	switch v.inspectEncoding(ctx, msg) {
 	case StructuredV03:
-		return v.decodeStructured(cloudevents.CloudEventsVersionV03, msg)
+		return v.decodeStructured(ctx, cloudevents.CloudEventsVersionV03, msg)
 	default:
 		return nil, transport.NewErrMessageEncodingUnknown("v03", TransportName)
 	}
 }
 
-func (v CodecV03) inspectEncoding(msg transport.Message) Encoding {
+func (v CodecV03) inspectEncoding(ctx context.Context, msg transport.Message) Encoding {
 	version := msg.CloudEventsVersion()
 	if version != cloudevents.CloudEventsVersionV03 {
 		return Unknown
diff --git a/pkg/cloudevents/transport/pubsub/encoding.go b/pkg/cloudevents/transport/pubsub/encoding.go
index 1f07b55..7f6ef14 100644
--- a/pkg/cloudevents/transport/pubsub/encoding.go
+++ b/pkg/cloudevents/transport/pubsub/encoding.go
@@ -1,11 +1,15 @@
 package pubsub
 
-import "github.com/cloudevents/sdk-go/pkg/cloudevents"
+import (
+	"context"
+
+	"github.com/cloudevents/sdk-go/pkg/cloudevents"
+)
 
 // Encoding to use for pubsub transport.
 type Encoding int32
 
-type EncodingSelector func(e cloudevents.Event) Encoding
+type EncodingSelector func(context.Context, cloudevents.Event) Encoding
 
 const (
 	// Default allows pubsub transport implementation to pick.
@@ -20,7 +24,7 @@ const (
 
 // DefaultBinaryEncodingSelectionStrategy implements a selection process for
 // which binary encoding to use based on spec version of the event.
-func DefaultBinaryEncodingSelectionStrategy(e cloudevents.Event) Encoding {
+func DefaultBinaryEncodingSelectionStrategy(ctx context.Context, e cloudevents.Event) Encoding {
 	switch e.SpecVersion() {
 	case cloudevents.CloudEventsVersionV01, cloudevents.CloudEventsVersionV02, cloudevents.CloudEventsVersionV03:
 		return BinaryV03
@@ -31,7 +35,7 @@ func DefaultBinaryEncodingSelectionStrategy(e cloudevents.Event) Encoding {
 
 // DefaultStructuredEncodingSelectionStrategy implements a selection process
 // for which structured encoding to use based on spec version of the event.
-func DefaultStructuredEncodingSelectionStrategy(e cloudevents.Event) Encoding {
+func DefaultStructuredEncodingSelectionStrategy(ctx context.Context, e cloudevents.Event) Encoding {
 	switch e.SpecVersion() {
 	case cloudevents.CloudEventsVersionV01, cloudevents.CloudEventsVersionV02, cloudevents.CloudEventsVersionV03:
 		return StructuredV03
diff --git a/pkg/cloudevents/transport/pubsub/transport.go b/pkg/cloudevents/transport/pubsub/transport.go
index 626800d..f249aa4 100644
--- a/pkg/cloudevents/transport/pubsub/transport.go
+++ b/pkg/cloudevents/transport/pubsub/transport.go
@@ -223,7 +223,7 @@ func (t *Transport) Send(ctx context.Context, event cloudevents.Event) (*cloudev
 		return nil, fmt.Errorf("unknown encoding set on transport: %d", t.Encoding)
 	}
 
-	msg, err := t.codec.Encode(event)
+	msg, err := t.codec.Encode(ctx, event)
 	if err != nil {
 		return nil, err
 	}
@@ -287,7 +287,7 @@ func (t *Transport) StartReceiver(ctx context.Context) error {
 			Attributes: m.Attributes,
 			Data:       m.Data,
 		}
-		event, err := t.codec.Decode(msg)
+		event, err := t.codec.Decode(ctx, msg)
 		// If codec returns and error, try with the converter if it is set.
 		if err != nil && t.HasConverter() {
 			event, err = t.Converter.Convert(ctx, msg, err)


Test patch:
diff --git a/pkg/cloudevents/datacodec/codec_test.go b/pkg/cloudevents/datacodec/codec_test.go
index 49611de..0267802 100644
--- a/pkg/cloudevents/datacodec/codec_test.go
+++ b/pkg/cloudevents/datacodec/codec_test.go
@@ -1,6 +1,7 @@
 package datacodec_test
 
 import (
+	"context"
 	"fmt"
 	"strings"
 	"testing"
@@ -46,7 +47,7 @@ func TestCodecDecode(t *testing.T) {
 		"custom content type": {
 			contentType: "unit/testing",
 			in:          []byte("Hello, Testing"),
-			decoder: func(in, out interface{}) error {
+			decoder: func(ctx context.Context, in, out interface{}) error {
 				if b, ok := in.([]byte); ok {
 					if s, k := out.(*map[string]string); k {
 						if (*s) == nil {
@@ -66,7 +67,7 @@ func TestCodecDecode(t *testing.T) {
 		"custom content type error": {
 			contentType: "unit/testing",
 			in:          []byte("Hello, Testing"),
-			decoder: func(in, out interface{}) error {
+			decoder: func(ctx context.Context, in, out interface{}) error {
 				return fmt.Errorf("expecting unit test error")
 			},
 			wantErr: "expecting unit test error",
@@ -81,7 +82,7 @@ func TestCodecDecode(t *testing.T) {
 
 			got, _ := types.Allocate(tc.want)
 
-			err := datacodec.Decode(tc.contentType, tc.in, got)
+			err := datacodec.Decode(context.TODO(), tc.contentType, tc.in, got)
 
 			if tc.wantErr != "" || err != nil {
 				if diff := cmp.Diff(tc.wantErr, err.Error()); diff != "" {
@@ -133,7 +134,7 @@ func TestCodecEncode(t *testing.T) {
 				"Hello,",
 				"Testing",
 			},
-			encoder: func(in interface{}) ([]byte, error) {
+			encoder: func(ctx context.Context, in interface{}) ([]byte, error) {
 				if s, ok := in.([]string); ok {
 					sb := strings.Builder{}
 					for _, v := range s {
@@ -151,7 +152,7 @@ func TestCodecEncode(t *testing.T) {
 		"custom content type error": {
 			contentType: "unit/testing",
 			in:          []byte("Hello, Testing"),
-			encoder: func(in interface{}) ([]byte, error) {
+			encoder: func(ctx context.Context, in interface{}) ([]byte, error) {
 				return nil, fmt.Errorf("expecting unit test error")
 			},
 			wantErr: "expecting unit test error",
@@ -164,7 +165,7 @@ func TestCodecEncode(t *testing.T) {
 				datacodec.AddEncoder(tc.contentType, tc.encoder)
 			}
 
-			got, err := datacodec.Encode(tc.contentType, tc.in)
+			got, err := datacodec.Encode(context.TODO(), tc.contentType, tc.in)
 
 			if tc.wantErr != "" || err != nil {
 				if diff := cmp.Diff(tc.wantErr, err.Error()); diff != "" {
diff --git a/pkg/cloudevents/datacodec/json/data_test.go b/pkg/cloudevents/datacodec/json/data_test.go
index 957ecc3..d9541e3 100644
--- a/pkg/cloudevents/datacodec/json/data_test.go
+++ b/pkg/cloudevents/datacodec/json/data_test.go
@@ -1,6 +1,7 @@
 package json_test
 
 import (
+	"context"
 	"encoding/json"
 	"fmt"
 	"testing"
@@ -125,7 +126,7 @@ func TestCodecDecode(t *testing.T) {
 		t.Run(n, func(t *testing.T) {
 			got, _ := types.Allocate(tc.want)
 
-			err := cej.Decode(tc.in, got)
+			err := cej.Decode(context.TODO(), tc.in, got)
 			if tc.wantErr != "" || err != nil {
 				var gotErr string
 				if err != nil {
@@ -229,7 +230,7 @@ func TestCodecEncode(t *testing.T) {
 	}
 	for n, tc := range testCases {
 		t.Run(n, func(t *testing.T) {
-			got, err := cej.Encode(tc.in)
+			got, err := cej.Encode(context.TODO(), tc.in)
 			if tc.wantErr != "" || err != nil {
 				var gotErr string
 				if err != nil {
diff --git a/pkg/cloudevents/datacodec/xml/data_test.go b/pkg/cloudevents/datacodec/xml/data_test.go
index 700bfd1..14dfee2 100644
--- a/pkg/cloudevents/datacodec/xml/data_test.go
+++ b/pkg/cloudevents/datacodec/xml/data_test.go
@@ -1,6 +1,7 @@
 package xml_test
 
 import (
+	"context"
 	"encoding/xml"
 	"fmt"
 	"strings"
@@ -103,7 +104,7 @@ func TestCodecDecode(t *testing.T) {
 
 			got, _ := types.Allocate(tc.want)
 
-			err := cex.Decode(tc.in, got)
+			err := cex.Decode(context.TODO(), tc.in, got)
 
 			if tc.wantErr != "" {
 				if err != nil {
diff --git a/pkg/cloudevents/transport/amqp/codec_test.go b/pkg/cloudevents/transport/amqp/codec_test.go
index 6a3bd48..f49ffa2 100644
--- a/pkg/cloudevents/transport/amqp/codec_test.go
+++ b/pkg/cloudevents/transport/amqp/codec_test.go
@@ -1,6 +1,7 @@
 package amqp_test
 
 import (
+	"context"
 	"encoding/json"
 	"fmt"
 	"net/url"
@@ -119,7 +120,7 @@ func TestCodecEncode(t *testing.T) {
 	for n, tc := range testCases {
 		t.Run(n, func(t *testing.T) {
 
-			got, err := tc.codec.Encode(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 != "" {
@@ -244,7 +245,7 @@ func TestCodecDecode(t *testing.T) {
 	for n, tc := range testCases {
 		t.Run(n, func(t *testing.T) {
 
-			got, err := tc.codec.Decode(tc.msg)
+			got, err := tc.codec.Decode(context.TODO(), tc.msg)
 
 			if tc.wantErr != nil || err != nil {
 				if diff := cmp.Diff(tc.wantErr, err); diff != "" {
@@ -341,7 +342,7 @@ func TestCodecRoundTrip(t *testing.T) {
 			n = fmt.Sprintf("%s, %s", encoding, n)
 			t.Run(n, func(t *testing.T) {
 
-				msg, err := tc.codec.Encode(tc.event)
+				msg, 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)
@@ -349,7 +350,7 @@ func TestCodecRoundTrip(t *testing.T) {
 					return
 				}
 
-				got, err := tc.codec.Decode(msg)
+				got, err := tc.codec.Decode(context.TODO(), msg)
 				if err != nil {
 					if diff := cmp.Diff(tc.wantErr, err); diff != "" {
 						t.Errorf("unexpected error (-want, +got) = %v", diff)
@@ -463,7 +464,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(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)
@@ -471,7 +472,7 @@ func TestCodecAsMiddleware(t *testing.T) {
 					return
 				}
 
-				midEvent, err := tc.codec.Decode(msg1)
+				midEvent, err := tc.codec.Decode(context.TODO(), msg1)
 				if err != nil {
 					if diff := cmp.Diff(tc.wantErr, err); diff != "" {
 						t.Errorf("unexpected error (-want, +got) = %v", diff)
@@ -479,7 +480,7 @@ func TestCodecAsMiddleware(t *testing.T) {
 					return
 				}
 
-				msg2, err := tc.codec.Encode(*midEvent)
+				msg2, err := tc.codec.Encode(context.TODO(), *midEvent)
 				if err != nil {
 					if diff := cmp.Diff(tc.wantErr, err); diff != "" {
 						t.Errorf("unexpected error (-want, +got) = %v", diff)
@@ -487,7 +488,7 @@ func TestCodecAsMiddleware(t *testing.T) {
 					return
 				}
 
-				got, err := tc.codec.Decode(msg2)
+				got, err := tc.codec.Decode(context.TODO(), msg2)
 				if err != nil {
 					if diff := cmp.Diff(tc.wantErr, err); diff != "" {
 						t.Errorf("unexpected error (-want, +got) = %v", diff)
diff --git a/pkg/cloudevents/transport/amqp/codec_v02_test.go b/pkg/cloudevents/transport/amqp/codec_v02_test.go
index ae591ec..ce881ef 100644
--- a/pkg/cloudevents/transport/amqp/codec_v02_test.go
+++ b/pkg/cloudevents/transport/amqp/codec_v02_test.go
@@ -1,6 +1,7 @@
 package amqp_test
 
 import (
+	"context"
 	"encoding/json"
 	"net/url"
 	"testing"
@@ -152,7 +153,7 @@ func TestCodecV02_Encode(t *testing.T) {
 	for n, tc := range testCases {
 		t.Run(n, func(t *testing.T) {
 
-			got, err := tc.codec.Encode(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 != "" {
@@ -256,7 +257,7 @@ func TestCodecV02_Decode(t *testing.T) {
 	for n, tc := range testCases {
 		t.Run(n, func(t *testing.T) {
 
-			got, err := tc.codec.Decode(tc.msg)
+			got, err := tc.codec.Decode(context.TODO(), tc.msg)
 
 			if tc.wantErr != nil || err != nil {
 				if diff := cmp.Diff(tc.wantErr, err); diff != "" {
diff --git a/pkg/cloudevents/transport/amqp/codec_v03_test.go b/pkg/cloudevents/transport/amqp/codec_v03_test.go
index e0a3f08..0b896de 100644
--- a/pkg/cloudevents/transport/amqp/codec_v03_test.go
+++ b/pkg/cloudevents/transport/amqp/codec_v03_test.go
@@ -1,6 +1,7 @@
 package amqp_test
 
 import (
+	"context"
 	"encoding/json"
 	"net/url"
 	"testing"
@@ -152,7 +153,7 @@ func TestCodecV03_Encode(t *testing.T) {
 	for n, tc := range testCases {
 		t.Run(n, func(t *testing.T) {
 
-			got, err := tc.codec.Encode(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 != "" {
@@ -256,7 +257,7 @@ func TestCodecV03_Decode(t *testing.T) {
 	for n, tc := range testCases {
 		t.Run(n, func(t *testing.T) {
 
-			got, err := tc.codec.Decode(tc.msg)
+			got, err := tc.codec.Decode(context.TODO(), tc.msg)
 
 			if tc.wantErr != nil || err != nil {
 				if diff := cmp.Diff(tc.wantErr, err); diff != "" {
diff --git a/pkg/cloudevents/transport/http/codec_test.go b/pkg/cloudevents/transport/http/codec_test.go
index 7eefc96..72bc208 100644
--- a/pkg/cloudevents/transport/http/codec_test.go
+++ b/pkg/cloudevents/transport/http/codec_test.go
@@ -1,6 +1,7 @@
 package http_test
 
 import (
+	"context"
 	"encoding/json"
 	"fmt"
 	nethttp "net/http"
@@ -53,7 +54,7 @@ func TestDefaultBinaryEncodingSelectionStrategy(t *testing.T) {
 	for n, tc := range testCases {
 		t.Run(n, func(t *testing.T) {
 
-			got := http.DefaultBinaryEncodingSelectionStrategy(tc.event)
+			got := http.DefaultBinaryEncodingSelectionStrategy(context.TODO(), tc.event)
 
 			if got != tc.want {
 				t.Errorf("unexpected selection want: %s, got: %s", tc.want, got)
@@ -97,7 +98,7 @@ func TestDefaultStructuredEncodingSelectionStrategy(t *testing.T) {
 	for n, tc := range testCases {
 		t.Run(n, func(t *testing.T) {
 
-			got := http.DefaultStructuredEncodingSelectionStrategy(tc.event)
+			got := http.DefaultStructuredEncodingSelectionStrategy(context.TODO(), tc.event)
 
 			if got != tc.want {
 				t.Errorf("unexpected selection want: %s, got: %s", tc.want, got)
@@ -396,7 +397,7 @@ func TestCodecEncode(t *testing.T) {
 	for n, tc := range testCases {
 		t.Run(n, func(t *testing.T) {
 
-			got, err := tc.codec.Encode(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 != "" {
@@ -682,7 +683,7 @@ func TestCodecDecode(t *testing.T) {
 	for n, tc := range testCases {
 		t.Run(n, func(t *testing.T) {
 
-			got, err := tc.codec.Decode(tc.msg)
+			got, err := tc.codec.Decode(context.TODO(), tc.msg)
 
 			if tc.wantErr != nil || err != nil {
 				if diff := cmp.Diff(tc.wantErr, err); diff != "" {
@@ -789,7 +790,7 @@ func TestCodecRoundTrip(t *testing.T) {
 			n = fmt.Sprintf("%s, %s", encoding, n)
 			t.Run(n, func(t *testing.T) {
 
-				msg, err := tc.codec.Encode(tc.event)
+				msg, 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)
@@ -797,7 +798,7 @@ func TestCodecRoundTrip(t *testing.T) {
 					return
 				}
 
-				got, err := tc.codec.Decode(msg)
+				got, err := tc.codec.Decode(context.TODO(), msg)
 				if err != nil {
 					if diff := cmp.Diff(tc.wantErr, err); diff != "" {
 						t.Errorf("unexpected error (-want, +got) = %v", diff)
@@ -915,7 +916,7 @@ func TestCodecAsMiddleware(t *testing.T) {
 				n = fmt.Sprintf("%s[%s],%s", encoding, contentType, n)
 				t.Run(n, func(t *testing.T) {
 
-					msg1, err := tc.codec.Encode(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)
@@ -923,7 +924,7 @@ func TestCodecAsMiddleware(t *testing.T) {
 						return
 					}
 
-					midEvent, err := tc.codec.Decode(msg1)
+					midEvent, err := tc.codec.Decode(context.TODO(), msg1)
 					if err != nil {
 						if diff := cmp.Diff(tc.wantErr, err); diff != "" {
 							t.Errorf("unexpected error (-want, +got) = %v", diff)
@@ -931,7 +932,7 @@ func TestCodecAsMiddleware(t *testing.T) {
 						return
 					}
 
-					msg2, err := tc.codec.Encode(*midEvent)
+					msg2, err := tc.codec.Encode(context.TODO(), *midEvent)
 					if err != nil {
 						if diff := cmp.Diff(tc.wantErr, err); diff != "" {
 							t.Errorf("unexpected error (-want, +got) = %v", diff)
@@ -939,7 +940,7 @@ func TestCodecAsMiddleware(t *testing.T) {
 						return
 					}
 
-					got, err := tc.codec.Decode(msg2)
+					got, err := tc.codec.Decode(context.TODO(), msg2)
 					if err != nil {
 						if diff := cmp.Diff(tc.wantErr, err); diff != "" {
 							t.Errorf("unexpected error (-want, +got) = %v", diff)
diff --git a/pkg/cloudevents/transport/http/codec_v01_test.go b/pkg/cloudevents/transport/http/codec_v01_test.go
index 2bb2d36..4808c85 100644
--- a/pkg/cloudevents/transport/http/codec_v01_test.go
+++ b/pkg/cloudevents/transport/http/codec_v01_test.go
@@ -1,6 +1,7 @@
 package http_test
 
 import (
+	"context"
 	"net/url"
 	"testing"
 	"time"
@@ -205,7 +206,7 @@ func TestCodecV01_Encode(t *testing.T) {
 	for n, tc := range testCases {
 		t.Run(n, func(t *testing.T) {
 
-			got, err := tc.codec.Encode(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 != "" {
@@ -395,7 +396,7 @@ func TestCodecV01_Decode(t *testing.T) {
 	for n, tc := range testCases {
 		t.Run(n, func(t *testing.T) {
 
-			got, err := tc.codec.Decode(tc.msg)
+			got, err := tc.codec.Decode(context.TODO(), tc.msg)
 
 			if tc.wantErr != nil || err != nil {
 				if diff := cmp.Diff(tc.wantErr, err); diff != "" {
diff --git a/pkg/cloudevents/transport/http/codec_v02_test.go b/pkg/cloudevents/transport/http/codec_v02_test.go
index ae608bf..5b641b8 100644
--- a/pkg/cloudevents/transport/http/codec_v02_test.go
+++ b/pkg/cloudevents/transport/http/codec_v02_test.go
@@ -1,6 +1,7 @@
 package http_test
 
 import (
+	"context"
 	"encoding/json"
 	"net/url"
 	"testing"
@@ -208,7 +209,7 @@ func TestCodecV02_Encode(t *testing.T) {
 	for n, tc := range testCases {
 		t.Run(n, func(t *testing.T) {
 
-			got, err := tc.codec.Encode(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 != "" {
@@ -403,7 +404,7 @@ func TestCodecV02_Decode(t *testing.T) {
 	for n, tc := range testCases {
 		t.Run(n, func(t *testing.T) {
 
-			got, err := tc.codec.Decode(tc.msg)
+			got, err := tc.codec.Decode(context.TODO(), tc.msg)
 
 			if tc.wantErr != nil || err != nil {
 				if diff := cmp.Diff(tc.wantErr, err); diff != "" {
diff --git a/pkg/cloudevents/transport/http/codec_v03_test.go b/pkg/cloudevents/transport/http/codec_v03_test.go
index fddd8ac..e77cd39 100644
--- a/pkg/cloudevents/transport/http/codec_v03_test.go
+++ b/pkg/cloudevents/transport/http/codec_v03_test.go
@@ -1,6 +1,7 @@
 package http_test
 
 import (
+	"context"
 	"encoding/json"
 	"net/url"
 	"testing"
@@ -305,7 +306,7 @@ func TestCodecV03_Encode(t *testing.T) {
 	for n, tc := range testCases {
 		t.Run(n, func(t *testing.T) {
 
-			got, err := tc.codec.Encode(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 != "" {
@@ -589,7 +590,7 @@ func TestCodecV03_Decode(t *testing.T) {
 	for n, tc := range testCases {
 		t.Run(n, func(t *testing.T) {
 
-			got, err := tc.codec.Decode(tc.msg)
+			got, err := tc.codec.Decode(context.TODO(), tc.msg)
 
 			if tc.wantErr != nil || err != nil {
 				if diff := cmp.Diff(tc.wantErr, err); diff != "" {
diff --git a/pkg/cloudevents/transport/http/options_test.go b/pkg/cloudevents/transport/http/options_test.go
index 4a928bb..4aaa02d 100644
--- a/pkg/cloudevents/transport/http/options_test.go
+++ b/pkg/cloudevents/transport/http/options_test.go
@@ -498,7 +498,7 @@ func TestWithEncoding(t *testing.T) {
 
 func TestWithDefaultEncodingSelector(t *testing.T) {
 
-	fn := func(e cloudevents.Event) Encoding {
+	fn := func(ctx context.Context, e cloudevents.Event) Encoding {
 		return Default
 	}
 
diff --git a/pkg/cloudevents/transport/nats/codec_test.go b/pkg/cloudevents/transport/nats/codec_test.go
index 6413d60..cc42aa2 100644
--- a/pkg/cloudevents/transport/nats/codec_test.go
+++ b/pkg/cloudevents/transport/nats/codec_test.go
@@ -1,6 +1,7 @@
 package nats_test
 
 import (
+	"context"
 	"encoding/json"
 	"fmt"
 	"net/url"
@@ -53,7 +54,7 @@ func TestCodecEncode(t *testing.T) {
 	for n, tc := range testCases {
 		t.Run(n, func(t *testing.T) {
 
-			got, err := tc.codec.Encode(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 != "" {
@@ -114,7 +115,7 @@ func TestCodecDecode(t *testing.T) {
 	for n, tc := range testCases {
 		t.Run(n, func(t *testing.T) {
 
-			got, err := tc.codec.Decode(tc.msg)
+			got, err := tc.codec.Decode(context.TODO(), tc.msg)
 
 			if tc.wantErr != nil || err != nil {
 				if diff := cmp.Diff(tc.wantErr, err); diff != "" {
@@ -211,7 +212,7 @@ func TestCodecRoundTrip(t *testing.T) {
 			n = fmt.Sprintf("%s, %s", encoding, n)
 			t.Run(n, func(t *testing.T) {
 
-				msg, err := tc.codec.Encode(tc.event)
+				msg, 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)
@@ -219,7 +220,7 @@ func TestCodecRoundTrip(t *testing.T) {
 					return
 				}
 
-				got, err := tc.codec.Decode(msg)
+				got, err := tc.codec.Decode(context.TODO(), msg)
 				if err != nil {
 					if diff := cmp.Diff(tc.wantErr, err); diff != "" {
 						t.Errorf("unexpected error (-want, +got) = %v", diff)
@@ -333,7 +334,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(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)
@@ -341,7 +342,7 @@ func TestCodecAsMiddleware(t *testing.T) {
 					return
 				}
 
-				midEvent, err := tc.codec.Decode(msg1)
+				midEvent, err := tc.codec.Decode(context.TODO(), msg1)
 				if err != nil {
 					if diff := cmp.Diff(tc.wantErr, err); diff != "" {
 						t.Errorf("unexpected error (-want, +got) = %v", diff)
@@ -349,7 +350,7 @@ func TestCodecAsMiddleware(t *testing.T) {
 					return
 				}
 
-				msg2, err := tc.codec.Encode(*midEvent)
+				msg2, err := tc.codec.Encode(context.TODO(), *midEvent)
 				if err != nil {
 					if diff := cmp.Diff(tc.wantErr, err); diff != "" {
 						t.Errorf("unexpected error (-want, +got) = %v", diff)
@@ -357,7 +358,7 @@ func TestCodecAsMiddleware(t *testing.T) {
 					return
 				}
 
-				got, err := tc.codec.Decode(msg2)
+				got, err := tc.codec.Decode(context.TODO(), msg2)
 				if err != nil {
 					if diff := cmp.Diff(tc.wantErr, err); diff != "" {
 						t.Errorf("unexpected error (-want, +got) = %v", diff)
diff --git a/pkg/cloudevents/transport/nats/codec_v02_test.go b/pkg/cloudevents/transport/nats/codec_v02_test.go
index b61809b..0c3145b 100644
--- a/pkg/cloudevents/transport/nats/codec_v02_test.go
+++ b/pkg/cloudevents/transport/nats/codec_v02_test.go
@@ -1,6 +1,7 @@
 package nats_test
 
 import (
+	"context"
 	"encoding/json"
 	"net/url"
 	"testing"
@@ -148,7 +149,7 @@ func TestCodecV02_Encode(t *testing.T) {
 	for n, tc := range testCases {
 		t.Run(n, func(t *testing.T) {
 
-			got, err := tc.codec.Encode(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 != "" {
@@ -250,7 +251,7 @@ func TestCodecV02_Decode(t *testing.T) {
 	for n, tc := range testCases {
 		t.Run(n, func(t *testing.T) {
 
-			got, err := tc.codec.Decode(tc.msg)
+			got, err := tc.codec.Decode(context.TODO(), tc.msg)
 
 			if tc.wantErr != nil || err != nil {
 				if diff := cmp.Diff(tc.wantErr, err); diff != "" {
diff --git a/pkg/cloudevents/transport/nats/codec_v03_test.go b/pkg/cloudevents/transport/nats/codec_v03_test.go
index e37cc83..7039169 100644
--- a/pkg/cloudevents/transport/nats/codec_v03_test.go
+++ b/pkg/cloudevents/transport/nats/codec_v03_test.go
@@ -1,6 +1,7 @@
 package nats_test
 
 import (
+	"context"
 	"encoding/json"
 	"net/url"
 	"testing"
@@ -148,7 +149,7 @@ func TestCodecV03_Encode(t *testing.T) {
 	for n, tc := range testCases {
 		t.Run(n, func(t *testing.T) {
 
-			got, err := tc.codec.Encode(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 != "" {
@@ -250,7 +251,7 @@ func TestCodecV03_Decode(t *testing.T) {
 	for n, tc := range testCases {
 		t.Run(n, func(t *testing.T) {
 
-			got, err := tc.codec.Decode(tc.msg)
+			got, err := tc.codec.Decode(context.TODO(), tc.msg)
 
 			if tc.wantErr != nil || err != nil {
 				if diff := cmp.Diff(tc.wantErr, err); diff != "" {
diff --git a/pkg/cloudevents/transport/pubsub/codec_test.go b/pkg/cloudevents/transport/pubsub/codec_test.go
index 5cc4053..20acaf4 100644
--- a/pkg/cloudevents/transport/pubsub/codec_test.go
+++ b/pkg/cloudevents/transport/pubsub/codec_test.go
@@ -1,6 +1,7 @@
 package pubsub_test
 
 import (
+	"context"
 	"encoding/json"
 	"fmt"
 	"net/url"
@@ -79,7 +80,7 @@ func TestCodecEncode(t *testing.T) {
 	for n, tc := range testCases {
 		t.Run(n, func(t *testing.T) {
 
-			got, err := tc.codec.Encode(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 != "" {
@@ -145,7 +146,7 @@ func TestCodecDecode(t *testing.T) {
 	for n, tc := range testCases {
 		t.Run(n, func(t *testing.T) {
 
-			got, err := tc.codec.Decode(tc.msg)
+			got, err := tc.codec.Decode(context.TODO(), tc.msg)
 
 			if tc.wantErr != nil || err != nil {
 				if diff := cmp.Diff(tc.wantErr, err); diff != "" {
@@ -242,7 +243,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(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)
@@ -250,7 +251,7 @@ func TestCodecAsMiddleware(t *testing.T) {
 					return
 				}
 
-				midEvent, err := tc.codec.Decode(msg1)
+				midEvent, err := tc.codec.Decode(context.TODO(), msg1)
 				if err != nil {
 					if diff := cmp.Diff(tc.wantErr, err); diff != "" {
 						t.Errorf("unexpected error (-want, +got) = %v", diff)
@@ -258,7 +259,7 @@ func TestCodecAsMiddleware(t *testing.T) {
 					return
 				}
 
-				msg2, err := tc.codec.Encode(*midEvent)
+				msg2, err := tc.codec.Encode(context.TODO(), *midEvent)
 				if err != nil {
 					if diff := cmp.Diff(tc.wantErr, err); diff != "" {
 						t.Errorf("unexpected error (-want, +got) = %v", diff)
@@ -266,7 +267,7 @@ func TestCodecAsMiddleware(t *testing.T) {
 					return
 				}
 
-				got, err := tc.codec.Decode(msg2)
+				got, err := tc.codec.Decode(context.TODO(), msg2)
 				if err != nil {
 					if diff := cmp.Diff(tc.wantErr, err); diff != "" {
 						t.Errorf("unexpected error (-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 ecb0902..993c180 100644
--- a/pkg/cloudevents/transport/pubsub/codec_v03_test.go
+++ b/pkg/cloudevents/transport/pubsub/codec_v03_test.go
@@ -1,6 +1,7 @@
 package pubsub_test
 
 import (
+	"context"
 	"encoding/json"
 	"net/url"
 	"testing"
@@ -160,7 +161,7 @@ func TestCodecV03_Encode(t *testing.T) {
 	for n, tc := range testCases {
 		t.Run(n, func(t *testing.T) {
 
-			got, err := tc.codec.Encode(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 != "" {
@@ -268,7 +269,7 @@ func TestCodecV03_Decode(t *testing.T) {
 	for n, tc := range testCases {
 		t.Run(n, func(t *testing.T) {
 
-			got, err := tc.codec.Decode(tc.msg)
+			got, err := tc.codec.Decode(context.TODO(), tc.msg)
 
 			if tc.wantErr != nil || err != nil {
 				if diff := cmp.Diff(tc.wantErr, err); diff != "" {