Repository: cloudevents/sdk-go
Pull request #381 (2020-03-16 09:09:51)

Removed wrong implementation of the skip key extension
After discussion during Serverless WG on Thu, we figure it out that our kafka implementation is wrong. This PR removes the bad code.

Signed-off-by: Francesco Guardiani <francescoguard@gmail.com>


Patch:
diff --git a/alias.go b/alias.go
index c078c7f..50b3670 100644
--- a/alias.go
+++ b/alias.go
@@ -36,9 +36,8 @@ type URIRef = types.URIRef
 // HTTP Protocol
 
 type HTTPOption http.Option
-type HTTPProtocolOption http.ProtocolOption
 
-type HTTPTransport = http.Protocol
+type HTTPProtocol = http.Protocol
 type HTTPEncoding = http.Encoding
 
 const (
@@ -72,8 +71,9 @@ var (
 
 	// Client Creation
 
-	NewClient        = client.New
-	NewDefaultClient = client.NewDefault
+	NewClient             = client.New
+	NewDefaultClient      = client.NewDefault
+	NewHTTPReceiveHandler = client.NewHTTPReceiveHandler
 
 	// Client Options
 
@@ -108,19 +108,18 @@ var (
 
 	// HTTP Protocol
 
-	NewHTTPTransport = http.New
-	NewHTTPProtocol  = http.NewProtocol
+	NewHTTP = http.New
 
 	// HTTP Protocol Options
 
-	WithTarget             = http.WithTarget
-	WithHeader             = http.WithHeader
-	WithShutdownTimeout    = http.WithShutdownTimeout
-	WithEncoding           = http.WithEncoding
-	WithStructuredEncoding = http.WithStructuredEncoding
-	WithPort               = http.WithPort
-	WithPath               = http.WithPath
-	WithMiddleware         = http.WithMiddleware
-	WithListener           = http.WithListener
-	WithHTTPTransport      = http.WithHTTPTransport
+	WithTarget          = http.WithTarget
+	WithHeader          = http.WithHeader
+	WithShutdownTimeout = http.WithShutdownTimeout
+	//WithEncoding           = http.WithEncoding
+	//WithStructuredEncoding = http.WithStructuredEncoding // TODO: expose new way
+	WithPort          = http.WithPort
+	WithPath          = http.WithPath
+	WithMiddleware    = http.WithMiddleware
+	WithListener      = http.WithListener
+	WithHTTPTransport = http.WithHTTPTransport
 )
diff --git a/cmd/samples/complex/sender/main.go b/cmd/samples/complex/sender/main.go
deleted file mode 100644
index 39049fc..0000000
--- a/cmd/samples/complex/sender/main.go
+++ /dev/null
@@ -1,172 +0,0 @@
-package main
-
-import (
-	"context"
-	"fmt"
-	"log"
-	"net/url"
-	"os"
-	"time"
-
-	cloudevents "github.com/cloudevents/sdk-go"
-	"github.com/cloudevents/sdk-go/pkg/client"
-	"github.com/cloudevents/sdk-go/pkg/event"
-	"github.com/cloudevents/sdk-go/pkg/transport"
-	cloudeventshttp "github.com/cloudevents/sdk-go/pkg/transport/http"
-	cloudeventsnats "github.com/cloudevents/sdk-go/pkg/transport/nats"
-	"github.com/cloudevents/sdk-go/pkg/types"
-	"github.com/kelseyhightower/envconfig"
-)
-
-const (
-	count = 1
-	delay = 100 * time.Millisecond
-)
-
-type envConfig struct {
-	// HTTPTarget is the target URL where to send cloudevents
-	HTTPTarget string `envconfig:"HTTP_TARGET" default:"http://localhost:8080" required:"true"`
-
-	// NATSServer URL to connect to the nats server.
-	NATSServer string `envconfig:"NATS_SERVER" default:"http://localhost:4222" required:"true"`
-
-	// Subject is the nats subject to publish cloudevents on.
-	Subject string `envconfig:"SUBJECT" default:"sample" required:"true"`
-}
-
-func main() {
-	var env envConfig
-	if err := envconfig.Process("", &env); err != nil {
-		log.Printf("[ERROR] Failed to process env var: %s", err)
-		os.Exit(1)
-	}
-	os.Exit(_main(os.Args[1:], env))
-}
-
-// Simple holder for the sending sample.
-type Demo struct {
-	Client client.Client
-
-	// Content
-	EventType   string
-	Source      url.URL
-	ContentType string
-
-	// Data
-	Message string
-}
-
-var seq int
-
-// Basic data struct.
-type Example struct {
-	Sequence int    `json:"id"`
-	Message  string `json:"message"`
-}
-
-func (d *Demo) Send() error {
-	e := cloudevents.Event{
-		Context: d.context(),
-	}
-	_ = e.SetData(cloudevents.ApplicationJSON, &Example{
-		Sequence: seq,
-		Message:  d.Message,
-	})
-	seq++
-	return d.Client.Send(context.Background(), e)
-}
-
-func (d *Demo) context() cloudevents.EventContext {
-	ctx := event.EventContextV1{
-		Type:            d.EventType,
-		Source:          types.URIRef{URL: d.Source},
-		DataContentType: &d.ContentType,
-	}.AsV1()
-	return ctx
-}
-
-func _main(args []string, env envConfig) int {
-	source, err := url.Parse("https://github.com/cloudevents/sdk-go/cmd/samples/sender")
-	if err != nil {
-		log.Printf("failed to parse source url, %v", err)
-		return 1
-	}
-
-	for _, contentType := range []string{"application/json", "application/xml"} {
-		// HTTP
-		for _, encoding := range []cloudeventshttp.Encoding{cloudeventshttp.Default, cloudeventshttp.Binary, cloudeventshttp.Structured} {
-
-			p, err := cloudeventshttp.NewProtocol(cloudeventshttp.WithTarget(env.HTTPTarget))
-			if err != nil {
-				log.Printf("failed to create protocol, %v", err)
-				return 1
-			}
-
-			t, err := cloudeventshttp.New(p,
-				cloudeventshttp.WithEncoding(encoding),
-			)
-			if err != nil {
-				log.Printf("failed to create transport, %v", err)
-				return 1
-			}
-
-			if err := doDemo(
-				t,
-				"com.cloudevents.sample.http.sent",
-				fmt.Sprintf("Hello %d, using %s!", encoding, contentType),
-				contentType,
-				*source,
-			); err != nil {
-				log.Printf("failed to do http demo: %v, %s", err, contentType)
-				return 1
-			}
-		}
-
-		// NATS
-		t, err := cloudeventsnats.New(
-			env.NATSServer,
-			env.Subject,
-		)
-		if err != nil {
-			log.Printf("failed to create client, %v", err)
-			return 1
-		}
-		if err := doDemo(
-			t.Transport(),
-			"com.cloudevents.sample.nats.sent",
-			fmt.Sprintf("Hello NATS, using %s!", contentType),
-			contentType,
-			*source,
-		); err != nil {
-			log.Printf("failed to do nats demo: %v, %s", err, contentType)
-			return 1
-		}
-	}
-
-	return 0
-}
-
-func doDemo(t transport.Transport, eventType, message, contentType string, source url.URL) error {
-	c, err := client.New(t,
-		client.WithUUIDs(),
-		client.WithTimeNow(),
-	)
-	if err != nil {
-		return err
-	}
-
-	d := &Demo{
-		Message:     message,
-		Client:      c,
-		Source:      source,
-		EventType:   eventType,
-		ContentType: contentType,
-	}
-	for i := 0; i < count; i++ {
-		if err := d.Send(); err != nil {
-			return err
-		}
-		time.Sleep(delay)
-	}
-	return nil
-}
diff --git a/cmd/samples/http/receiver/main.go b/cmd/samples/http/receiver/main.go
index 5ad29a4..821dac3 100644
--- a/cmd/samples/http/receiver/main.go
+++ b/cmd/samples/http/receiver/main.go
@@ -45,20 +45,12 @@ func gotEvent(ctx context.Context, event cloudevents.Event) error {
 func _main(args []string, env envConfig) int {
 	ctx := context.Background()
 
-	p, err := cloudevents.NewHTTPProtocol()
+	p, err := cloudevents.NewHTTP(cloudevents.WithPort(env.Port), cloudevents.WithPath(env.Path))
 	if err != nil {
 		log.Fatalf("failed to create protocol: %s", err.Error())
 	}
 
-	t, err := cloudevents.NewHTTPTransport(p,
-		cloudevents.WithPort(env.Port),
-		cloudevents.WithPath(env.Path),
-	)
-	if err != nil {
-		log.Printf("failed to create transport, %v", err)
-		return 1
-	}
-	c, err := cloudevents.NewClient(t)
+	c, err := cloudevents.NewClient(p)
 	if err != nil {
 		log.Printf("failed to create client, %v", err)
 		return 1
diff --git a/cmd/samples/http/requester-with-custom-client/main.go b/cmd/samples/http/requester-with-custom-client/main.go
index 09ed9a0..7b2ed38 100644
--- a/cmd/samples/http/requester-with-custom-client/main.go
+++ b/cmd/samples/http/requester-with-custom-client/main.go
@@ -7,7 +7,6 @@ import (
 	"fmt"
 	"io/ioutil"
 	"log"
-	"net/http"
 	"net/url"
 	"os"
 	"time"
@@ -76,25 +75,26 @@ func _main(args []string, env envConfig) int {
 	for _, dataContentType := range []string{"application/json", "application/xml"} {
 		for _, encoding := range []cloudevents.HTTPEncoding{cloudevents.HTTPBinaryEncoding, cloudevents.HTTPStructuredEncoding} {
 
-			p, err := cloudevents.NewHTTPProtocol(cloudevents.WithTarget(env.Target))
+			p, err := cloudevents.NewHTTP(cloudevents.WithTarget(env.Target))
 			if err != nil {
 				log.Printf("failed to create protocol, %v", err)
 				return 1
 			}
 
-			t, err := cloudevents.NewHTTPTransport(
-				p,
-				cloudevents.WithEncoding(encoding),
-				// Use custom transport
-				cloudevents.WithHTTPTransport(&http.Transport{TLSClientConfig: tlsConfig}),
-			)
-
-			if err != nil {
-				log.Printf("failed to create transport, %v", err)
-				return 1
-			}
-
-			c, err := cloudevents.NewClient(t,
+			// TODO: redo this demo
+			//t, err := cloudevents.NewHTTPTransport(
+			//	p,
+			//	cloudevents.WithEncoding(encoding),
+			//	// Use custom transport
+			//	cloudevents.WithHTTPTransport(&http.Transport{TLSClientConfig: tlsConfig}),
+			//)
+			//
+			//if err != nil {
+			//	log.Printf("failed to create transport, %v", err)
+			//	return 1
+			//}
+
+			c, err := cloudevents.NewClient(p,
 				cloudevents.WithTimeNow(),
 			)
 			if err != nil {
diff --git a/cmd/samples/http/requester/main.go b/cmd/samples/http/requester/main.go
index 6a7be91..2da7c8c 100644
--- a/cmd/samples/http/requester/main.go
+++ b/cmd/samples/http/requester/main.go
@@ -46,23 +46,15 @@ func _main(args []string, env envConfig) int {
 
 	seq := 0
 	for _, contentType := range []string{"application/json", "application/xml", "text/plain"} {
-		for _, encoding := range []cloudevents.HTTPEncoding{cloudevents.HTTPBinaryEncoding, cloudevents.HTTPStructuredEncoding} {
+		for _, encoding := range []cloudevents.HTTPEncoding{cloudevents.HTTPBinaryEncoding, cloudevents.HTTPStructuredEncoding} { // TODO: force this again or remove
 
-			p, err := cloudevents.NewHTTPProtocol(cloudevents.WithTarget(env.Target))
+			p, err := cloudevents.NewHTTP(cloudevents.WithTarget(env.Target))
 			if err != nil {
 				log.Printf("failed to create protocol, %v", err)
 				return 1
 			}
 
-			t, err := cloudevents.NewHTTPTransport(p,
-				cloudevents.WithEncoding(encoding),
-			)
-			if err != nil {
-				log.Printf("failed to create transport, %v", err)
-				return 1
-			}
-
-			c, err := cloudevents.NewClient(t,
+			c, err := cloudevents.NewClient(p,
 				cloudevents.WithTimeNow(),
 			)
 			if err != nil {
diff --git a/cmd/samples/http/responder/main.go b/cmd/samples/http/responder/main.go
index 362ef33..79a1611 100644
--- a/cmd/samples/http/responder/main.go
+++ b/cmd/samples/http/responder/main.go
@@ -59,19 +59,11 @@ func gotEvent(ctx context.Context, event cloudevents.Event) (*cloudevents.Event,
 func _main(args []string, env envConfig) int {
 	ctx := context.Background()
 
-	p, err := cloudevents.NewHTTPProtocol()
+	p, err := cloudevents.NewHTTP(cloudevents.WithPort(env.Port), cloudevents.WithPath(env.Path))
 	if err != nil {
 		log.Fatalf("failed to create protocol: %s", err.Error())
 	}
-
-	t, err := cloudevents.NewHTTPTransport(p,
-		cloudevents.WithPort(env.Port),
-		cloudevents.WithPath(env.Path),
-	)
-	if err != nil {
-		log.Fatalf("failed to create transport: %s", err.Error())
-	}
-	c, err := cloudevents.NewClient(t,
+	c, err := cloudevents.NewClient(p,
 		cloudevents.WithUUIDs(),
 		cloudevents.WithTimeNow(),
 	)
diff --git a/cmd/samples/http/sleepy/main.go b/cmd/samples/http/sleepy/main.go
index 972a224..4f23912 100644
--- a/cmd/samples/http/sleepy/main.go
+++ b/cmd/samples/http/sleepy/main.go
@@ -43,20 +43,11 @@ func gotEvent(event cloudevents.Event) error {
 }
 
 func _main(args []string, env envConfig) int {
-	p, err := cloudevents.NewHTTPProtocol()
+	p, err := cloudevents.NewHTTP(cloudevents.WithPort(env.Port), cloudevents.WithPath(env.Path))
 	if err != nil {
 		log.Fatalf("failed to create protocol: %s", err.Error())
 	}
-
-	t, err := cloudevents.NewHTTPTransport(p,
-		cloudevents.WithPort(env.Port),
-		cloudevents.WithPath(env.Path),
-	)
-	if err != nil {
-		log.Printf("failed to create transport, %v", err)
-		return 1
-	}
-	c, err := cloudevents.NewClient(t)
+	c, err := cloudevents.NewClient(p)
 	if err != nil {
 		log.Printf("failed to create client, %v", err)
 		return 1
diff --git a/cmd/samples/httpb/receiver-direct/main.go b/cmd/samples/httpb/receiver-direct/main.go
new file mode 100644
index 0000000..ac4aaf9
--- /dev/null
+++ b/cmd/samples/httpb/receiver-direct/main.go
@@ -0,0 +1,32 @@
+package main
+
+import (
+	"context"
+	"fmt"
+	"log"
+	"net/http"
+
+	cloudevents "github.com/cloudevents/sdk-go"
+)
+
+func main() {
+	ctx := context.Background()
+	p, err := cloudevents.NewHTTP()
+	if err != nil {
+		log.Fatalf("failed to create protocol: %s", err.Error())
+	}
+
+	h, err := cloudevents.NewHTTPReceiveHandler(ctx, p, receive)
+	if err != nil {
+		log.Fatalf("failed to create handler: %s", err.Error())
+	}
+
+	log.Printf("will listen on :8080\n")
+	if err := http.ListenAndServe(":8080", h); err != nil {
+		log.Fatalf("unable to start http server, %s", err)
+	}
+}
+
+func receive(ctx context.Context, event cloudevents.Event) {
+	fmt.Printf("%s", event)
+}
diff --git a/cmd/samples/httpb/receiver/main.go b/cmd/samples/httpb/receiver/main.go
index fc49133..77d8c31 100644
--- a/cmd/samples/httpb/receiver/main.go
+++ b/cmd/samples/httpb/receiver/main.go
@@ -5,24 +5,17 @@ import (
 	"fmt"
 	"log"
 
-	"github.com/cloudevents/sdk-go/pkg/transport/http"
-
 	cloudevents "github.com/cloudevents/sdk-go"
 )
 
 func main() {
 	ctx := context.Background()
-	p, err := cloudevents.NewHTTPProtocol()
+	p, err := cloudevents.NewHTTP()
 	if err != nil {
 		log.Fatalf("failed to create protocol: %s", err.Error())
 	}
 
-	t, err := http.New(p)
-	if err != nil {
-		log.Fatalf("failed to create transport, %v", err)
-	}
-
-	c, err := cloudevents.NewClient(t)
+	c, err := cloudevents.NewClient(p)
 	if err != nil {
 		log.Fatalf("failed to create client, %v", err)
 	}
diff --git a/cmd/samples/httpb/requester/main.go b/cmd/samples/httpb/requester/main.go
index a02c8de..9c2f32c 100644
--- a/cmd/samples/httpb/requester/main.go
+++ b/cmd/samples/httpb/requester/main.go
@@ -5,23 +5,17 @@ import (
 	"log"
 
 	cloudevents "github.com/cloudevents/sdk-go"
-	"github.com/cloudevents/sdk-go/pkg/transport/http"
 )
 
 func main() {
 	ctx := cloudevents.ContextWithTarget(context.Background(), "http://localhost:8080/")
 
-	p, err := cloudevents.NewHTTPProtocol()
+	p, err := cloudevents.NewHTTP()
 	if err != nil {
 		log.Fatalf("failed to create protocol: %s", err.Error())
 	}
 
-	t, err := http.New(p)
-	if err != nil {
-		log.Fatalf("failed to create transport, %v", err)
-	}
-
-	c, err := cloudevents.NewClient(t, cloudevents.WithTimeNow(), cloudevents.WithUUIDs())
+	c, err := cloudevents.NewClient(p, cloudevents.WithTimeNow(), cloudevents.WithUUIDs())
 	if err != nil {
 		log.Fatalf("failed to create client, %v", err)
 	}
diff --git a/cmd/samples/httpb/responder/main.go b/cmd/samples/httpb/responder/main.go
index e3d17e9..97f2e3b 100644
--- a/cmd/samples/httpb/responder/main.go
+++ b/cmd/samples/httpb/responder/main.go
@@ -28,17 +28,12 @@ func main() {
 	}
 	ctx := context.Background()
 
-	p, err := cloudevents.NewHTTPProtocol()
+	p, err := cloudevents.NewHTTP()
 	if err != nil {
 		log.Fatalf("failed to create protocol: %s", err.Error())
 	}
 
-	t, err := cloudevents.NewHTTPTransport(p)
-	if err != nil {
-		log.Fatalf("failed to create transport, %v", err)
-	}
-
-	c, err := cloudevents.NewClient(t,
+	c, err := cloudevents.NewClient(p,
 		cloudevents.WithUUIDs(),
 		cloudevents.WithTimeNow(),
 		cloudevents.WithoutTracePropagation(),
@@ -47,7 +42,7 @@ func main() {
 		log.Fatalf("failed to create client: %s", err.Error())
 	}
 
-	log.Printf("listening on :%d%s\n", t.GetPort(), t.GetPath())
+	log.Printf("listening on :%d%s\n", p.GetPort(), p.GetPath())
 	if err := c.StartReceiver(ctx, gotEvent); err != nil {
 		log.Fatalf("failed to start receiver: %s", err.Error())
 	}
diff --git a/cmd/samples/httpb/sender/main.go b/cmd/samples/httpb/sender/main.go
index 0fd5cf2..cf21ef4 100644
--- a/cmd/samples/httpb/sender/main.go
+++ b/cmd/samples/httpb/sender/main.go
@@ -5,23 +5,17 @@ import (
 	"log"
 
 	cloudevents "github.com/cloudevents/sdk-go"
-	"github.com/cloudevents/sdk-go/pkg/transport/http"
 )
 
 func main() {
 	ctx := cloudevents.ContextWithTarget(context.Background(), "http://localhost:8080/")
 
-	p, err := cloudevents.NewHTTPProtocol()
+	p, err := cloudevents.NewHTTP()
 	if err != nil {
 		log.Fatalf("failed to create protocol: %s", err.Error())
 	}
 
-	t, err := http.New(p)
-	if err != nil {
-		log.Fatalf("failed to create transport, %v", err)
-	}
-
-	c, err := cloudevents.NewClient(t, cloudevents.WithTimeNow(), cloudevents.WithUUIDs())
+	c, err := cloudevents.NewClient(p, cloudevents.WithTimeNow(), cloudevents.WithUUIDs())
 	if err != nil {
 		log.Fatalf("failed to create client, %v", err)
 	}
diff --git a/cmd/samples/complex/httptonats/main.go b/cmd/samples/httptonats/main.go
similarity index 86%
rename from cmd/samples/complex/httptonats/main.go
rename to cmd/samples/httptonats/main.go
index fef738a..f771c80 100644
--- a/cmd/samples/complex/httptonats/main.go
+++ b/cmd/samples/httptonats/main.go
@@ -64,11 +64,11 @@ func (r *Receiver) Receive(event event.Event) error {
 func _main(args []string, env envConfig) int {
 	ctx := context.Background()
 
-	nt, err := cloudeventsnats.New(env.NATSServer, env.Subject)
+	np, err := cloudeventsnats.New(env.NATSServer, env.Subject)
 	if err != nil {
 		log.Fatalf("failed to create nats transport, %s", err.Error())
 	}
-	nc, err := client.New(nt.Transport())
+	nc, err := client.New(np)
 	if err != nil {
 		log.Printf("failed to create client, %v", err)
 		return 1
@@ -76,19 +76,12 @@ func _main(args []string, env envConfig) int {
 
 	r := &Receiver{Client: nc}
 
-	p, err := cloudeventshttp.NewProtocol()
+	p, err := cloudeventshttp.New(cloudeventshttp.WithPort(env.Port))
 	if err != nil {
 		log.Fatalf("failed to create protocol: %s", err.Error())
 	}
 
-	t, err := cloudeventshttp.New(p,
-		cloudeventshttp.WithPort(env.Port),
-	)
-	if err != nil {
-		log.Printf("failed to create transport, %v", err)
-		return 1
-	}
-	c, err := client.New(t)
+	c, err := client.New(p)
 	if err != nil {
 		log.Printf("failed to create client, %v", err)
 		return 1
diff --git a/cmd/samples/nats/receiver/main.go b/cmd/samples/nats/receiver/main.go
index 23292ba..f28277e 100644
--- a/cmd/samples/nats/receiver/main.go
+++ b/cmd/samples/nats/receiver/main.go
@@ -28,11 +28,11 @@ func main() {
 	}
 	ctx := context.Background()
 
-	t, err := cloudeventsnats.New(env.NATSServer, env.Subject)
+	p, err := cloudeventsnats.New(env.NATSServer, env.Subject)
 	if err != nil {
 		log.Fatalf("failed to create nats transport, %s", err.Error())
 	}
-	c, err := client.New(t.Transport())
+	c, err := client.New(p)
 	if err != nil {
 		log.Fatalf("failed to create client, %s", err.Error())
 	}
diff --git a/cmd/samples/nats/sender/main.go b/cmd/samples/nats/sender/main.go
index b2c75d4..0415094 100644
--- a/cmd/samples/nats/sender/main.go
+++ b/cmd/samples/nats/sender/main.go
@@ -3,16 +3,13 @@ package main
 import (
 	"context"
 	"fmt"
-	cloudevents "github.com/cloudevents/sdk-go"
 	"log"
-	"net/url"
 	"os"
 	"time"
 
 	"github.com/cloudevents/sdk-go/pkg/client"
 	"github.com/cloudevents/sdk-go/pkg/event"
 	cloudeventsnats "github.com/cloudevents/sdk-go/pkg/transport/nats"
-	"github.com/cloudevents/sdk-go/pkg/types"
 	"github.com/google/uuid"
 	"github.com/kelseyhightower/envconfig"
 )
@@ -29,82 +26,50 @@ type envConfig struct {
 	Subject string `envconfig:"SUBJECT" default:"sample" required:"true"`
 }
 
-func main() {
-	var env envConfig
-	if err := envconfig.Process("", &env); err != nil {
-		log.Printf("[ERROR] Failed to process env var: %s", err)
-		os.Exit(1)
-	}
-	os.Exit(_main(os.Args[1:], env))
-}
-
-// Simple holder for the sending sample.
-type Demo struct {
-	Message string
-	Source  url.URL
-	Target  url.URL
-
-	Client client.Client
-}
-
-// Basic data struct.
 type Example struct {
 	Sequence int    `json:"id"`
 	Message  string `json:"message"`
 }
 
-func (d *Demo) Send(eventContext event.EventContext, i int) error {
-	e := event.Event{Context: eventContext}
-	_ = e.SetData(cloudevents.ApplicationJSON, &Example{
-		Sequence: i,
-		Message:  d.Message,
-	})
-	return d.Client.Send(context.Background(), e)
-}
-
-func _main(args []string, env envConfig) int {
-	source, err := url.Parse("https://github.com/cloudevents/sdk-go/cmd/samples/sender")
-	if err != nil {
-		log.Printf("failed to parse source url, %v", err)
-		return 1
+func main() {
+	var env envConfig
+	if err := envconfig.Process("", &env); err != nil {
+		log.Printf("[ERROR] Failed to process env var: %s", err)
+		os.Exit(1)
 	}
 
-	seq := 0
 	for _, contentType := range []string{"application/json", "application/xml"} {
-		t, err := cloudeventsnats.New(env.NATSServer, env.Subject)
+		p, err := cloudeventsnats.New(env.NATSServer, env.Subject)
 		if err != nil {
 			log.Printf("failed to create nats transport, %s", err.Error())
-			return 1
+			os.Exit(1)
 		}
-		c, err := client.New(t.Transport())
+		c, err := client.New(p)
 		if err != nil {
 			log.Printf("failed to create client, %s", err.Error())
-			return 1
-		}
-
-		d := &Demo{
-			Message: fmt.Sprintf("Hello, %s!", contentType),
-			Source:  *source,
-			Client:  c,
+			os.Exit(1)
 		}
 
 		for i := 0; i < count; i++ {
 			now := time.Now()
-			ctx := event.EventContextV1{
-				ID:              uuid.New().String(),
-				Type:            "com.cloudevents.sample.sent",
-				Time:            &types.Timestamp{Time: now},
-				Source:          types.URIRef{URL: d.Source},
-				DataContentType: &contentType,
-			}.AsV1()
-			if err := d.Send(ctx, seq); err != nil {
+			e := event.New()
+			e.SetID(uuid.New().String())
+
+			e.SetType("com.cloudevents.sample.sent")
+			e.SetTime(now)
+			e.SetSource("https://github.com/cloudevents/sdk-go/cmd/samples/sender")
+			_ = e.SetData(contentType, &Example{
+				Sequence: i,
+				Message:  fmt.Sprintf("Hello, %s!", contentType),
+			})
+
+			if err := c.Send(context.Background(), e); err != nil {
 				log.Printf("failed to send: %v", err)
-				return 1
+				os.Exit(1)
 			}
-			seq++
 			time.Sleep(100 * time.Millisecond)
 		}
 	}
 
-	return 0
+	os.Exit(0)
 }
diff --git a/pkg/binding/binary_writer.go b/pkg/binding/binary_writer.go
index 3ebb254..36211d3 100644
--- a/pkg/binding/binary_writer.go
+++ b/pkg/binding/binary_writer.go
@@ -35,5 +35,5 @@ type BinaryWriter interface {
 
 	// End method is invoked only after the whole encoding process ends successfully.
 	// If it fails, it's never invoked. It can be used to finalize the message.
-	End() error
+	End(ctx context.Context) error
 }
diff --git a/pkg/binding/buffering/binary_buffer_message.go b/pkg/binding/buffering/binary_buffer_message.go
index 91454f1..8386b1a 100644
--- a/pkg/binding/buffering/binary_buffer_message.go
+++ b/pkg/binding/buffering/binary_buffer_message.go
@@ -27,7 +27,7 @@ func (m *binaryBufferedMessage) Start(ctx context.Context) error {
 	return nil
 }
 
-func (m *binaryBufferedMessage) End() error {
+func (m *binaryBufferedMessage) End(ctx context.Context) error {
 	return nil
 }
 
@@ -66,7 +66,7 @@ func (m *binaryBufferedMessage) ReadBinary(ctx context.Context, b binding.Binary
 			return
 		}
 	}
-	return b.End()
+	return b.End(ctx)
 }
 
 func (b *binaryBufferedMessage) Finish(error) error {
diff --git a/pkg/binding/event_message.go b/pkg/binding/event_message.go
index 8450ec4..b515144 100644
--- a/pkg/binding/event_message.go
+++ b/pkg/binding/event_message.go
@@ -55,7 +55,7 @@ func (m *EventMessage) ReadBinary(ctx context.Context, b BinaryWriter) (err erro
 			return err
 		}
 	}
-	return b.End()
+	return b.End(ctx)
 }
 
 func eventContextToBinaryWriter(c event.EventContext, b BinaryWriter) (err error) {
diff --git a/pkg/binding/to_event.go b/pkg/binding/to_event.go
index afa9fb5..b62262b 100644
--- a/pkg/binding/to_event.go
+++ b/pkg/binding/to_event.go
@@ -76,7 +76,7 @@ func (b *messageToEventBuilder) Start(ctx context.Context) error {
 	return nil
 }
 
-func (b *messageToEventBuilder) End() error {
+func (b *messageToEventBuilder) End(ctx context.Context) error {
 	return nil
 }
 
diff --git a/pkg/binding/transformer/add_metadata.go b/pkg/binding/transformer/add_metadata.go
index 23805ed..b94faf3 100644
--- a/pkg/binding/transformer/add_metadata.go
+++ b/pkg/binding/transformer/add_metadata.go
@@ -1,6 +1,7 @@
 package transformer
 
 import (
+	"context"
 	"fmt"
 
 	"github.com/cloudevents/sdk-go/pkg/binding"
@@ -92,14 +93,14 @@ func (b *addAttributeTransformer) SetAttribute(attribute spec.Attribute, value i
 	return b.BinaryWriter.SetAttribute(attribute, value)
 }
 
-func (b *addAttributeTransformer) End() error {
+func (b *addAttributeTransformer) End(ctx context.Context) error {
 	if !b.found {
 		err := b.BinaryWriter.SetAttribute(b.version.AttributeFromKind(b.attributeKind), b.value)
 		if err != nil {
 			return err
 		}
 	}
-	return b.BinaryWriter.End()
+	return b.BinaryWriter.End(ctx)
 }
 
 type addExtensionTransformer struct {
@@ -116,12 +117,12 @@ func (b *addExtensionTransformer) SetExtension(name string, value interface{}) e
 	return b.BinaryWriter.SetExtension(name, value)
 }
 
-func (b *addExtensionTransformer) End() error {
+func (b *addExtensionTransformer) End(ctx context.Context) error {
 	if !b.found {
 		err := b.BinaryWriter.SetExtension(b.name, b.value)
 		if err != nil {
 			return err
 		}
 	}
-	return b.BinaryWriter.End()
+	return b.BinaryWriter.End(ctx)
 }
diff --git a/pkg/binding/transformer/default_metadata.go b/pkg/binding/transformer/default_metadata.go
index a68becb..367190e 100644
--- a/pkg/binding/transformer/default_metadata.go
+++ b/pkg/binding/transformer/default_metadata.go
@@ -1,6 +1,7 @@
 package transformer
 
 import (
+	"context"
 	"time"
 
 	"github.com/google/uuid"
@@ -83,12 +84,12 @@ func (b *addTimeNowTransformer) SetAttribute(attribute spec.Attribute, value int
 	return b.BinaryWriter.SetAttribute(attribute, value)
 }
 
-func (b *addTimeNowTransformer) End() error {
+func (b *addTimeNowTransformer) End(ctx context.Context) error {
 	if !b.found {
 		err := b.BinaryWriter.SetAttribute(b.version.AttributeFromKind(spec.Time), types.Timestamp{Time: time.Now()})
 		if err != nil {
 			return err
 		}
 	}
-	return b.BinaryWriter.End()
+	return b.BinaryWriter.End(ctx)
 }
diff --git a/pkg/client/client.go b/pkg/client/client.go
index 1c62cd5..bfc386c 100644
--- a/pkg/client/client.go
+++ b/pkg/client/client.go
@@ -2,15 +2,16 @@ package client
 
 import (
 	"context"
+	"errors"
 	"fmt"
+	"github.com/cloudevents/sdk-go/pkg/binding"
+	cecontext "github.com/cloudevents/sdk-go/pkg/context"
+	"go.uber.org/zap"
+	"io"
 	"sync"
 
 	"github.com/cloudevents/sdk-go/pkg/event"
-	"github.com/cloudevents/sdk-go/pkg/extensions"
-	"github.com/cloudevents/sdk-go/pkg/observability"
 	"github.com/cloudevents/sdk-go/pkg/transport"
-	"github.com/cloudevents/sdk-go/pkg/transport/http"
-	"go.opencensus.io/trace"
 )
 
 // Client interface defines the runtime contract the CloudEvents client supports.
@@ -44,196 +45,111 @@ type Client interface {
 
 // New produces a new client with the provided transport object and applied
 // client options.
-func New(t transport.Transport, opts ...Option) (Client, error) {
-	c := &ceClient{
-		transport: t,
+func New(protocol interface{}, opts ...Option) (Client, error) {
+	c := &ceClient{}
+
+	if p, ok := protocol.(transport.Sender); ok {
+		c.sender = p
 	}
-	if err := c.applyOptions(opts...); err != nil {
-		return nil, err
+	if p, ok := protocol.(transport.Requester); ok {
+		c.requester = p
 	}
-	t.SetDelivery(c)
-	return c, nil
-}
-
-// NewDefault provides the good defaults for the common case using an HTTP
-// Protocol client. The http transport has had WithBinaryEncoding http
-// transport option applied to it. The client will always send Binary
-// encoding but will inspect the outbound event context and match the version.
-// The WithTimeNow, and WithUUIDs client options are also applied to the
-// client, all outbound events will have a time and id set if not already
-// present.
-func NewDefault() (Client, error) {
-	p, err := http.NewProtocol()
-	if err != nil {
-		return nil, err
+	if p, ok := protocol.(transport.Responder); ok {
+		c.responder = p
 	}
-	t, err := http.New(p, http.WithEncoding(http.Binary))
-	if err != nil {
-		return nil, err
+	if p, ok := protocol.(transport.Receiver); ok {
+		c.receiver = p
 	}
-	c, err := New(t, WithTimeNow(), WithUUIDs())
-	if err != nil {
+	if p, ok := protocol.(transport.Engine); ok {
+		c.engine = p
+	}
+
+	if err := c.applyOptions(opts...); err != nil {
 		return nil, err
 	}
 	return c, nil
 }
 
 type ceClient struct {
-	transport transport.Transport
-	fn        *receiverFn
-
-	receiverMu        sync.Mutex
-	eventDefaulterFns []EventDefaulter
-
-	disableTracePropagation bool
+	sender    transport.Sender
+	requester transport.Requester
+	receiver  transport.Receiver
+	responder transport.Responder
+	// Optional.
+	engine transport.Engine
+
+	outboundContextDecorators []func(context.Context) context.Context
+	invoker                   Invoker
+	receiverMu                sync.Mutex
+	eventDefaulterFns         []EventDefaulter
 }
 
-// Send transmits the provided event on a preconfigured Protocol. 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))
-	defer span.End()
-	if span.IsRecordingEvents() {
-		span.AddAttributes(eventTraceAttributes(event.Context)...)
+func (c *ceClient) applyOptions(opts ...Option) error {
+	for _, fn := range opts {
+		if err := fn(c); err != nil {
+			return err
+		}
 	}
+	return nil
+}
 
-	err := c.obsSend(ctx, event)
-	if err != nil {
-		r.Error()
-	} else {
-		r.OK()
+func (c *ceClient) Send(ctx context.Context, e event.Event) error {
+	if c.sender == nil {
+		return errors.New("sender not set")
 	}
-	return err
-}
 
-func (c *ceClient) obsSend(ctx context.Context, event event.Event) error {
-	// Confirm we have a transport set.
-	if c.transport == nil {
-		return fmt.Errorf("client not ready, transport not initialized")
+	for _, f := range c.outboundContextDecorators {
+		ctx = f(ctx)
 	}
-	// Apply the defaulter chain to the incoming event.
+
 	if len(c.eventDefaulterFns) > 0 {
 		for _, fn := range c.eventDefaulterFns {
-			event = fn(ctx, event)
+			e = fn(ctx, e)
 		}
 	}
 
-	// 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 fmt.Errorf("error setting distributed tracing extension: %w", err)
-			}
-		}
-	}
-
-	// Validate the event conforms to the CloudEvents Spec.
-	if err := event.Validate(); err != nil {
+	if err := e.Validate(); err != nil {
 		return err
 	}
-	// Send the event over the transport.
-	return c.transport.Send(ctx, event)
-}
 
-// Request transmits the provided event on a preconfigured Protocol. 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)
+	return c.sender.Send(ctx, (*binding.EventMessage)(&e))
+}
 
-	ctx, span := trace.StartSpan(ctx, clientSpanName, trace.WithSpanKind(trace.SpanKindClient))
-	defer span.End()
-	if span.IsRecordingEvents() {
-		span.AddAttributes(eventTraceAttributes(event.Context)...)
+func (c *ceClient) Request(ctx context.Context, e event.Event) (*event.Event, error) {
+	if c.requester == nil {
+		return nil, errors.New("requester not set")
 	}
-
-	resp, err := c.obsRequest(ctx, event)
-	if err != nil {
-		r.Error()
-	} else {
-		r.OK()
+	for _, f := range c.outboundContextDecorators {
+		ctx = f(ctx)
 	}
-	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)
+			e = fn(ctx, e)
 		}
 	}
 
-	// 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 {
+	if err := e.Validate(); err != nil {
 		return nil, err
 	}
-	// Send the event over the transport.
-	return c.transport.Request(ctx, event)
-}
-
-// Delivery is called from from the transport on event delivery.
-func (c *ceClient) Delivery(ctx context.Context, e event.Event) (*event.Event, transport.Result) {
-	ctx, r := observability.NewReporter(ctx, reportReceive)
 
-	var span *trace.Span
-	if !c.transport.HasTracePropagation() {
-		if ext, ok := extensions.GetDistributedTracingExtension(e); ok {
-			ctx, span = ext.StartChildSpan(ctx, clientSpanName, trace.WithSpanKind(trace.SpanKindServer))
+	// If provided a requester, use it to do request/response.
+	var resp *event.Event
+	msg, err := c.requester.Request(ctx, (*binding.EventMessage)(&e))
+	defer func() {
+		if err := msg.Finish(err); err != nil {
+			cecontext.LoggerFrom(ctx).Warnw("failed calling message.Finish", zap.Error(err))
 		}
-	}
-	if span == nil {
-		ctx, span = trace.StartSpan(ctx, clientSpanName, trace.WithSpanKind(trace.SpanKindServer))
-	}
-	defer span.End()
-	if span.IsRecordingEvents() {
-		span.AddAttributes(eventTraceAttributes(e.Context)...)
-	}
-
-	resp, result := c.obsDelivery(ctx, e)
-	if result != nil { // TODO: test result for Ack/Nack
-		r.Error()
-	} else {
-		r.OK()
-	}
-	return resp, result
-}
-
-func (c *ceClient) obsDelivery(ctx context.Context, e event.Event) (*event.Event, transport.Result) {
-	if c.fn != nil {
-		resp, err := c.fn.invoke(ctx, e)
-
-		// Apply the defaulter chain to the outgoing event.
-		if err == nil && resp != nil && len(c.eventDefaulterFns) > 0 {
-			for _, fn := range c.eventDefaulterFns {
-				*resp = fn(ctx, *resp)
-			}
-			// Validate the event conforms to the CloudEvents Spec.
-			if verr := resp.Validate(); verr != nil {
-				return nil, fmt.Errorf("cloudevent validation failed on response event: %v, %w", verr, err)
-			}
+	}()
+	if err == nil {
+		fmt.Printf("%#v", msg)
+		if rs, err := binding.ToEvent(ctx, msg, nil); err != nil {
+			cecontext.LoggerFrom(ctx).Infow("failed calling ToEvent", zap.Error(err), zap.Any("resp", msg))
+		} else {
+			resp = rs
 		}
-		return resp, err
 	}
-	return nil, nil
+	return resp, err
 }
 
 // StartReceiver sets up the given fn to handle Receive.
@@ -242,31 +158,55 @@ func (c *ceClient) StartReceiver(ctx context.Context, fn interface{}) error {
 	c.receiverMu.Lock()
 	defer c.receiverMu.Unlock()
 
-	if c.transport == nil {
-		return fmt.Errorf("client not ready, transport not initialized")
-	}
-	if c.fn != nil {
+	if c.invoker != nil {
 		return fmt.Errorf("client already has a receiver")
 	}
 
-	if fn, err := receiver(fn); err != nil {
+	invoker, err := newReceiveInvoker(fn, c.eventDefaulterFns...) // TODO: this will have to pick between a observed invoker or not.
+	if err != nil {
 		return err
-	} else {
-		c.fn = fn
 	}
+	if invoker.IsReceiver() && c.receiver == nil {
+		return fmt.Errorf("mismatched receiver callback without transport.Receiver supported by protocol")
+	}
+	if invoker.IsResponder() && c.responder == nil {
+		return fmt.Errorf("mismatched receiver callback without transport.Responder supported by protocol")
+	}
+	c.invoker = invoker
 
 	defer func() {
-		c.fn = nil
+		c.invoker = nil
 	}()
 
-	return c.transport.StartReceiver(ctx)
-}
+	// Start the engine, if set.
+	if c.engine != nil {
+		go func() {
+			// TODO: handle error correctly here.
+			if err := c.engine.StartInbound(ctx); err != nil {
+				panic(err)
+			}
+		}()
+	}
+
+	var msg binding.Message
+	var respFn transport.ResponseFn
+	// Start Polling.
+	for {
+		if c.responder != nil {
+			msg, respFn, err = c.responder.Respond(ctx)
+		} else if c.receiver != nil {
+			msg, err = c.receiver.Receive(ctx)
+		} else {
+			return errors.New("responder and receiver not set")
+		}
 
-func (c *ceClient) applyOptions(opts ...Option) error {
-	for _, fn := range opts {
-		if err := fn(c); err != nil {
+		if err == io.EOF { // Normal close
+			return nil
+		} else if err != nil {
+			return err
+		}
+		if err := c.invoker.Invoke(ctx, msg, respFn); err != nil {
 			return err
 		}
 	}
-	return nil
 }
diff --git a/pkg/client/client_default.go b/pkg/client/client_default.go
new file mode 100644
index 0000000..fbd9bbe
--- /dev/null
+++ b/pkg/client/client_default.go
@@ -0,0 +1,26 @@
+package client
+
+import (
+	"github.com/cloudevents/sdk-go/pkg/transport/http"
+)
+
+// NewDefault provides the good defaults for the common case using an HTTP
+// Protocol client. The http transport has had WithBinaryEncoding http
+// transport option applied to it. The client will always send Binary
+// encoding but will inspect the outbound event context and match the version.
+// The WithTimeNow, and WithUUIDs client options are also applied to the
+// client, all outbound events will have a time and id set if not already
+// present.
+func NewDefault() (Client, error) {
+	p, err := http.New()
+	if err != nil {
+		return nil, err
+	}
+
+	c, err := NewObserved(p, WithTimeNow(), WithUUIDs())
+	if err != nil {
+		return nil, err
+	}
+
+	return c, nil
+}
diff --git a/pkg/client/client_observed.go b/pkg/client/client_observed.go
new file mode 100644
index 0000000..99445e3
--- /dev/null
+++ b/pkg/client/client_observed.go
@@ -0,0 +1,89 @@
+package client
+
+import (
+	"context"
+	"github.com/cloudevents/sdk-go/pkg/event"
+	"github.com/cloudevents/sdk-go/pkg/observability"
+	"go.opencensus.io/trace"
+)
+
+// New produces a new client with the provided transport object and applied
+// client options.
+func NewObserved(protocol interface{}, opts ...Option) (Client, error) {
+	client, err := New(protocol, opts...)
+	if err != nil {
+		return nil, err
+	}
+
+	return &obsClient{client: client}, nil
+}
+
+type obsClient struct {
+	client Client
+
+	disableTracePropagation bool // TODO?
+}
+
+//
+//func (c *ceClient) applyOptions(opts ...Option) error {
+//	for _, fn := range opts {
+//		if err := fn(c); err != nil {
+//			return err
+//		}
+//	}
+//	return nil
+//}
+
+// Send transmits the provided event on a preconfigured Protocol. Send returns
+// an error if there was an an issue validating the outbound event or the
+// transport returns an error.
+func (c *obsClient) Send(ctx context.Context, e 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(e.Context)...)
+	}
+
+	err := c.client.Send(ctx, e)
+
+	if err != nil {
+		r.Error()
+	} else {
+		r.OK()
+	}
+	return err
+}
+
+func (c *obsClient) Request(ctx context.Context, e event.Event) (*event.Event, error) {
+	ctx, r := observability.NewReporter(ctx, reportRequest)
+	ctx, span := trace.StartSpan(ctx, clientSpanName, trace.WithSpanKind(trace.SpanKindClient))
+	defer span.End()
+	if span.IsRecordingEvents() {
+		span.AddAttributes(eventTraceAttributes(e.Context)...)
+	}
+
+	resp, err := c.client.Request(ctx, e)
+
+	if err != nil {
+		r.Error()
+	} else {
+		r.OK()
+	}
+	return resp, err
+}
+
+// StartReceiver sets up the given fn to handle Receive.
+// See Client.StartReceiver for details. This is a blocking call.
+func (c *obsClient) StartReceiver(ctx context.Context, fn interface{}) error {
+	ctx, r := observability.NewReporter(ctx, reportStartReceiver)
+
+	err := c.client.StartReceiver(ctx, fn)
+
+	if err != nil {
+		r.Error()
+	} else {
+		r.OK()
+	}
+	return err
+}
diff --git a/pkg/client/http_receiver.go b/pkg/client/http_receiver.go
new file mode 100644
index 0000000..7e69c7f
--- /dev/null
+++ b/pkg/client/http_receiver.go
@@ -0,0 +1,38 @@
+package client
+
+import (
+	"context"
+	thttp "github.com/cloudevents/sdk-go/pkg/transport/http"
+	"net/http"
+)
+
+func NewHTTPReceiveHandler(ctx context.Context, p *thttp.Protocol, fn interface{}) (*EventReceiver, error) {
+	invoker, err := newReceiveInvoker(fn)
+	if err != nil {
+		return nil, err
+	}
+
+	return &EventReceiver{
+		p:       p,
+		invoker: invoker,
+	}, nil
+}
+
+type EventReceiver struct {
+	p       *thttp.Protocol
+	invoker Invoker
+}
+
+func (r *EventReceiver) ServeHTTP(rw http.ResponseWriter, req *http.Request) {
+	go func() {
+		r.p.ServeHTTP(rw, req)
+	}()
+
+	ctx := context.Background()
+	msg, respFn, err := r.p.Respond(ctx)
+	if err != nil {
+		// TODO
+	} else if err := r.invoker.Invoke(ctx, msg, respFn); err != nil {
+		// TODO
+	}
+}
diff --git a/pkg/client/invoker.go b/pkg/client/invoker.go
new file mode 100644
index 0000000..07cb866
--- /dev/null
+++ b/pkg/client/invoker.go
@@ -0,0 +1,81 @@
+package client
+
+import (
+	"context"
+	"fmt"
+	"github.com/cloudevents/sdk-go/pkg/binding"
+	"github.com/cloudevents/sdk-go/pkg/transport"
+)
+
+type Invoker interface {
+	Invoke(context.Context, binding.Message, transport.ResponseFn) error
+	IsReceiver() bool
+	IsResponder() bool
+}
+
+var _ Invoker = (*receiveInvoker)(nil)
+
+func newReceiveInvoker(fn interface{}, fns ...EventDefaulter) (Invoker, error) {
+	r := &receiveInvoker{
+		eventDefaulterFns: fns,
+	}
+
+	if fn, err := receiver(fn); err != nil {
+		return nil, err
+	} else {
+		r.fn = fn
+	}
+
+	return r, nil
+}
+
+type receiveInvoker struct {
+	fn                *receiverFn
+	eventDefaulterFns []EventDefaulter // TODO: set these.
+}
+
+func (r *receiveInvoker) Invoke(ctx context.Context, m binding.Message, respFn transport.ResponseFn) (err error) {
+	defer func() {
+		if err2 := m.Finish(err); err2 == nil {
+			err = err2
+		}
+	}()
+
+	e, err := binding.ToEvent(ctx, m, nil)
+	if err != nil {
+		return err
+	}
+
+	if e != nil && r.fn != nil {
+		resp, result := r.fn.invoke(ctx, *e)
+
+		// Apply the defaulter chain to the outgoing event.
+		if resp != nil && len(r.eventDefaulterFns) > 0 {
+			for _, fn := range r.eventDefaulterFns {
+				*resp = fn(ctx, *resp)
+			}
+			// Validate the event conforms to the CloudEvents Spec.
+			if verr := resp.Validate(); verr != nil {
+				return fmt.Errorf("cloudevent validation failed on response event: %v, %w", verr, err)
+			}
+		}
+		if respFn != nil {
+			var rm binding.Message
+			if resp != nil {
+				rm = (*binding.EventMessage)(resp)
+			}
+
+			return respFn(ctx, rm, result) // TODO: there is a chance this never gets called. Is that ok?
+		}
+	}
+
+	return nil
+}
+
+func (r *receiveInvoker) IsReceiver() bool {
+	return !r.fn.hasEventOut
+}
+
+func (r *receiveInvoker) IsResponder() bool {
+	return r.fn.hasEventOut
+}
diff --git a/pkg/client/observability.go b/pkg/client/observability.go
index dfbffb6..cc360df 100644
--- a/pkg/client/observability.go
+++ b/pkg/client/observability.go
@@ -41,7 +41,7 @@ const (
 
 	reportSend observed = iota
 	reportRequest
-	reportReceive
+	reportStartReceiver
 )
 
 // MethodName implements Observable.MethodName
@@ -51,8 +51,8 @@ func (o observed) MethodName() string {
 		return "send"
 	case reportRequest:
 		return "request"
-	case reportReceive:
-		return "receive"
+	case reportStartReceiver:
+		return "start_receiver"
 	default:
 		return "unknown"
 	}
diff --git a/pkg/client/options.go b/pkg/client/options.go
index 2c4ffcf..4a5539f 100644
--- a/pkg/client/options.go
+++ b/pkg/client/options.go
@@ -2,6 +2,7 @@ package client
 
 import (
 	"fmt"
+	"github.com/cloudevents/sdk-go/pkg/binding"
 )
 
 // Option is the function signature required to be considered an client.Option.
@@ -18,6 +19,20 @@ func WithEventDefaulter(fn EventDefaulter) Option {
 	}
 }
 
+func WithForceBinary() Option {
+	return func(c *ceClient) error {
+		c.outboundContextDecorators = append(c.outboundContextDecorators, binding.WithForceBinary)
+		return nil
+	}
+}
+
+func WithForceStructured() Option {
+	return func(c *ceClient) error {
+		c.outboundContextDecorators = append(c.outboundContextDecorators, binding.WithForceStructured)
+		return nil
+	}
+}
+
 // WithUUIDs adds DefaultIDToUUIDIfNotSet event defaulter to the end of the
 // defaulter chain.
 func WithUUIDs() Option {
@@ -40,7 +55,7 @@ func WithTimeNow() Option {
 // the distributed tracing extension.
 func WithoutTracePropagation() Option {
 	return func(c *ceClient) error {
-		c.disableTracePropagation = true
+		//c.disableTracePropagation = true
 		return nil
 	}
 }
diff --git a/pkg/transport/amqp/message.go b/pkg/transport/amqp/message.go
index a21fbd1..462b9f5 100644
--- a/pkg/transport/amqp/message.go
+++ b/pkg/transport/amqp/message.go
@@ -105,7 +105,7 @@ func (m *Message) ReadBinary(ctx context.Context, encoder binding.BinaryWriter)
 			return err
 		}
 	}
-	return encoder.End()
+	return encoder.End(ctx)
 }
 
 func (m *Message) Finish(err error) error {
diff --git a/pkg/transport/amqp/options.go b/pkg/transport/amqp/options.go
index 21b9b74..ba4bce3 100644
--- a/pkg/transport/amqp/options.go
+++ b/pkg/transport/amqp/options.go
@@ -3,11 +3,11 @@ package amqp
 import "pack.ag/amqp"
 
 // Option is the function signature required to be considered an amqp.Option.
-type Option func(*Transport) error
+type Option func(*Protocol) error
 
 // WithEncoding sets the encoding for amqp transport.
 func WithEncoding(encoding Encoding) Option {
-	return func(t *Transport) error {
+	return func(t *Protocol) error {
 		t.Encoding = encoding
 		return nil
 	}
@@ -15,7 +15,7 @@ func WithEncoding(encoding Encoding) Option {
 
 // WithConnOpt sets a connection option for amqp
 func WithConnOpt(opt amqp.ConnOption) Option {
-	return func(t *Transport) error {
+	return func(t *Protocol) error {
 		t.connOpts = append(t.connOpts, opt)
 		return nil
 	}
@@ -28,7 +28,7 @@ func WithConnSASLPlain(username, password string) Option {
 
 // WithSessionOpt sets a session option for amqp
 func WithSessionOpt(opt amqp.SessionOption) Option {
-	return func(t *Transport) error {
+	return func(t *Protocol) error {
 		t.sessionOpts = append(t.sessionOpts, opt)
 		return nil
 	}
@@ -36,7 +36,7 @@ func WithSessionOpt(opt amqp.SessionOption) Option {
 
 // WithSenderLinkOption sets a link option for amqp
 func WithSenderLinkOption(opt amqp.LinkOption) Option {
-	return func(t *Transport) error {
+	return func(t *Protocol) error {
 		t.senderLinkOpts = append(t.senderLinkOpts, opt)
 		return nil
 	}
@@ -44,7 +44,7 @@ func WithSenderLinkOption(opt amqp.LinkOption) Option {
 
 // WithReceiverLinkOption sets a link option for amqp
 func WithReceiverLinkOption(opt amqp.LinkOption) Option {
-	return func(t *Transport) error {
+	return func(t *Protocol) error {
 		t.receiverLinkOpts = append(t.receiverLinkOpts, opt)
 		return nil
 	}
diff --git a/pkg/transport/amqp/sender.go b/pkg/transport/amqp/sender.go
index ad804dd..f12dacc 100644
--- a/pkg/transport/amqp/sender.go
+++ b/pkg/transport/amqp/sender.go
@@ -3,10 +3,10 @@ package amqp
 import (
 	"context"
 
-	bindings "github.com/cloudevents/sdk-go/pkg/transport"
 	"pack.ag/amqp"
 
 	"github.com/cloudevents/sdk-go/pkg/binding"
+	"github.com/cloudevents/sdk-go/pkg/transport"
 )
 
 // sender wraps an amqp.Sender as a binding.Sender
@@ -34,7 +34,7 @@ func (s *sender) Send(ctx context.Context, in binding.Message) error {
 func (s *sender) Close(ctx context.Context) error { return s.amqp.Close(ctx) }
 
 // Create a new Sender which wraps an amqp.Sender in a binding.Sender
-func NewSender(amqpSender *amqp.Sender, options ...SenderOptionFunc) bindings.Sender {
+func NewSender(amqpSender *amqp.Sender, options ...SenderOptionFunc) transport.Sender {
 	s := &sender{amqp: amqpSender, transformers: make(binding.TransformerFactories, 0)}
 	for _, o := range options {
 		o(s)
diff --git a/pkg/transport/amqp/transport.go b/pkg/transport/amqp/transport.go
index 1b3326a..9e3f761 100644
--- a/pkg/transport/amqp/transport.go
+++ b/pkg/transport/amqp/transport.go
@@ -10,19 +10,9 @@ import (
 	"github.com/cloudevents/sdk-go/pkg/binding/transformer"
 	cecontext "github.com/cloudevents/sdk-go/pkg/context"
 	"github.com/cloudevents/sdk-go/pkg/transport"
-	"github.com/cloudevents/sdk-go/pkg/transport/bindings"
 )
 
-// Transport adheres to transport.Transport.
-var _ transport.Transport = (*Transport)(nil)
-
-const (
-	// TransportName is the name of this transport.
-	TransportName = "AMQP"
-)
-
-type Transport struct {
-	bindings.BindingTransport
+type Protocol struct {
 	connOpts         []amqp.ConnOption
 	sessionOpts      []amqp.SessionOption
 	senderLinkOpts   []amqp.LinkOption
@@ -34,16 +24,19 @@ type Transport struct {
 	// AMQP
 	Client  *amqp.Client
 	Session *amqp.Session
-	Sender  *amqp.Sender
 	Node    string
 
+	// Sender
+	Sender                  transport.Sender
+	SenderContextDecorators []func(context.Context) context.Context
+
 	// Receiver
 	Receiver transport.Receiver
 }
 
 // New creates a new amqp transport.
-func New(server, queue string, opts ...Option) (*Transport, error) {
-	t := &Transport{
+func New(server, queue string, opts ...Option) (*Protocol, error) {
+	t := &Protocol{
 		Node:             queue,
 		connOpts:         []amqp.ConnOption(nil),
 		sessionOpts:      []amqp.SessionOption(nil),
@@ -78,11 +71,11 @@ func New(server, queue string, opts ...Option) (*Transport, error) {
 		return nil, err
 	}
 	// TODO: in the future we might have more than one sender.
-	t.BindingTransport.Sender, t.BindingTransport.SenderContextDecorators = t.applyEncoding(sender)
+	t.Sender, t.SenderContextDecorators = t.applyEncoding(sender)
 	return t, nil
 }
 
-func (t *Transport) applyEncoding(amqpSender *amqp.Sender) (transport.Sender, []func(context.Context) context.Context) {
+func (t *Protocol) applyEncoding(amqpSender *amqp.Sender) (transport.Sender, []func(context.Context) context.Context) {
 	switch t.Encoding {
 	case BinaryV03:
 		return NewSender(
@@ -108,7 +101,7 @@ func (t *Transport) applyEncoding(amqpSender *amqp.Sender) (transport.Sender, []
 	return NewSender(amqpSender), []func(context.Context) context.Context{}
 }
 
-func (t *Transport) applyOptions(opts ...Option) error {
+func (t *Protocol) applyOptions(opts ...Option) error {
 	for _, fn := range opts {
 		if err := fn(t); err != nil {
 			return err
@@ -117,9 +110,9 @@ func (t *Transport) applyOptions(opts ...Option) error {
 	return nil
 }
 
-// StartReceiver implements Transport.StartReceiver
+// StartReceiver implements Protocol.StartReceiver
 // NOTE: This is a blocking call.
-func (t *Transport) StartReceiver(ctx context.Context) error {
+func (t *Protocol) StartInbound(ctx context.Context) error {
 	logger := cecontext.LoggerFrom(ctx)
 	logger.Info("StartReceiver on ", t.Node)
 
@@ -128,15 +121,23 @@ func (t *Transport) StartReceiver(ctx context.Context) error {
 	if err != nil {
 		return err
 	}
-	t.BindingTransport.Receiver = NewReceiver(receiver)
-	return t.BindingTransport.StartReceiver(ctx)
+	t.Receiver = NewReceiver(receiver)
+	return nil
 }
 
-// HasTracePropagation implements Transport.HasTracePropagation
-func (t *Transport) HasTracePropagation() bool {
+// HasTracePropagation implements Protocol.HasTracePropagation
+func (t *Protocol) HasTracePropagation() bool {
 	return false
 }
 
-func (t *Transport) Close() error {
+func (t *Protocol) Close() error {
 	return t.Client.Close()
 }
+
+func (t *Protocol) Send(ctx context.Context, in binding.Message) error {
+	return t.Sender.Send(ctx, in)
+}
+
+func (t *Protocol) Receive(ctx context.Context) (binding.Message, error) {
+	return t.Receiver.Receive(ctx)
+}
diff --git a/pkg/transport/amqp/write_message.go b/pkg/transport/amqp/write_message.go
index b7a5841..eb55a4d 100644
--- a/pkg/transport/amqp/write_message.go
+++ b/pkg/transport/amqp/write_message.go
@@ -47,7 +47,7 @@ func (b *amqpMessageWriter) Start(ctx context.Context) error {
 	return nil
 }
 
-func (b *amqpMessageWriter) End() error {
+func (b *amqpMessageWriter) End(ctx context.Context) error {
 	return nil
 }
 
diff --git a/pkg/transport/bindings/transport.go b/pkg/transport/bindings/transport.go
deleted file mode 100644
index f97cc36..0000000
--- a/pkg/transport/bindings/transport.go
+++ /dev/null
@@ -1,140 +0,0 @@
-package bindings
-
-import (
-	"context"
-	"errors"
-	"io"
-
-	"github.com/cloudevents/sdk-go/pkg/binding"
-	bindings "github.com/cloudevents/sdk-go/pkg/transport"
-
-	"go.uber.org/zap"
-
-	cecontext "github.com/cloudevents/sdk-go/pkg/context"
-	"github.com/cloudevents/sdk-go/pkg/event"
-	"github.com/cloudevents/sdk-go/pkg/transport"
-)
-
-// BindingTransport adapts the transport.Transport interface to binding.Sender and binding.Receiver.
-// This adapter doesn't support the request/response model
-type BindingTransport struct {
-	Sender    bindings.Sender
-	Requester bindings.Requester
-	Receiver  bindings.Receiver
-	Responder bindings.Responder
-	// SenderContextDecorators can be used to decorate the context passed to the Sender.Send() method
-	SenderContextDecorators []func(context.Context) context.Context
-	handler                 transport.Delivery
-}
-
-var _ transport.Transport = (*BindingTransport)(nil) // Conforms to the interface
-
-func NewRequestingTransport(r bindings.Requester, rx bindings.Receiver, senderContextDecorators []func(context.Context) context.Context) *BindingTransport {
-	if senderContextDecorators == nil {
-		senderContextDecorators = []func(ctx context.Context) context.Context{}
-	}
-	return &BindingTransport{Requester: r, Receiver: rx, SenderContextDecorators: senderContextDecorators}
-}
-
-func NewSendingTransport(s bindings.Sender, rx bindings.Receiver, senderContextDecorators []func(context.Context) context.Context) *BindingTransport {
-	if senderContextDecorators == nil {
-		senderContextDecorators = []func(ctx context.Context) context.Context{}
-	}
-	return &BindingTransport{Sender: s, Receiver: rx, SenderContextDecorators: senderContextDecorators}
-}
-
-func (t *BindingTransport) Send(ctx context.Context, e event.Event) error {
-	if t.Sender == nil {
-		return errors.New("sender not set")
-	}
-	for _, f := range t.SenderContextDecorators {
-		ctx = f(ctx)
-	}
-	return t.Sender.Send(ctx, (*binding.EventMessage)(&e))
-}
-
-func (t *BindingTransport) Request(ctx context.Context, e event.Event) (*event.Event, error) {
-	if t.Requester == nil {
-		return nil, errors.New("requester not set")
-	}
-	for _, f := range t.SenderContextDecorators {
-		ctx = f(ctx)
-	}
-
-	// If provided a requester, use it to do request/response.
-	var resp *event.Event
-	msg, err := t.Requester.Request(ctx, (*binding.EventMessage)(&e))
-	defer func() {
-		if err := msg.Finish(err); err != nil {
-			cecontext.LoggerFrom(ctx).Warnw("failed calling message.Finish", zap.Error(err))
-		}
-	}()
-	if err == nil {
-		if rs, err := binding.ToEvent(ctx, msg, nil); err != nil {
-			cecontext.LoggerFrom(ctx).Warnw("failed calling ToEvent", zap.Error(err), zap.Any("resp", msg))
-		} else {
-			resp = rs
-		}
-	}
-	return resp, err
-}
-
-func (t *BindingTransport) SetDelivery(r transport.Delivery) {
-	t.handler = r
-}
-
-// Legacy Transport StartReceiver is really start Responder.
-func (t *BindingTransport) StartReceiver(ctx context.Context) error {
-	var msg binding.Message
-	var err error
-	var respFn transport.ResponseFn
-	for {
-		if t.Responder != nil {
-			msg, respFn, err = t.Responder.Respond(ctx)
-		} else if t.Receiver != nil {
-			msg, err = t.Receiver.Receive(ctx)
-		} else {
-			return errors.New("responder and receiver not set")
-		}
-
-		if err == io.EOF { // Normal close
-			return nil
-		} else if err != nil {
-			return err
-		}
-		if err := t.handle(ctx, msg, respFn); err != nil {
-			return err
-		}
-	}
-}
-
-func (t *BindingTransport) handle(ctx context.Context, m binding.Message, respFn transport.ResponseFn) (err error) {
-	defer func() {
-		if err2 := m.Finish(err); err2 == nil {
-			err = err2
-		}
-	}()
-
-	if t.handler == nil {
-		return nil
-	}
-
-	e, err := binding.ToEvent(ctx, m, nil)
-	if err != nil {
-		return err
-	}
-
-	resp, result := t.handler.Delivery(ctx, *e)
-	if respFn != nil {
-		var rm binding.Message
-		if resp != nil {
-			rm = (*binding.EventMessage)(resp)
-		}
-
-		return respFn(ctx, rm, result)
-	}
-
-	return nil
-}
-
-func (t *BindingTransport) HasTracePropagation() bool { return false } // TODO
diff --git a/pkg/transport/engine.go b/pkg/transport/engine.go
new file mode 100644
index 0000000..1f84bac
--- /dev/null
+++ b/pkg/transport/engine.go
@@ -0,0 +1,11 @@
+package transport
+
+import (
+	"context"
+)
+
+// TODO: document
+type Engine interface {
+	// Blocking call. Context is used to cancel.
+	StartInbound(ctx context.Context) error
+}
diff --git a/pkg/transport/http/encoding.go b/pkg/transport/http/encoding.go
index b91d21f..4984fb4 100644
--- a/pkg/transport/http/encoding.go
+++ b/pkg/transport/http/encoding.go
@@ -5,11 +5,14 @@ type Encoding int32
 
 const (
 	// Default is binary
+	// Deprecated: use binding
 	Default Encoding = iota
 
 	// Binary
+	// Deprecated: use binding
 	Binary
 
 	// Structured
+	// Deprecated: use binding
 	Structured
 )
diff --git a/pkg/transport/http/engine.go b/pkg/transport/http/engine.go
new file mode 100644
index 0000000..9785254
--- /dev/null
+++ b/pkg/transport/http/engine.go
@@ -0,0 +1,138 @@
+package http
+
+import (
+	"context"
+	"fmt"
+	"net"
+	"net/http"
+	"strings"
+
+	"github.com/cloudevents/sdk-go/pkg/transport"
+	"go.opencensus.io/plugin/ochttp"
+	"go.opencensus.io/plugin/ochttp/propagation/tracecontext"
+)
+
+var _ transport.Engine = (*Protocol)(nil)
+
+func (e *Protocol) StartInbound(ctx context.Context) error {
+	e.reMu.Lock()
+	defer e.reMu.Unlock()
+
+	if e.Handler == nil {
+		e.Handler = http.NewServeMux()
+	}
+
+	if !e.handlerRegistered {
+		// handler.Handle might panic if the user tries to use the same path as the sdk.
+		e.Handler.Handle(e.GetPath(), e)
+		e.handlerRegistered = true
+	}
+
+	addr, err := e.listen()
+	if err != nil {
+		return err
+	}
+
+	e.server = &http.Server{
+		Addr: addr.String(),
+		Handler: &ochttp.Handler{
+			Propagation:    &tracecontext.HTTPFormat{},
+			Handler:        attachMiddleware(e.Handler, e.middleware),
+			FormatSpanName: formatSpanName,
+		},
+	}
+
+	// Shutdown
+	defer func() {
+		_ = e.server.Close()
+		e.server = nil
+	}()
+
+	errChan := make(chan error, 1)
+	go func() {
+		errChan <- e.server.Serve(e.listener)
+	}()
+
+	// wait for the server to return or ctx.Done().
+	select {
+	case <-ctx.Done():
+		// Try a gracefully shutdown.
+		timeout := *e.ShutdownTimeout
+		ctx, cancel := context.WithTimeout(context.Background(), timeout)
+		defer cancel()
+		err := e.server.Shutdown(ctx)
+		<-errChan // Wait for server goroutine to exit
+		return err
+	case err := <-errChan:
+		return err
+	}
+}
+
+// HasTracePropagation implements Protocol.HasTracePropagation
+func (e *Protocol) HasTracePropagation() bool { // TODO: clean this all up.
+	return false
+}
+
+// GetPort returns the listening port.
+// Returns -1 if there is a listening error.
+// Note this will call net.Listen() if  the listener is not already started.
+func (e *Protocol) GetPort() int {
+	// Ensure we have a listener and therefore a port.
+	if _, err := e.listen(); err == nil || e.Port != nil {
+		return *e.Port
+	}
+	return -1
+}
+
+func formatSpanName(r *http.Request) string {
+	return "cloudevents.http." + r.URL.Path
+}
+
+func (e *Protocol) setPort(port int) {
+	if e.Port == nil {
+		e.Port = new(int)
+	}
+	*e.Port = port
+}
+
+// listen if not already listening, update t.Port
+func (e *Protocol) listen() (net.Addr, error) {
+	if e.listener == nil {
+		port := 8080
+		if e.Port != nil {
+			port = *e.Port
+			if port < 0 || port > 65535 {
+				return nil, fmt.Errorf("invalid port %d", port)
+			}
+		}
+		var err error
+		if e.listener, err = net.Listen("tcp", fmt.Sprintf(":%d", port)); err != nil {
+			return nil, err
+		}
+	}
+	addr := e.listener.Addr()
+	if tcpAddr, ok := addr.(*net.TCPAddr); ok {
+		e.setPort(tcpAddr.Port)
+	}
+	return addr, nil
+}
+
+// GetPath returns the path the transport is hosted on. If the path is '/',
+// the transport will handle requests on any URI. To discover the true path
+// a request was received on, inspect the context from Receive(cxt, ...) with
+// TransportContextFrom(ctx).
+func (e *Protocol) GetPath() string {
+	path := strings.TrimSpace(e.Path)
+	if len(path) > 0 {
+		return path
+	}
+	return "/" // default
+}
+
+// attachMiddleware attaches the HTTP middleware to the specified handler.
+func attachMiddleware(h http.Handler, middleware []Middleware) http.Handler {
+	for _, m := range middleware {
+		h = m(h)
+	}
+	return h
+}
diff --git a/pkg/transport/http/message.go b/pkg/transport/http/message.go
index 190c447..663a5af 100644
--- a/pkg/transport/http/message.go
+++ b/pkg/transport/http/message.go
@@ -109,7 +109,7 @@ func (m *Message) ReadBinary(ctx context.Context, encoder binding.BinaryWriter)
 		}
 	}
 
-	return encoder.End()
+	return encoder.End(ctx)
 }
 
 func (m *Message) Finish(err error) error {
@@ -119,5 +119,7 @@ func (m *Message) Finish(err error) error {
 	if m.OnFinish != nil {
 		return m.OnFinish(err)
 	}
+	// TODO: if in binary mode, there could be nothing in this request or
+	// response. Meaning Message is not nil but never going to be a valid event.
 	return nil
 }
diff --git a/pkg/transport/http/options.go b/pkg/transport/http/options.go
index af9f8fe..82032d6 100644
--- a/pkg/transport/http/options.go
+++ b/pkg/transport/http/options.go
@@ -9,15 +9,12 @@ import (
 	"time"
 )
 
-// ProtocolOption is the function signature required to be considered an http.ProtocolOption.
-type ProtocolOption func(*Protocol) error
-
 // Option is the function signature required to be considered an http.Option.
-type Option func(*Transport) error
+type Option func(*Protocol) error
 
 // WithTarget sets the outbound recipient of cloudevents when using an HTTP
 // request.
-func WithTarget(targetUrl string) ProtocolOption {
+func WithTarget(targetUrl string) Option {
 	return func(p *Protocol) error {
 		if p == nil {
 			return fmt.Errorf("http protocol option can not set nil protocol")
@@ -48,7 +45,7 @@ func WithTarget(targetUrl string) ProtocolOption {
 
 // WithHeader sets an additional default outbound header for all cloudevents
 // when using an HTTP request.
-func WithHeader(key, value string) ProtocolOption {
+func WithHeader(key, value string) Option {
 	return func(p *Protocol) error {
 		if p == nil {
 			return fmt.Errorf("http header option can not set nil transport")
@@ -72,7 +69,7 @@ func WithHeader(key, value string) ProtocolOption {
 
 // WithShutdownTimeout sets the shutdown timeout when the http server is being shutdown.
 func WithShutdownTimeout(timeout time.Duration) Option {
-	return func(t *Transport) error {
+	return func(t *Protocol) error {
 		if t == nil {
 			return fmt.Errorf("http shutdown timeout option can not set nil transport")
 		}
@@ -81,46 +78,7 @@ func WithShutdownTimeout(timeout time.Duration) Option {
 	}
 }
 
-// WithEncoding sets the encoding for clients with HTTP transports.
-func WithEncoding(encoding Encoding) Option {
-	return func(t *Transport) error {
-		if t == nil {
-			return fmt.Errorf("http encoding option can not set nil transport")
-		}
-		t.Encoding = encoding
-		return nil
-	}
-}
-
-// WithBinaryEncoding sets the encoding selection strategy for
-// default encoding selections based on Event, the encoded event will be the
-// given version in Binary form.
-func WithBinaryEncoding() Option {
-	return func(t *Transport) error {
-		if t == nil {
-			return fmt.Errorf("http binary encoding option can not set nil transport")
-		}
-
-		t.Encoding = Binary
-		return nil
-	}
-}
-
-// WithStructuredEncoding sets the encoding selection strategy for
-// default encoding selections based on Event, the encoded event will be the
-// given version in Structured form.
-func WithStructuredEncoding() Option {
-	return func(t *Transport) error {
-		if t == nil {
-			return fmt.Errorf("http structured encoding option can not set nil transport")
-		}
-
-		t.Encoding = Structured
-		return nil
-	}
-}
-
-func checkListen(t *Transport, prefix string) error {
+func checkListen(t *Protocol, prefix string) error {
 	switch {
 	case t.Port != nil:
 		return fmt.Errorf("%v port already set", prefix)
@@ -133,7 +91,7 @@ func checkListen(t *Transport, prefix string) error {
 // WithPort sets the listening port for StartReceiver.
 // Only one of WithListener  or WithPort is allowed.
 func WithPort(port int) Option {
-	return func(t *Transport) error {
+	return func(t *Protocol) error {
 		if t == nil {
 			return fmt.Errorf("http port option can not set nil transport")
 		}
@@ -151,7 +109,7 @@ func WithPort(port int) Option {
 // WithListener sets the listener for StartReceiver.
 // Only one of WithListener or WithPort is allowed.
 func WithListener(l net.Listener) Option {
-	return func(t *Transport) error {
+	return func(t *Protocol) error {
 		if t == nil {
 			return fmt.Errorf("http listener option can not set nil transport")
 		}
@@ -166,7 +124,7 @@ func WithListener(l net.Listener) Option {
 
 // WithPath sets the path to receive cloudevents on for HTTP transports.
 func WithPath(path string) Option {
-	return func(t *Transport) error {
+	return func(t *Protocol) error {
 		if t == nil {
 			return fmt.Errorf("http path option can not set nil transport")
 		}
@@ -188,7 +146,7 @@ type Middleware func(next nethttp.Handler) nethttp.Handler
 // Middleware is applied to everything before it. For example
 // `NewClient(WithMiddleware(foo), WithMiddleware(bar))` would result in `bar(foo(original))`.
 func WithMiddleware(middleware Middleware) Option {
-	return func(t *Transport) error {
+	return func(t *Protocol) error {
 		if t == nil {
 			return fmt.Errorf("http middleware option can not set nil transport")
 		}
@@ -199,7 +157,7 @@ func WithMiddleware(middleware Middleware) Option {
 
 // WithHTTPTransport sets the HTTP client transport.
 func WithHTTPTransport(httpTransport nethttp.RoundTripper) Option {
-	return func(t *Transport) error {
+	return func(t *Protocol) error {
 		t.transport = httpTransport
 		return nil
 	}
diff --git a/pkg/transport/http/protocol.go b/pkg/transport/http/protocol.go
index 1935976..41820b8 100644
--- a/pkg/transport/http/protocol.go
+++ b/pkg/transport/http/protocol.go
@@ -4,8 +4,10 @@ import (
 	"context"
 	"fmt"
 	"io"
+	"net"
 	"net/http"
 	"net/url"
+	"sync"
 	"time"
 
 	"github.com/cloudevents/sdk-go/pkg/transport"
@@ -26,9 +28,31 @@ type Protocol struct {
 	transformers    binding.TransformerFactories
 	Client          *http.Client
 	incoming        chan msgErr
+
+	// To support engine:
+
+	// ShutdownTimeout defines the timeout given to the http.Server when calling Shutdown.
+	// If nil, DefaultShutdownTimeout is used.
+	ShutdownTimeout *time.Duration
+
+	// Port is the port to bind the receiver to. Defaults to 8080.
+	Port *int
+	// Path is the path to bind the receiver to. Defaults to "/".
+	Path string
+
+	// Receive Mutex
+	reMu sync.Mutex
+	// Handler is the handler the http Server will use. Use this to reuse the
+	// http server. If nil, the Protocol will create a one.
+	Handler           *http.ServeMux
+	listener          net.Listener
+	transport         http.RoundTripper // TODO: use this.
+	server            *http.Server
+	handlerRegistered bool
+	middleware        []Middleware
 }
 
-func NewProtocol(opts ...ProtocolOption) (*Protocol, error) {
+func New(opts ...Option) (*Protocol, error) {
 	p := &Protocol{
 		transformers: make(binding.TransformerFactories, 0),
 		incoming:     make(chan msgErr),
@@ -41,10 +65,15 @@ func NewProtocol(opts ...ProtocolOption) (*Protocol, error) {
 		p.Client = http.DefaultClient
 	}
 
+	if p.ShutdownTimeout == nil {
+		timeout := DefaultShutdownTimeout
+		p.ShutdownTimeout = &timeout
+	}
+
 	return p, nil
 }
 
-func (p *Protocol) applyOptions(opts ...ProtocolOption) error {
+func (p *Protocol) applyOptions(opts ...Option) error {
 	for _, fn := range opts {
 		if err := fn(p); err != nil {
 			return err
diff --git a/pkg/transport/http/protocol_server.go b/pkg/transport/http/protocol_server.go
deleted file mode 100644
index bb53c38..0000000
--- a/pkg/transport/http/protocol_server.go
+++ /dev/null
@@ -1,237 +0,0 @@
-package http
-
-import (
-	"context"
-	"fmt"
-	"net"
-	"net/http"
-	"strings"
-	"sync"
-	"time"
-
-	"github.com/cloudevents/sdk-go/pkg/binding"
-	"github.com/cloudevents/sdk-go/pkg/event"
-	"github.com/cloudevents/sdk-go/pkg/transport"
-	"github.com/cloudevents/sdk-go/pkg/transport/bindings"
-	"go.opencensus.io/plugin/ochttp"
-	"go.opencensus.io/plugin/ochttp/propagation/tracecontext"
-)
-
-type Transport struct {
-	bindings.BindingTransport
-
-	// The encoding used to select the codec for outbound events.
-	Encoding Encoding
-
-	// ShutdownTimeout defines the timeout given to the http.Server when calling Shutdown.
-	// If nil, DefaultShutdownTimeout is used.
-	ShutdownTimeout *time.Duration
-
-	// Port is the port to bind the receiver to. Defaults to 8080.
-	Port *int
-	// Path is the path to bind the receiver to. Defaults to "/".
-	Path string
-
-	// Receive Mutex
-	reMu sync.Mutex
-	// Handler is the handler the http Server will use. Use this to reuse the
-	// http server. If nil, the Protocol will create a one.
-	Handler           *http.ServeMux
-	listener          net.Listener
-	transport         http.RoundTripper // TODO: use this.
-	server            *http.Server
-	handlerRegistered bool
-	middleware        []Middleware
-
-	protocol *Protocol
-}
-
-func New(p *Protocol, opts ...Option) (*Transport, error) {
-	t := &Transport{}
-	if err := t.applyOptions(opts...); err != nil {
-		return nil, err
-	}
-
-	t.Sender = p
-	t.Requester = p
-	t.Receiver = p
-	t.Responder = p
-	t.protocol = p
-
-	if t.ShutdownTimeout == nil {
-		timeout := DefaultShutdownTimeout
-		t.ShutdownTimeout = &timeout
-	}
-
-	return t, nil
-}
-
-// Protocol returns a CloudEvents transport.Protocol.
-// Deprecated: This is for legacy transition until client uses Sender/Receiver directly.
-func (t *Transport) Transport() transport.Transport {
-	return t
-}
-
-func (t *Transport) applyOptions(opts ...Option) error {
-	for _, fn := range opts {
-		if err := fn(t); err != nil {
-			return err
-		}
-	}
-	return nil
-}
-
-// Send implements Transport.Send
-func (t *Transport) Send(ctx context.Context, e event.Event) error {
-	switch t.Encoding {
-	case Default, Binary:
-		ctx = binding.WithForceBinary(ctx)
-	case Structured:
-		ctx = binding.WithForceStructured(ctx)
-	}
-
-	return t.BindingTransport.Send(ctx, e)
-}
-
-// Request implements Transport.Request
-func (t *Transport) Request(ctx context.Context, e event.Event) (*event.Event, error) {
-	switch t.Encoding {
-	case Default, Binary:
-		ctx = binding.WithForceBinary(ctx)
-	case Structured:
-		ctx = binding.WithForceStructured(ctx)
-	}
-
-	return t.BindingTransport.Request(ctx, e)
-}
-
-// HasTracePropagation implements Protocol.HasTracePropagation
-func (t *Transport) HasTracePropagation() bool {
-	return false
-}
-
-// StartReceiver implements Transport.StartReceiver
-// NOTE: This is a blocking call.
-func (t *Transport) StartReceiver(ctx context.Context) error {
-	t.reMu.Lock()
-	defer t.reMu.Unlock()
-
-	if t.Handler == nil {
-		t.Handler = http.NewServeMux()
-	}
-
-	if !t.handlerRegistered {
-		// handler.Handle might panic if the user tries to use the same path as the sdk.
-		t.Handler.Handle(t.GetPath(), t.protocol)
-		t.handlerRegistered = true
-	}
-
-	addr, err := t.listen()
-	if err != nil {
-		return err
-	}
-
-	t.server = &http.Server{
-		Addr: addr.String(),
-		Handler: &ochttp.Handler{
-			Propagation:    &tracecontext.HTTPFormat{},
-			Handler:        attachMiddleware(t.Handler, t.middleware),
-			FormatSpanName: formatSpanName,
-		},
-	}
-
-	// Shutdown
-	defer func() {
-		_ = t.server.Close()
-		t.server = nil
-	}()
-
-	errChan := make(chan error, 1)
-	go func() {
-		errChan <- t.server.Serve(t.listener)
-	}()
-
-	go func() {
-		if err := t.BindingTransport.StartReceiver(ctx); err != nil {
-			errChan <- err
-		}
-	}()
-
-	// wait for the server to return or ctx.Done().
-	select {
-	case <-ctx.Done():
-		// Try a gracefully shutdown.
-		timeout := *t.ShutdownTimeout
-		ctx, cancel := context.WithTimeout(context.Background(), timeout)
-		defer cancel()
-		err := t.server.Shutdown(ctx)
-		<-errChan // Wait for server goroutine to exit
-		return err
-	case err := <-errChan:
-		return err
-	}
-}
-
-// GetPort returns the listening port.
-// Returns -1 if there is a listening error.
-// Note this will call net.Listen() if  the listener is not already started.
-func (t *Transport) GetPort() int {
-	// Ensure we have a listener and therefore a port.
-	if _, err := t.listen(); err == nil || t.Port != nil {
-		return *t.Port
-	}
-	return -1
-}
-
-func formatSpanName(r *http.Request) string {
-	return "cloudevents.http." + r.URL.Path
-}
-
-func (t *Transport) setPort(port int) {
-	if t.Port == nil {
-		t.Port = new(int)
-	}
-	*t.Port = port
-}
-
-// listen if not already listening, update t.Port
-func (t *Transport) listen() (net.Addr, error) {
-	if t.listener == nil {
-		port := 8080
-		if t.Port != nil {
-			port = *t.Port
-			if port < 0 || port > 65535 {
-				return nil, fmt.Errorf("invalid port %d", port)
-			}
-		}
-		var err error
-		if t.listener, err = net.Listen("tcp", fmt.Sprintf(":%d", port)); err != nil {
-			return nil, err
-		}
-	}
-	addr := t.listener.Addr()
-	if tcpAddr, ok := addr.(*net.TCPAddr); ok {
-		t.setPort(tcpAddr.Port)
-	}
-	return addr, nil
-}
-
-// GetPath returns the path the transport is hosted on. If the path is '/',
-// the transport will handle requests on any URI. To discover the true path
-// a request was received on, inspect the context from Receive(cxt, ...) with
-// TransportContextFrom(ctx).
-func (t *Transport) GetPath() string {
-	path := strings.TrimSpace(t.Path)
-	if len(path) > 0 {
-		return path
-	}
-	return "/" // default
-}
-
-// attachMiddleware attaches the HTTP middleware to the specified handler.
-func attachMiddleware(h http.Handler, middleware []Middleware) http.Handler {
-	for _, m := range middleware {
-		h = m(h)
-	}
-	return h
-}
diff --git a/pkg/transport/http/write_request.go b/pkg/transport/http/write_request.go
index d7668b2..7773a08 100644
--- a/pkg/transport/http/write_request.go
+++ b/pkg/transport/http/write_request.go
@@ -40,7 +40,7 @@ func (b *httpRequestWriter) Start(ctx context.Context) error {
 	return nil
 }
 
-func (b *httpRequestWriter) End() error {
+func (b *httpRequestWriter) End(ctx context.Context) error {
 	return nil
 }
 
diff --git a/pkg/transport/http/write_responsewriter.go b/pkg/transport/http/write_responsewriter.go
index c9f90b7..a075a83 100644
--- a/pkg/transport/http/write_responsewriter.go
+++ b/pkg/transport/http/write_responsewriter.go
@@ -44,7 +44,7 @@ func (b *httpResponseWriter) Start(ctx context.Context) error {
 	return nil
 }
 
-func (b *httpResponseWriter) End() error {
+func (b *httpResponseWriter) End(ctx context.Context) error {
 	return nil
 }
 
diff --git a/pkg/transport/kafka_sarama/message.go b/pkg/transport/kafka_sarama/message.go
index bc42a3b..b5a41e4 100644
--- a/pkg/transport/kafka_sarama/message.go
+++ b/pkg/transport/kafka_sarama/message.go
@@ -5,9 +5,6 @@ import (
 	"context"
 	"strings"
 
-	"github.com/pkg/errors"
-
-	ce "github.com/cloudevents/sdk-go"
 	"github.com/cloudevents/sdk-go/pkg/binding"
 	"github.com/cloudevents/sdk-go/pkg/binding/format"
 	"github.com/cloudevents/sdk-go/pkg/binding/spec"
@@ -38,7 +35,7 @@ var _ binding.Message = (*Message)(nil)
 // Returns a binding.Message that holds the provided ConsumerMessage.
 // The returned binding.Message *can* be read several times safely
 // This function *doesn't* guarantee that the returned binding.Message is always a kafka_sarama.Message instance
-func NewMessageFromConsumerMessage(cm *sarama.ConsumerMessage) (binding.Message, error) {
+func NewMessageFromConsumerMessage(cm *sarama.ConsumerMessage) *Message {
 	var contentType string
 	headers := make(map[string][]byte, len(cm.Headers))
 	for _, r := range cm.Headers {
@@ -54,27 +51,14 @@ func NewMessageFromConsumerMessage(cm *sarama.ConsumerMessage) (binding.Message,
 // Returns a binding.Message that holds the provided kafka message components.
 // The returned binding.Message *can* be read several times safely
 // This function *doesn't* guarantee that the returned binding.Message is always a kafka_sarama.Message instance
-func NewMessage(key []byte, value []byte, contentType string, headers map[string][]byte) (binding.Message, error) {
+func NewMessage(key []byte, value []byte, contentType string, headers map[string][]byte) *Message {
 	if ft := format.Lookup(contentType); ft != nil {
-		// If the message is structured and has a key,
-		// then it's cheaper to go through event message
-		// because we need to add the key as extension
-		if key != nil {
-			event := ce.Event{}
-			err := ft.Unmarshal(value, &event)
-			if err != nil {
-				return nil, errors.Wrap(err, "Error while trying to convert the kafka message to event")
-			}
-			event.SetExtension("key", string(key))
-			return (*binding.EventMessage)(&event), nil
-		} else {
-			return &Message{
-				Key:         key,
-				Value:       value,
-				ContentType: contentType,
-				Headers:     headers,
-				format:      ft,
-			}, nil
+		return &Message{
+			Key:         key,
+			Value:       value,
+			ContentType: contentType,
+			Headers:     headers,
+			format:      ft,
 		}
 	} else if v := specs.Version(string(headers[specs.PrefixedSpecVersionName()])); v != nil {
 		return &Message{
@@ -83,7 +67,7 @@ func NewMessage(key []byte, value []byte, contentType string, headers map[string
 			ContentType: contentType,
 			Headers:     headers,
 			version:     v,
-		}, nil
+		}
 	}
 
 	return &Message{
@@ -91,7 +75,7 @@ func NewMessage(key []byte, value []byte, contentType string, headers map[string
 		Value:       value,
 		ContentType: contentType,
 		Headers:     headers,
-	}, nil
+	}
 }
 
 func (m *Message) ReadEncoding() binding.Encoding {
@@ -151,7 +135,7 @@ func (m *Message) ReadBinary(ctx context.Context, encoder binding.BinaryWriter)
 		}
 	}
 
-	return encoder.End()
+	return encoder.End(ctx)
 }
 
 func (m *Message) Finish(error) error {
diff --git a/pkg/transport/kafka_sarama/receiver.go b/pkg/transport/kafka_sarama/receiver.go
index 731af25..53c0a91 100644
--- a/pkg/transport/kafka_sarama/receiver.go
+++ b/pkg/transport/kafka_sarama/receiver.go
@@ -46,16 +46,11 @@ func (r *Receiver) Cleanup(sarama.ConsumerGroupSession) error {
 
 func (r *Receiver) ConsumeClaim(session sarama.ConsumerGroupSession, claim sarama.ConsumerGroupClaim) error {
 	for message := range claim.Messages() {
-		m, err := NewMessageFromConsumerMessage(message)
+		m := NewMessageFromConsumerMessage(message)
 
-		if err != nil {
-			r.incoming <- msgErr{err: err}
-		} else {
-			r.incoming <- msgErr{
-				msg: binding.WithFinish(m, func(err error) { session.MarkMessage(message, "") }),
-			}
+		r.incoming <- msgErr{
+			msg: binding.WithFinish(m, func(err error) { session.MarkMessage(message, "") }),
 		}
-
 	}
 	return nil
 }
diff --git a/pkg/transport/kafka_sarama/write_producer_message.go b/pkg/transport/kafka_sarama/write_producer_message.go
index e393125..6f763d0 100644
--- a/pkg/transport/kafka_sarama/write_producer_message.go
+++ b/pkg/transport/kafka_sarama/write_producer_message.go
@@ -7,93 +7,28 @@ import (
 
 	"github.com/Shopify/sarama"
 
-	ce "github.com/cloudevents/sdk-go"
 	"github.com/cloudevents/sdk-go/pkg/binding"
 	"github.com/cloudevents/sdk-go/pkg/binding/format"
 	"github.com/cloudevents/sdk-go/pkg/binding/spec"
 	"github.com/cloudevents/sdk-go/pkg/types"
 )
 
-const (
-	SKIP_KEY_EXTENSION = "SKIP_KEY_EXTENSION"
-)
-
-// Skip the key extension while encoding to Kafka ProducerMessage
-func WithSkipKeyExtension(ctx context.Context) context.Context {
-	return context.WithValue(ctx, SKIP_KEY_EXTENSION, true)
-}
-
 // Fill the provided producerMessage with the message m.
 // Using context you can tweak the encoding processing (more details on binding.Write documentation).
-// You can skip the key extension handling decorating the context using WithSkipKeyExtension:
-// https://github.com/cloudevents/spec/blob/master/kafka-protocol-binding.md#31-key-attribute
 func WriteProducerMessage(ctx context.Context, m binding.Message, producerMessage *sarama.ProducerMessage, transformerFactories binding.TransformerFactories) error {
-	skipKey := binding.GetOrDefaultFromCtx(ctx, SKIP_KEY_EXTENSION, false).(bool)
-
-	if skipKey {
-		enc := &kafkaProducerMessageWriter{
-			producerMessage,
-			skipKey,
-		}
-
-		_, err := binding.Write(
-			ctx,
-			m,
-			enc,
-			enc,
-			transformerFactories,
-		)
-		return err
-	}
-
-	enc := m.ReadEncoding()
-	var err error
-	// Skip direct encoding if the event is an event message
-	if enc == binding.EncodingBinary {
-		encoder := &kafkaProducerMessageWriter{
-			producerMessage,
-			skipKey,
-		}
-		enc, err = binding.DirectWrite(ctx, m, nil, encoder, transformerFactories)
-		if enc != binding.EncodingUnknown {
-			// Message directly encoded binary -> binary, nothing else to do here
-			return err
-		}
-	}
-
-	var e *ce.Event
-	e, err = binding.ToEvent(ctx, m, transformerFactories)
-	if err != nil {
-		return err
-	}
-
-	if val, ok := e.Extensions()["key"]; ok {
-		s, err := types.Format(val)
-		if err != nil {
-			return err
-		}
-
-		producerMessage.Key = sarama.StringEncoder(s)
-	}
-
-	eventMessage := binding.EventMessage(*e)
-
-	encoder := &kafkaProducerMessageWriter{
-		producerMessage,
-		skipKey,
-	}
-
-	if binding.GetOrDefaultFromCtx(ctx, binding.PREFERRED_EVENT_ENCODING, binding.EncodingBinary).(binding.Encoding) == binding.EncodingStructured {
-		return eventMessage.ReadStructured(ctx, encoder)
-	} else {
-		return eventMessage.ReadBinary(ctx, encoder)
-	}
+	enc := (*kafkaProducerMessageWriter)(producerMessage)
+
+	_, err := binding.Write(
+		ctx,
+		m,
+		enc,
+		enc,
+		transformerFactories,
+	)
+	return err
 }
 
-type kafkaProducerMessageWriter struct {
-	*sarama.ProducerMessage
-	skipKey bool
-}
+type kafkaProducerMessageWriter sarama.ProducerMessage
 
 func (b *kafkaProducerMessageWriter) SetStructuredEvent(ctx context.Context, format format.Format, event io.Reader) error {
 	b.Headers = []sarama.RecordHeader{{
@@ -116,7 +51,7 @@ func (b *kafkaProducerMessageWriter) Start(ctx context.Context) error {
 	return nil
 }
 
-func (b *kafkaProducerMessageWriter) End() error {
+func (b *kafkaProducerMessageWriter) End(ctx context.Context) error {
 	return nil
 }
 
@@ -147,20 +82,7 @@ func (b *kafkaProducerMessageWriter) SetAttribute(attribute spec.Attribute, valu
 }
 
 func (b *kafkaProducerMessageWriter) SetExtension(name string, value interface{}) error {
-	if !b.skipKey && name == "key" {
-		if v, ok := value.([]byte); ok {
-			b.Key = sarama.ByteEncoder(v)
-		} else {
-			s, err := types.Format(value)
-			if err != nil {
-				return err
-			}
-			b.Key = sarama.ByteEncoder(s)
-		}
-		return nil
-	}
-
-	// Http headers, everything is a string!
+	// Kafka headers, everything is a string!
 	s, err := types.Format(value)
 	if err != nil {
 		return err
diff --git a/pkg/transport/nats/protocol.go b/pkg/transport/nats/protocol.go
index 41bae2a..a882a3a 100644
--- a/pkg/transport/nats/protocol.go
+++ b/pkg/transport/nats/protocol.go
@@ -3,8 +3,6 @@ package nats
 import (
 	"context"
 	"github.com/cloudevents/sdk-go/pkg/binding"
-	"github.com/cloudevents/sdk-go/pkg/transport"
-	"github.com/cloudevents/sdk-go/pkg/transport/bindings"
 	"time"
 
 	"github.com/nats-io/nats.go"
@@ -13,8 +11,6 @@ import (
 // Protocol is a reference implementation for using the CloudEvents binding
 // integration. Protocol acts as both a NATS client and a NATS handler.
 type Protocol struct {
-	bindings.BindingTransport
-
 	Conn         *nats.Conn
 	ConnOptions  []nats.Option
 	NatsURL      string
@@ -43,18 +39,9 @@ func New(natsURL, subject string, opts ...Option) (*Protocol, error) {
 		return nil, err
 	}
 
-	t.Sender = t
-	t.Receiver = t
-
 	return t, nil
 }
 
-// Transport returns a CloudEvents transport.Transport.
-// Deprecated: This is for legacy transition until client uses Sender/Receiver directly.
-func (p *Protocol) Transport() transport.Transport {
-	return &p.BindingTransport
-}
-
 func (t *Protocol) connect() error {
 	var err error
 
diff --git a/pkg/transport/nats/write_message.go b/pkg/transport/nats/write_message.go
index a45873f..4596e69 100644
--- a/pkg/transport/nats/write_message.go
+++ b/pkg/transport/nats/write_message.go
@@ -41,7 +41,7 @@ func (b *natsMessageWriter) Start(ctx context.Context) error {
 	return nil
 }
 
-func (b *natsMessageWriter) End() error {
+func (b *natsMessageWriter) End(ctx context.Context) error {
 	return nil
 }
 
diff --git a/pkg/transport/requester.go b/pkg/transport/requester.go
new file mode 100644
index 0000000..a536b86
--- /dev/null
+++ b/pkg/transport/requester.go
@@ -0,0 +1,22 @@
+package transport
+
+import (
+	"context"
+
+	"github.com/cloudevents/sdk-go/pkg/binding"
+)
+
+// Requester sends a message and receives a response
+//
+// Optional interface that may be implemented by protocols that support
+// request/response correlation.
+type Requester interface {
+	// Request sends m like Sender.Send() but also arranges to receive a response.
+	Request(ctx context.Context, m binding.Message) (binding.Message, error)
+}
+
+// RequesterCloser is a Requester that can be closed.
+type RequesterCloser interface {
+	Requester
+	Closer
+}
diff --git a/pkg/transport/sender.go b/pkg/transport/sender.go
index f636acc..d7887a3 100644
--- a/pkg/transport/sender.go
+++ b/pkg/transport/sender.go
@@ -19,17 +19,6 @@ type Sender interface {
 	Send(ctx context.Context, m binding.Message) error
 }
 
-// Requester sends a message and receives a response
-//
-// Optional interface that may be implemented by protocols that support
-// request/response correlation.
-type Requester interface {
-	Sender
-
-	// Request sends m like Sender.Send() but also arranges to receive a response.
-	Request(ctx context.Context, m binding.Message) (binding.Message, error)
-}
-
 // SendCloser is a Sender that can be closed.
 type SendCloser interface {
 	Sender
diff --git a/pkg/transport/transport.go b/pkg/transport/transport.go
deleted file mode 100644
index 01ac16d..0000000
--- a/pkg/transport/transport.go
+++ /dev/null
@@ -1,40 +0,0 @@
-package transport
-
-import (
-	"context"
-
-	"github.com/cloudevents/sdk-go/pkg/event"
-)
-
-// Transport is the interface for transport sender to send the converted Message
-// over the underlying transport.
-type Transport interface {
-	Send(context.Context, event.Event) error
-	Request(context.Context, event.Event) (*event.Event, error)
-
-	SetDelivery(Delivery)
-	StartReceiver(context.Context) error
-
-	// HasTracePropagation is true when the transport implements
-	// in-band trace propagation. When false, the client receiver
-	// will propagate trace context from distributed tracing
-	// extension attributes when available.
-	HasTracePropagation() bool
-}
-
-// REMOVE:
-// Receiver is an interface to define how a transport will invoke a listener
-// of incoming events.
-// Deprecated: use bindings.
-type Delivery interface {
-	// Deprecated: use bindings.
-	Delivery(context.Context, event.Event) (*event.Event, Result)
-}
-
-// DeliveryFunc wraps a function as a Receiver object.
-type DeliveryFunc func(ctx context.Context, e event.Event) (*event.Event, error)
-
-// Receive implements Receiver.Receive
-func (f DeliveryFunc) Receive(ctx context.Context, e event.Event) (*event.Event, error) {
-	return f(ctx, e)
-}


Test patch:
diff --git a/pkg/binding/example_implementing_test.go b/pkg/binding/example_implementing_test.go
deleted file mode 100644
index 98b86fc..0000000
--- a/pkg/binding/example_implementing_test.go
+++ /dev/null
@@ -1,98 +0,0 @@
-package binding_test
-
-import (
-	"bytes"
-	"context"
-	"encoding/json"
-	"io"
-	"io/ioutil"
-
-	bindings "github.com/cloudevents/sdk-go/pkg/transport"
-	bindings2 "github.com/cloudevents/sdk-go/pkg/transport/bindings"
-
-	"github.com/cloudevents/sdk-go/pkg/binding"
-	"github.com/cloudevents/sdk-go/pkg/binding/format"
-	"github.com/cloudevents/sdk-go/pkg/transport"
-)
-
-// ExMessage is a json.RawMessage, a byte slice containing a JSON encoded event.
-// It implements binding.Message
-type ExMessage json.RawMessage
-
-func NewMessage(m json.RawMessage) binding.Message {
-	return ExMessage(m)
-}
-
-func (m ExMessage) ReadEncoding() binding.Encoding {
-	return binding.EncodingStructured
-}
-
-func (m ExMessage) ReadStructured(ctx context.Context, b binding.StructuredWriter) error {
-	return b.SetStructuredEvent(ctx, format.JSON, bytes.NewReader(m))
-}
-
-func (m ExMessage) ReadBinary(context.Context, binding.BinaryWriter) error {
-	return binding.ErrNotBinary
-}
-
-func (m ExMessage) Finish(error) error { return nil }
-
-var _ binding.Message = (*ExMessage)(nil)
-
-// ExSender sends by writing JSON encoded events to an io.Writer
-// ExSender implements directly StructuredWriter in order to perform the conversion
-type ExSender struct {
-	encoder      *json.Encoder
-	transformers binding.TransformerFactories
-}
-
-func NewExSender(w io.Writer, factories ...binding.TransformerFactory) bindings.Sender {
-	return &ExSender{encoder: json.NewEncoder(w), transformers: factories}
-}
-
-func (s *ExSender) Send(ctx context.Context, m binding.Message) error {
-	// Write tries to perform the encoding directly, otherwise
-	// It fallbacks to converting to event.Event and then convert back to the provided encoders
-	_, err := binding.Write(
-		ctx,
-		m,
-		s,
-		nil,
-		s.transformers,
-	)
-
-	return err
-}
-
-func (s *ExSender) SetStructuredEvent(ctx context.Context, f format.Format, event io.Reader) error {
-	if f == format.JSON {
-		b, err := ioutil.ReadAll(event)
-		if err != nil {
-			return err
-		}
-		return s.encoder.Encode(json.RawMessage(b))
-	} else {
-		return binding.ErrNotStructured
-	}
-}
-
-var _ bindings.Sender = (*ExSender)(nil)
-var _ binding.StructuredWriter = (*ExSender)(nil)
-
-// ExReceiver receives by reading JSON encoded events from an io.Reader
-type ExReceiver struct{ decoder *json.Decoder }
-
-func NewExReceiver(r io.Reader) bindings.Receiver { return &ExReceiver{json.NewDecoder(r)} }
-
-func (r *ExReceiver) Receive(context.Context) (binding.Message, error) {
-	var rm json.RawMessage
-	err := r.decoder.Decode(&rm) // This is just a byte copy.
-	return ExMessage(rm), err
-}
-func (r *ExReceiver) Close(context.Context) error { return nil }
-
-// NewExTransport returns a transport.Transport which is implemented by
-// an ExSender and an ExReceiver
-func NewExTransport(r io.Reader, w io.Writer) transport.Transport {
-	return bindings2.NewSendingTransport(NewExSender(w), NewExReceiver(r), []func(ctx context.Context) context.Context{})
-}
diff --git a/pkg/binding/example_using_test.go b/pkg/binding/example_using_test.go
deleted file mode 100644
index b70a3b0..0000000
--- a/pkg/binding/example_using_test.go
+++ /dev/null
@@ -1,125 +0,0 @@
-package binding_test
-
-import (
-	"context"
-	"fmt"
-	"io"
-	"strconv"
-
-	"github.com/cloudevents/sdk-go/pkg/client"
-	"github.com/cloudevents/sdk-go/pkg/event"
-)
-
-const count = 3 // Example ends after this many events.
-
-// The sender uses the cloudevents.Client API, not the transport APIs directly.
-func runSender(w io.Writer) error {
-	c, err := client.New(NewExTransport(nil, w), client.WithoutTracePropagation())
-	if err != nil {
-		return err
-	}
-	for i := 0; i < count; i++ {
-		e := event.New()
-		e.SetType("example.com/event")
-		e.SetSource("example.com/source")
-		e.SetID(strconv.Itoa(i))
-		if err := e.SetData(event.TextJSON, fmt.Sprintf("hello %d", i)); err != nil {
-			return err
-		}
-		if err := c.Send(context.TODO(), e); err != nil {
-			return err
-		}
-	}
-	return nil
-}
-
-// The receiver uses the cloudevents.Client API, not the transport APIs directly.
-func runReceiver(r io.Reader) error {
-	i := 0
-	process := func(e event.Event) error {
-		fmt.Printf("%s\n", e)
-		i++
-		if i == count {
-			return io.EOF
-		}
-		return nil
-	}
-	c, err := client.New(NewExTransport(r, nil), client.WithoutTracePropagation())
-	if err != nil {
-		return err
-	}
-	return c.StartReceiver(context.TODO(), process)
-}
-
-// The intermediary receives events and forwards them to another
-// process using ExReceiver and ExSender directly.
-//
-// By forwarding a transport.Message instead of a event.Event,
-// it allows the transports to avoid un-necessary decoding of
-// structured events, and to exchange delivery status between reliable
-// transports. Even transports using different protocols can ensure
-// reliable delivery.
-//
-func runIntermediary(r io.Reader, w io.WriteCloser) error {
-	defer w.Close()
-	receiver := NewExReceiver(r)
-	sender := NewExSender(w)
-	for i := 0; i < count; i++ {
-		if m, err := receiver.Receive(context.TODO()); err != nil {
-			return err
-		} else if err := sender.Send(context.TODO(), m); err != nil {
-			return err
-		}
-	}
-	return nil
-}
-
-// This example shows how to use a transport in sender, receiver,
-// and intermediary processes.
-//
-// The sender and receiver use the client.Client API to send and
-// receive messages.  the transport.  Only the intermediary example
-// actually uses the transport APIs for efficiency and reliability in
-// forwarding events.
-func Exampl_using() {
-	r1, w1 := io.Pipe() // The sender-to-intermediary pipe
-	r2, w2 := io.Pipe() // The intermediary-to-receiver pipe
-
-	done := make(chan error)
-	go func() { done <- runReceiver(r2) }()
-	go func() { done <- runIntermediary(r1, w2) }()
-	go func() { done <- runSender(w1) }()
-	for i := 0; i < 2; i++ {
-		if err := <-done; err != nil && err != io.EOF {
-			fmt.Println(err)
-		}
-	}
-
-	// Output:
-	// Validation: valid
-	// Context Attributes,
-	//   specversion: 1.0
-	//   type: example.com/event
-	//   source: example.com/source
-	//   id: 0
-	// Data,
-	//   "hello 0"
-	//
-	// Validation: valid
-	// Context Attributes,
-	//   specversion: 1.0
-	//   type: example.com/event
-	//   source: example.com/source
-	//   id: 1
-	// Data,
-	//   "hello 1"
-	//
-	// Validation: valid
-	// Context Attributes,
-	//   specversion: 1.0
-	//   type: example.com/event
-	//   source: example.com/source
-	//   id: 2
-	// Data,
-	//   "hello 2"
-}
diff --git a/pkg/binding/test/mock_binary_message.go b/pkg/binding/test/mock_binary_message.go
index 812df17..b3a7631 100644
--- a/pkg/binding/test/mock_binary_message.go
+++ b/pkg/binding/test/mock_binary_message.go
@@ -71,7 +71,7 @@ func (bm *MockBinaryMessage) ReadBinary(ctx context.Context, b binding.BinaryWri
 			return err
 		}
 	}
-	return b.End()
+	return b.End(ctx)
 }
 
 func (bm *MockBinaryMessage) ReadEncoding() binding.Encoding {
@@ -101,7 +101,7 @@ func (bm *MockBinaryMessage) SetData(data io.Reader) (err error) {
 	return err
 }
 
-func (bm *MockBinaryMessage) End() error {
+func (bm *MockBinaryMessage) End(ctx context.Context) error {
 	return nil
 }
 
diff --git a/pkg/client/client_test.go b/pkg/client/client_test.go
index 1219461..673878d 100644
--- a/pkg/client/client_test.go
+++ b/pkg/client/client_test.go
@@ -37,20 +37,13 @@ var (
 )
 
 func simpleBinaryClient(target string) client.Client {
-	p, err := cehttp.NewProtocol(cehttp.WithTarget(target))
+	p, err := cehttp.New(cehttp.WithTarget(target))
 	if err != nil {
 		log.Printf("failed to create protocol, %v", err)
 		return nil
 	}
 
-	t, err := cehttp.New(p,
-		cehttp.WithBinaryEncoding(),
-	)
-	if err != nil {
-		return nil
-	}
-
-	c, err := client.New(t, client.WithoutTracePropagation())
+	c, err := client.New(p, client.WithoutTracePropagation(), client.WithForceBinary())
 	if err != nil {
 		return nil
 	}
@@ -58,20 +51,13 @@ func simpleBinaryClient(target string) client.Client {
 }
 
 func simpleTracingBinaryClient(target string) client.Client {
-	p, err := cehttp.NewProtocol(cehttp.WithTarget(target))
+	p, err := cehttp.New(cehttp.WithTarget(target))
 	if err != nil {
 		log.Printf("failed to create protocol, %v", err)
 		return nil
 	}
 
-	t, err := cehttp.New(p,
-		cehttp.WithBinaryEncoding(),
-	)
-	if err != nil {
-		return nil
-	}
-
-	c, err := client.New(t)
+	c, err := client.New(p)
 	if err != nil {
 		return nil
 	}
@@ -79,20 +65,13 @@ func simpleTracingBinaryClient(target string) client.Client {
 }
 
 func simpleStructuredClient(target string) client.Client {
-	p, err := cehttp.NewProtocol(cehttp.WithTarget(target))
+	p, err := cehttp.New(cehttp.WithTarget(target))
 	if err != nil {
 		log.Printf("failed to create protocol, %v", err)
 		return nil
 	}
 
-	t, err := cehttp.New(p,
-		cehttp.WithStructuredEncoding(),
-	)
-	if err != nil {
-		return nil
-	}
-
-	c, err := client.New(t, client.WithoutTracePropagation())
+	c, err := client.New(p, client.WithoutTracePropagation(), client.WithForceStructured())
 	if err != nil {
 		return nil
 	}
@@ -208,6 +187,8 @@ func TestClientSend(t *testing.T) {
 }
 
 func TestTracingClientSend(t *testing.T) {
+	t.Skip("skipping tracing tests for now, need to rework this for sdk v2")
+
 	now := time.Now()
 
 	testCases := map[string]struct {
@@ -315,7 +296,7 @@ func TestTracingClientSend(t *testing.T) {
 func simpleBinaryOptions(port int, path string) []cehttp.Option {
 	opts := []cehttp.Option{
 		cehttp.WithPort(port),
-		cehttp.WithBinaryEncoding(),
+		//cehttp.WithBinaryEncoding(),
 	}
 	if len(path) > 0 {
 		opts = append(opts, cehttp.WithPath(path))
@@ -326,7 +307,7 @@ func simpleBinaryOptions(port int, path string) []cehttp.Option {
 func simpleStructuredOptions(port int, path string) []cehttp.Option {
 	opts := []cehttp.Option{
 		cehttp.WithPort(port),
-		cehttp.WithStructuredEncoding(),
+		//cehttp.WithStructuredEncoding(),
 	}
 	if len(path) > 0 {
 		opts = append(opts, cehttp.WithPath(path))
@@ -405,17 +386,12 @@ func TestClientReceive(t *testing.T) {
 
 				events := make(chan event.Event)
 
-				p, err := cehttp.NewProtocol()
+				p, err := cehttp.New(tc.optsFn(0, "")...)
 				if err != nil {
 					t.Fatal(err)
 				}
 
-				tp, err := cehttp.New(p, tc.optsFn(0, path)...)
-				if err != nil {
-					t.Errorf("failed to make http transport %s", err.Error())
-				}
-
-				c, err := client.New(tp)
+				c, err := client.New(p)
 				if err != nil {
 					t.Errorf("failed to make client %s", err.Error())
 				}
@@ -434,7 +410,7 @@ func TestClientReceive(t *testing.T) {
 				}()
 				time.Sleep(5 * time.Millisecond) // let the server start
 
-				target, _ := url.Parse(fmt.Sprintf("http://localhost:%d%s", tp.GetPort(), tp.GetPath()))
+				target, _ := url.Parse(fmt.Sprintf("http://localhost:%d%s", p.GetPort(), p.GetPath()))
 
 				if tc.wantErr != "" {
 					if err == nil {
@@ -530,17 +506,12 @@ func TestTracedClientReceive(t *testing.T) {
 		t.Run(n, func(t *testing.T) {
 			spanContexts := make(chan trace.SpanContext)
 
-			p, err := cehttp.NewProtocol()
+			p, err := cehttp.New(tc.optsFn(0, "")...)
 			if err != nil {
 				t.Fatal(err)
 			}
 
-			tp, err := cehttp.New(p, tc.optsFn(0, "")...)
-			if err != nil {
-				t.Errorf("failed to make http transport %s", err.Error())
-			}
-
-			c, err := client.New(tp)
+			c, err := client.New(p)
 			if err != nil {
 				t.Errorf("failed to make client %s", err.Error())
 			}
@@ -559,7 +530,7 @@ func TestTracedClientReceive(t *testing.T) {
 			}()
 			time.Sleep(5 * time.Millisecond) // let the server start
 
-			target := fmt.Sprintf("http://localhost:%d", tp.GetPort())
+			target := fmt.Sprintf("http://localhost:%d", p.GetPort())
 			client := simpleBinaryClient(target)
 
 			ctx, span := trace.StartSpan(context.TODO(), "test-span")
diff --git a/pkg/transport/amqp/transport_test.go b/pkg/transport/amqp/transport_test.go
deleted file mode 100644
index 2808d60..0000000
--- a/pkg/transport/amqp/transport_test.go
+++ /dev/null
@@ -1,110 +0,0 @@
-// +build amqp
-
-package amqp_test
-
-import (
-	"context"
-	"net/url"
-	"os"
-	"testing"
-
-	"github.com/stretchr/testify/require"
-
-	"github.com/cloudevents/sdk-go/pkg/binding/spec"
-	"github.com/cloudevents/sdk-go/pkg/event"
-	"github.com/cloudevents/sdk-go/pkg/transport"
-	"github.com/cloudevents/sdk-go/pkg/transport/amqp"
-	"github.com/cloudevents/sdk-go/pkg/types"
-
-	. "github.com/cloudevents/sdk-go/pkg/binding/test"
-)
-
-// Requires an external AMQP broker or router, skip if not available.
-// The env variable TEST_AMQP_URL provides the URL, default is "/test"
-//
-// One option is http://qpid.apache.org/components/dispatch-router/indext.html.
-// It can be installed from source or from RPMs, see https://qpid.apache.org/packages.html
-// Run `qdrouterd` and the tests will work with no further config.
-func testTransport(t *testing.T, opts ...amqp.Option) *amqp.Transport {
-	t.Helper()
-	addr := "test"
-	s := os.Getenv("TEST_AMQP_URL")
-	if u, err := url.Parse(s); err == nil && u.Path != "" {
-		addr = u.Path
-	}
-	tx, err := amqp.New(s, addr, opts...)
-	if err != nil {
-		t.Skipf("ampq.New(%#v): %v", s, err)
-	}
-	return tx
-}
-
-type tester struct {
-	s, r transport.Transport
-	got  chan interface{} // ce.Event or error
-}
-
-func (t *tester) Delivery(_ context.Context, e event.Event) (*event.Event, transport.Result) {
-	t.got <- e
-	return nil, nil
-}
-
-func (t *tester) Close() {
-	_ = t.s.(*amqp.Transport).Close()
-	_ = t.r.(*amqp.Transport).Close()
-}
-
-func newTester(t *testing.T, sendOpts, recvOpts []amqp.Option) *tester {
-	t.Helper()
-	tester := &tester{
-		s:   testTransport(t, sendOpts...),
-		r:   testTransport(t, recvOpts...),
-		got: make(chan interface{}),
-	}
-	got := make(chan interface{}, 100)
-	go func() {
-		defer func() { close(got) }()
-		tester.r.SetDelivery(tester)
-		err := tester.r.StartReceiver(context.Background())
-		if err != nil {
-			got <- err
-		}
-	}()
-	return tester
-}
-
-func exurl(e event.Event) event.Event {
-	// Flatten exurl to string, AMQP doesn't preserve the URL type.
-	// It should preserve other attribute types.
-	if s, _ := types.Format(e.Extensions()["exurl"]); s != "" {
-		e.SetExtension("exurl", s)
-	}
-	return e
-}
-
-func TestSendReceive(t *testing.T) {
-	ctx := context.Background()
-	tester := newTester(t, nil, nil)
-	defer tester.Close()
-	EachEvent(t, Events(), func(t *testing.T, e event.Event) {
-		err := tester.s.Send(ctx, e)
-		require.NoError(t, err)
-		got := <-tester.got
-		AssertEventEquals(t, exurl(e), got.(event.Event))
-	})
-}
-
-func TestWithEncoding(t *testing.T) {
-	ctx := context.Background()
-	tester := newTester(t, []amqp.Option{amqp.WithEncoding(amqp.StructuredV03)}, nil)
-	defer tester.Close()
-	// FIXME(alanconway) problem with JSON round-tripping extensions
-	events := NoExtensions(Events())
-	EachEvent(t, events, func(t *testing.T, e event.Event) {
-		err := tester.s.Send(ctx, e)
-		require.NoError(t, err)
-		got := <-tester.got
-		e.Context = spec.V03.Convert(e.Context)
-		AssertEventEquals(t, exurl(e), got.(event.Event))
-	})
-}
diff --git a/pkg/transport/bindings/transport_test.go b/pkg/transport/bindings/transport_test.go
deleted file mode 100644
index f709c81..0000000
--- a/pkg/transport/bindings/transport_test.go
+++ /dev/null
@@ -1,54 +0,0 @@
-package bindings_test
-
-import (
-	"context"
-	"testing"
-
-	bindings2 "github.com/cloudevents/sdk-go/pkg/transport/bindings"
-
-	"github.com/stretchr/testify/require"
-
-	"github.com/cloudevents/sdk-go/pkg/binding"
-	"github.com/cloudevents/sdk-go/pkg/binding/test"
-	client "github.com/cloudevents/sdk-go/pkg/client"
-	"github.com/cloudevents/sdk-go/pkg/event"
-)
-
-func TestTransportSend(t *testing.T) {
-	messageChannel := make(chan binding.Message, 1)
-	transport := bindings2.NewSendingTransport(binding.ChanSender(messageChannel), binding.ChanReceiver(messageChannel), nil)
-	ev := test.MinEvent()
-
-	c, err := client.New(transport, client.WithoutTracePropagation())
-	require.NoError(t, err)
-
-	err = c.Send(context.Background(), ev)
-	require.NoError(t, err)
-
-	result := <-messageChannel
-
-	test.AssertEventEquals(t, ev, event.Event(*(result.(*binding.EventMessage))))
-}
-
-func TestTransportReceive(t *testing.T) {
-	messageChannel := make(chan binding.Message, 1)
-	eventReceivedChannel := make(chan event.Event, 1)
-	transport := bindings2.NewSendingTransport(binding.ChanSender(messageChannel), binding.ChanReceiver(messageChannel), nil)
-	ev := test.MinEvent()
-
-	c, err := client.New(transport)
-	require.NoError(t, err)
-
-	messageChannel <- (*binding.EventMessage)(&ev)
-
-	go func() {
-		err = c.StartReceiver(context.Background(), func(event event.Event) {
-			eventReceivedChannel <- event
-		})
-		require.NoError(t, err)
-	}()
-
-	result := <-eventReceivedChannel
-
-	test.AssertEventEquals(t, ev, result)
-}
diff --git a/pkg/transport/kafka_sarama/message_benchmark_test.go b/pkg/transport/kafka_sarama/message_benchmark_test.go
index bd132ed..f343528 100644
--- a/pkg/transport/kafka_sarama/message_benchmark_test.go
+++ b/pkg/transport/kafka_sarama/message_benchmark_test.go
@@ -19,43 +19,20 @@ var Event *event.Event
 var Err error
 
 var (
-	e                                   = test.FullEvent()
-	structuredConsumerMessageWithoutKey = &sarama.ConsumerMessage{
-		Value: test.MustJSON(e),
+	benchEvent                = test.FullEvent()
+	structuredConsumerMessage = &sarama.ConsumerMessage{
+		Value: test.MustJSON(benchEvent),
 		Headers: []*sarama.RecordHeader{{
 			Key:   []byte("content-type"),
 			Value: []byte(cloudevents.ApplicationCloudEventsJSON),
 		}},
 	}
-	structuredConsumerMessageWithKey = &sarama.ConsumerMessage{
-		Key:   []byte("aaa"),
-		Value: test.MustJSON(e),
-		Headers: []*sarama.RecordHeader{{
-			Key:   []byte("content-type"),
-			Value: []byte(cloudevents.ApplicationCloudEventsJSON),
-		}},
-	}
-	binaryConsumerMessageWithoutKey = &sarama.ConsumerMessage{
+	binaryConsumerMessage = &sarama.ConsumerMessage{
 		Value: []byte("hello world!"),
 		Headers: mustToSaramaConsumerHeaders(map[string]string{
-			"ce_type":            e.Type(),
-			"ce_source":          e.Source(),
-			"ce_id":              e.ID(),
-			"ce_time":            test.Timestamp.String(),
-			"ce_specversion":     "1.0",
-			"ce_dataschema":      test.Schema.String(),
-			"ce_datacontenttype": "text/json",
-			"ce_subject":         "topic",
-			"ce_exta":            "someext",
-		}),
-	}
-	binaryConsumerMessageWithKey = &sarama.ConsumerMessage{
-		Key:   []byte("akey"),
-		Value: []byte("hello world!"),
-		Headers: mustToSaramaConsumerHeaders(map[string]string{
-			"ce_type":            e.Type(),
-			"ce_source":          e.Source(),
-			"ce_id":              e.ID(),
+			"ce_type":            benchEvent.Type(),
+			"ce_source":          benchEvent.Source(),
+			"ce_id":              benchEvent.ID(),
 			"ce_time":            test.Timestamp.String(),
 			"ce_specversion":     "1.0",
 			"ce_dataschema":      test.Schema.String(),
@@ -66,54 +43,28 @@ var (
 	}
 )
 
-func BenchmarkNewStructuredMessageWithoutKey(b *testing.B) {
+func BenchmarkNewStructuredMessage(b *testing.B) {
 	for i := 0; i < b.N; i++ {
-		M, Err = kafka_sarama.NewMessageFromConsumerMessage(structuredConsumerMessageWithoutKey)
+		M = kafka_sarama.NewMessageFromConsumerMessage(structuredConsumerMessage)
 	}
 }
 
-func BenchmarkNewStructuredMessageWithKey(b *testing.B) {
+func BenchmarkNewBinaryMessage(b *testing.B) {
 	for i := 0; i < b.N; i++ {
-		M, Err = kafka_sarama.NewMessageFromConsumerMessage(structuredConsumerMessageWithKey)
-	}
-}
-
-func BenchmarkNewBinaryMessageWithoutKey(b *testing.B) {
-	for i := 0; i < b.N; i++ {
-		M, Err = kafka_sarama.NewMessageFromConsumerMessage(binaryConsumerMessageWithoutKey)
-	}
-}
-
-func BenchmarkNewBinaryMessageWithKey(b *testing.B) {
-	for i := 0; i < b.N; i++ {
-		M, Err = kafka_sarama.NewMessageFromConsumerMessage(binaryConsumerMessageWithKey)
-	}
-}
-
-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)
-	}
-}
-
-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)
+		M = kafka_sarama.NewMessageFromConsumerMessage(binaryConsumerMessage)
 	}
 }
 
-func BenchmarkNewBinaryMessageWithoutKeyToEvent(b *testing.B) {
+func BenchmarkNewStructuredMessageToEvent(b *testing.B) {
 	for i := 0; i < b.N; i++ {
-		M, Err = kafka_sarama.NewMessageFromConsumerMessage(binaryConsumerMessageWithoutKey)
+		M = kafka_sarama.NewMessageFromConsumerMessage(structuredConsumerMessage)
 		Event, Err = binding.ToEvent(context.TODO(), M, nil)
 	}
 }
 
-func BenchmarkNewBinaryMessageWithKeyToEvent(b *testing.B) {
+func BenchmarkNewBinaryMessageToEvent(b *testing.B) {
 	for i := 0; i < b.N; i++ {
-		M, Err = kafka_sarama.NewMessageFromConsumerMessage(binaryConsumerMessageWithKey)
+		M = kafka_sarama.NewMessageFromConsumerMessage(binaryConsumerMessage)
 		Event, Err = binding.ToEvent(context.TODO(), M, nil)
 	}
 }
diff --git a/pkg/transport/kafka_sarama/write_producer_message_benchmark_test.go b/pkg/transport/kafka_sarama/write_producer_message_benchmark_test.go
index 88c9838..eaac7ab 100644
--- a/pkg/transport/kafka_sarama/write_producer_message_benchmark_test.go
+++ b/pkg/transport/kafka_sarama/write_producer_message_benchmark_test.go
@@ -16,54 +16,31 @@ import (
 var ProducerMessage *sarama.ProducerMessage
 
 var (
-	ctxSkipKey                  context.Context
-	ctx                         context.Context
-	eventWithoutKey             event.Event
-	eventWithKey                event.Event
-	structuredMessageWithoutKey binding.Message
-	structuredMessageWithKey    binding.Message
-	binaryMessageWithoutKey     binding.Message
-	binaryMessageWithKey        binding.Message
+	ctx               context.Context
+	initialEvent      event.Event
+	structuredMessage binding.Message
+	binaryMessage     binding.Message
 )
 
 func init() {
-	ctxSkipKey = kafka_sarama.WithSkipKeyExtension(context.TODO())
 	ctx = context.TODO()
 
-	eventWithoutKey = test.FullEvent()
-	eventWithKey = test.FullEvent()
-	eventWithKey.SetExtension("key", "aaaaaa")
+	initialEvent = test.FullEvent()
 
-	structuredMessageWithoutKey = test.MustCreateMockStructuredMessage(eventWithoutKey)
-	structuredMessageWithKey = test.MustCreateMockStructuredMessage(eventWithKey)
-	binaryMessageWithoutKey = test.MustCreateMockBinaryMessage(eventWithoutKey)
-	binaryMessageWithKey = test.MustCreateMockBinaryMessage(eventWithKey)
-}
-
-func BenchmarkEncodeStructuredMessageSkipKey(b *testing.B) {
-	for i := 0; i < b.N; i++ {
-		ProducerMessage = &sarama.ProducerMessage{}
-		Err = kafka_sarama.WriteProducerMessage(ctxSkipKey, structuredMessageWithoutKey, ProducerMessage, nil)
-	}
+	structuredMessage = test.MustCreateMockStructuredMessage(initialEvent)
+	binaryMessage = test.MustCreateMockBinaryMessage(initialEvent)
 }
 
 func BenchmarkEncodeStructuredMessage(b *testing.B) {
 	for i := 0; i < b.N; i++ {
 		ProducerMessage = &sarama.ProducerMessage{}
-		Err = kafka_sarama.WriteProducerMessage(ctx, structuredMessageWithKey, ProducerMessage, nil)
-	}
-}
-
-func BenchmarkEncodeBinaryMessageSkipKey(b *testing.B) {
-	for i := 0; i < b.N; i++ {
-		ProducerMessage = &sarama.ProducerMessage{}
-		Err = kafka_sarama.WriteProducerMessage(ctxSkipKey, binaryMessageWithoutKey, ProducerMessage, nil)
+		Err = kafka_sarama.WriteProducerMessage(ctx, structuredMessage, ProducerMessage, nil)
 	}
 }
 
 func BenchmarkEncodeBinaryMessage(b *testing.B) {
 	for i := 0; i < b.N; i++ {
 		ProducerMessage = &sarama.ProducerMessage{}
-		Err = kafka_sarama.WriteProducerMessage(ctx, binaryMessageWithKey, ProducerMessage, nil)
+		Err = kafka_sarama.WriteProducerMessage(ctx, binaryMessage, ProducerMessage, nil)
 	}
 }
diff --git a/pkg/transport/kafka_sarama/write_producer_message_test.go b/pkg/transport/kafka_sarama/write_producer_message_test.go
index 06c1177..f99ca46 100644
--- a/pkg/transport/kafka_sarama/write_producer_message_test.go
+++ b/pkg/transport/kafka_sarama/write_producer_message_test.go
@@ -19,63 +19,30 @@ func TestEncodeKafkaProducerMessage(t *testing.T) {
 		context          context.Context
 		messageFactory   func(e event.Event) binding.Message
 		expectedEncoding binding.Encoding
-		skipKey          bool
 	}{
 		{
-			name:             "Structured to Structured with Skip key",
+			name:             "Structured to Structured",
 			context:          context.TODO(),
 			messageFactory:   test.MustCreateMockStructuredMessage,
 			expectedEncoding: binding.EncodingStructured,
-			skipKey:          true,
-		},
-		{
-			name:             "Binary to Binary with Skip key",
-			context:          context.TODO(),
-			messageFactory:   test.MustCreateMockBinaryMessage,
-			expectedEncoding: binding.EncodingBinary,
-			skipKey:          true,
-		},
-		{
-			name:             "Event to Structured with Skip key",
-			context:          binding.WithPreferredEventEncoding(context.TODO(), binding.EncodingStructured),
-			messageFactory:   func(e event.Event) binding.Message { return (*binding.EventMessage)(&e) },
-			expectedEncoding: binding.EncodingStructured,
-			skipKey:          true,
-		},
-		{
-			name:             "Event to Binary with Skip key",
-			context:          binding.WithPreferredEventEncoding(context.TODO(), binding.EncodingBinary),
-			messageFactory:   func(e event.Event) binding.Message { return (*binding.EventMessage)(&e) },
-			expectedEncoding: binding.EncodingBinary,
-			skipKey:          true,
-		},
-		{
-			name:             "Structured to Structured",
-			context:          binding.WithPreferredEventEncoding(context.TODO(), binding.EncodingStructured),
-			messageFactory:   func(e event.Event) binding.Message { return test.MustCreateMockStructuredMessage(e) },
-			expectedEncoding: binding.EncodingEvent,
-			skipKey:          false,
 		},
 		{
 			name:             "Binary to Binary",
 			context:          context.TODO(),
-			messageFactory:   func(e event.Event) binding.Message { return test.MustCreateMockBinaryMessage(e) },
+			messageFactory:   test.MustCreateMockBinaryMessage,
 			expectedEncoding: binding.EncodingBinary,
-			skipKey:          false,
 		},
 		{
 			name:             "Event to Structured",
 			context:          binding.WithPreferredEventEncoding(context.TODO(), binding.EncodingStructured),
 			messageFactory:   func(e event.Event) binding.Message { return (*binding.EventMessage)(&e) },
-			expectedEncoding: binding.EncodingEvent,
-			skipKey:          false,
+			expectedEncoding: binding.EncodingStructured,
 		},
 		{
 			name:             "Event to Binary",
 			context:          binding.WithPreferredEventEncoding(context.TODO(), binding.EncodingBinary),
 			messageFactory:   func(e event.Event) binding.Message { return (*binding.EventMessage)(&e) },
 			expectedEncoding: binding.EncodingBinary,
-			skipKey:          false,
 		},
 	}
 	for _, tt := range tests {
@@ -83,12 +50,6 @@ func TestEncodeKafkaProducerMessage(t *testing.T) {
 			t.Run(tt.name, func(t *testing.T) {
 				ctx := tt.context
 
-				if tt.skipKey {
-					ctx = WithSkipKeyExtension(ctx)
-				} else {
-					eventIn.SetExtension("key", "bla")
-				}
-
 				kafkaMessage := &sarama.ProducerMessage{
 					Topic: "aaa",
 				}
@@ -117,13 +78,8 @@ func TestEncodeKafkaProducerMessage(t *testing.T) {
 					require.NoError(t, err)
 				}
 
-				if !tt.skipKey {
-					require.Equal(t, []byte("bla"), key)
-				}
-
-				messageOut, err := NewMessage(key, value, string(headers[contentTypeHeader]), headers)
+				messageOut := NewMessage(key, value, string(headers[contentTypeHeader]), headers)
 				require.Equal(t, tt.expectedEncoding, messageOut.ReadEncoding())
-				require.NoError(t, err)
 
 				eventOut, err := binding.ToEvent(context.TODO(), messageOut, nil)
 				require.NoError(t, err)
diff --git a/test/benchmark/e2e/http/http_mock.go b/test/benchmark/e2e/http/http_mock.go
index 8b32aca..a9445c1 100644
--- a/test/benchmark/e2e/http/http_mock.go
+++ b/test/benchmark/e2e/http/http_mock.go
@@ -27,10 +27,10 @@ func NewTestClient(fn RoundTripFunc) *nethttp.Client {
 	}
 }
 
-func MockedSender(options ...http.ProtocolOption) bindings.Sender {
+func MockedSender(options ...http.Option) bindings.Sender {
 	u, _ := url.Parse("http://localhost")
 
-	p, err := http.NewProtocol(options...)
+	p, err := http.New(options...)
 	if err != nil {
 		panic(err)
 	}
@@ -44,7 +44,7 @@ func MockedSender(options ...http.ProtocolOption) bindings.Sender {
 	return p
 }
 
-func MockedClient() (cloudevents.Client, *cehttp.Protocol, *cehttp.Transport) {
+func MockedClient() (cloudevents.Client, *cehttp.Protocol) {
 	mockTransport := RoundTripFunc(func(req *nethttp.Request) *nethttp.Response {
 		return &nethttp.Response{
 			StatusCode: 202,
@@ -53,22 +53,17 @@ func MockedClient() (cloudevents.Client, *cehttp.Protocol, *cehttp.Transport) {
 		}
 	})
 
-	p, err := cehttp.NewProtocol(cehttp.WithTarget("http://localhost"))
+	p, err := cehttp.New(cehttp.WithTarget("http://localhost"), cehttp.WithHTTPTransport(mockTransport))
 	if err != nil {
 		panic(err)
 	}
 
-	t, err := cehttp.New(p, cehttp.WithHTTPTransport(mockTransport))
+	client, err := cloudevents.NewClient(p)
 	if err != nil {
 		panic(err)
 	}
 
-	client, err := cloudevents.NewClient(t)
-	if err != nil {
-		panic(err)
-	}
-
-	return client, p, t
+	return client, p
 }
 
 func MockedBinaryRequest(body []byte) *nethttp.Request {
diff --git a/test/benchmark/e2e/http/main.go b/test/benchmark/e2e/http/main.go
index 9ecf7f4..ad3022b 100644
--- a/test/benchmark/e2e/http/main.go
+++ b/test/benchmark/e2e/http/main.go
@@ -12,16 +12,12 @@ import (
 	"os"
 	"runtime"
 	"runtime/pprof"
-	"sync"
 	"testing"
 	"time"
 
-	"github.com/cloudevents/sdk-go/pkg/event"
-
 	cloudevents "github.com/cloudevents/sdk-go"
 	"github.com/cloudevents/sdk-go/pkg/binding"
 	"github.com/cloudevents/sdk-go/pkg/binding/buffering"
-	"github.com/cloudevents/sdk-go/pkg/transport"
 	http "github.com/cloudevents/sdk-go/pkg/transport/http"
 	"github.com/cloudevents/sdk-go/test/benchmark/e2e"
 )
@@ -67,7 +63,7 @@ func benchmarkBaseline(cases []e2e.BenchmarkCase, requestFactory func([]byte) *n
 	return results
 }
 
-func pipeLoopDirect(r *http.Protocol, sendCtx context.Context, endCtx context.Context, opts ...http.ProtocolOption) {
+func pipeLoopDirect(r *http.Protocol, sendCtx context.Context, endCtx context.Context, opts ...http.Option) {
 	s := MockedSender(opts...)
 	var err error
 	var m binding.Message
@@ -85,7 +81,7 @@ func pipeLoopDirect(r *http.Protocol, sendCtx context.Context, endCtx context.Co
 	}
 }
 
-func pipeLoopMulti(r *http.Protocol, sendCtx context.Context, endCtx context.Context, outputSenders int, opts ...http.ProtocolOption) {
+func pipeLoopMulti(r *http.Protocol, sendCtx context.Context, endCtx context.Context, outputSenders int, opts ...http.Option) {
 	s := MockedSender(opts...)
 	var err error
 	var m binding.Message
@@ -120,7 +116,7 @@ func benchmarkReceiverSender(cases []e2e.BenchmarkCase, requestFactory func([]by
 		fmt.Printf("%+v\n", c)
 
 		ctx, cancel := context.WithCancel(context.TODO())
-		receiver, err := http.NewProtocol()
+		receiver, err := http.New()
 		if err != nil {
 			panic(err)
 		}
@@ -156,28 +152,29 @@ func benchmarkReceiverSender(cases []e2e.BenchmarkCase, requestFactory func([]by
 	return results
 }
 
-type benchDelivery struct {
-	fn transport.DeliveryFunc
-}
-
-func (b *benchDelivery) Delivery(ctx context.Context, e event.Event) (*event.Event, transport.Result) {
-	return b.fn(ctx, e)
-}
-
-func dispatchReceiver(clients []cloudevents.Client, outputSenders int) transport.Delivery {
-	return &benchDelivery{fn: func(ctx context.Context, e cloudevents.Event) (*cloudevents.Event, error) {
-		var wg sync.WaitGroup
-		for i := 0; i < outputSenders; i++ {
-			wg.Add(1)
-			go func(client cloudevents.Client) {
-				_ = client.Send(ctx, e)
-				wg.Done()
-			}(clients[i])
-		}
-		wg.Wait()
-		return nil, nil
-	}}
-}
+//
+//type benchDelivery struct {
+//	fn transport.DeliveryFunc
+//}
+//
+//func (b *benchDelivery) Delivery(ctx context.Context, e event.Event) (*event.Event, transport.Result) {
+//	return b.fn(ctx, e)
+//}
+
+//func dispatchReceiver(clients []cloudevents.Client, outputSenders int) transport.Delivery {
+//	return &benchDelivery{fn: func(ctx context.Context, e cloudevents.Event) (*cloudevents.Event, error) {
+//		var wg sync.WaitGroup
+//		for i := 0; i < outputSenders; i++ {
+//			wg.Add(1)
+//			go func(client cloudevents.Client) {
+//				_ = client.Send(ctx, e)
+//				wg.Done()
+//			}(clients[i])
+//		}
+//		wg.Wait()
+//		return nil, nil
+//	}}
+//}
 
 func benchmarkClient(cases []e2e.BenchmarkCase, requestFactory func([]byte) *nethttp.Request) e2e.BenchmarkResults {
 	var results e2e.BenchmarkResults
@@ -186,14 +183,14 @@ func benchmarkClient(cases []e2e.BenchmarkCase, requestFactory func([]byte) *net
 	for _, c := range cases {
 		fmt.Printf("%+v\n", c)
 
-		_, mockedReceiverProtocol, mockedReceiverTransport := MockedClient()
+		//_, mockedReceiverProtocol, mockedReceiverTransport := MockedClient()
 
 		senderClients := make([]cloudevents.Client, c.OutputSenders)
 		for i := 0; i < c.OutputSenders; i++ {
-			senderClients[i], _, _ = MockedClient()
+			senderClients[i], _ = MockedClient()
 		}
 
-		mockedReceiverTransport.SetDelivery(dispatchReceiver(senderClients, c.OutputSenders))
+		//mockedReceiverTransport.SetDelivery(dispatchReceiver(senderClients, c.OutputSenders))
 
 		buffer := make([]byte, c.PayloadSize)
 		fillRandom(buffer, random)
@@ -203,8 +200,8 @@ func benchmarkClient(cases []e2e.BenchmarkCase, requestFactory func([]byte) *net
 			b.SetParallelism(c.Parallelism)
 			b.RunParallel(func(pb *testing.PB) {
 				for pb.Next() {
-					w := httptest.NewRecorder()
-					mockedReceiverProtocol.ServeHTTP(w, requestFactory(buffer))
+					//w := httptest.NewRecorder()
+					//mockedReceiverProtocol.ServeHTTP(w, requestFactory(buffer))
 				}
 			})
 		})
diff --git a/test/benchmark/http_request_to_kafka_sarama_encode/benchmark_test.go b/test/benchmark/http_request_to_kafka_sarama_encode/benchmark_test.go
index e3be5ad..97f0268 100644
--- a/test/benchmark/http_request_to_kafka_sarama_encode/benchmark_test.go
+++ b/test/benchmark/http_request_to_kafka_sarama_encode/benchmark_test.go
@@ -1,4 +1,4 @@
-package kafka_sarama_to_http_request_encode
+package http_request_to_kafka_sarama_encode
 
 import (
 	"context"
@@ -15,46 +15,27 @@ import (
 )
 
 var (
-	eventWithoutKey                 cloudevents.Event
-	eventWithKey                    cloudevents.Event
-	structuredHttpRequestWithoutKey *nethttp.Request
-	structuredHttpRequestWithKey    *nethttp.Request
-	binaryHttpRequestWithoutKey     *nethttp.Request
-	binaryHttpRequestWithKey        *nethttp.Request
+	event                 cloudevents.Event
+	structuredHttpRequest *nethttp.Request
+	binaryHttpRequest     *nethttp.Request
 
-	ctxSkipKey = kafka_sarama.WithSkipKeyExtension(context.TODO())
-	ctx        = context.TODO()
+	ctx = context.TODO()
 )
 
 func init() {
-	eventWithoutKey = test.FullEvent()
-	eventWithKey = test.FullEvent()
-	eventWithKey.SetExtension("key", "aaa")
+	event = test.FullEvent()
 
-	structuredHttpRequestWithoutKey, _ = nethttp.NewRequest("POST", "http://localhost", nil)
-	Err = http.WriteRequest(context.TODO(), (*binding.EventMessage)(&eventWithoutKey), structuredHttpRequestWithoutKey, binding.TransformerFactories{})
+	structuredHttpRequest, _ = nethttp.NewRequest("POST", "http://localhost", nil)
+	Err = http.WriteRequest(context.TODO(), (*binding.EventMessage)(&event), structuredHttpRequest, binding.TransformerFactories{})
 	if Err != nil {
 		panic(Err)
 	}
 
-	structuredHttpRequestWithKey, _ = nethttp.NewRequest("POST", "http://localhost", nil)
-	Err = http.WriteRequest(context.TODO(), (*binding.EventMessage)(&eventWithKey), structuredHttpRequestWithKey, binding.TransformerFactories{})
+	binaryHttpRequest, _ = nethttp.NewRequest("POST", "http://localhost", nil)
+	Err = http.WriteRequest(context.TODO(), (*binding.EventMessage)(&event), binaryHttpRequest, binding.TransformerFactories{})
 	if Err != nil {
 		panic(Err)
 	}
-
-	binaryHttpRequestWithoutKey, _ = nethttp.NewRequest("POST", "http://localhost", nil)
-	Err = http.WriteRequest(context.TODO(), (*binding.EventMessage)(&eventWithoutKey), binaryHttpRequestWithoutKey, binding.TransformerFactories{})
-	if Err != nil {
-		panic(Err)
-	}
-
-	binaryHttpRequestWithKey, _ = nethttp.NewRequest("POST", "http://localhost", nil)
-	Err = http.WriteRequest(context.TODO(), (*binding.EventMessage)(&eventWithKey), binaryHttpRequestWithKey, binding.TransformerFactories{})
-	if Err != nil {
-		panic(Err)
-	}
-
 }
 
 // Avoid DCE
@@ -65,14 +46,14 @@ var Err error
 func BenchmarkBaselineStructured(b *testing.B) {
 	for i := 0; i < b.N; i++ {
 		Req, _ = nethttp.NewRequest("POST", "http://localhost", nil)
-		Err = http.WriteRequest(context.TODO(), (*binding.EventMessage)(&eventWithKey), Req, binding.TransformerFactories{})
+		Err = http.WriteRequest(context.TODO(), (*binding.EventMessage)(&event), Req, binding.TransformerFactories{})
 	}
 }
 
 func BenchmarkStructured(b *testing.B) {
 	for i := 0; i < b.N; i++ {
 		tempReq, _ := nethttp.NewRequest("POST", "http://localhost", nil)
-		Err = http.WriteRequest(context.TODO(), (*binding.EventMessage)(&eventWithKey), tempReq, binding.TransformerFactories{})
+		Err = http.WriteRequest(context.TODO(), (*binding.EventMessage)(&event), tempReq, binding.TransformerFactories{})
 
 		M = http.NewMessageFromHttpRequest(tempReq)
 		Req, Err = nethttp.NewRequest("POST", "http://localhost", nil)
@@ -81,36 +62,17 @@ func BenchmarkStructured(b *testing.B) {
 	}
 }
 
-func BenchmarkBaselineStructuredSkipKey(b *testing.B) {
-	for i := 0; i < b.N; i++ {
-		Req, _ = nethttp.NewRequest("POST", "http://localhost", nil)
-		Err = http.WriteRequest(context.TODO(), (*binding.EventMessage)(&eventWithoutKey), Req, binding.TransformerFactories{})
-	}
-}
-
-func BenchmarkStructuredSkipKey(b *testing.B) {
-	for i := 0; i < b.N; i++ {
-		tempReq, _ := nethttp.NewRequest("POST", "http://localhost", nil)
-		Err = http.WriteRequest(context.TODO(), (*binding.EventMessage)(&eventWithoutKey), tempReq, binding.TransformerFactories{})
-
-		M = http.NewMessageFromHttpRequest(tempReq)
-		Req, Err = nethttp.NewRequest("POST", "http://localhost", nil)
-		producerMessage := &sarama.ProducerMessage{}
-		Err = kafka_sarama.WriteProducerMessage(ctxSkipKey, M, producerMessage, binding.TransformerFactories{})
-	}
-}
-
 func BenchmarkBaselineBinary(b *testing.B) {
 	for i := 0; i < b.N; i++ {
 		Req, _ = nethttp.NewRequest("POST", "http://localhost", nil)
-		Err = http.WriteRequest(context.TODO(), (*binding.EventMessage)(&eventWithKey), Req, binding.TransformerFactories{})
+		Err = http.WriteRequest(context.TODO(), (*binding.EventMessage)(&event), Req, binding.TransformerFactories{})
 	}
 }
 
 func BenchmarkBinary(b *testing.B) {
 	for i := 0; i < b.N; i++ {
 		tempReq, _ := nethttp.NewRequest("POST", "http://localhost", nil)
-		Err = http.WriteRequest(context.TODO(), (*binding.EventMessage)(&eventWithKey), tempReq, binding.TransformerFactories{})
+		Err = http.WriteRequest(context.TODO(), (*binding.EventMessage)(&event), tempReq, binding.TransformerFactories{})
 
 		M = http.NewMessageFromHttpRequest(tempReq)
 		Req, Err = nethttp.NewRequest("POST", "http://localhost", nil)
@@ -118,22 +80,3 @@ func BenchmarkBinary(b *testing.B) {
 		Err = kafka_sarama.WriteProducerMessage(ctx, M, producerMessage, binding.TransformerFactories{})
 	}
 }
-
-func BenchmarkBaselineBinarySkipKey(b *testing.B) {
-	for i := 0; i < b.N; i++ {
-		Req, _ = nethttp.NewRequest("POST", "http://localhost", nil)
-		Err = http.WriteRequest(context.TODO(), (*binding.EventMessage)(&eventWithoutKey), Req, binding.TransformerFactories{})
-	}
-}
-
-func BenchmarkBinarySkipKey(b *testing.B) {
-	for i := 0; i < b.N; i++ {
-		tempReq, _ := nethttp.NewRequest("POST", "http://localhost", nil)
-		Err = http.WriteRequest(context.TODO(), (*binding.EventMessage)(&eventWithoutKey), tempReq, binding.TransformerFactories{})
-
-		M = http.NewMessageFromHttpRequest(tempReq)
-		Req, Err = nethttp.NewRequest("POST", "http://localhost", nil)
-		producerMessage := &sarama.ProducerMessage{}
-		Err = kafka_sarama.WriteProducerMessage(ctxSkipKey, M, producerMessage, binding.TransformerFactories{})
-	}
-}
diff --git a/test/benchmark/kafka_sarama_to_http_request_encode/benchmark_test.go b/test/benchmark/kafka_sarama_to_http_request_encode/benchmark_test.go
index e461ed7..772cbb1 100644
--- a/test/benchmark/kafka_sarama_to_http_request_encode/benchmark_test.go
+++ b/test/benchmark/kafka_sarama_to_http_request_encode/benchmark_test.go
@@ -17,43 +17,20 @@ import (
 )
 
 var (
-	e                                   = test.FullEvent()
-	structuredConsumerMessageWithoutKey = &sarama.ConsumerMessage{
-		Value: test2.MustJSON(e),
+	event                     = test.FullEvent()
+	structuredConsumerMessage = &sarama.ConsumerMessage{
+		Value: test2.MustJSON(event),
 		Headers: []*sarama.RecordHeader{{
 			Key:   []byte("Content-Type"),
 			Value: []byte(cloudevents.ApplicationCloudEventsJSON),
 		}},
 	}
-	structuredConsumerMessageWithKey = &sarama.ConsumerMessage{
-		Key:   []byte("aaa"),
-		Value: test2.MustJSON(e),
-		Headers: []*sarama.RecordHeader{{
-			Key:   []byte("Content-Type"),
-			Value: []byte(cloudevents.ApplicationCloudEventsJSON),
-		}},
-	}
-	binaryConsumerMessageWithoutKey = &sarama.ConsumerMessage{
+	binaryConsumerMessage = &sarama.ConsumerMessage{
 		Value: []byte("hello world!"),
 		Headers: mustToSaramaConsumerHeaders(map[string]string{
-			"ce_type":            e.Type(),
-			"ce_source":          e.Source(),
-			"ce_id":              e.ID(),
-			"ce_time":            test.Timestamp.String(),
-			"ce_specversion":     "1.0",
-			"ce_dataschema":      test.Schema.String(),
-			"ce_datacontenttype": "text/json",
-			"ce_subject":         "topic",
-			"ce_exta":            "someext",
-		}),
-	}
-	binaryConsumerMessageWithKey = &sarama.ConsumerMessage{
-		Key:   []byte("akey"),
-		Value: []byte("hello world!"),
-		Headers: mustToSaramaConsumerHeaders(map[string]string{
-			"ce_type":            e.Type(),
-			"ce_source":          e.Source(),
-			"ce_id":              e.ID(),
+			"ce_type":            event.Type(),
+			"ce_source":          event.Source(),
+			"ce_id":              event.ID(),
 			"ce_time":            test.Timestamp.String(),
 			"ce_specversion":     "1.0",
 			"ce_dataschema":      test.Schema.String(),
@@ -79,33 +56,17 @@ var M binding.Message
 var Req *nethttp.Request
 var Err error
 
-func BenchmarkStructuredWithKey(b *testing.B) {
-	for i := 0; i < b.N; i++ {
-		M, Err = kafka_sarama.NewMessageFromConsumerMessage(structuredConsumerMessageWithKey)
-		Req, Err = nethttp.NewRequest("POST", "http://localhost", nil)
-		Err = http.WriteRequest(context.TODO(), M, Req, binding.TransformerFactories{})
-	}
-}
-
-func BenchmarkStructuredWithoutKey(b *testing.B) {
-	for i := 0; i < b.N; i++ {
-		M, Err = kafka_sarama.NewMessageFromConsumerMessage(structuredConsumerMessageWithoutKey)
-		Req, Err = nethttp.NewRequest("POST", "http://localhost", nil)
-		Err = http.WriteRequest(context.TODO(), M, Req, binding.TransformerFactories{})
-	}
-}
-
-func BenchmarkBinaryWithKey(b *testing.B) {
+func BenchmarkStructured(b *testing.B) {
 	for i := 0; i < b.N; i++ {
-		M, Err = kafka_sarama.NewMessageFromConsumerMessage(binaryConsumerMessageWithKey)
+		M = kafka_sarama.NewMessageFromConsumerMessage(structuredConsumerMessage)
 		Req, Err = nethttp.NewRequest("POST", "http://localhost", nil)
 		Err = http.WriteRequest(context.TODO(), M, Req, binding.TransformerFactories{})
 	}
 }
 
-func BenchmarkBinaryWithoutKey(b *testing.B) {
+func BenchmarkBinary(b *testing.B) {
 	for i := 0; i < b.N; i++ {
-		M, Err = kafka_sarama.NewMessageFromConsumerMessage(binaryConsumerMessageWithoutKey)
+		M = kafka_sarama.NewMessageFromConsumerMessage(binaryConsumerMessage)
 		Req, Err = nethttp.NewRequest("POST", "http://localhost", nil)
 		Err = http.WriteRequest(context.TODO(), M, Req, binding.TransformerFactories{})
 	}
diff --git a/test/integration/http/direct.go b/test/integration/http/direct.go
index 7864f80..80caed2 100644
--- a/test/integration/http/direct.go
+++ b/test/integration/http/direct.go
@@ -2,6 +2,7 @@ package http
 
 import (
 	"context"
+	"github.com/cloudevents/sdk-go/pkg/client"
 	"net/http/httptest"
 	"testing"
 	"time"
@@ -28,45 +29,26 @@ type DirectTapTest struct {
 
 type DirectTapTestCases map[string]DirectTapTest
 
-func ClientDirect(t *testing.T, tc DirectTapTest, opts ...interface{}) {
+func ClientDirect(t *testing.T, tc DirectTapTest, copts ...client.Option) {
 	tap := NewTap()
 	server := httptest.NewServer(tap)
 	defer server.Close()
 
-	topts := make([]cehttp.Option, 0)
-	popts := make([]cehttp.ProtocolOption, 0)
-	for _, v := range opts {
-		switch o := v.(type) {
-		case cehttp.Option:
-			topts = append(topts, o)
-		case cehttp.ProtocolOption:
-			popts = append(popts, o)
-		}
-	}
-
-	popts = append(popts, cloudevents.WithTarget(server.URL))
+	opts := make([]cehttp.Option, 0)
+	opts = append(opts, cloudevents.WithTarget(server.URL))
+	opts = append(opts, cloudevents.WithPort(0)) // random port
 
-	protocol, err := cloudevents.NewHTTPProtocol(popts...)
-	if err != nil {
-		t.Fatal(err)
-	}
-
-	if len(topts) == 0 {
-		topts = append(topts, cloudevents.WithEncoding(cloudevents.HTTPBinaryEncoding))
-	}
-	topts = append(topts, cloudevents.WithPort(0)) // random port
-	transport, err := cloudevents.NewHTTPTransport(protocol, topts...)
+	protocol, err := cloudevents.NewHTTP(opts...)
 	if err != nil {
 		t.Fatal(err)
 	}
 
 	tap.handler = protocol
 
-	ce, err := cloudevents.NewClient(
-		transport,
-		cloudevents.WithEventDefaulter(AlwaysThen(tc.now)),
-		cloudevents.WithoutTracePropagation(),
-	)
+	copts = append(copts, cloudevents.WithEventDefaulter(AlwaysThen(tc.now)))
+	copts = append(copts, cloudevents.WithoutTracePropagation())
+
+	ce, err := cloudevents.NewClient(protocol, copts...)
 	if err != nil {
 		t.Fatal(err)
 	}
diff --git a/test/integration/http/direct_v1_test.go b/test/integration/http/direct_v1_test.go
index 71d0f05..65dd1da 100644
--- a/test/integration/http/direct_v1_test.go
+++ b/test/integration/http/direct_v1_test.go
@@ -2,6 +2,7 @@ package http
 
 import (
 	"fmt"
+	"github.com/cloudevents/sdk-go/pkg/client"
 	"testing"
 	"time"
 
@@ -101,7 +102,7 @@ func TestSenderReceiver_structured_v1(t *testing.T) {
 
 	for n, tc := range testCases {
 		t.Run(n, func(t *testing.T) {
-			ClientDirect(t, tc, cloudevents.WithStructuredEncoding())
+			ClientDirect(t, tc, client.WithForceStructured())
 		})
 	}
 }
@@ -148,7 +149,7 @@ func TestSenderReceiver_data_base64_v1(t *testing.T) {
 
 	for n, tc := range testCases {
 		t.Run(n, func(t *testing.T) {
-			ClientDirect(t, tc, cloudevents.WithStructuredEncoding())
+			ClientDirect(t, tc, client.WithForceStructured())
 		})
 	}
 }
diff --git a/test/integration/http/loopback.go b/test/integration/http/loopback.go
index e2c12f7..d9cc2fe 100644
--- a/test/integration/http/loopback.go
+++ b/test/integration/http/loopback.go
@@ -41,46 +41,26 @@ type TapTest struct {
 
 type TapTestCases map[string]TapTest
 
-func ClientLoopback(t *testing.T, tc TapTest, opts ...interface{}) {
+func ClientLoopback(t *testing.T, tc TapTest, copts ...client.Option) {
 	tap := NewTap()
 	server := httptest.NewServer(tap)
 	defer server.Close()
 
-	topts := make([]cehttp.Option, 0)
-	popts := make([]cehttp.ProtocolOption, 0)
-	for _, v := range opts {
-		switch o := v.(type) {
-		case cehttp.Option:
-			topts = append(topts, o)
-		case cehttp.ProtocolOption:
-			popts = append(popts, o)
-		}
-	}
-
-	popts = append(popts, cloudevents.WithTarget(server.URL))
-
-	protocol, err := cloudevents.NewHTTPProtocol(popts...)
-	if err != nil {
-		t.Fatal(err)
-	}
+	opts := make([]cehttp.Option, 0)
+	opts = append(opts, cloudevents.WithTarget(server.URL))
+	opts = append(opts, cloudevents.WithPort(0)) // random port
 
-	if len(topts) == 0 {
-		topts = append(topts, cloudevents.WithEncoding(cloudevents.HTTPBinaryEncoding))
-	}
-	topts = append(topts, cloudevents.WithPort(0)) // random port
-
-	transport, err := cloudevents.NewHTTPTransport(protocol, topts...)
+	protocol, err := cloudevents.NewHTTP(opts...)
 	if err != nil {
 		t.Fatal(err)
 	}
 
 	tap.handler = protocol
 
-	ce, err := cloudevents.NewClient(
-		transport,
-		cloudevents.WithEventDefaulter(AlwaysThen(tc.now)),
-		cloudevents.WithoutTracePropagation(),
-	)
+	copts = append(copts, cloudevents.WithEventDefaulter(AlwaysThen(tc.now)))
+	copts = append(copts, cloudevents.WithoutTracePropagation())
+
+	ce, err := cloudevents.NewClient(protocol, copts...)
 	if err != nil {
 		t.Fatal(err)
 	}
diff --git a/test/integration/http/loopback_setters_test.go b/test/integration/http/loopback_setters_test.go
index a15237f..3b90dba 100644
--- a/test/integration/http/loopback_setters_test.go
+++ b/test/integration/http/loopback_setters_test.go
@@ -2,6 +2,7 @@ package http
 
 import (
 	"fmt"
+	"github.com/cloudevents/sdk-go/pkg/client"
 	"testing"
 	"time"
 
@@ -377,7 +378,7 @@ func TestClientLoopback_setters_structured_json(t *testing.T) {
 				testcase.asSent.ContentLength = int64(len(testcase.asSent.Body))
 				testcase.asRecv.ContentLength = int64(len(testcase.asRecv.Body))
 
-				ClientLoopback(t, testcase, cloudevents.WithStructuredEncoding())
+				ClientLoopback(t, testcase, client.WithForceStructured())
 			})
 		}
 	}
@@ -471,7 +472,7 @@ func TestClientLoopback_setters_structured_json_base64(t *testing.T) {
 				testcase.asSent.ContentLength = int64(len(testcase.asSent.Body))
 				testcase.asRecv.ContentLength = int64(len(testcase.asRecv.Body))
 
-				ClientLoopback(t, testcase, cloudevents.WithStructuredEncoding())
+				ClientLoopback(t, testcase, client.WithForceStructured())
 			})
 		}
 	}
diff --git a/test/integration/http/loopback_v03_test.go b/test/integration/http/loopback_v03_test.go
index 4ef164e..c5b6171 100644
--- a/test/integration/http/loopback_v03_test.go
+++ b/test/integration/http/loopback_v03_test.go
@@ -2,6 +2,7 @@ package http
 
 import (
 	"fmt"
+	"github.com/cloudevents/sdk-go/pkg/client"
 	"testing"
 	"time"
 
@@ -223,7 +224,7 @@ func TestClientLoopback_structured_base64_v03tov03(t *testing.T) {
 			tc.asSent.ContentLength = int64(len(tc.asSent.Body))
 			tc.asRecv.ContentLength = int64(len(tc.asRecv.Body))
 
-			ClientLoopback(t, tc, cloudevents.WithStructuredEncoding())
+			ClientLoopback(t, tc, client.WithForceStructured())
 		})
 	}
 }
diff --git a/test/integration/http/loopback_v1_test.go b/test/integration/http/loopback_v1_test.go
index d6b4c25..ac2125d 100644
--- a/test/integration/http/loopback_v1_test.go
+++ b/test/integration/http/loopback_v1_test.go
@@ -3,6 +3,7 @@ package http
 import (
 	"encoding/json"
 	"fmt"
+	"github.com/cloudevents/sdk-go/pkg/client"
 	"testing"
 	"time"
 
@@ -222,7 +223,7 @@ func TestClientLoopback_structured_base64_v03tov1(t *testing.T) {
 			tc.asSent.ContentLength = int64(len(tc.asSent.Body))
 			tc.asRecv.ContentLength = int64(len(tc.asRecv.Body))
 
-			ClientLoopback(t, tc, cloudevents.WithStructuredEncoding())
+			ClientLoopback(t, tc, client.WithForceStructured())
 		})
 	}
 }
@@ -286,7 +287,7 @@ func TestClientLoopback_structured_base64_v1tov1(t *testing.T) {
 			tc.asSent.ContentLength = int64(len(tc.asSent.Body))
 			tc.asRecv.ContentLength = int64(len(tc.asRecv.Body))
 
-			ClientLoopback(t, tc, cloudevents.WithStructuredEncoding())
+			ClientLoopback(t, tc, client.WithForceStructured())
 		})
 	}
 }
diff --git a/test/integration/http/middleware.go b/test/integration/http/middleware.go
index 4744e3b..dcf81c0 100644
--- a/test/integration/http/middleware.go
+++ b/test/integration/http/middleware.go
@@ -2,54 +2,36 @@ package http
 
 import (
 	"context"
+	"github.com/cloudevents/sdk-go/pkg/client"
 	"net/http/httptest"
 	"testing"
 
 	"github.com/google/uuid"
 
 	cloudevents "github.com/cloudevents/sdk-go"
-	"github.com/cloudevents/sdk-go/pkg/transport/http"
+	cehttp "github.com/cloudevents/sdk-go/pkg/transport/http"
 )
 
-func ClientMiddleware(t *testing.T, tc TapTest, opts ...interface{}) {
+func ClientMiddleware(t *testing.T, tc TapTest, copts ...client.Option) {
 	tap := NewTap()
 	server := httptest.NewServer(tap)
 	defer server.Close()
 
-	topts := make([]http.Option, 0)
-	popts := make([]http.ProtocolOption, 0)
-	for _, v := range opts {
-		switch o := v.(type) {
-		case http.Option:
-			topts = append(topts, o)
-		case http.ProtocolOption:
-			popts = append(popts, o)
-		}
-	}
-
-	popts = append(popts, cloudevents.WithTarget(server.URL))
+	opts := make([]cehttp.Option, 0)
+	opts = append(opts, cloudevents.WithTarget(server.URL))
+	opts = append(opts, cloudevents.WithPort(0))
 
-	protocol, err := cloudevents.NewHTTPProtocol(popts...)
-	if err != nil {
-		t.Fatal(err)
-	}
-
-	if len(topts) == 0 {
-		topts = append(topts, cloudevents.WithEncoding(cloudevents.HTTPBinaryEncoding))
-	}
-	topts = append(topts, cloudevents.WithPort(0)) // random port
-	transport, err := cloudevents.NewHTTPTransport(protocol, topts...)
+	protocol, err := cloudevents.NewHTTP(opts...)
 	if err != nil {
 		t.Fatal(err)
 	}
 
 	tap.handler = protocol
 
-	ce, err := cloudevents.NewClient(
-		transport,
-		cloudevents.WithEventDefaulter(AlwaysThen(tc.now)),
-		cloudevents.WithoutTracePropagation(),
-	)
+	copts = append(copts, cloudevents.WithEventDefaulter(AlwaysThen(tc.now)))
+	copts = append(copts, cloudevents.WithoutTracePropagation())
+
+	ce, err := cloudevents.NewClient(protocol, copts...)
 	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 ad22445..d46289b 100644
--- a/test/integration/http_binding/http_test.go
+++ b/test/integration/http_binding/http_test.go
@@ -88,8 +88,8 @@ func TestSendEventReceiveBinary(t *testing.T) {
 	})
 }
 
-func testSenderReceiver(t testing.TB, options ...http.ProtocolOption) (func(), bindings.Sender, bindings.Receiver) {
-	p, err := http.NewProtocol(options...)
+func testSenderReceiver(t testing.TB, options ...http.Option) (func(), bindings.Sender, bindings.Receiver) {
+	p, err := http.New(options...)
 	require.NoError(t, err)
 	srv := httptest.NewServer(p)
 	u, err := url.Parse(srv.URL)
diff --git a/test/integration/kafka_sarama_binding/kafka_test.go b/test/integration/kafka_sarama_binding/kafka_test.go
index 0cc9882..7ba93b8 100644
--- a/test/integration/kafka_sarama_binding/kafka_test.go
+++ b/test/integration/kafka_sarama_binding/kafka_test.go
@@ -23,23 +23,7 @@ const (
 	TEST_GROUP_ID = "test_group_id"
 )
 
-func TestSendStructuredMessageToStructuredWithKey(t *testing.T) {
-	close, s, r := testSenderReceiver(t)
-	defer close()
-	EachEvent(t, Events(), func(t *testing.T, eventIn event.Event) {
-		eventIn = ExToStr(t, eventIn)
-		require.NoError(t, eventIn.Context.SetExtension("key", "aaa"))
-
-		in := MustCreateMockStructuredMessage(eventIn)
-		test.SendReceive(t, binding.WithPreferredEventEncoding(context.TODO(), binding.EncodingStructured), in, s, r, func(out binding.Message) {
-			eventOut := MustToEvent(t, context.Background(), out)
-			assert.Equal(t, binding.EncodingEvent, out.ReadEncoding())
-			AssertEventEquals(t, eventIn, ExToStr(t, eventOut))
-		})
-	})
-}
-
-func TestSendStructuredMessageToStructuredWithoutKey(t *testing.T) {
+func TestSendStructuredMessageToStructured(t *testing.T) {
 	close, s, r := testSenderReceiver(t)
 	defer close()
 	EachEvent(t, Events(), func(t *testing.T, eventIn event.Event) {
@@ -54,23 +38,7 @@ func TestSendStructuredMessageToStructuredWithoutKey(t *testing.T) {
 	})
 }
 
-func TestSendBinaryMessageToBinaryWithKey(t *testing.T) {
-	close, s, r := testSenderReceiver(t)
-	defer close()
-	EachEvent(t, Events(), func(t *testing.T, eventIn event.Event) {
-		eventIn = ExToStr(t, eventIn)
-		require.NoError(t, eventIn.Context.SetExtension("key", "aaa"))
-
-		in := MustCreateMockBinaryMessage(eventIn)
-		test.SendReceive(t, binding.WithPreferredEventEncoding(context.TODO(), binding.EncodingBinary), in, s, r, func(out binding.Message) {
-			eventOut := MustToEvent(t, context.Background(), out)
-			assert.Equal(t, binding.EncodingBinary, out.ReadEncoding())
-			AssertEventEquals(t, eventIn, ExToStr(t, eventOut))
-		})
-	})
-}
-
-func TestSendBinaryMessageToBinaryWithoutKey(t *testing.T) {
+func TestSendBinaryMessageToBinary(t *testing.T) {
 	close, s, r := testSenderReceiver(t)
 	defer close()
 	EachEvent(t, Events(), func(t *testing.T, eventIn event.Event) {