Repository: cloudevents/sdk-go
Pull request #356 (2020-03-06 19:17:00)

Move extensions to pkg.
Moving extensions out of `pkg/cloudevents`

Signed-off-by: Scott Nichols <snichols@vmware.com>


Patch:
diff --git a/cmd/samples/amqp/sender/main.go b/cmd/samples/amqp/sender/main.go
index 1549a5f..0df5a37 100644
--- a/cmd/samples/amqp/sender/main.go
+++ b/cmd/samples/amqp/sender/main.go
@@ -55,15 +55,15 @@ type Example struct {
 	Message  string `json:"message"`
 }
 
-func (d *Demo) Send(eventContext event.EventContext, i int) (context.Context, *event.Event, error) {
-	event := event.Event{
+func (d *Demo) Send(eventContext event.EventContext, i int) error {
+	e := event.Event{
 		Context: eventContext,
 		Data: &Example{
 			Sequence: i,
 			Message:  d.Message,
 		},
 	}
-	return d.Client.Send(context.Background(), event)
+	return d.Client.Send(context.Background(), e)
 }
 
 func main() {
@@ -99,7 +99,7 @@ func main() {
 			Source:          types.URLRef{URL: d.Source},
 			DataContentType: &contentType,
 		}.AsV03()
-		if _, _, err := d.Send(ctx, seq); err != nil {
+		if err := d.Send(ctx, seq); err != nil {
 			log.Fatalf("Failed to send: %v", err)
 		}
 		seq++
diff --git a/cmd/samples/complex/httptonats/main.go b/cmd/samples/complex/httptonats/main.go
index aef6615..65832da 100644
--- a/cmd/samples/complex/httptonats/main.go
+++ b/cmd/samples/complex/httptonats/main.go
@@ -53,7 +53,7 @@ func (r *Receiver) Receive(event event.Event) error {
 
 	fmt.Printf("forwarding...")
 
-	if _, _, err := r.Client.Send(context.Background(), event); err != nil {
+	if err := r.Client.Send(context.Background(), event); err != nil {
 		fmt.Printf("forwarding failed: %s", err.Error())
 	}
 
diff --git a/cmd/samples/complex/sender/main.go b/cmd/samples/complex/sender/main.go
index 1a3194c..be82603 100644
--- a/cmd/samples/complex/sender/main.go
+++ b/cmd/samples/complex/sender/main.go
@@ -64,8 +64,8 @@ type Example struct {
 	Message  string `json:"message"`
 }
 
-func (d *Demo) Send() (context.Context, *cloudevents.Event, error) {
-	event := cloudevents.Event{
+func (d *Demo) Send() error {
+	e := cloudevents.Event{
 		Context: d.context(),
 		Data: &Example{
 			Sequence: seq,
@@ -73,7 +73,7 @@ func (d *Demo) Send() (context.Context, *cloudevents.Event, error) {
 		},
 	}
 	seq++
-	return d.Client.Send(context.Background(), event)
+	return d.Client.Send(context.Background(), e)
 }
 
 func (d *Demo) context() cloudevents.EventContext {
@@ -163,7 +163,7 @@ func doDemo(t transport.Transport, eventType, message, contentType string, sourc
 		ContentType: contentType,
 	}
 	for i := 0; i < count; i++ {
-		if _, _, err := d.Send(); err != nil {
+		if err := d.Send(); err != nil {
 			return err
 		}
 		time.Sleep(delay)
diff --git a/cmd/samples/http/sender-with-custom-client/main.go b/cmd/samples/http/requester-with-custom-client/main.go
similarity index 98%
rename from cmd/samples/http/sender-with-custom-client/main.go
rename to cmd/samples/http/requester-with-custom-client/main.go
index 145832b..5914dbe 100644
--- a/cmd/samples/http/sender-with-custom-client/main.go
+++ b/cmd/samples/http/requester-with-custom-client/main.go
@@ -112,7 +112,7 @@ func _main(args []string, env envConfig) int {
 					},
 				}
 
-				if _, resp, err := c.Send(context.Background(), event); err != nil {
+				if resp, err := c.Request(context.Background(), event); err != nil {
 					log.Printf("failed to send: %v", err)
 				} else if resp != nil {
 					fmt.Printf("Response:\n%s\n", resp)
diff --git a/cmd/samples/http/sender/main.go b/cmd/samples/http/requester/main.go
similarity index 95%
rename from cmd/samples/http/sender/main.go
rename to cmd/samples/http/requester/main.go
index 768bf33..cf7d11f 100644
--- a/cmd/samples/http/sender/main.go
+++ b/cmd/samples/http/requester/main.go
@@ -38,7 +38,7 @@ type Example struct {
 }
 
 func _main(args []string, env envConfig) int {
-	source, err := url.Parse("https://github.com/cloudevents/sdk-go/cmd/samples/sender")
+	source, err := url.Parse("https://github.com/cloudevents/sdk-go/cmd/samples/requester")
 	if err != nil {
 		log.Printf("failed to parse source url, %v", err)
 		return 1
@@ -84,8 +84,8 @@ func _main(args []string, env envConfig) int {
 					},
 				}
 
-				if _, resp, err := c.Send(ctx, event); err != nil {
-					log.Printf("failed to send: %v", err)
+				if resp, err := c.Request(ctx, event); err != nil {
+					log.Printf("failed to request: %v", err)
 				} else if resp != nil {
 					fmt.Printf("Response:\n%s\n", resp)
 					fmt.Printf("Got Event Response Context: %+v\n", resp.Context)
diff --git a/cmd/samples/nats/sender/main.go b/cmd/samples/nats/sender/main.go
index f131c32..675e45e 100644
--- a/cmd/samples/nats/sender/main.go
+++ b/cmd/samples/nats/sender/main.go
@@ -52,7 +52,7 @@ type Example struct {
 	Message  string `json:"message"`
 }
 
-func (d *Demo) Send(eventContext event.EventContext, i int) (context.Context, *event.Event, error) {
+func (d *Demo) Send(eventContext event.EventContext, i int) error {
 	event := event.Event{
 		Context: eventContext,
 		Data: &Example{
@@ -98,7 +98,7 @@ func _main(args []string, env envConfig) int {
 				Source:          types.URIRef{URL: d.Source},
 				DataContentType: &contentType,
 			}.AsV1()
-			if _, _, err := d.Send(ctx, seq); err != nil {
+			if err := d.Send(ctx, seq); err != nil {
 				log.Printf("failed to send: %v", err)
 				return 1
 			}
diff --git a/cmd/samples/pubsub/multisender/main.go b/cmd/samples/pubsub/multisender/main.go
index 7b24a30..1e7b2e0 100644
--- a/cmd/samples/pubsub/multisender/main.go
+++ b/cmd/samples/pubsub/multisender/main.go
@@ -66,7 +66,7 @@ func main() {
 			Message:  "HELLO " + topic,
 		})
 
-		_, _, err = c.Send(ctx, event)
+		err = c.Send(ctx, event)
 
 		if err != nil {
 			log.Printf("failed to send: %v", err)
diff --git a/cmd/samples/pubsub/sender/main.go b/cmd/samples/pubsub/sender/main.go
index b448bb6..d9b2904 100644
--- a/cmd/samples/pubsub/sender/main.go
+++ b/cmd/samples/pubsub/sender/main.go
@@ -50,9 +50,7 @@ func main() {
 		Message:  "HELLO",
 	})
 
-	_, _, err = c.Send(context.Background(), event)
-
-	if err != nil {
+	if err = c.Send(context.Background(), event); err != nil {
 		log.Printf("failed to send: %v", err)
 		os.Exit(1)
 	}
diff --git a/cmd/samples/simple/http/sender/main.go b/cmd/samples/simple/http/requester/main.go
similarity index 95%
rename from cmd/samples/simple/http/sender/main.go
rename to cmd/samples/simple/http/requester/main.go
index d70f9b8..df29d35 100644
--- a/cmd/samples/simple/http/sender/main.go
+++ b/cmd/samples/simple/http/requester/main.go
@@ -43,7 +43,7 @@ func main() {
 		event.SetSource("https://github.com/cloudevents/sdk-go/cmd/samples/sender")
 		event.Data = data
 
-		if _, resp, err := c.Send(ctx, event); err != nil {
+		if resp, err := c.Request(ctx, event); err != nil {
 			log.Printf("failed to send: %v", err)
 		} else if resp != nil {
 			fmt.Printf("got back a response: \n%s", resp)
diff --git a/cmd/samples/simple/http/sender/v03/main_v03.go b/cmd/samples/simple/http/sender/v03/main_v03.go
deleted file mode 100644
index 2dde8e6..0000000
--- a/cmd/samples/simple/http/sender/v03/main_v03.go
+++ /dev/null
@@ -1,53 +0,0 @@
-package main
-
-import (
-	"context"
-	"fmt"
-	"log"
-
-	"github.com/cloudevents/sdk-go"
-)
-
-var source = cloudevents.ParseURLRef("https://github.com/cloudevents/sdk-go/cmd/samples/sender")
-
-// Basic data struct.
-type Example struct {
-	Sequence int    `json:"id"`
-	Message  string `json:"message"`
-}
-
-var subject = "this_thing"
-
-func main() {
-	ctx := cloudevents.ContextWithTarget(context.Background(), "http://localhost:8080/")
-
-	ctx = cloudevents.ContextWithHeader(ctx, "demo", "header value")
-
-	c, err := cloudevents.NewDefaultClient()
-	if err != nil {
-		log.Fatalf("failed to create client, %v", err)
-	}
-
-	for i := 0; i < 10; i++ {
-		data := &Example{
-			Sequence: i,
-			Message:  "Hello, World!",
-		}
-		event := cloudevents.Event{
-			Context: cloudevents.EventContextV03{
-				Type:    "com.cloudevents.sample.sent",
-				Source:  *source,
-				Subject: &subject,
-			}.AsV03(),
-			Data: data,
-		}
-
-		if _, resp, err := c.Send(ctx, event); err != nil {
-			log.Printf("failed to send: %v", err)
-		} else if resp != nil {
-			fmt.Printf("got back a response: \n%s", resp)
-		} else {
-			log.Printf("%s: %d - %s", event.Context.GetType(), data.Sequence, data.Message)
-		}
-	}
-}
diff --git a/cmd/samples/stats/main.go b/cmd/samples/stats/main.go
index 7baa1d8..f1b8a59 100644
--- a/cmd/samples/stats/main.go
+++ b/cmd/samples/stats/main.go
@@ -81,7 +81,7 @@ func mainSender() {
 				Data: data,
 			}
 
-			if _, resp, err := c.Send(ctx, e); err != nil {
+			if resp, err := c.Request(ctx, e); err != nil {
 				log.Printf("failed to send: %v", err)
 			} else if resp != nil {
 				fmt.Printf("got back a response event of type %s", resp.Context.GetType())
diff --git a/pkg/binding/buffering/copy_message.go b/pkg/binding/buffering/copy_message.go
index 5b3d24d..c513d7a 100644
--- a/pkg/binding/buffering/copy_message.go
+++ b/pkg/binding/buffering/copy_message.go
@@ -28,7 +28,7 @@ func CopyMessage(ctx context.Context, m binding.Message, transformers binding.Tr
 		return nil, binding.ErrUnknownEncoding
 	}
 	if originalMessageEncoding == binding.EncodingEvent {
-		e, _, err := binding.ToEvent(ctx, m, transformers)
+		e, err := binding.ToEvent(ctx, m, transformers)
 		if err != nil {
 			return nil, err
 		}
@@ -44,7 +44,7 @@ func CopyMessage(ctx context.Context, m binding.Message, transformers binding.Tr
 	} else if encoding == binding.EncodingBinary {
 		return &bm, err
 	} else {
-		e, _, err := binding.ToEvent(ctx, m, transformers)
+		e, err := binding.ToEvent(ctx, m, transformers)
 		if err != nil {
 			return nil, err
 		}
diff --git a/pkg/binding/to_event.go b/pkg/binding/to_event.go
index c778f26..d8971a4 100644
--- a/pkg/binding/to_event.go
+++ b/pkg/binding/to_event.go
@@ -19,7 +19,7 @@ var ErrCannotConvertToEvent = errors.New("cannot convert message to event")
 // This function returns the Event generated from the Message and the original encoding of the message or
 // an error that points the conversion error.
 // transformers can be nil and this function guarantees that they are invoked only once during the encoding process.
-func ToEvent(ctx context.Context, message Message, transformers TransformerFactories) (e event.Event, encoding Encoding, err error) {
+func ToEvent(ctx context.Context, message Message, transformers TransformerFactories) (e event.Event, err error) {
 	e = event.New()
 
 	messageEncoding := message.Encoding()
@@ -27,7 +27,6 @@ func ToEvent(ctx context.Context, message Message, transformers TransformerFacto
 		for m := message; m != nil; {
 			if em, ok := m.(EventMessage); ok {
 				e = event.Event(em)
-				encoding = EncodingEvent
 				err = transformers.EventTransformer()(&e)
 				return
 			}
@@ -42,7 +41,7 @@ func ToEvent(ctx context.Context, message Message, transformers TransformerFacto
 	}
 
 	encoder := &messageToEventBuilder{event: &e}
-	encoding, err = RunDirectEncoding(
+	_, err = RunDirectEncoding(
 		context.TODO(),
 		message,
 		encoder,
@@ -50,7 +49,7 @@ func ToEvent(ctx context.Context, message Message, transformers TransformerFacto
 		[]TransformerFactory{},
 	)
 	if err != nil {
-		return e, encoding, err
+		return e, err
 	}
 	err = transformers.EventTransformer()(&e)
 	return
diff --git a/pkg/binding/translate.go b/pkg/binding/translate.go
index 425878c..4742674 100644
--- a/pkg/binding/translate.go
+++ b/pkg/binding/translate.go
@@ -87,7 +87,7 @@ func Encode(
 	}
 
 	var e event.Event
-	e, enc, err = ToEvent(ctx, message, transformers)
+	e, err = ToEvent(ctx, message, transformers)
 	if err != nil {
 		return enc, err
 	}
diff --git a/pkg/binding/transport.go b/pkg/binding/transport.go
index 0daf312..a70cf2e 100644
--- a/pkg/binding/transport.go
+++ b/pkg/binding/transport.go
@@ -2,6 +2,7 @@ package binding
 
 import (
 	"context"
+	"errors"
 	"io"
 
 	"github.com/cloudevents/sdk-go/pkg/cloudevents/transport"
@@ -27,11 +28,15 @@ func NewTransportAdapter(s Sender, r Receiver, senderContextDecorators []func(co
 	return &BindingTransport{Sender: s, Receiver: r, SenderContextDecorators: senderContextDecorators}
 }
 
-func (t *BindingTransport) Send(ctx context.Context, e event.Event) (context.Context, *event.Event, error) {
+func (t *BindingTransport) Send(ctx context.Context, e event.Event) error {
 	for _, f := range t.SenderContextDecorators {
 		ctx = f(ctx)
 	}
-	return ctx, nil, t.Sender.Send(ctx, EventMessage(e))
+	return t.Sender.Send(ctx, EventMessage(e))
+}
+
+func (t *BindingTransport) Request(ctx context.Context, e event.Event) (*event.Event, error) {
+	return nil, errors.New("Transport.Request is not yet supported by binding transport.")
 }
 
 func (t *BindingTransport) SetReceiver(r transport.Receiver) { t.handler = r }
@@ -56,7 +61,7 @@ func (t *BindingTransport) handle(ctx context.Context, m Message) (err error) {
 			err = err2
 		}
 	}()
-	e, _, err := ToEvent(ctx, m, nil)
+	e, err := ToEvent(ctx, m, nil)
 	if err != nil {
 		return err
 	}
diff --git a/pkg/bindings/kafka_sarama/producer_message_encoder.go b/pkg/bindings/kafka_sarama/producer_message_encoder.go
index 7fdf066..060fdc3 100644
--- a/pkg/bindings/kafka_sarama/producer_message_encoder.go
+++ b/pkg/bindings/kafka_sarama/producer_message_encoder.go
@@ -62,7 +62,7 @@ func EncodeKafkaProducerMessage(ctx context.Context, m binding.Message, producer
 	}
 
 	var e ce.Event
-	e, _, err = binding.ToEvent(ctx, m, transformerFactories)
+	e, err = binding.ToEvent(ctx, m, transformerFactories)
 	if err != nil {
 		return err
 	}
diff --git a/pkg/client/client.go b/pkg/client/client.go
index a8edc72..656808a 100644
--- a/pkg/client/client.go
+++ b/pkg/client/client.go
@@ -7,17 +7,21 @@ import (
 
 	"github.com/cloudevents/sdk-go/pkg/event"
 
-	"github.com/cloudevents/sdk-go/pkg/cloudevents/extensions"
 	"github.com/cloudevents/sdk-go/pkg/cloudevents/observability"
 	"github.com/cloudevents/sdk-go/pkg/cloudevents/transport"
 	"github.com/cloudevents/sdk-go/pkg/cloudevents/transport/http"
+	"github.com/cloudevents/sdk-go/pkg/extensions"
 	"go.opencensus.io/trace"
 )
 
 // Client interface defines the runtime contract the CloudEvents client supports.
 type Client interface {
 	// Send will transmit the given event over the client's configured transport.
-	Send(ctx context.Context, event event.Event) (context.Context, *event.Event, error)
+	Send(ctx context.Context, event event.Event) error
+
+	// Request will transmit the given event over the client's configured
+	// transport and return any response event.
+	Request(ctx context.Context, event event.Event) (*event.Event, error)
 
 	// StartReceiver will register the provided function for callback on receipt
 	// of a cloudevent. It will also start the underlying transport as it has
@@ -85,11 +89,10 @@ type ceClient struct {
 	disableTracePropagation bool
 }
 
-// Send transmits the provided event on a preconfigured Transport.
-// Send returns a response event if there is a response or an error if there
-// was an an issue validating the outbound event or the transport returns an
-// error.
-func (c *ceClient) Send(ctx context.Context, event event.Event) (context.Context, *event.Event, error) {
+// Send transmits the provided event on a preconfigured Transport. Send returns
+// an error if there was an an issue validating the outbound event or the
+// transport returns an error.
+func (c *ceClient) Send(ctx context.Context, event event.Event) error {
 	ctx, r := observability.NewReporter(ctx, reportSend)
 
 	ctx, span := trace.StartSpan(ctx, clientSpanName, trace.WithSpanKind(trace.SpanKindClient))
@@ -98,19 +101,19 @@ func (c *ceClient) Send(ctx context.Context, event event.Event) (context.Context
 		span.AddAttributes(eventTraceAttributes(event.Context)...)
 	}
 
-	rctx, resp, err := c.obsSend(ctx, event)
+	err := c.obsSend(ctx, event)
 	if err != nil {
 		r.Error()
 	} else {
 		r.OK()
 	}
-	return rctx, resp, err
+	return err
 }
 
-func (c *ceClient) obsSend(ctx context.Context, event event.Event) (context.Context, *event.Event, error) {
+func (c *ceClient) obsSend(ctx context.Context, event event.Event) error {
 	// Confirm we have a transport set.
 	if c.transport == nil {
-		return ctx, nil, fmt.Errorf("client not ready, transport not initialized")
+		return fmt.Errorf("client not ready, transport not initialized")
 	}
 	// Apply the defaulter chain to the incoming event.
 	if len(c.eventDefaulterFns) > 0 {
@@ -124,19 +127,70 @@ func (c *ceClient) obsSend(ctx context.Context, event event.Event) (context.Cont
 		if span := trace.FromContext(ctx); span != nil {
 			event.Context = event.Context.Clone()
 			if err := extensions.FromSpanContext(span.SpanContext()).AddTracingAttributes(event.Context); err != nil {
-				return ctx, nil, fmt.Errorf("error setting distributed tracing extension: %w", err)
+				return fmt.Errorf("error setting distributed tracing extension: %w", err)
 			}
 		}
 	}
 
 	// Validate the event conforms to the CloudEvents Spec.
 	if err := event.Validate(); err != nil {
-		return ctx, nil, err
+		return err
 	}
 	// Send the event over the transport.
 	return c.transport.Send(ctx, event)
 }
 
+// Request transmits the provided event on a preconfigured Transport. Request
+// returns a response event if there is a response or an error if there was an
+// an issue validating the outbound event or the transport returns an error.
+func (c *ceClient) Request(ctx context.Context, event event.Event) (*event.Event, error) {
+	ctx, r := observability.NewReporter(ctx, reportSend)
+
+	ctx, span := trace.StartSpan(ctx, clientSpanName, trace.WithSpanKind(trace.SpanKindClient))
+	defer span.End()
+	if span.IsRecordingEvents() {
+		span.AddAttributes(eventTraceAttributes(event.Context)...)
+	}
+
+	resp, err := c.obsRequest(ctx, event)
+	if err != nil {
+		r.Error()
+	} else {
+		r.OK()
+	}
+	return resp, err
+}
+
+func (c *ceClient) obsRequest(ctx context.Context, event event.Event) (*event.Event, error) {
+	// Confirm we have a transport set.
+	if c.transport == nil {
+		return nil, fmt.Errorf("client not ready, transport not initialized")
+	}
+	// Apply the defaulter chain to the incoming event.
+	if len(c.eventDefaulterFns) > 0 {
+		for _, fn := range c.eventDefaulterFns {
+			event = fn(ctx, event)
+		}
+	}
+
+	// Set distributed tracing extension.
+	if !c.disableTracePropagation {
+		if span := trace.FromContext(ctx); span != nil {
+			event.Context = event.Context.Clone()
+			if err := extensions.FromSpanContext(span.SpanContext()).AddTracingAttributes(event.Context); err != nil {
+				return nil, fmt.Errorf("error setting distributed tracing extension: %w", err)
+			}
+		}
+	}
+
+	// Validate the event conforms to the CloudEvents Spec.
+	if err := event.Validate(); err != nil {
+		return nil, err
+	}
+	// Send the event over the transport.
+	return c.transport.Request(ctx, event)
+}
+
 // Receive is called from from the transport on event delivery.
 func (c *ceClient) Receive(ctx context.Context, event event.Event, resp *event.EventResponse) error {
 	ctx, r := observability.NewReporter(ctx, reportReceive)
diff --git a/pkg/client/observability.go b/pkg/client/observability.go
index a5fed5d..f55db10 100644
--- a/pkg/client/observability.go
+++ b/pkg/client/observability.go
@@ -40,6 +40,7 @@ const (
 	datacontenttypeAttr = "cloudevents.datacontenttype"
 
 	reportSend observed = iota
+	reportRequest
 	reportReceive
 )
 
@@ -48,6 +49,8 @@ func (o observed) MethodName() string {
 	switch o {
 	case reportSend:
 		return "send"
+	case reportRequest:
+		return "request"
 	case reportReceive:
 		return "receive"
 	default:
diff --git a/pkg/cloudevents/transport/http/transport.go b/pkg/cloudevents/transport/http/transport.go
index 620f3a5..6132e33 100644
--- a/pkg/cloudevents/transport/http/transport.go
+++ b/pkg/cloudevents/transport/http/transport.go
@@ -170,18 +170,30 @@ func copyHeadersEnsure(from http.Header, to *http.Header) {
 }
 
 // Send implements Transport.Send
-func (t *Transport) Send(ctx context.Context, event event2.Event) (context.Context, *event2.Event, error) {
+func (t *Transport) Send(ctx context.Context, event event2.Event) error {
 	ctx, r := observability.NewReporter(ctx, reportSend)
-	rctx, resp, err := t.obsSend(ctx, event)
+	_, err := t.obsSendRequest(ctx, event)
 	if err != nil {
 		r.Error()
 	} else {
 		r.OK()
 	}
-	return rctx, resp, err
+	return err
 }
 
-func (t *Transport) obsSend(ctx context.Context, event event2.Event) (context.Context, *event2.Event, error) {
+// Send implements Transport.Request
+func (t *Transport) Request(ctx context.Context, event event2.Event) (*event2.Event, error) {
+	ctx, r := observability.NewReporter(ctx, reportSend)
+	resp, err := t.obsSendRequest(ctx, event)
+	if err != nil {
+		r.Error()
+	} else {
+		r.OK()
+	}
+	return resp, err
+}
+
+func (t *Transport) obsSendRequest(ctx context.Context, event event2.Event) (*event2.Event, error) {
 	req := http.Request{
 		Header: HeaderFrom(ctx),
 	}
@@ -199,12 +211,12 @@ func (t *Transport) obsSend(ctx context.Context, event event2.Event) (context.Co
 	}
 
 	if ok := t.loadCodec(ctx); !ok {
-		return WithTransportContext(ctx, NewTransportContextFromResponse(nil)), nil, fmt.Errorf("unknown encoding set on transport: %d", t.Encoding)
+		return nil, fmt.Errorf("unknown encoding set on transport: %d", t.Encoding)
 	}
 
 	msg, err := t.codec.Encode(ctx, event)
 	if err != nil {
-		return WithTransportContext(ctx, NewTransportContextFromResponse(nil)), nil, err
+		return nil, err
 	}
 
 	if m, ok := msg.(*Message); ok {
@@ -213,10 +225,9 @@ func (t *Transport) obsSend(ctx context.Context, event event2.Event) (context.Co
 		if client == nil {
 			client = &http.Client{Transport: t.transport}
 		}
-		return httpDo(ctx, client, &req, func(resp *http.Response, err error) (context.Context, *event2.Event, error) {
-			rctx := WithTransportContext(ctx, NewTransportContextFromResponse(resp))
+		return httpDo(ctx, client, &req, func(resp *http.Response, err error) (*event2.Event, error) {
 			if err != nil {
-				return rctx, nil, err
+				return nil, err
 			}
 			defer resp.Body.Close()
 
@@ -237,19 +248,19 @@ func (t *Transport) obsSend(ctx context.Context, event event2.Event) (context.Co
 					}
 				}
 				if isErr {
-					return rctx, nil, err
+					return nil, err
 				}
 				if handled {
-					return rctx, nil, nil
+					return nil, nil
 				}
 			}
 			if accepted(resp) {
-				return rctx, respEvent, nil
+				return respEvent, nil
 			}
-			return rctx, respEvent, fmt.Errorf("error sending cloudevent: %s", resp.Status)
+			return respEvent, fmt.Errorf("error sending cloudevent: %s", resp.Status)
 		})
 	}
-	return WithTransportContext(ctx, NewTransportContextFromResponse(nil)), nil, fmt.Errorf("failed to encode Event into a Message")
+	return nil, fmt.Errorf("failed to encode Event into a Message")
 }
 
 func (t *Transport) MessageToEvent(ctx context.Context, msg *Message) (*event2.Event, error) {
@@ -485,19 +496,19 @@ type eventError struct {
 	err   error
 }
 
-func httpDo(ctx context.Context, client *http.Client, req *http.Request, fn func(*http.Response, error) (context.Context, *event2.Event, error)) (context.Context, *event2.Event, error) {
+func httpDo(ctx context.Context, client *http.Client, req *http.Request, fn func(*http.Response, error) (*event2.Event, error)) (*event2.Event, error) {
 	// Run the HTTP request in a goroutine and pass the response to fn.
 	c := make(chan eventError, 1)
 	req = req.WithContext(ctx)
 	go func() {
-		rctx, event, err := fn(client.Do(req))
-		c <- eventError{ctx: rctx, event: event, err: err}
+		event, err := fn(client.Do(req))
+		c <- eventError{event: event, err: err}
 	}()
 	select {
 	case <-ctx.Done():
-		return ctx, nil, ctx.Err()
+		return nil, ctx.Err()
 	case ee := <-c:
-		return ee.ctx, ee.event, ee.err
+		return ee.event, ee.err
 	}
 }
 
diff --git a/pkg/cloudevents/transport/nats/transport.go b/pkg/cloudevents/transport/nats/transport.go
index 54db071..97e3d3b 100644
--- a/pkg/cloudevents/transport/nats/transport.go
+++ b/pkg/cloudevents/transport/nats/transport.go
@@ -93,22 +93,27 @@ func (t *Transport) loadCodec() bool {
 }
 
 // Send implements Transport.Send
-func (t *Transport) Send(ctx context.Context, event event.Event) (context.Context, *event.Event, error) {
+func (t *Transport) Send(ctx context.Context, event event.Event) error {
 	// TODO populate response context properly.
 	if ok := t.loadCodec(); !ok {
-		return ctx, nil, fmt.Errorf("unknown encoding set on transport: %d", t.Encoding)
+		return fmt.Errorf("unknown encoding set on transport: %d", t.Encoding)
 	}
 
 	msg, err := t.codec.Encode(ctx, event)
 	if err != nil {
-		return ctx, nil, err
+		return err
 	}
 
 	if m, ok := msg.(*Message); ok {
-		return ctx, nil, t.Conn.Publish(t.Subject, m.Body)
+		return t.Conn.Publish(t.Subject, m.Body)
 	}
 
-	return ctx, nil, fmt.Errorf("failed to encode Event into a Message")
+	return fmt.Errorf("failed to encode Event into a Message")
+}
+
+// Request implements Transport.Request
+func (t *Transport) Request(ctx context.Context, event event.Event) (*event.Event, error) {
+	return nil, fmt.Errorf("Transport.Request is not supported for NATS")
 }
 
 // SetReceiver implements Transport.SetReceiver
@@ -152,10 +157,10 @@ func (t *Transport) StartReceiver(ctx context.Context) (err error) {
 		msg := &Message{
 			Body: m.Data,
 		}
-		event, err := t.codec.Decode(ctx, msg)
+		e, 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)
+			e, err = t.Converter.Convert(ctx, msg, err)
 		}
 		if err != nil {
 			logger.Errorw("failed to decode message", zap.Error(err)) // TODO: create an error channel to pass this up
@@ -163,7 +168,7 @@ func (t *Transport) StartReceiver(ctx context.Context) (err error) {
 		}
 		// TODO: I do not know enough about NATS to implement reply.
 		// For now, NATS does not support reply.
-		if err := t.Receiver.Receive(context.TODO(), *event, nil); err != nil {
+		if err := t.Receiver.Receive(context.TODO(), *e, nil); err != nil {
 			logger.Warnw("nats receiver return err", zap.Error(err))
 		}
 	})
diff --git a/pkg/cloudevents/transport/pubsub/internal/connection.go b/pkg/cloudevents/transport/pubsub/internal/connection.go
index fc020c0..d28db22 100644
--- a/pkg/cloudevents/transport/pubsub/internal/connection.go
+++ b/pkg/cloudevents/transport/pubsub/internal/connection.go
@@ -7,8 +7,6 @@ import (
 	"sync"
 	"time"
 
-	"github.com/cloudevents/sdk-go/pkg/event"
-
 	"cloud.google.com/go/pubsub"
 	pscontext "github.com/cloudevents/sdk-go/pkg/cloudevents/transport/pubsub/context"
 )
@@ -185,15 +183,15 @@ func (c *Connection) DeleteSubscription(ctx context.Context) error {
 }
 
 // Publish
-func (c *Connection) Publish(ctx context.Context, msg *pubsub.Message) (*event.Event, error) {
+func (c *Connection) Publish(ctx context.Context, msg *pubsub.Message) error {
 	topic, err := c.getOrCreateTopic(ctx)
 	if err != nil {
-		return nil, err
+		return err
 	}
 
 	r := topic.Publish(ctx, msg)
 	_, err = r.Get(ctx)
-	return nil, err
+	return err
 }
 
 // Start
diff --git a/pkg/cloudevents/transport/pubsub/transport.go b/pkg/cloudevents/transport/pubsub/transport.go
index 94013b6..c2f11ac 100644
--- a/pkg/cloudevents/transport/pubsub/transport.go
+++ b/pkg/cloudevents/transport/pubsub/transport.go
@@ -177,10 +177,10 @@ func (t *Transport) getOrCreateConnection(ctx context.Context, topic, subscripti
 }
 
 // Send implements Transport.Send
-func (t *Transport) Send(ctx context.Context, event event.Event) (context.Context, *event.Event, error) {
+func (t *Transport) Send(ctx context.Context, event event.Event) error {
 	// TODO populate response context properly.
 	if ok := t.loadCodec(ctx); !ok {
-		return ctx, nil, fmt.Errorf("unknown encoding set on transport: %d", t.Encoding)
+		return fmt.Errorf("unknown encoding set on transport: %d", t.Encoding)
 	}
 
 	topic := cecontext.TopicFrom(ctx)
@@ -192,18 +192,23 @@ func (t *Transport) Send(ctx context.Context, event event.Event) (context.Contex
 
 	msg, err := t.codec.Encode(ctx, event)
 	if err != nil {
-		return ctx, nil, err
+		return err
 	}
 
 	if m, ok := msg.(*Message); ok {
-		respEvent, err := conn.Publish(ctx, &pubsub.Message{
+		err := conn.Publish(ctx, &pubsub.Message{
 			Attributes: m.Attributes,
 			Data:       m.Data,
 		})
-		return ctx, respEvent, err
+		return err
 	}
 
-	return ctx, nil, fmt.Errorf("failed to encode Event into a Message")
+	return fmt.Errorf("failed to encode Event into a Message")
+}
+
+// Request implements Transport.Request
+func (t *Transport) Request(ctx context.Context, event event.Event) (*event.Event, error) {
+	return nil, fmt.Errorf("Transport.Request is not supported for Pub/Sub")
 }
 
 // SetReceiver implements Transport.SetReceiver
diff --git a/pkg/cloudevents/transport/transport.go b/pkg/cloudevents/transport/transport.go
index 5b2a106..e5cb28a 100644
--- a/pkg/cloudevents/transport/transport.go
+++ b/pkg/cloudevents/transport/transport.go
@@ -9,7 +9,8 @@ import (
 // Transport is the interface for transport sender to send the converted Message
 // over the underlying transport.
 type Transport interface {
-	Send(context.Context, event.Event) (context.Context, *event.Event, error)
+	Send(context.Context, event.Event) error
+	Request(context.Context, event.Event) (*event.Event, error)
 
 	SetReceiver(Receiver)
 	StartReceiver(context.Context) error


Test patch:
diff --git a/pkg/binding/buffering/copy_message_test.go b/pkg/binding/buffering/copy_message_test.go
index da4cf62..1416a5a 100644
--- a/pkg/binding/buffering/copy_message_test.go
+++ b/pkg/binding/buffering/copy_message_test.go
@@ -74,9 +74,9 @@ func TestCopyMessage(t *testing.T) {
 			require.NoError(t, err)
 			// The copy can be read any number of times
 			for i := 0; i < 3; i++ {
-				got, encoding, err := binding.ToEvent(context.Background(), cpy, nil)
+				got, err := binding.ToEvent(context.Background(), cpy, nil)
 				assert.NoError(t, err)
-				require.Equal(t, tt.encoding, encoding)
+				require.Equal(t, tt.encoding, cpy.Encoding())
 				test.AssertEventEquals(t, test.ExToStr(t, tt.want), test.ExToStr(t, got))
 			}
 			require.NoError(t, cpy.Finish(nil))
@@ -91,9 +91,9 @@ func TestCopyMessage(t *testing.T) {
 			require.NoError(t, err)
 			// The copy can be read any number of times
 			for i := 0; i < 3; i++ {
-				got, encoding, err := binding.ToEvent(context.Background(), cpy, nil)
+				got, err := binding.ToEvent(context.Background(), cpy, nil)
 				assert.NoError(t, err)
-				require.Equal(t, tt.encoding, encoding)
+				require.Equal(t, tt.encoding, cpy.Encoding())
 				test.AssertEventEquals(t, test.ExToStr(t, tt.want), test.ExToStr(t, got))
 			}
 			require.NoError(t, cpy.Finish(nil))
diff --git a/pkg/binding/example_using_test.go b/pkg/binding/example_using_test.go
index 6c8da9e..07e3a90 100644
--- a/pkg/binding/example_using_test.go
+++ b/pkg/binding/example_using_test.go
@@ -26,7 +26,7 @@ func runSender(w io.Writer) error {
 		if err := e.SetData(fmt.Sprintf("hello %d", i)); err != nil {
 			return err
 		}
-		if _, _, err := c.Send(context.TODO(), e); err != nil {
+		if err := c.Send(context.TODO(), e); err != nil {
 			return err
 		}
 	}
diff --git a/pkg/binding/test/mock_structured_message.go b/pkg/binding/test/mock_structured_message.go
index 9969499..b2dab45 100644
--- a/pkg/binding/test/mock_structured_message.go
+++ b/pkg/binding/test/mock_structured_message.go
@@ -20,12 +20,8 @@ type MockStructuredMessage struct {
 
 // Create a new MockStructuredMessage starting from an event.Event. Panics in case of error
 func MustCreateMockStructuredMessage(e event.Event) binding.Message {
-	testEventSerialized, err := format.JSON.Marshal(e)
-	if err != nil {
-		panic(err)
-	}
 	return &MockStructuredMessage{
-		Bytes:  testEventSerialized,
+		Bytes:  MustJSON(e),
 		Format: format.JSON,
 	}
 }
diff --git a/pkg/binding/test/test.go b/pkg/binding/test/test.go
index 974a226..ccb1dc3 100644
--- a/pkg/binding/test/test.go
+++ b/pkg/binding/test/test.go
@@ -129,13 +129,10 @@ func MustJSON(e event.Event) []byte {
 }
 
 // Must convert the Message to event.Event
-func MustToEvent(ctx context.Context, m binding.Message) (e event.Event, encoding binding.Encoding) {
-	var err error
-	e, encoding, err = binding.ToEvent(ctx, m, nil)
-	if err != nil {
-		panic(err)
-	}
-	return
+func MustToEvent(t *testing.T, ctx context.Context, m binding.Message) event.Event {
+	e, err := binding.ToEvent(ctx, m, nil)
+	require.NoError(t, err)
+	return e
 }
 
 // Returns a copy of the event.Event with only the event.EventContext copied
diff --git a/pkg/binding/test/transformer.go b/pkg/binding/test/transformer.go
index db257d8..af94bb8 100644
--- a/pkg/binding/test/transformer.go
+++ b/pkg/binding/test/transformer.go
@@ -31,10 +31,10 @@ func RunTransformerTests(t *testing.T, ctx context.Context, tests []TransformerT
 
 			var e event.Event
 			if enc == binding.EncodingStructured {
-				e, _, err = binding.ToEvent(ctx, &mockStructured, nil)
+				e, err = binding.ToEvent(ctx, &mockStructured, nil)
 				require.NoError(t, err)
 			} else if enc == binding.EncodingBinary {
-				e, _, err = binding.ToEvent(ctx, &mockBinary, nil)
+				e, err = binding.ToEvent(ctx, &mockBinary, nil)
 				require.NoError(t, err)
 			} else {
 				t.Fatalf("Unexpected encoding %v", enc)
diff --git a/pkg/binding/to_event_test.go b/pkg/binding/to_event_test.go
index 86668aa..ed2e5d9 100644
--- a/pkg/binding/to_event_test.go
+++ b/pkg/binding/to_event_test.go
@@ -12,10 +12,9 @@ import (
 )
 
 type toEventTestCase struct {
-	name     string
-	encoding binding.Encoding
-	message  binding.Message
-	want     event.Event
+	name    string
+	message binding.Message
+	want    event.Event
 }
 
 func TestToEvent(t *testing.T) {
@@ -24,49 +23,42 @@ func TestToEvent(t *testing.T) {
 	for _, v := range test.Events() {
 		tests = append(tests, []toEventTestCase{
 			{
-				name:     "From mock structured with payload/" + test.NameOf(v),
-				encoding: binding.EncodingStructured,
-				message:  test.MustCreateMockStructuredMessage(v),
-				want:     v,
+				name:    "From mock structured with payload/" + test.NameOf(v),
+				message: test.MustCreateMockStructuredMessage(v),
+				want:    v,
 			},
 			{
-				name:     "From mock structured without payload/" + test.NameOf(v),
-				encoding: binding.EncodingStructured,
-				message:  test.MustCreateMockStructuredMessage(v),
-				want:     v,
+				name:    "From mock structured without payload/" + test.NameOf(v),
+				message: test.MustCreateMockStructuredMessage(v),
+				want:    v,
 			},
 			{
-				name:     "From mock binary with payload/" + test.NameOf(v),
-				encoding: binding.EncodingBinary,
-				message:  test.MustCreateMockBinaryMessage(v),
-				want:     v,
+				name:    "From mock binary with payload/" + test.NameOf(v),
+				message: test.MustCreateMockBinaryMessage(v),
+				want:    v,
 			},
 			{
-				name:     "From mock binary without payload/" + test.NameOf(v),
-				encoding: binding.EncodingBinary,
-				message:  test.MustCreateMockBinaryMessage(v),
-				want:     v,
+				name:    "From mock binary without payload/" + test.NameOf(v),
+				message: test.MustCreateMockBinaryMessage(v),
+				want:    v,
 			},
 			{
-				name:     "From event with payload/" + test.NameOf(v),
-				encoding: binding.EncodingEvent,
-				message:  binding.EventMessage(v),
-				want:     v,
+				name:    "From event with payload/" + test.NameOf(v),
+				message: binding.EventMessage(v),
+				want:    v,
 			},
 			{
-				name:     "From event without payload/" + test.NameOf(v),
-				encoding: binding.EncodingEvent,
-				message:  binding.EventMessage(v),
-				want:     v,
+				name:    "From event without payload/" + test.NameOf(v),
+				message: binding.EventMessage(v),
+				want:    v,
 			},
 		}...)
 	}
 	for _, tt := range tests {
 		tt := tt // Don't use range variable in Run() scope
 		t.Run(tt.name, func(t *testing.T) {
-			got, encoding, err := binding.ToEvent(context.Background(), tt.message, nil)
+			got, err := binding.ToEvent(context.Background(), tt.message, nil)
 			require.NoError(t, err)
-			require.Equal(t, tt.encoding, encoding)
 			test.AssertEventEquals(t, test.ExToStr(t, tt.want), test.ExToStr(t, got))
 		})
 	}
diff --git a/pkg/binding/transport_test.go b/pkg/binding/transport_test.go
index c8fedbd..0708851 100644
--- a/pkg/binding/transport_test.go
+++ b/pkg/binding/transport_test.go
@@ -20,7 +20,7 @@ func TestTransportSend(t *testing.T) {
 	c, err := client.New(transport, client.WithoutTracePropagation())
 	require.NoError(t, err)
 
-	_, _, err = c.Send(context.Background(), ev)
+	err = c.Send(context.Background(), ev)
 	require.NoError(t, err)
 
 	result := <-messageChannel
diff --git a/pkg/bindings/http/response_encoder_test.go b/pkg/bindings/http/response_encoder_test.go
index e58eccc..452b19f 100644
--- a/pkg/bindings/http/response_encoder_test.go
+++ b/pkg/bindings/http/response_encoder_test.go
@@ -59,10 +59,9 @@ func TestEncodeHttpResponse(t *testing.T) {
 
 				//Little hack to go back to Message
 				messageOut := NewMessageFromHttpResponse(res)
-				require.NoError(t, err)
+				require.Equal(t, tt.expectedEncoding, messageOut.Encoding())
 
-				eventOut, encoding, err := binding.ToEvent(context.TODO(), messageOut, nil)
-				require.Equal(t, encoding, tt.expectedEncoding)
+				eventOut, err := binding.ToEvent(context.TODO(), messageOut, nil)
 				test.AssertEventEquals(t, eventIn, eventOut)
 			})
 		})
diff --git a/pkg/bindings/kafka_sarama/message_benchmark_test.go b/pkg/bindings/kafka_sarama/message_benchmark_test.go
index 3d5d88e..544613e 100644
--- a/pkg/bindings/kafka_sarama/message_benchmark_test.go
+++ b/pkg/bindings/kafka_sarama/message_benchmark_test.go
@@ -93,28 +93,28 @@ func BenchmarkNewBinaryMessageWithKey(b *testing.B) {
 func BenchmarkNewStructuredMessageWithoutKeyToEvent(b *testing.B) {
 	for i := 0; i < b.N; i++ {
 		M, Err = kafka_sarama.NewMessageFromConsumerMessage(structuredConsumerMessageWithoutKey)
-		Event, _, Err = binding.ToEvent(context.TODO(), M, nil)
+		Event, Err = binding.ToEvent(context.TODO(), M, nil)
 	}
 }
 
 func BenchmarkNewStructuredMessageWithKeyToEvent(b *testing.B) {
 	for i := 0; i < b.N; i++ {
 		M, Err = kafka_sarama.NewMessageFromConsumerMessage(structuredConsumerMessageWithKey)
-		Event, _, Err = binding.ToEvent(context.TODO(), M, nil)
+		Event, Err = binding.ToEvent(context.TODO(), M, nil)
 	}
 }
 
 func BenchmarkNewBinaryMessageWithoutKeyToEvent(b *testing.B) {
 	for i := 0; i < b.N; i++ {
 		M, Err = kafka_sarama.NewMessageFromConsumerMessage(binaryConsumerMessageWithoutKey)
-		Event, _, Err = binding.ToEvent(context.TODO(), M, nil)
+		Event, Err = binding.ToEvent(context.TODO(), M, nil)
 	}
 }
 
 func BenchmarkNewBinaryMessageWithKeyToEvent(b *testing.B) {
 	for i := 0; i < b.N; i++ {
 		M, Err = kafka_sarama.NewMessageFromConsumerMessage(binaryConsumerMessageWithKey)
-		Event, _, Err = binding.ToEvent(context.TODO(), M, nil)
+		Event, Err = binding.ToEvent(context.TODO(), M, nil)
 	}
 }
 
diff --git a/pkg/bindings/kafka_sarama/producer_message_encoder_test.go b/pkg/bindings/kafka_sarama/producer_message_encoder_test.go
index 7ceb604..5e63a9d 100644
--- a/pkg/bindings/kafka_sarama/producer_message_encoder_test.go
+++ b/pkg/bindings/kafka_sarama/producer_message_encoder_test.go
@@ -122,11 +122,11 @@ func TestEncodeKafkaProducerMessage(t *testing.T) {
 				}
 
 				messageOut, err := NewMessage(key, value, string(headers[contentTypeHeader]), headers)
+				require.Equal(t, tt.expectedEncoding, messageOut.Encoding())
 				require.NoError(t, err)
 
-				eventOut, encoding, err := binding.ToEvent(context.TODO(), messageOut, nil)
+				eventOut, err := binding.ToEvent(context.TODO(), messageOut, nil)
 				require.NoError(t, err)
-				require.Equal(t, tt.expectedEncoding, encoding)
 				test.AssertEventEquals(t, eventIn, eventOut)
 			})
 		})
diff --git a/pkg/client/client_test.go b/pkg/client/client_test.go
index 13c273b..d0ccb95 100644
--- a/pkg/client/client_test.go
+++ b/pkg/client/client_test.go
@@ -159,7 +159,7 @@ func TestClientSend(t *testing.T) {
 
 			c := tc.c(server.URL)
 
-			_, _, err := c.Send(context.TODO(), tc.event) // TODO: update test with new returned event and returned context
+			err := c.Send(context.TODO(), tc.event)
 			if tc.wantErr != "" {
 				if err == nil {
 					t.Fatalf("failed to return expected error, got nil")
@@ -254,7 +254,7 @@ func TestTracingClientSend(t *testing.T) {
 			ctx, span := trace.StartSpan(context.TODO(), "test-span", trace.WithSampler(sampler))
 			sc := span.SpanContext()
 
-			_, _, err := c.Send(ctx, tc.event)
+			err := c.Send(ctx, tc.event)
 			span.End()
 
 			if err != nil {
@@ -521,7 +521,7 @@ func TestTracedClientReceive(t *testing.T) {
 			client := simpleBinaryClient(target)
 
 			ctx, span := trace.StartSpan(context.TODO(), "test-span")
-			_, _, err = client.Send(ctx, tc.event)
+			err = client.Send(ctx, tc.event)
 			span.End()
 
 			if err != nil {
diff --git a/pkg/cloudevents/transport/http/transport_test.go b/pkg/cloudevents/transport/http/transport_test.go
index 3e021a4..0eb7887 100644
--- a/pkg/cloudevents/transport/http/transport_test.go
+++ b/pkg/cloudevents/transport/http/transport_test.go
@@ -116,14 +116,10 @@ func TestStableConnectionsToSingleHost(t *testing.T) {
 	duration := 1 * time.Second
 	var sent uint64
 	err = doConcurrently(concurrency, duration, func() error {
-		rctx, _, err := ceClient.Send(ctx, e)
+		err := ceClient.Send(ctx, e)
 		if err != nil {
 			return fmt.Errorf("unexpected error sending CloudEvent: %v", err.Error())
 		}
-		trctx := cehttp.TransportContextFrom(rctx)
-		if trctx.StatusCode != http.StatusAccepted {
-			return fmt.Errorf("unexpected status code: want %d, got %d", http.StatusAccepted, trctx.StatusCode)
-		}
 		atomic.AddUint64(&sent, 1)
 		return nil
 	})
diff --git a/pkg/cloudevents/transport/pubsub/internal/connection_test.go b/pkg/cloudevents/transport/pubsub/internal/connection_test.go
index be4eebc..191f648 100644
--- a/pkg/cloudevents/transport/pubsub/internal/connection_test.go
+++ b/pkg/cloudevents/transport/pubsub/internal/connection_test.go
@@ -59,7 +59,7 @@ func TestPublishCreateTopic(t *testing.T) {
 		ID:   "msg-id-1",
 		Data: []byte("msg-data-1"),
 	}
-	if _, err := psconn.Publish(ctx, msg); err != nil {
+	if err := psconn.Publish(ctx, msg); err != nil {
 		t.Errorf("failed to publish message: %v", err)
 	}
 
@@ -180,7 +180,7 @@ func TestPublishReceiveRoundtrip(t *testing.T) {
 		data := fmt.Sprintf("data-%d", i)
 		wantMsgs[data] = data
 
-		if _, err := psconn.Publish(ctx, &pubsub.Message{Data: []byte(data)}); err != nil {
+		if err := psconn.Publish(ctx, &pubsub.Message{Data: []byte(data)}); err != nil {
 			t.Errorf("failed to publish message: %v", err)
 		}
 		wg.Add(1)
diff --git a/pkg/cloudevents/extensions/distributed_tracing_extension.go b/pkg/extensions/distributed_tracing_extension.go
similarity index 100%
rename from pkg/cloudevents/extensions/distributed_tracing_extension.go
rename to pkg/extensions/distributed_tracing_extension.go
diff --git a/pkg/cloudevents/extensions/distributed_tracing_extension_test.go b/pkg/extensions/distributed_tracing_extension_test.go
similarity index 99%
rename from pkg/cloudevents/extensions/distributed_tracing_extension_test.go
rename to pkg/extensions/distributed_tracing_extension_test.go
index 2d7e2c4..8c3d5ac 100644
--- a/pkg/cloudevents/extensions/distributed_tracing_extension_test.go
+++ b/pkg/extensions/distributed_tracing_extension_test.go
@@ -8,7 +8,7 @@ import (
 
 	"github.com/cloudevents/sdk-go/pkg/event"
 
-	"github.com/cloudevents/sdk-go/pkg/cloudevents/extensions"
+	"github.com/cloudevents/sdk-go/pkg/extensions"
 	"github.com/cloudevents/sdk-go/pkg/types"
 	"github.com/google/go-cmp/cmp"
 	"go.opencensus.io/trace"
diff --git a/test/benchmark/e2e/http/main.go b/test/benchmark/e2e/http/main.go
index 9feaf79..6864baa 100644
--- a/test/benchmark/e2e/http/main.go
+++ b/test/benchmark/e2e/http/main.go
@@ -157,7 +157,7 @@ func dispatchReceiver(clients []cloudevents.Client, outputSenders int) transport
 		for i := 0; i < outputSenders; i++ {
 			wg.Add(1)
 			go func(client cloudevents.Client) {
-				_, _, _ = client.Send(ctx, e)
+				_ = client.Send(ctx, e)
 				wg.Done()
 			}(clients[i])
 		}
diff --git a/test/integration/amqp_binding/amqp_test.go b/test/integration/amqp_binding/amqp_test.go
index 21104d1..fd6e55a 100644
--- a/test/integration/amqp_binding/amqp_test.go
+++ b/test/integration/amqp_binding/amqp_test.go
@@ -24,8 +24,8 @@ func TestSendSkipBinary(t *testing.T) {
 		eventIn := test.ExToStr(t, e)
 		in := test.MustCreateMockBinaryMessage(eventIn)
 		test.SendReceive(t, binding.WithSkipDirectBinaryEncoding(binding.WithPreferredEventEncoding(context.Background(), binding.EncodingStructured), true), in, s, r, func(out binding.Message) {
-			eventOut, encoding := test.MustToEvent(context.TODO(), out)
-			assert.Equal(t, encoding, binding.EncodingStructured)
+			eventOut := test.MustToEvent(t, context.TODO(), out)
+			assert.Equal(t, out.Encoding(), binding.EncodingStructured)
 			test.AssertEventEquals(t, eventIn, test.ExToStr(t, eventOut))
 		})
 	})
@@ -38,8 +38,8 @@ func TestSendSkipStructured(t *testing.T) {
 		eventIn := test.ExToStr(t, e)
 		in := test.MustCreateMockStructuredMessage(eventIn)
 		test.SendReceive(t, binding.WithSkipDirectStructuredEncoding(context.Background(), true), in, s, r, func(out binding.Message) {
-			eventOut, encoding := test.MustToEvent(context.Background(), out)
-			assert.Equal(t, encoding, binding.EncodingBinary)
+			eventOut := test.MustToEvent(t, context.Background(), out)
+			assert.Equal(t, out.Encoding(), binding.EncodingBinary)
 			test.AssertEventEquals(t, eventIn, test.ExToStr(t, eventOut))
 		})
 	})
@@ -52,8 +52,8 @@ func TestSendEventReceiveStruct(t *testing.T) {
 		eventIn := test.ExToStr(t, e)
 		in := binding.EventMessage(eventIn)
 		test.SendReceive(t, binding.WithPreferredEventEncoding(context.TODO(), binding.EncodingStructured), in, s, r, func(out binding.Message) {
-			eventOut, encoding := test.MustToEvent(context.Background(), out)
-			assert.Equal(t, encoding, binding.EncodingStructured)
+			eventOut := test.MustToEvent(t, context.Background(), out)
+			assert.Equal(t, out.Encoding(), binding.EncodingStructured)
 			test.AssertEventEquals(t, eventIn, test.ExToStr(t, eventOut))
 		})
 	})
@@ -66,8 +66,8 @@ func TestSendEventReceiveBinary(t *testing.T) {
 		eventIn := test.ExToStr(t, e)
 		in := binding.EventMessage(eventIn)
 		test.SendReceive(t, context.Background(), in, s, r, func(out binding.Message) {
-			eventOut, encoding := test.MustToEvent(context.Background(), out)
-			assert.Equal(t, encoding, binding.EncodingBinary)
+			eventOut := test.MustToEvent(t, context.Background(), out)
+			assert.Equal(t, out.Encoding(), binding.EncodingBinary)
 			test.AssertEventEquals(t, eventIn, test.ExToStr(t, eventOut))
 		})
 	})
diff --git a/test/integration/http/direct.go b/test/integration/http/direct.go
index 22fdf9e..5385b39 100644
--- a/test/integration/http/direct.go
+++ b/test/integration/http/direct.go
@@ -72,7 +72,7 @@ func ClientDirect(t *testing.T, tc DirectTapTest, topts ...cehttp.Option) {
 		}
 	}()
 
-	_, _, err = ce.Send(ctx, *tc.event)
+	err = ce.Send(ctx, *tc.event)
 	if err != nil {
 		t.Fatal(err)
 	}
diff --git a/test/integration/http/loopback.go b/test/integration/http/loopback.go
index 872450e..7f4eef6 100644
--- a/test/integration/http/loopback.go
+++ b/test/integration/http/loopback.go
@@ -83,7 +83,7 @@ func ClientLoopback(t *testing.T, tc TapTest, topts ...cehttp.Option) {
 		}
 	}()
 
-	_, got, err := ce.Send(ctx, *tc.event)
+	got, err := ce.Request(ctx, *tc.event)
 	if err != nil {
 		t.Fatal(err)
 	}
diff --git a/test/integration/http/middleware.go b/test/integration/http/middleware.go
index 3dc3bc9..c73e15b 100644
--- a/test/integration/http/middleware.go
+++ b/test/integration/http/middleware.go
@@ -52,7 +52,7 @@ func ClientMiddleware(t *testing.T, tc TapTest, topts ...cehttp.Option) {
 		}
 	}()
 
-	_, got, err := ce.Send(ctx, *tc.event)
+	got, err := ce.Request(ctx, *tc.event)
 	if err != nil {
 		t.Fatal(err)
 	}
diff --git a/test/integration/http_binding/http_test.go b/test/integration/http_binding/http_test.go
index 9c2fc7d..54721c5 100644
--- a/test/integration/http_binding/http_test.go
+++ b/test/integration/http_binding/http_test.go
@@ -24,8 +24,8 @@ func TestSendSkipBinary(t *testing.T) {
 		eventIn = test.ExToStr(t, eventIn)
 		in := test.MustCreateMockBinaryMessage(eventIn)
 		test.SendReceive(t, binding.WithSkipDirectBinaryEncoding(binding.WithPreferredEventEncoding(context.Background(), binding.EncodingStructured), true), in, s, r, func(out binding.Message) {
-			eventOut, encoding := test.MustToEvent(context.Background(), out)
-			assert.Equal(t, encoding, binding.EncodingStructured)
+			eventOut := test.MustToEvent(t, context.Background(), out)
+			assert.Equal(t, binding.EncodingStructured, out.Encoding())
 			test.AssertEventEquals(t, eventIn, test.ExToStr(t, eventOut))
 		})
 	})
@@ -38,8 +38,8 @@ func TestSendSkipStructured(t *testing.T) {
 		eventIn = test.ExToStr(t, eventIn)
 		in := test.MustCreateMockStructuredMessage(eventIn)
 		test.SendReceive(t, binding.WithSkipDirectStructuredEncoding(context.Background(), true), in, s, r, func(out binding.Message) {
-			eventOut, encoding := test.MustToEvent(context.Background(), out)
-			assert.Equal(t, encoding, binding.EncodingBinary)
+			eventOut := test.MustToEvent(t, context.Background(), out)
+			assert.Equal(t, binding.EncodingBinary, out.Encoding())
 			test.AssertEventEquals(t, eventIn, test.ExToStr(t, eventOut))
 		})
 	})
@@ -52,8 +52,8 @@ func TestSendBinaryReceiveBinary(t *testing.T) {
 		eventIn = test.ExToStr(t, eventIn)
 		in := test.MustCreateMockBinaryMessage(eventIn)
 		test.SendReceive(t, context.Background(), in, s, r, func(out binding.Message) {
-			eventOut, encoding := test.MustToEvent(context.Background(), out)
-			assert.Equal(t, encoding, binding.EncodingBinary)
+			eventOut := test.MustToEvent(t, context.Background(), out)
+			assert.Equal(t, binding.EncodingBinary, out.Encoding())
 			test.AssertEventEquals(t, eventIn, test.ExToStr(t, eventOut))
 		})
 	})
@@ -66,8 +66,8 @@ func TestSendStructReceiveStruct(t *testing.T) {
 		eventIn = test.ExToStr(t, eventIn)
 		in := test.MustCreateMockStructuredMessage(eventIn)
 		test.SendReceive(t, context.Background(), in, s, r, func(out binding.Message) {
-			eventOut, encoding := test.MustToEvent(context.Background(), out)
-			assert.Equal(t, encoding, binding.EncodingStructured)
+			eventOut := test.MustToEvent(t, context.Background(), out)
+			require.Equal(t, binding.EncodingStructured, out.Encoding())
 			test.AssertEventEquals(t, eventIn, test.ExToStr(t, eventOut))
 		})
 	})
@@ -80,8 +80,8 @@ func TestSendEventReceiveBinary(t *testing.T) {
 		eventIn = test.ExToStr(t, eventIn)
 		in := binding.EventMessage(eventIn)
 		test.SendReceive(t, context.Background(), in, s, r, func(out binding.Message) {
-			eventOut, encoding := test.MustToEvent(context.Background(), out)
-			assert.Equal(t, encoding, binding.EncodingBinary)
+			eventOut := test.MustToEvent(t, context.Background(), out)
+			require.Equal(t, binding.EncodingBinary, out.Encoding())
 			test.AssertEventEquals(t, eventIn, test.ExToStr(t, eventOut))
 		})
 	})
diff --git a/test/integration/kafka_sarama_binding/kafka_test.go b/test/integration/kafka_sarama_binding/kafka_test.go
index 2f21f32..0d3c721 100644
--- a/test/integration/kafka_sarama_binding/kafka_test.go
+++ b/test/integration/kafka_sarama_binding/kafka_test.go
@@ -30,8 +30,8 @@ func TestSendStructuredMessageToStructuredWithKey(t *testing.T) {
 
 		in := test.MustCreateMockStructuredMessage(eventIn)
 		test.SendReceive(t, binding.WithPreferredEventEncoding(context.TODO(), binding.EncodingStructured), in, s, r, func(out binding.Message) {
-			eventOut, encoding := test.MustToEvent(context.Background(), out)
-			assert.Equal(t, binding.EncodingEvent, encoding)
+			eventOut := test.MustToEvent(t, context.Background(), out)
+			assert.Equal(t, binding.EncodingEvent, out.Encoding())
 			test.AssertEventEquals(t, eventIn, test.ExToStr(t, eventOut))
 		})
 	})
@@ -45,8 +45,8 @@ func TestSendStructuredMessageToStructuredWithoutKey(t *testing.T) {
 
 		in := test.MustCreateMockStructuredMessage(eventIn)
 		test.SendReceive(t, binding.WithPreferredEventEncoding(context.TODO(), binding.EncodingStructured), in, s, r, func(out binding.Message) {
-			eventOut, encoding := test.MustToEvent(context.Background(), out)
-			assert.Equal(t, binding.EncodingStructured, encoding)
+			eventOut := test.MustToEvent(t, context.Background(), out)
+			assert.Equal(t, binding.EncodingStructured, out.Encoding())
 			test.AssertEventEquals(t, eventIn, test.ExToStr(t, eventOut))
 		})
 	})
@@ -61,8 +61,8 @@ func TestSendBinaryMessageToBinaryWithKey(t *testing.T) {
 
 		in := test.MustCreateMockBinaryMessage(eventIn)
 		test.SendReceive(t, binding.WithPreferredEventEncoding(context.TODO(), binding.EncodingBinary), in, s, r, func(out binding.Message) {
-			eventOut, encoding := test.MustToEvent(context.Background(), out)
-			assert.Equal(t, binding.EncodingBinary, encoding)
+			eventOut := test.MustToEvent(t, context.Background(), out)
+			assert.Equal(t, binding.EncodingBinary, out.Encoding())
 			test.AssertEventEquals(t, eventIn, test.ExToStr(t, eventOut))
 		})
 	})
@@ -76,8 +76,8 @@ func TestSendBinaryMessageToBinaryWithoutKey(t *testing.T) {
 
 		in := test.MustCreateMockBinaryMessage(eventIn)
 		test.SendReceive(t, binding.WithPreferredEventEncoding(context.TODO(), binding.EncodingBinary), in, s, r, func(out binding.Message) {
-			eventOut, encoding := test.MustToEvent(context.Background(), out)
-			assert.Equal(t, binding.EncodingBinary, encoding)
+			eventOut := test.MustToEvent(t, context.Background(), out)
+			assert.Equal(t, binding.EncodingBinary, out.Encoding())
 			test.AssertEventEquals(t, eventIn, test.ExToStr(t, eventOut))
 		})
 	})