Repository: cloudevents/sdk-go
Pull request #364 (2020-03-10 23:13:58)

Update HTTP to combine Sender and Receiver into a single impl.
This combines HTTP Sender and Requester, Receiver into a new thing called Protocol that leverages Bindings.

This makes a new Transport object that handles all the http server details. This is to enable Transport for legacy integration that will be removed before the real 2.0 cut.


Patch:
diff --git a/alias.go b/alias.go
index 07a68a7..14194ac 100644
--- a/alias.go
+++ b/alias.go
@@ -35,11 +35,12 @@ type Timestamp = types.Timestamp
 type URLRef = types.URLRef
 type URIRef = types.URIRef
 
-// HTTP Transport
+// HTTP Protocol
 
 type HTTPOption http.Option
+type HTTPProtocolOption http.ProtocolOption
 
-type HTTPTransport = http.Transport
+type HTTPTransport = http.Protocol
 type HTTPEncoding = http.Encoding
 
 const (
@@ -104,11 +105,12 @@ var (
 	ParseURIRef    = types.ParseURIRef
 	ParseURI       = types.ParseURI
 
-	// HTTP Transport
+	// HTTP Protocol
 
 	NewHTTPTransport = http.New
+	NewHTTPProtocol  = http.NewProtocol
 
-	// HTTP Transport Options
+	// HTTP Protocol Options
 
 	WithTarget             = http.WithTarget
 	WitHHeader             = http.WithHeader
diff --git a/cmd/longpollserver/main.go b/cmd/longpollserver/main.go
deleted file mode 100644
index b7ab2a4..0000000
--- a/cmd/longpollserver/main.go
+++ /dev/null
@@ -1,60 +0,0 @@
-package main
-
-import (
-	"context"
-	"fmt"
-	"log"
-	"math/rand"
-	"net/http"
-	"time"
-)
-
-// This is a simple example of a long poll server.
-
-var timeout = time.Second * 5 //30
-var maxOpMS = 8000            //60 * 1000 // 60 seconds -> 50% chance of StatusNotModified.
-
-func init() {
-	rand.Seed(time.Now().Unix())
-}
-
-func operation(ctx context.Context, ch chan<- string) {
-	wait := time.Millisecond * time.Duration(rand.Intn(maxOpMS))
-	fmt.Printf(" -> Wait %v\n", wait)
-	select {
-	case <-time.After(wait):
-		ch <- fmt.Sprintf(`{"waited":"%v"}`, wait)
-	case <-ctx.Done():
-		fmt.Println(" -> Op canceled")
-		return
-	}
-	fmt.Println(" -> Op done")
-}
-
-func handler(w http.ResponseWriter, req *http.Request) {
-	fmt.Println("Request")
-	ch := make(chan string) // TODO: send real struct.
-	go operation(req.Context(), ch)
-
-	select {
-	case result := <-ch:
-		fmt.Println("Response")
-		_, _ = w.Write([]byte(result))
-	case <-time.After(timeout):
-		fmt.Println("Timeout")
-		w.WriteHeader(http.StatusNotModified)
-	case <-req.Context().Done():
-		fmt.Println("Client has disconnected")
-	}
-	close(ch)
-	fmt.Println("Done")
-}
-
-func main() {
-	addr := ":8181"
-	http.HandleFunc("/", handler)
-	server := &http.Server{Addr: addr, Handler: nil}
-
-	log.Printf("Starting on %s\n", addr)
-	log.Print(server.ListenAndServe())
-}
diff --git a/cmd/samples/complex/httptonats/main.go b/cmd/samples/complex/httptonats/main.go
index 85ac076..fef738a 100644
--- a/cmd/samples/complex/httptonats/main.go
+++ b/cmd/samples/complex/httptonats/main.go
@@ -68,7 +68,7 @@ func _main(args []string, env envConfig) int {
 	if err != nil {
 		log.Fatalf("failed to create nats transport, %s", err.Error())
 	}
-	nc, err := client.New(nt)
+	nc, err := client.New(nt.Transport())
 	if err != nil {
 		log.Printf("failed to create client, %v", err)
 		return 1
@@ -76,7 +76,12 @@ func _main(args []string, env envConfig) int {
 
 	r := &Receiver{Client: nc}
 
-	t, err := cloudeventshttp.New(
+	p, err := cloudeventshttp.NewProtocol()
+	if err != nil {
+		log.Fatalf("failed to create protocol: %s", err.Error())
+	}
+
+	t, err := cloudeventshttp.New(p,
 		cloudeventshttp.WithPort(env.Port),
 	)
 	if err != nil {
diff --git a/cmd/samples/complex/sender/main.go b/cmd/samples/complex/sender/main.go
index 34d5654..6468812 100644
--- a/cmd/samples/complex/sender/main.go
+++ b/cmd/samples/complex/sender/main.go
@@ -96,12 +96,17 @@ func _main(args []string, env envConfig) int {
 		// HTTP
 		for _, encoding := range []cloudeventshttp.Encoding{cloudeventshttp.Default, cloudeventshttp.Binary, cloudeventshttp.Structured} {
 
-			t, err := cloudeventshttp.New(
-				cloudeventshttp.WithTarget(env.HTTPTarget),
+			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 client, %v", err)
+				log.Printf("failed to create transport, %v", err)
 				return 1
 			}
 
@@ -118,27 +123,23 @@ func _main(args []string, env envConfig) int {
 		}
 
 		// NATS
-		for _, encoding := range []cloudeventsnats.Encoding{cloudeventsnats.Default, cloudeventsnats.StructuredV03} {
-
-			t, err := cloudeventsnats.New(
-				env.NATSServer,
-				env.Subject,
-				cloudeventsnats.WithEncoding(encoding),
-			)
-			if err != nil {
-				log.Printf("failed to create client, %v", err)
-				return 1
-			}
-			if err := doDemo(
-				t,
-				"com.cloudevents.sample.nats.sent",
-				fmt.Sprintf("Hello, %s using %s!", encoding, contentType),
-				contentType,
-				*source,
-			); err != nil {
-				log.Printf("failed to do nats demo: %v, %s", err, contentType)
-				return 1
-			}
+		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
 		}
 	}
 
@@ -146,7 +147,6 @@ func _main(args []string, env envConfig) int {
 }
 
 func doDemo(t transport.Transport, eventType, message, contentType string, source url.URL) error {
-
 	c, err := client.New(t,
 		client.WithUUIDs(),
 		client.WithTimeNow(),
diff --git a/cmd/samples/http/receiver/main.go b/cmd/samples/http/receiver/main.go
index e6df689..5ad29a4 100644
--- a/cmd/samples/http/receiver/main.go
+++ b/cmd/samples/http/receiver/main.go
@@ -45,7 +45,12 @@ func gotEvent(ctx context.Context, event cloudevents.Event) error {
 func _main(args []string, env envConfig) int {
 	ctx := context.Background()
 
-	t, err := cloudevents.NewHTTPTransport(
+	p, err := cloudevents.NewHTTPProtocol()
+	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),
 	)
diff --git a/cmd/samples/http/requester-with-custom-client/main.go b/cmd/samples/http/requester-with-custom-client/main.go
index f6777c5..75e1ec0 100644
--- a/cmd/samples/http/requester-with-custom-client/main.go
+++ b/cmd/samples/http/requester-with-custom-client/main.go
@@ -76,8 +76,14 @@ 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))
+			if err != nil {
+				log.Printf("failed to create protocol, %v", err)
+				return 1
+			}
+
 			t, err := cloudevents.NewHTTPTransport(
-				cloudevents.WithTarget(env.Target),
+				p,
 				cloudevents.WithEncoding(encoding),
 				// Use custom transport
 				cloudevents.WithHTTPTransport(&http.Transport{TLSClientConfig: tlsConfig}),
diff --git a/cmd/samples/http/requester/main.go b/cmd/samples/http/requester/main.go
index 1bfd5b6..1b89722 100644
--- a/cmd/samples/http/requester/main.go
+++ b/cmd/samples/http/requester/main.go
@@ -48,8 +48,13 @@ func _main(args []string, env envConfig) int {
 	for _, contentType := range []string{"application/json", "application/xml"} {
 		for _, encoding := range []cloudevents.HTTPEncoding{cloudevents.HTTPBinaryEncoding, cloudevents.HTTPStructuredEncoding} {
 
-			t, err := cloudevents.NewHTTPTransport(
-				cloudevents.WithTarget(env.Target),
+			p, err := cloudevents.NewHTTPProtocol(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 {
diff --git a/cmd/samples/http/responder/main.go b/cmd/samples/http/responder/main.go
index bd92f47..4f285bd 100644
--- a/cmd/samples/http/responder/main.go
+++ b/cmd/samples/http/responder/main.go
@@ -60,7 +60,12 @@ func gotEvent(ctx context.Context, event cloudevents.Event, resp *cloudevents.Ev
 func _main(args []string, env envConfig) int {
 	ctx := context.Background()
 
-	t, err := cloudevents.NewHTTPTransport(
+	p, err := cloudevents.NewHTTPProtocol()
+	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),
 	)
diff --git a/cmd/samples/http/sleepy/main.go b/cmd/samples/http/sleepy/main.go
index d4bd819..972a224 100644
--- a/cmd/samples/http/sleepy/main.go
+++ b/cmd/samples/http/sleepy/main.go
@@ -43,7 +43,12 @@ func gotEvent(event cloudevents.Event) error {
 }
 
 func _main(args []string, env envConfig) int {
-	t, err := cloudevents.NewHTTPTransport(
+	p, err := cloudevents.NewHTTPProtocol()
+	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),
 	)
diff --git a/cmd/samples/httpb/receiver/main.go b/cmd/samples/httpb/receiver/main.go
index 9585042..fc49133 100644
--- a/cmd/samples/httpb/receiver/main.go
+++ b/cmd/samples/httpb/receiver/main.go
@@ -12,8 +12,12 @@ import (
 
 func main() {
 	ctx := context.Background()
+	p, err := cloudevents.NewHTTPProtocol()
+	if err != nil {
+		log.Fatalf("failed to create protocol: %s", err.Error())
+	}
 
-	t, err := http.New()
+	t, err := http.New(p)
 	if err != nil {
 		log.Fatalf("failed to create transport, %v", err)
 	}
diff --git a/cmd/samples/httpb/requester/main.go b/cmd/samples/httpb/requester/main.go
index 793a99a..4238591 100644
--- a/cmd/samples/httpb/requester/main.go
+++ b/cmd/samples/httpb/requester/main.go
@@ -12,7 +12,13 @@ import (
 
 func main() {
 	ctx := cloudevents.ContextWithTarget(context.Background(), "http://localhost:8080/")
-	t, err := http.New()
+
+	p, err := cloudevents.NewHTTPProtocol()
+	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)
 	}
diff --git a/cmd/samples/httpb/responder/main.go b/cmd/samples/httpb/responder/main.go
index 50e4a15..69be799 100644
--- a/cmd/samples/httpb/responder/main.go
+++ b/cmd/samples/httpb/responder/main.go
@@ -26,7 +26,12 @@ func main() {
 	}
 	ctx := context.Background()
 
-	t, err := http.New()
+	p, err := cloudevents.NewHTTPProtocol()
+	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)
 	}
diff --git a/cmd/samples/httpb/sender/main.go b/cmd/samples/httpb/sender/main.go
index 15c6440..47ff609 100644
--- a/cmd/samples/httpb/sender/main.go
+++ b/cmd/samples/httpb/sender/main.go
@@ -12,7 +12,13 @@ import (
 
 func main() {
 	ctx := cloudevents.ContextWithTarget(context.Background(), "http://localhost:8080/")
-	t, err := http.New()
+
+	p, err := cloudevents.NewHTTPProtocol()
+	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)
 	}
diff --git a/cmd/samples/nats/receiver/main.go b/cmd/samples/nats/receiver/main.go
index f7ee5a9..d4518c0 100644
--- a/cmd/samples/nats/receiver/main.go
+++ b/cmd/samples/nats/receiver/main.go
@@ -32,7 +32,7 @@ func main() {
 	if err != nil {
 		log.Fatalf("failed to create nats transport, %s", err.Error())
 	}
-	c, err := client.New(t)
+	c, err := client.New(t.Transport())
 	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 74e3255..604df3b 100644
--- a/cmd/samples/nats/sender/main.go
+++ b/cmd/samples/nats/sender/main.go
@@ -77,7 +77,7 @@ func _main(args []string, env envConfig) int {
 			log.Printf("failed to create nats transport, %s", err.Error())
 			return 1
 		}
-		c, err := client.New(t)
+		c, err := client.New(t.Transport())
 		if err != nil {
 			log.Printf("failed to create client, %s", err.Error())
 			return 1
diff --git a/pkg/client/client.go b/pkg/client/client.go
index 3635bad..2fe89a2 100644
--- a/pkg/client/client.go
+++ b/pkg/client/client.go
@@ -59,14 +59,18 @@ func New(t transport.Transport, opts ...Option) (Client, error) {
 }
 
 // NewDefault provides the good defaults for the common case using an HTTP
-// Transport client. The http transport has had WithBinaryEncoding 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, WithUUIDs and WithDataContentType("application/json")
 // 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) {
-	t, err := http.New(http.WithEncoding(http.Binary))
+	p, err := http.NewProtocol()
+	if err != nil {
+		return nil, err
+	}
+	t, err := http.New(p, http.WithEncoding(http.Binary))
 	if err != nil {
 		return nil, err
 	}
@@ -89,7 +93,7 @@ type ceClient struct {
 	disableTracePropagation bool
 }
 
-// Send transmits the provided event on a preconfigured Transport. Send returns
+// 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 {
@@ -140,7 +144,7 @@ func (c *ceClient) obsSend(ctx context.Context, event event.Event) error {
 	return c.transport.Send(ctx, event)
 }
 
-// Request transmits the provided event on a preconfigured Transport. Request
+// 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) {
diff --git a/pkg/transport/http/encoding.go b/pkg/transport/http/encoding.go
index adc1434..b91d21f 100644
--- a/pkg/transport/http/encoding.go
+++ b/pkg/transport/http/encoding.go
@@ -3,8 +3,6 @@ package http
 // Encoding to use for HTTP transport.
 type Encoding int32
 
-// TODO: there is some legacy stuff in here still, clean it up.
-
 const (
 	// Default is binary
 	Default Encoding = iota
diff --git a/pkg/transport/http/request_encoder.go b/pkg/transport/http/message_write_request.go
similarity index 100%
rename from pkg/transport/http/request_encoder.go
rename to pkg/transport/http/message_write_request.go
diff --git a/pkg/transport/http/response_encoder.go b/pkg/transport/http/message_write_response.go
similarity index 53%
rename from pkg/transport/http/response_encoder.go
rename to pkg/transport/http/message_write_response.go
index 0f52db2..3abf77d 100644
--- a/pkg/transport/http/response_encoder.go
+++ b/pkg/transport/http/message_write_response.go
@@ -14,9 +14,9 @@ import (
 
 // Fill the provided httpResponse with the message m.
 // Using context you can tweak the encoding processing (more details on binding.Write documentation).
-func EncodeHttpResponse(ctx context.Context, m binding.Message, httpResponse *http.Response, transformers binding.TransformerFactories) error {
-	structuredWriter := (*httpResponseEncoder)(httpResponse)
-	binaryWriter := (*httpResponseEncoder)(httpResponse)
+func WriteHttpResponse(ctx context.Context, m binding.Message, httpResponse *http.Response, transformers binding.TransformerFactories) error {
+	structuredWriter := (*httpResponseWriter)(httpResponse)
+	binaryWriter := (*httpResponseWriter)(httpResponse)
 
 	_, err := binding.Write(
 		ctx,
@@ -28,28 +28,28 @@ func EncodeHttpResponse(ctx context.Context, m binding.Message, httpResponse *ht
 	return err
 }
 
-type httpResponseEncoder http.Response
+type httpResponseWriter http.Response
 
-func (b *httpResponseEncoder) SetStructuredEvent(ctx context.Context, format format.Format, event io.Reader) error {
+func (b *httpResponseWriter) SetStructuredEvent(ctx context.Context, format format.Format, event io.Reader) error {
 	b.Header.Set(ContentType, format.MediaType())
 	b.Body = ioutil.NopCloser(event)
 	return nil
 }
 
-func (b *httpResponseEncoder) Start(ctx context.Context) error {
+func (b *httpResponseWriter) Start(ctx context.Context) error {
 	return nil
 }
 
-func (b *httpResponseEncoder) End() error {
+func (b *httpResponseWriter) End() error {
 	return nil
 }
 
-func (b *httpResponseEncoder) SetData(reader io.Reader) error {
+func (b *httpResponseWriter) SetData(reader io.Reader) error {
 	b.Body = ioutil.NopCloser(reader)
 	return nil
 }
 
-func (b *httpResponseEncoder) SetAttribute(attribute spec.Attribute, value interface{}) error {
+func (b *httpResponseWriter) SetAttribute(attribute spec.Attribute, value interface{}) error {
 	// Http headers, everything is a string!
 	s, err := types.Format(value)
 	if err != nil {
@@ -64,7 +64,7 @@ func (b *httpResponseEncoder) SetAttribute(attribute spec.Attribute, value inter
 	return nil
 }
 
-func (b *httpResponseEncoder) SetExtension(name string, value interface{}) error {
+func (b *httpResponseWriter) SetExtension(name string, value interface{}) error {
 	// Http headers, everything is a string!
 	s, err := types.Format(value)
 	if err != nil {
@@ -74,5 +74,5 @@ func (b *httpResponseEncoder) SetExtension(name string, value interface{}) error
 	return nil
 }
 
-var _ binding.StructuredWriter = (*httpResponseEncoder)(nil) // Test it conforms to the interface
-var _ binding.BinaryWriter = (*httpResponseEncoder)(nil)     // Test it conforms to the interface
+var _ binding.StructuredWriter = (*httpResponseWriter)(nil) // Test it conforms to the interface
+var _ binding.BinaryWriter = (*httpResponseWriter)(nil)     // Test it conforms to the interface
diff --git a/pkg/transport/http/responsewriter_encoder.go b/pkg/transport/http/message_write_responsewriter.go
similarity index 85%
rename from pkg/transport/http/responsewriter_encoder.go
rename to pkg/transport/http/message_write_responsewriter.go
index e4657fd..a14bf7c 100644
--- a/pkg/transport/http/responsewriter_encoder.go
+++ b/pkg/transport/http/message_write_responsewriter.go
@@ -15,7 +15,7 @@ import (
 
 // Write out to the the provided httpResponseWriter with the message m.
 // Using context you can tweak the encoding processing (more details on binding.Write documentation).
-func EncodeHttpResponseWriter(ctx context.Context, m binding.Message, rw http.ResponseWriter, transformers binding.TransformerFactories) error {
+func WriteHttpResponseWriter(ctx context.Context, m binding.Message, rw http.ResponseWriter, transformers binding.TransformerFactories) error {
 	structuredWriter := &httpResponseWriterEncoder{rw: rw}
 	binaryWriter := &httpResponseWriterEncoder{rw: rw}
 
@@ -82,5 +82,5 @@ func (b *httpResponseWriterEncoder) SetExtension(name string, value interface{})
 	return nil
 }
 
-var _ binding.StructuredWriter = (*httpResponseEncoder)(nil) // Test it conforms to the interface
-var _ binding.BinaryWriter = (*httpResponseEncoder)(nil)     // Test it conforms to the interface
+var _ binding.StructuredWriter = (*httpResponseWriter)(nil) // Test it conforms to the interface
+var _ binding.BinaryWriter = (*httpResponseWriter)(nil)     // Test it conforms to the interface
diff --git a/pkg/transport/http/option.go b/pkg/transport/http/option.go
deleted file mode 100644
index 5c33a21..0000000
--- a/pkg/transport/http/option.go
+++ /dev/null
@@ -1,13 +0,0 @@
-package http
-
-import "github.com/cloudevents/sdk-go/pkg/binding"
-
-// http.Sender options
-type SenderOptionFunc func(sender *Sender)
-
-// Add a transformer, which Sender uses while encoding a binding.Message to a nethttp.Request
-func WithTransformer(transformer binding.TransformerFactory) SenderOptionFunc {
-	return func(sender *Sender) {
-		sender.transformers = append(sender.transformers, transformer)
-	}
-}
diff --git a/pkg/transport/http/options.go b/pkg/transport/http/options.go
index 49ba34e..b8e50ff 100644
--- a/pkg/transport/http/options.go
+++ b/pkg/transport/http/options.go
@@ -9,15 +9,18 @@ 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
 
 // WithTarget sets the outbound recipient of cloudevents when using an HTTP
 // request.
-func WithTarget(targetUrl string) Option {
-	return func(t *Transport) error {
-		if t == nil {
-			return fmt.Errorf("http target option can not set nil transport")
+func WithTarget(targetUrl string) ProtocolOption {
+	return func(p *Protocol) error {
+		if p == nil {
+			return fmt.Errorf("http protocol option can not set nil protocol")
 		}
 		targetUrl = strings.TrimSpace(targetUrl)
 		if targetUrl != "" {
@@ -28,15 +31,14 @@ func WithTarget(targetUrl string) Option {
 				return fmt.Errorf("http target option failed to parse target url: %s", err.Error())
 			}
 
-			t.Target = target
-			// TODO: it might make sense to go back to using an http request that is pre-built to allow customization.
-			//
-			//if t.Req == nil {
-			//	t.Req = &nethttp.Request{
-			//		Method: nethttp.MethodPost,
-			//	}
-			//}
-			//t.Req.URL = target
+			p.Target = target
+
+			if p.RequestTemplate == nil {
+				p.RequestTemplate = &nethttp.Request{
+					Method: nethttp.MethodPost,
+				}
+			}
+			p.RequestTemplate.URL = target
 
 			return nil
 		}
@@ -46,20 +48,20 @@ func WithTarget(targetUrl string) Option {
 
 // WithHeader sets an additional default outbound header for all cloudevents
 // when using an HTTP request.
-func WithHeader(key, value string) Option {
-	return func(t *Transport) error {
-		if t == nil {
+func WithHeader(key, value string) ProtocolOption {
+	return func(p *Protocol) error {
+		if p == nil {
 			return fmt.Errorf("http header option can not set nil transport")
 		}
 		key = strings.TrimSpace(key)
 		if key != "" {
-			if t.RequestTemplate == nil {
-				t.RequestTemplate = &nethttp.Request{}
+			if p.RequestTemplate == nil {
+				p.RequestTemplate = &nethttp.Request{}
 			}
-			if t.RequestTemplate.Header == nil {
-				t.RequestTemplate.Header = nethttp.Header{}
+			if p.RequestTemplate.Header == nil {
+				p.RequestTemplate.Header = nethttp.Header{}
 			}
-			t.RequestTemplate.Header.Add(key, value)
+			p.RequestTemplate.Header.Add(key, value)
 			return nil
 		}
 		return fmt.Errorf("http header option was empty string")
diff --git a/pkg/transport/http/protocol.go b/pkg/transport/http/protocol.go
new file mode 100644
index 0000000..abd5944
--- /dev/null
+++ b/pkg/transport/http/protocol.go
@@ -0,0 +1,170 @@
+package http
+
+import (
+	"context"
+	"fmt"
+	"io"
+	"net/http"
+	"net/url"
+	"time"
+
+	"github.com/cloudevents/sdk-go/pkg/binding"
+	cecontext "github.com/cloudevents/sdk-go/pkg/context"
+)
+
+const (
+	// DefaultShutdownTimeout defines the default timeout given to the http.Server when calling Shutdown.
+	DefaultShutdownTimeout = time.Minute * 1
+)
+
+// Protocol acts as both a http client and a http handler.
+type Protocol struct {
+	Target          *url.URL
+	RequestTemplate *http.Request
+	transformers    binding.TransformerFactories
+	Client          *http.Client
+	incoming        chan msgErr
+}
+
+func NewProtocol(opts ...ProtocolOption) (*Protocol, error) {
+	p := &Protocol{
+		transformers: make(binding.TransformerFactories, 0),
+		incoming:     make(chan msgErr),
+	}
+	if err := p.applyOptions(opts...); err != nil {
+		return nil, err
+	}
+
+	if p.Client == nil {
+		p.Client = http.DefaultClient
+	}
+
+	return p, nil
+}
+
+func (p *Protocol) applyOptions(opts ...ProtocolOption) error {
+	for _, fn := range opts {
+		if err := fn(p); err != nil {
+			return err
+		}
+	}
+	return nil
+}
+
+// Send implements binding.Sender
+func (p *Protocol) Send(ctx context.Context, m binding.Message) error {
+	_, err := p.Request(ctx, m)
+	return err
+}
+
+// Request implements binding.Requester
+func (p *Protocol) Request(ctx context.Context, m binding.Message) (binding.Message, error) {
+	var err error
+	defer func() { _ = m.Finish(err) }()
+
+	req := p.makeRequest(ctx)
+
+	if p.Client == nil || req == nil || req.URL == nil {
+		return nil, fmt.Errorf("not initialized: %#v", p)
+	}
+
+	if err = WriteHttpRequest(ctx, m, req, p.transformers); err != nil {
+		return nil, err
+	}
+	resp, err := p.Client.Do(req)
+	if err != nil {
+		return nil, err
+	}
+	if resp.StatusCode/100 != 2 {
+		return nil, fmt.Errorf("%d %s", resp.StatusCode, http.StatusText(resp.StatusCode))
+	}
+
+	return NewMessage(resp.Header, resp.Body), nil
+}
+
+func (p *Protocol) makeRequest(ctx context.Context) *http.Request {
+	// TODO: support custom headers from context?
+	req := &http.Request{
+		Header: make(http.Header),
+		// TODO: HeaderFrom(ctx),
+	}
+
+	if p.RequestTemplate != nil {
+		req.Method = p.RequestTemplate.Method
+		req.URL = p.RequestTemplate.URL
+		req.Close = p.RequestTemplate.Close
+		req.Host = p.RequestTemplate.Host
+		copyHeadersEnsure(p.RequestTemplate.Header, &req.Header)
+	}
+
+	if p.Target != nil {
+		req.URL = p.Target
+	}
+
+	// Override the default request with target from context.
+	if target := cecontext.TargetFrom(ctx); target != nil {
+		req.URL = target
+	}
+	return req.WithContext(ctx)
+}
+
+// Ensure to is a non-nil map before copying
+func copyHeadersEnsure(from http.Header, to *http.Header) {
+	if len(from) > 0 {
+		if *to == nil {
+			*to = http.Header{}
+		}
+		copyHeaders(from, *to)
+	}
+}
+
+func copyHeaders(from, to http.Header) {
+	if from == nil || to == nil {
+		return
+	}
+	for header, values := range from {
+		for _, value := range values {
+			to.Add(header, value)
+		}
+	}
+}
+
+// Receive the next incoming HTTP request as a CloudEvent.
+// Returns non-nil error if the incoming HTTP request fails to parse as a CloudEvent
+// Returns io.EOF if the receiver is closed.
+func (p *Protocol) Receive(ctx context.Context) (binding.Message, error) {
+	msgErr, ok := <-p.incoming
+	if !ok {
+		return nil, io.EOF
+	}
+	return msgErr.msg, msgErr.err
+}
+
+type msgErr struct {
+	msg *Message
+	err error
+}
+
+// ServeHTTP implements http.Handler.
+// Blocks until Message.Finish is called.
+func (p *Protocol) ServeHTTP(rw http.ResponseWriter, req *http.Request) {
+	var err error
+	m := NewMessageFromHttpRequest(req)
+	if m.ReadEncoding() == binding.EncodingUnknown {
+		p.incoming <- msgErr{nil, binding.ErrUnknownEncoding}
+	}
+	done := make(chan error)
+	m.OnFinish = func(err error) error {
+		if m.resp != nil {
+			err := WriteHttpResponseWriter(context.Background(), m.resp, rw, p.transformers)
+			_ = m.resp.Finish(err)
+		}
+		m.resp = nil
+		done <- err
+		return nil
+	}
+	p.incoming <- msgErr{m, err} // Send to Receive()
+	if err = <-done; err != nil {
+		http.Error(rw, fmt.Sprintf("cannot forward CloudEvent: %v", err), http.StatusInternalServerError)
+	}
+}
diff --git a/pkg/transport/http/transport.go b/pkg/transport/http/protocol_server.go
similarity index 75%
rename from pkg/transport/http/transport.go
rename to pkg/transport/http/protocol_server.go
index adcb0f4..0264327 100644
--- a/pkg/transport/http/transport.go
+++ b/pkg/transport/http/protocol_server.go
@@ -4,30 +4,19 @@ import (
 	"context"
 	"fmt"
 	"net"
-	nethttp "net/http"
-	"net/url"
+	"net/http"
 	"strings"
 	"sync"
 	"time"
 
-	"go.opencensus.io/plugin/ochttp"
-	"go.opencensus.io/plugin/ochttp/propagation/tracecontext"
-
 	"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"
 )
 
-// Transport adheres to transport.Transport.
-var _ transport.Transport = (*Transport)(nil)
-
-const (
-	// DefaultShutdownTimeout defines the default timeout given to the http.Server when calling Shutdown.
-	DefaultShutdownTimeout = time.Minute * 1
-)
-
-// Transport acts as both a http client and a http handler.
 type Transport struct {
 	bindings.BindingTransport
 
@@ -46,40 +35,42 @@ type Transport struct {
 	// Receive Mutex
 	reMu sync.Mutex
 	// Handler is the handler the http Server will use. Use this to reuse the
-	// http server. If nil, the Transport will create a one.
-	Handler           *nethttp.ServeMux
+	// http server. If nil, the Protocol will create a one.
+	Handler           *http.ServeMux
 	listener          net.Listener
-	transport         nethttp.RoundTripper // TODO: use this.
-	server            *nethttp.Server
+	transport         http.RoundTripper // TODO: use this.
+	server            *http.Server
 	handlerRegistered bool
 	middleware        []Middleware
-	Target            *url.URL         // TODO: this is here just to allow the options to mutate it.
-	RequestTemplate   *nethttp.Request // TODO: this is here just to allow the options to mutate it.
+
+	protocol *Protocol
 }
 
-func New(opts ...Option) (*Transport, error) {
+func New(p *Protocol, opts ...Option) (*Transport, error) {
 	t := &Transport{}
 	if err := t.applyOptions(opts...); err != nil {
 		return nil, err
 	}
 
-	if t.Requester == nil {
-		client := nethttp.DefaultClient
-		t.Requester = NewRequester(client, t.Target)
-	}
+	t.Sender = p
+	t.Requester = p
+	t.Receiver = p
+	t.protocol = p
 
-	if t.Sender == nil {
-		client := nethttp.DefaultClient
-		t.Sender = NewSender(client, t.Target)
-	}
-
-	if t.Receiver == nil {
-		t.Receiver = NewReceiver()
+	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 {
@@ -113,6 +104,11 @@ func (t *Transport) Request(ctx context.Context, e event.Event) (*event.Event, e
 	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 {
@@ -120,12 +116,12 @@ func (t *Transport) StartReceiver(ctx context.Context) error {
 	defer t.reMu.Unlock()
 
 	if t.Handler == nil {
-		t.Handler = nethttp.NewServeMux()
+		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)
+		t.Handler.Handle(t.GetPath(), t.protocol)
 		t.handlerRegistered = true
 	}
 
@@ -134,7 +130,7 @@ func (t *Transport) StartReceiver(ctx context.Context) error {
 		return err
 	}
 
-	t.server = &nethttp.Server{
+	t.server = &http.Server{
 		Addr: addr.String(),
 		Handler: &ochttp.Handler{
 			Propagation:    &tracecontext.HTTPFormat{},
@@ -164,10 +160,7 @@ func (t *Transport) StartReceiver(ctx context.Context) error {
 	select {
 	case <-ctx.Done():
 		// Try a gracefully shutdown.
-		timeout := DefaultShutdownTimeout
-		if t.ShutdownTimeout != nil {
-			timeout = *t.ShutdownTimeout
-		}
+		timeout := *t.ShutdownTimeout
 		ctx, cancel := context.WithTimeout(context.Background(), timeout)
 		defer cancel()
 		err := t.server.Shutdown(ctx)
@@ -178,17 +171,6 @@ func (t *Transport) StartReceiver(ctx context.Context) error {
 	}
 }
 
-func (t *Transport) ServeHTTP(rw nethttp.ResponseWriter, req *nethttp.Request) {
-	if r, ok := t.BindingTransport.Receiver.(*Receiver); ok {
-		r.ServeHTTP(rw, req)
-	}
-}
-
-// HasTracePropagation implements Transport.HasTracePropagation
-func (t *Transport) HasTracePropagation() bool {
-	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.
@@ -200,7 +182,7 @@ func (t *Transport) GetPort() int {
 	return -1
 }
 
-func formatSpanName(r *nethttp.Request) string {
+func formatSpanName(r *http.Request) string {
 	return "cloudevents.http." + r.URL.Path
 }
 
@@ -246,7 +228,7 @@ func (t *Transport) GetPath() string {
 }
 
 // attachMiddleware attaches the HTTP middleware to the specified handler.
-func attachMiddleware(h nethttp.Handler, middleware []Middleware) nethttp.Handler {
+func attachMiddleware(h http.Handler, middleware []Middleware) http.Handler {
 	for _, m := range middleware {
 		h = m(h)
 	}
diff --git a/pkg/transport/http/receiver.go b/pkg/transport/http/receiver.go
deleted file mode 100644
index 2749265..0000000
--- a/pkg/transport/http/receiver.go
+++ /dev/null
@@ -1,66 +0,0 @@
-package http
-
-import (
-	"context"
-	"fmt"
-	"io"
-	"net/http"
-	nethttp "net/http"
-
-	"github.com/cloudevents/sdk-go/pkg/binding"
-)
-
-type msgErr struct {
-	msg *Message
-	err error
-}
-
-// Receiver for CloudEvents as HTTP requests which implements http.Handler.
-// To receive messages, associate it with a http.Server.
-type Receiver struct {
-	incoming chan msgErr
-
-	transformers binding.TransformerFactories
-}
-
-// ServeHTTP implements http.Handler.
-// Blocks until Message.Finish is called.
-func (r *Receiver) ServeHTTP(rw http.ResponseWriter, req *http.Request) {
-	var err error
-	m := NewMessageFromHttpRequest(req)
-	if m.ReadEncoding() == binding.EncodingUnknown {
-		r.incoming <- msgErr{nil, binding.ErrUnknownEncoding}
-	}
-	done := make(chan error)
-	m.OnFinish = func(err error) error {
-		//status := http.StatusNoContent
-		if m.resp != nil {
-			err := EncodeHttpResponseWriter(context.Background(), m.resp, rw, r.transformers)
-			_ = m.resp.Finish(err)
-		}
-		m.resp = nil
-		done <- err
-		return nil
-	}
-	r.incoming <- msgErr{m, err} // Send to Receive()
-	if err = <-done; err != nil {
-		nethttp.Error(rw, fmt.Sprintf("cannot forward CloudEvent: %v", err), http.StatusInternalServerError)
-	}
-}
-
-// NewReceiver creates a Receiver which implements http.Handler.
-// To receive messages, associate it with a http.Server.
-func NewReceiver() *Receiver {
-	return &Receiver{incoming: make(chan msgErr)}
-}
-
-// Receive the next incoming HTTP request as a CloudEvent.
-// Returns non-nil error if the incoming HTTP request fails to parse as a CloudEvent
-// Returns io.EOF if the receiver is closed.
-func (r *Receiver) Receive(ctx context.Context) (binding.Message, error) {
-	msgErr, ok := <-r.incoming
-	if !ok {
-		return nil, io.EOF
-	}
-	return msgErr.msg, msgErr.err
-}
diff --git a/pkg/transport/http/sender.go b/pkg/transport/http/sender.go
deleted file mode 100644
index 0aab5ae..0000000
--- a/pkg/transport/http/sender.go
+++ /dev/null
@@ -1,121 +0,0 @@
-package http
-
-import (
-	"context"
-	"fmt"
-	"net/http"
-	nethttp "net/http"
-	"net/url"
-
-	cecontext "github.com/cloudevents/sdk-go/pkg/context"
-	bindings "github.com/cloudevents/sdk-go/pkg/transport"
-
-	"github.com/cloudevents/sdk-go/pkg/binding"
-)
-
-// Sender implements binding.Sender wrapping a nethttp.Client and a target URL
-type Sender struct {
-	// Client is the HTTP client used to send events as HTTP requests
-	Client *http.Client
-
-	// RequestTemplate is the base http request that is used for http.Do.
-	// Only .Method, .URL, .Close, and .Header is considered.
-	// If not set, Req.Method defaults to POST.
-	// Req.URL or context.WithTarget(url) are required for sending.
-	RequestTemplate *http.Request
-
-	transformers binding.TransformerFactories
-}
-
-func NewRequester(client *http.Client, target *url.URL, options ...SenderOptionFunc) bindings.Requester {
-	s := &Sender{
-		Client:          client,
-		RequestTemplate: &http.Request{Method: http.MethodPost, URL: target},
-		transformers:    make(binding.TransformerFactories, 0),
-	}
-	for _, o := range options {
-		o(s)
-	}
-	return s
-}
-
-func NewSender(client *http.Client, target *url.URL, options ...SenderOptionFunc) bindings.Sender {
-	return NewRequester(client, target, options...)
-}
-
-// Confirm Sender implements binding.Requester
-var _ bindings.Requester = (*Sender)(nil)
-
-// Send implements binding.Sender
-func (s *Sender) Send(ctx context.Context, m binding.Message) error {
-	_, err := s.Request(ctx, m)
-	return err
-}
-
-// Request implements binding.Requester
-func (s *Sender) Request(ctx context.Context, m binding.Message) (binding.Message, error) {
-	var err error
-	defer func() { _ = m.Finish(err) }()
-
-	req := s.makeRequest(ctx)
-
-	if s.Client == nil || req == nil || req.URL == nil {
-		return nil, fmt.Errorf("not initialized: %#v", s)
-	}
-
-	if err = WriteHttpRequest(ctx, m, req, s.transformers); err != nil {
-		return nil, err
-	}
-	resp, err := s.Client.Do(req)
-	if err != nil {
-		return nil, err
-	}
-	if resp.StatusCode/100 != 2 {
-		return nil, fmt.Errorf("%d %s", resp.StatusCode, nethttp.StatusText(resp.StatusCode))
-	}
-
-	return NewMessage(resp.Header, resp.Body), nil
-}
-
-func (s *Sender) makeRequest(ctx context.Context) *http.Request {
-	// TODO: support custom headers from context?
-	req := &http.Request{
-		Header: make(http.Header),
-		// TODO: HeaderFrom(ctx),
-	}
-
-	if s.RequestTemplate != nil {
-		req.Method = s.RequestTemplate.Method
-		req.URL = s.RequestTemplate.URL
-		req.Close = s.RequestTemplate.Close
-		req.Host = s.RequestTemplate.Host
-		copyHeadersEnsure(s.RequestTemplate.Header, &req.Header)
-	}
-
-	// Override the default request with target from context.
-	if target := cecontext.TargetFrom(ctx); target != nil {
-		req.URL = target
-	}
-	return req.WithContext(ctx)
-}
-
-// Ensure to is a non-nil map before copying
-func copyHeadersEnsure(from http.Header, to *http.Header) {
-	if len(from) > 0 {
-		if *to == nil {
-			*to = http.Header{}
-		}
-		copyHeaders(from, *to)
-	}
-}
-
-func copyHeaders(from, to http.Header) {
-	if from == nil || to == nil {
-		return
-	}
-	for header, values := range from {
-		for _, value := range values {
-			to.Add(header, value)
-		}
-	}
-}
diff --git a/pkg/transport/nats/encoder.go b/pkg/transport/nats/messge_write.go
similarity index 82%
rename from pkg/transport/nats/encoder.go
rename to pkg/transport/nats/messge_write.go
index e66e5e8..f13a832 100644
--- a/pkg/transport/nats/encoder.go
+++ b/pkg/transport/nats/messge_write.go
@@ -11,9 +11,9 @@ import (
 	"github.com/cloudevents/sdk-go/pkg/binding/format"
 )
 
-// Fill the provided amqpMessage with the message m.
+// Fill the provided natsMessage with the bindings.Message m.
 // Using context you can tweak the encoding processing (more details on binding.Write documentation).
-func WriteNATSMessage(ctx context.Context, m binding.Message, natsMessage *nats.Msg, transformers binding.TransformerFactories) error {
+func WriteMessage(ctx context.Context, m binding.Message, natsMessage *nats.Msg, transformers binding.TransformerFactories) error {
 	structuredWriter := (*natsMessageWriter)(natsMessage)
 
 	_, err := binding.Write(
diff --git a/pkg/transport/nats/nats.go b/pkg/transport/nats/nats.go
deleted file mode 100644
index c1e4935..0000000
--- a/pkg/transport/nats/nats.go
+++ /dev/null
@@ -1,47 +0,0 @@
-package nats
-
-// Encoding to use for NATS transport.
-type Encoding int32
-
-const (
-	// Default allows NATS transport implementation to pick.
-	Default Encoding = iota
-	// StructuredV03 is Structured CloudEvents spec v0.3.
-	StructuredV03
-	// StructuredV1 is Structured CloudEvents spec v1.0.
-	StructuredV1
-	// Unknown is unknown.
-	Unknown
-)
-
-// String pretty-prints the encoding as a string.
-func (e Encoding) String() string {
-	switch e {
-	case Default:
-		return "Default Encoding " + e.Version()
-
-	// Structured
-	case StructuredV03, StructuredV1:
-		return "Structured Encoding " + e.Version()
-
-	default:
-		return "Unknown Encoding"
-	}
-}
-
-// Version pretty-prints the encoding version as a string.
-func (e Encoding) Version() string {
-	switch e {
-	// Version 0.3
-	case StructuredV03:
-		return "v0.3"
-
-	// Version 1.0
-	case StructuredV1, Default:
-		return "v1.0"
-
-	// Unknown
-	default:
-		return "Unknown"
-	}
-}
diff --git a/pkg/transport/nats/options.go b/pkg/transport/nats/options.go
index 10e3ba7..edbc7e2 100644
--- a/pkg/transport/nats/options.go
+++ b/pkg/transport/nats/options.go
@@ -6,20 +6,12 @@ import (
 )
 
 // Option is the function signature required to be considered an nats.Option.
-type Option func(*Transport) error
-
-// WithEncoding sets the encoding for NATS transport.
-func WithEncoding(encoding Encoding) Option {
-	return func(t *Transport) error {
-		t.Encoding = encoding
-		return nil
-	}
-}
+type Option func(*Protocol) error
 
 // WithConnOptions supplies NATS connection options that will be used when setting
 // up the internal NATS connection
 func WithConnOptions(opts ...nats.Option) Option {
-	return func(t *Transport) error {
+	return func(t *Protocol) error {
 		for _, o := range opts {
 			t.ConnOptions = append(t.ConnOptions, o)
 		}
@@ -28,12 +20,10 @@ func WithConnOptions(opts ...nats.Option) Option {
 	}
 }
 
-// nats.Sender options
-type SenderOptionFunc func(sender *sender)
-
-// Add a transformer, which Sender uses while encoding a binding.Message to an amqp.Message
-func WithTransformer(transformer binding.TransformerFactory) SenderOptionFunc {
-	return func(sender *sender) {
-		sender.transformers = append(sender.transformers, transformer)
+// Add a transformer, which Protocol uses while encoding a binding.Message to an nats.Message
+func WithTransformer(transformer binding.TransformerFactory) Option {
+	return func(p *Protocol) error {
+		p.Transformers = append(p.Transformers, transformer)
+		return nil
 	}
 }
diff --git a/pkg/transport/nats/protocol.go b/pkg/transport/nats/protocol.go
new file mode 100644
index 0000000..5f1e80c
--- /dev/null
+++ b/pkg/transport/nats/protocol.go
@@ -0,0 +1,115 @@
+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"
+)
+
+// 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
+	Subject      string
+	Transformers binding.TransformerFactories
+
+	subscription *nats.Subscription
+}
+
+// New creates a new NATS protocol.
+func New(natsURL, subject string, opts ...Option) (*Protocol, error) {
+	t := &Protocol{
+		Subject:      subject,
+		NatsURL:      natsURL,
+		ConnOptions:  []nats.Option{},
+		Transformers: make(binding.TransformerFactories, 0),
+	}
+
+	err := t.applyOptions(opts...)
+	if err != nil {
+		return nil, err
+	}
+
+	err = t.connect()
+	if err != nil {
+		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
+
+	t.Conn, err = nats.Connect(t.NatsURL, t.ConnOptions...)
+
+	return err
+}
+
+func (t *Protocol) applyOptions(opts ...Option) error {
+	for _, fn := range opts {
+		if err := fn(t); err != nil {
+			return err
+		}
+	}
+	return nil
+}
+
+// Send implements Sender.Send
+func (t *Protocol) Send(ctx context.Context, in binding.Message) error {
+	msg := &nats.Msg{}
+	if err := WriteMessage(ctx, in, msg, t.Transformers); err != nil {
+		return err
+	}
+	msg.Subject = t.Subject
+	return t.Conn.PublishMsg(msg)
+}
+
+// Receive implements Receiver.Receive
+func (t *Protocol) Receive(ctx context.Context) (binding.Message, error) {
+	var err error
+	if t.subscription == nil {
+		t.subscription, err = t.Conn.SubscribeSync(t.Subject)
+		if err != nil {
+			return nil, err
+		}
+	}
+
+	// TODO: allow to pass in a timeout
+	timeout := time.Second * 60
+
+	msg, err := t.subscription.NextMsg(timeout)
+	if err != nil {
+		return nil, err
+	}
+	return NewMessage(msg), nil
+}
+
+// Close implements Closer.Close
+func (t *Protocol) Close(ctx context.Context) error {
+	defer t.Conn.Close()
+	if t.subscription != nil {
+		if err := t.subscription.Unsubscribe(); err != nil {
+			return err
+		}
+		t.subscription = nil
+	}
+	return nil
+}
diff --git a/pkg/transport/nats/receiver.go b/pkg/transport/nats/receiver.go
deleted file mode 100644
index e4cf800..0000000
--- a/pkg/transport/nats/receiver.go
+++ /dev/null
@@ -1,48 +0,0 @@
-package nats
-
-import (
-	"context"
-	"time"
-
-	"github.com/cloudevents/sdk-go/pkg/binding"
-	"github.com/cloudevents/sdk-go/pkg/transport"
-	"github.com/nats-io/nats.go"
-)
-
-var _ transport.Receiver
-
-// receiver for nats, implements transport.Receiver
-type receiver struct {
-	conn    *nats.Conn
-	subject string
-	sub     *nats.Subscription
-}
-
-func (r *receiver) Receive(ctx context.Context) (binding.Message, error) {
-	var err error
-	if r.sub == nil {
-		r.sub, err = r.conn.SubscribeSync(r.subject)
-		if err != nil {
-			return nil, err
-		}
-	}
-
-	// TODO: allow to pass in a timeout
-	timeout := time.Second * 60
-
-	msg, err := r.sub.NextMsg(timeout)
-	if err != nil {
-		return nil, err
-	}
-	return NewMessage(msg), nil
-}
-
-func (r *receiver) Close(ctx context.Context) error {
-	defer r.conn.Close()
-	return r.sub.Unsubscribe()
-}
-
-// Create a new Receiver which wraps an amqp.Receiver in a binding.Receiver
-func NewReceiver(conn *nats.Conn, subject string) transport.Receiver {
-	return &receiver{conn: conn, subject: subject}
-}
diff --git a/pkg/transport/nats/sender.go b/pkg/transport/nats/sender.go
deleted file mode 100644
index 450a046..0000000
--- a/pkg/transport/nats/sender.go
+++ /dev/null
@@ -1,39 +0,0 @@
-package nats
-
-import (
-	"context"
-
-	"github.com/cloudevents/sdk-go/pkg/binding"
-	bindings "github.com/cloudevents/sdk-go/pkg/transport"
-	"github.com/nats-io/nats.go"
-)
-
-// sender implements binding.Sender
-type sender struct {
-	conn         *nats.Conn
-	subject      string
-	transformers binding.TransformerFactories
-}
-
-func (s *sender) Send(ctx context.Context, in binding.Message) error {
-	msg := &nats.Msg{}
-	if err := WriteNATSMessage(ctx, in, msg, nil); err != nil {
-		return err
-	}
-	msg.Subject = s.subject // TODO: allow for overwriting this.
-	return s.conn.PublishMsg(msg)
-}
-
-func (s *sender) Close(ctx context.Context) error {
-	s.conn.Close()
-	return nil
-}
-
-// Create a new nats Sender, implements binding.Sender
-func NewSender(conn *nats.Conn, subject string, options ...SenderOptionFunc) bindings.Sender {
-	s := &sender{conn: conn, subject: subject, transformers: make(binding.TransformerFactories, 0)}
-	for _, o := range options {
-		o(s)
-	}
-	return s
-}
diff --git a/pkg/transport/nats/transport.go b/pkg/transport/nats/transport.go
deleted file mode 100644
index 1ce6560..0000000
--- a/pkg/transport/nats/transport.go
+++ /dev/null
@@ -1,68 +0,0 @@
-package nats
-
-import (
-	"github.com/cloudevents/sdk-go/pkg/transport/bindings"
-
-	"github.com/cloudevents/sdk-go/pkg/transport"
-	"github.com/nats-io/nats.go"
-)
-
-// Transport adheres to transport.Transport.
-var _ transport.Transport = (*Transport)(nil)
-
-const (
-	// TransportName is the name of this transport.
-	TransportName = "NATS"
-)
-
-// Transport acts as both a NATS client and a NATS handler.
-type Transport struct {
-	bindings.BindingTransport
-
-	Encoding    Encoding
-	Conn        *nats.Conn
-	ConnOptions []nats.Option
-	NatsURL     string
-	Subject     string
-}
-
-// New creates a new NATS transport.
-func New(natsURL, subject string, opts ...Option) (*Transport, error) {
-	t := &Transport{
-		Subject:     subject,
-		NatsURL:     natsURL,
-		ConnOptions: []nats.Option{},
-	}
-
-	err := t.applyOptions(opts...)
-	if err != nil {
-		return nil, err
-	}
-
-	err = t.connect()
-	if err != nil {
-		return nil, err
-	}
-
-	t.Sender = NewSender(t.Conn, t.Subject) // TODO: Allow sender options to be passed in.
-	t.Receiver = NewReceiver(t.Conn, t.Subject)
-
-	return t, nil
-}
-
-func (t *Transport) connect() error {
-	var err error
-
-	t.Conn, err = nats.Connect(t.NatsURL, t.ConnOptions...)
-
-	return err
-}
-
-func (t *Transport) applyOptions(opts ...Option) error {
-	for _, fn := range opts {
-		if err := fn(t); err != nil {
-			return err
-		}
-	}
-	return nil
-}


Test patch:
diff --git a/pkg/client/client_test.go b/pkg/client/client_test.go
index a5669e3..663c2f4 100644
--- a/pkg/client/client_test.go
+++ b/pkg/client/client_test.go
@@ -5,6 +5,7 @@ import (
 	"context"
 	"fmt"
 	"io/ioutil"
+	"log"
 	"net/http"
 	"net/http/httptest"
 	"net/url"
@@ -35,8 +36,13 @@ var (
 )
 
 func simpleBinaryClient(target string) client.Client {
-	t, err := cehttp.New(
-		cehttp.WithTarget(target),
+	p, err := cehttp.NewProtocol(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 {
@@ -51,8 +57,13 @@ func simpleBinaryClient(target string) client.Client {
 }
 
 func simpleTracingBinaryClient(target string) client.Client {
-	t, err := cehttp.New(
-		cehttp.WithTarget(target),
+	p, err := cehttp.NewProtocol(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 {
@@ -67,8 +78,13 @@ func simpleTracingBinaryClient(target string) client.Client {
 }
 
 func simpleStructuredClient(target string) client.Client {
-	t, err := cehttp.New(
-		cehttp.WithTarget(target),
+	p, err := cehttp.NewProtocol(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 {
@@ -371,7 +387,12 @@ func TestClientReceive(t *testing.T) {
 
 				events := make(chan event.Event)
 
-				tp, err := cehttp.New(tc.optsFn(0, path)...)
+				p, err := cehttp.NewProtocol()
+				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())
 				}
@@ -495,7 +516,12 @@ func TestTracedClientReceive(t *testing.T) {
 		t.Run(n, func(t *testing.T) {
 			spanContexts := make(chan trace.SpanContext)
 
-			tp, err := cehttp.New(tc.optsFn(0, "")...)
+			p, err := cehttp.NewProtocol()
+			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())
 			}
diff --git a/pkg/transport/http/response_encoder_test.go b/pkg/transport/http/message_write_response_test.go
similarity index 97%
rename from pkg/transport/http/response_encoder_test.go
rename to pkg/transport/http/message_write_response_test.go
index bbec55f..9f8a425 100644
--- a/pkg/transport/http/response_encoder_test.go
+++ b/pkg/transport/http/message_write_response_test.go
@@ -56,7 +56,7 @@ func TestEncodeHttpResponse(t *testing.T) {
 				eventIn = test2.ExToStr(t, eventIn)
 				messageIn := tt.messageFactory(eventIn)
 
-				err := EncodeHttpResponse(tt.context, messageIn, res, nil)
+				err := WriteHttpResponse(tt.context, messageIn, res, nil)
 				require.NoError(t, err)
 
 				//Little hack to go back to Message
diff --git a/pkg/transport/nats/options_test.go b/pkg/transport/nats/options_test.go
index 19e00c1..4ba3799 100644
--- a/pkg/transport/nats/options_test.go
+++ b/pkg/transport/nats/options_test.go
@@ -4,55 +4,10 @@ import (
 	"testing"
 	"time"
 
-	"github.com/cloudevents/sdk-go/pkg/transport/bindings"
-
-	"github.com/google/go-cmp/cmp"
-	"github.com/google/go-cmp/cmp/cmpopts"
 	natsd "github.com/nats-io/nats-server/v2/server"
 	"github.com/nats-io/nats.go"
 )
 
-func TestWithEncoding(t *testing.T) {
-	testCases := map[string]struct {
-		t        *Transport
-		encoding Encoding
-		want     *Transport
-		wantErr  string
-	}{
-		"valid encoding": {
-			t:        &Transport{},
-			encoding: StructuredV03,
-			want: &Transport{
-				Encoding: StructuredV03,
-			},
-		},
-	}
-	for n, tc := range testCases {
-		t.Run(n, func(t *testing.T) {
-
-			err := tc.t.applyOptions(WithEncoding(tc.encoding))
-
-			if tc.wantErr != "" || err != nil {
-				var gotErr string
-				if err != nil {
-					gotErr = err.Error()
-				}
-				if diff := cmp.Diff(tc.wantErr, gotErr); diff != "" {
-					t.Errorf("unexpected error (-want, +got) = %v", diff)
-				}
-				return
-			}
-
-			got := tc.t
-
-			if diff := cmp.Diff(tc.want, got,
-				cmpopts.IgnoreUnexported(Transport{}, bindings.BindingTransport{})); diff != "" {
-				t.Errorf("unexpected (-want, +got) = %v", diff)
-			}
-		})
-	}
-}
-
 func TestWithConnOptions(t *testing.T) {
 	opts := []nats.Option{
 		nats.DontRandomize(),
diff --git a/test/benchmark/e2e/http/http_mock.go b/test/benchmark/e2e/http/http_mock.go
index bff562f..8b32aca 100644
--- a/test/benchmark/e2e/http/http_mock.go
+++ b/test/benchmark/e2e/http/http_mock.go
@@ -27,17 +27,24 @@ func NewTestClient(fn RoundTripFunc) *nethttp.Client {
 	}
 }
 
-func MockedSender(options ...http.SenderOptionFunc) bindings.Sender {
+func MockedSender(options ...http.ProtocolOption) bindings.Sender {
 	u, _ := url.Parse("http://localhost")
-	return http.NewSender(NewTestClient(func(req *nethttp.Request) *nethttp.Response {
+
+	p, err := http.NewProtocol(options...)
+	if err != nil {
+		panic(err)
+	}
+	p.Client = NewTestClient(func(req *nethttp.Request) *nethttp.Response {
 		return &nethttp.Response{
 			StatusCode: 202,
 			Header:     make(nethttp.Header),
 		}
-	}), u, options...)
+	})
+	p.Target = u
+	return p
 }
 
-func MockedClient() (cloudevents.Client, *cehttp.Transport) {
+func MockedClient() (cloudevents.Client, *cehttp.Protocol, *cehttp.Transport) {
 	mockTransport := RoundTripFunc(func(req *nethttp.Request) *nethttp.Response {
 		return &nethttp.Response{
 			StatusCode: 202,
@@ -45,22 +52,23 @@ func MockedClient() (cloudevents.Client, *cehttp.Transport) {
 			Body:       ioutil.NopCloser(bytes.NewReader([]byte{})),
 		}
 	})
-	t, err := cehttp.New(
-		cehttp.WithTarget("http://localhost"),
-		cehttp.WithHTTPTransport(mockTransport),
-	)
 
+	p, err := cehttp.NewProtocol(cehttp.WithTarget("http://localhost"))
 	if err != nil {
 		panic(err)
 	}
 
-	client, err := cloudevents.NewClient(t)
+	t, err := cehttp.New(p, cehttp.WithHTTPTransport(mockTransport))
+	if err != nil {
+		panic(err)
+	}
 
+	client, err := cloudevents.NewClient(t)
 	if err != nil {
 		panic(err)
 	}
 
-	return client, t
+	return client, p, t
 }
 
 func MockedBinaryRequest(body []byte) *nethttp.Request {
diff --git a/test/benchmark/e2e/http/main.go b/test/benchmark/e2e/http/main.go
index 33cb6e4..867b5bf 100644
--- a/test/benchmark/e2e/http/main.go
+++ b/test/benchmark/e2e/http/main.go
@@ -67,7 +67,7 @@ func benchmarkBaseline(cases []e2e.BenchmarkCase, requestFactory func([]byte) *n
 	return results
 }
 
-func pipeLoopDirect(r *http.Receiver, sendCtx context.Context, endCtx context.Context, opts ...http.SenderOptionFunc) {
+func pipeLoopDirect(r *http.Protocol, sendCtx context.Context, endCtx context.Context, opts ...http.ProtocolOption) {
 	s := MockedSender(opts...)
 	var err error
 	var m binding.Message
@@ -85,7 +85,7 @@ func pipeLoopDirect(r *http.Receiver, sendCtx context.Context, endCtx context.Co
 	}
 }
 
-func pipeLoopMulti(r *http.Receiver, sendCtx context.Context, endCtx context.Context, outputSenders int, opts ...http.SenderOptionFunc) {
+func pipeLoopMulti(r *http.Protocol, sendCtx context.Context, endCtx context.Context, outputSenders int, opts ...http.ProtocolOption) {
 	s := MockedSender(opts...)
 	var err error
 	var m binding.Message
@@ -120,7 +120,10 @@ func benchmarkReceiverSender(cases []e2e.BenchmarkCase, requestFactory func([]by
 		fmt.Printf("%+v\n", c)
 
 		ctx, cancel := context.WithCancel(context.TODO())
-		receiver := http.NewReceiver()
+		receiver, err := http.NewProtocol()
+		if err != nil {
+			panic(err)
+		}
 
 		// Spawn dispatchers
 		for i := 0; i < c.Parallelism; i++ {
@@ -184,11 +187,11 @@ func benchmarkClient(cases []e2e.BenchmarkCase, requestFactory func([]byte) *net
 	for _, c := range cases {
 		fmt.Printf("%+v\n", c)
 
-		_, 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))
@@ -202,7 +205,7 @@ func benchmarkClient(cases []e2e.BenchmarkCase, requestFactory func([]byte) *net
 			b.RunParallel(func(pb *testing.PB) {
 				for pb.Next() {
 					w := httptest.NewRecorder()
-					mockedReceiverTransport.ServeHTTP(w, requestFactory(buffer))
+					mockedReceiverProtocol.ServeHTTP(w, requestFactory(buffer))
 				}
 			})
 		})
diff --git a/test/integration/http/direct.go b/test/integration/http/direct.go
index f47ab68..7864f80 100644
--- a/test/integration/http/direct.go
+++ b/test/integration/http/direct.go
@@ -28,24 +28,39 @@ type DirectTapTest struct {
 
 type DirectTapTestCases map[string]DirectTapTest
 
-func ClientDirect(t *testing.T, tc DirectTapTest, topts ...cehttp.Option) {
+func ClientDirect(t *testing.T, tc DirectTapTest, opts ...interface{}) {
 	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)
+	}
+
 	if len(topts) == 0 {
 		topts = append(topts, cloudevents.WithEncoding(cloudevents.HTTPBinaryEncoding))
 	}
-	topts = append(topts, cloudevents.WithTarget(server.URL))
 	topts = append(topts, cloudevents.WithPort(0)) // random port
-	transport, err := cloudevents.NewHTTPTransport(
-		topts...,
-	)
+	transport, err := cloudevents.NewHTTPTransport(protocol, topts...)
 	if err != nil {
 		t.Fatal(err)
 	}
 
-	tap.handler = transport
+	tap.handler = protocol
 
 	ce, err := cloudevents.NewClient(
 		transport,
diff --git a/test/integration/http/loopback.go b/test/integration/http/loopback.go
index 4450cf1..cdc6177 100644
--- a/test/integration/http/loopback.go
+++ b/test/integration/http/loopback.go
@@ -2,6 +2,7 @@ package http
 
 import (
 	"context"
+
 	"net/http/httptest"
 	"testing"
 	"time"
@@ -40,24 +41,40 @@ type TapTest struct {
 
 type TapTestCases map[string]TapTest
 
-func ClientLoopback(t *testing.T, tc TapTest, topts ...cehttp.Option) {
+func ClientLoopback(t *testing.T, tc TapTest, opts ...interface{}) {
 	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)
+	}
+
 	if len(topts) == 0 {
 		topts = append(topts, cloudevents.WithEncoding(cloudevents.HTTPBinaryEncoding))
 	}
-	topts = append(topts, cloudevents.WithTarget(server.URL))
 	topts = append(topts, cloudevents.WithPort(0)) // random port
-	transport, err := cloudevents.NewHTTPTransport(
-		topts...,
-	)
+
+	transport, err := cloudevents.NewHTTPTransport(protocol, topts...)
 	if err != nil {
 		t.Fatal(err)
 	}
 
-	tap.handler = transport
+	tap.handler = protocol
 
 	ce, err := cloudevents.NewClient(
 		transport,
diff --git a/test/integration/http/middleware.go b/test/integration/http/middleware.go
index f58367a..1375cb4 100644
--- a/test/integration/http/middleware.go
+++ b/test/integration/http/middleware.go
@@ -11,24 +11,39 @@ import (
 	"github.com/cloudevents/sdk-go/pkg/transport/http"
 )
 
-func ClientMiddleware(t *testing.T, tc TapTest, topts ...http.Option) {
+func ClientMiddleware(t *testing.T, tc TapTest, opts ...interface{}) {
 	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))
+
+	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.WithTarget(server.URL))
 	topts = append(topts, cloudevents.WithPort(0)) // random port
-	transport, err := cloudevents.NewHTTPTransport(
-		topts...,
-	)
+	transport, err := cloudevents.NewHTTPTransport(protocol, topts...)
 	if err != nil {
 		t.Fatal(err)
 	}
 
-	tap.handler = transport
+	tap.handler = protocol
 
 	ce, err := cloudevents.NewClient(
 		transport,
diff --git a/test/integration/http_binding/http_test.go b/test/integration/http_binding/http_test.go
index e70f9be..7c65e08 100644
--- a/test/integration/http_binding/http_test.go
+++ b/test/integration/http_binding/http_test.go
@@ -2,7 +2,6 @@ package http_binding_test
 
 import (
 	"context"
-	nethttp "net/http"
 	"net/http/httptest"
 	"net/url"
 	"testing"
@@ -89,13 +88,14 @@ func TestSendEventReceiveBinary(t *testing.T) {
 	})
 }
 
-func testSenderReceiver(t testing.TB, options ...http.SenderOptionFunc) (func(), bindings.Sender, bindings.Receiver) {
-	r := http.NewReceiver() // Parameters? Capacity, sync.
-	srv := httptest.NewServer(r)
+func testSenderReceiver(t testing.TB, options ...http.ProtocolOption) (func(), bindings.Sender, bindings.Receiver) {
+	p, err := http.NewProtocol(options...)
+	require.NoError(t, err)
+	srv := httptest.NewServer(p)
 	u, err := url.Parse(srv.URL)
 	require.NoError(t, err)
-	s := http.NewSender(&nethttp.Client{}, u, options...) // Capacity, sync etc.
-	return func() { srv.Close() }, s, r
+	p.Target = u
+	return func() { srv.Close() }, p, p
 }
 
 func BenchmarkSendReceive(b *testing.B) {