Repository: cloudevents/sdk-go
Pull request #229 (2019-10-01 16:03:43)

Do not modify the outbound event.
Do not assume json for empty content encoding. Preserve empty content encoding.

This adds but does not yet use a nop data encoder.

Fixes to tests to allow for non-defaulted app/json to be assumed.

Before doing these changes I touched the base64 encoding code. It is not quite finished.

fixes: https://github.com/cloudevents/sdk-go/issues/217
relates to https://github.com/cloudevents/sdk-go/issues/222


Patch:
diff --git a/pkg/binding/doc.go b/pkg/binding/doc.go
index 5aa1594..5b1b6ff 100644
--- a/pkg/binding/doc.go
+++ b/pkg/binding/doc.go
@@ -1,25 +1,36 @@
 /*
 
-Package binding is for implementing transport bindings and
-intermediaries like importers, brokers or channels that forward
-messages between bindings.
-
-A transport binding implements Message, Sender and Receiver interfaces.
-An intermediary uses those interfaces to transfer event messages.
-
-A Message is an abstract container for an Event. It provides
-additional methods for efficient forwarding of structured events
-and reliable delivery when the underlying transports support it.
-
-A Receiver can return instances of its own Message implementations. A
-Sender must be able to send any implementation of the Message
-interface, but it may provide optimized handling for its own Message
-implementations.
-
-For transports that support a reliable delivery QoS, the Message
-interface allows acknowledgment between sender and receiver for QoS
-level 0, 1 or 2. The effective QoS is the lowest provided by sender or
-receiver.
+Package binding defines interfaces for protocol bindings.
+
+NOTE: Most applications that emit or consume events can use the client
+package. This package is for implementing new protocol bindings and
+intermediaries; processes that forward events between protocols, rather than
+emitting or consuming events themselves.
+
+Protocol Bindings
+
+A protocol binding implements at least Message, Sender and Receiver, and usually
+Encoder.
+
+Receiver: receives protocol messages and wraps them to implement the Message interface.
+
+Message: converts to protocol-neutral cloudevents.Event or structured event
+data. It also provides methods to manage acknowledgment for reliable
+delivery across bindings.
+
+Sender: converts arbitrary Message implementations to a protocol-specific form
+and sends them.
+
+Message and ExactlyOnceMessage provide methods to allow acknowledgments to
+propagate when a reliable messages is forwarded from a Receiver to a Sender.
+QoS 0 (unreliable), 1 (at-least-once) and 2 (exactly-once) are supported.
+
+Intermediaries
+
+Intermediaries can forward Messages from a Receiver to a Sender without
+knowledge of the underlying protocols. The Message interface allows structured
+messages to be forwarded without decoding and re-encoding. It also allows any
+Message to be fully decoded and examined if needed.
 
 */
 package binding
diff --git a/pkg/binding/format/format.go b/pkg/binding/format/format.go
new file mode 100644
index 0000000..9ff22d1
--- /dev/null
+++ b/pkg/binding/format/format.go
@@ -0,0 +1,75 @@
+/*
+Package format formats structured events.
+
+The "application/cloudevents+json" format is built-in and always
+available. Other formats may be added.
+*/
+package format
+
+import (
+	"encoding/json"
+	"fmt"
+	"strings"
+
+	ce "github.com/cloudevents/sdk-go/pkg/cloudevents"
+)
+
+// Format marshals and unmarshals structured events to bytes.
+type Format interface {
+	// MediaType identifies the format
+	MediaType() string
+	// Marshal event to bytes
+	Marshal(ce.Event) ([]byte, error)
+	// Unmarshal bytes to event
+	Unmarshal([]byte, *ce.Event) error
+}
+
+// Prefix for event-format media types.
+const Prefix = "application/cloudevents"
+
+// IsFormat returns true if mediaType begins with "application/cloudevents"
+func IsFormat(mediaType string) bool { return strings.HasPrefix(mediaType, Prefix) }
+
+// JSON is the built-in "application/cloudevents+json" format.
+var JSON = jsonFmt{}
+
+type jsonFmt struct{}
+
+func (jsonFmt) MediaType() string { return ce.ApplicationCloudEventsJSON }
+
+func (jsonFmt) Marshal(e ce.Event) ([]byte, error)    { return json.Marshal(e) }
+func (jsonFmt) Unmarshal(b []byte, e *ce.Event) error { return json.Unmarshal(b, e) }
+
+// built-in formats
+var formats map[string]Format
+
+func init() {
+	formats = map[string]Format{}
+	Add(JSON)
+}
+
+// Lookup returns the format for mediaType, or nil if not found.
+func Lookup(mediaType string) Format { return formats[mediaType] }
+
+func unknown(mediaType string) error {
+	return fmt.Errorf("unknown event format media-type %#v", mediaType)
+}
+
+// Add a new Format. It can be retrieved by Lookup(f.MediaType())
+func Add(f Format) { formats[f.MediaType()] = f }
+
+// Marshal an event to bytes using the mediaType event format.
+func Marshal(mediaType string, e ce.Event) ([]byte, error) {
+	if f := formats[mediaType]; f != nil {
+		return f.Marshal(e)
+	}
+	return nil, unknown(mediaType)
+}
+
+// Unmarshal bytes to an event using the mediaType event format.
+func Unmarshal(mediaType string, b []byte, e *ce.Event) error {
+	if f := formats[mediaType]; f != nil {
+		return f.Unmarshal(b, e)
+	}
+	return unknown(mediaType)
+}
diff --git a/pkg/binding/interfaces.go b/pkg/binding/interfaces.go
new file mode 100644
index 0000000..c5355b1
--- /dev/null
+++ b/pkg/binding/interfaces.go
@@ -0,0 +1,112 @@
+package binding
+
+import (
+	"context"
+
+	ce "github.com/cloudevents/sdk-go/pkg/cloudevents"
+)
+
+// Message is the interface to a binding-specific message containing an event.
+//
+// Reliable Delivery
+//
+// There are 3 reliable qualities of service for messages:
+//
+// 0/at-most-once/unreliable: messages can be dropped silently.
+//
+// 1/at-least-once: messages are not dropped without signaling an error
+// to the sender, but they may be duplicated in the event of a re-send.
+//
+// 2/exactly-once: messages are never dropped (without error) or
+// duplicated, as long as both sending and receiving ends maintain
+// some binding-specific delivery state. Whether this is persisted
+// depends on the configuration of the binding implementations.
+//
+// The Message interface supports QoS 0 and 1, the ExactlyOnceMessage interface
+// supports QoS 2
+//
+type Message interface {
+	// Event decodes and returns the contained Event.
+	Event() (ce.Event, error)
+
+	// Structured optionally returns a structured event encoding and its
+	// format type, if the message contains such an encoding. Returns
+	// ("", nil) if not.
+	//
+	// This allows Senders to avoid re-encoding messages that are
+	// already in suitable structured form.
+	//
+	Structured() (formatMediaType string, encodedEvent []byte)
+
+	// Finish *must* be called when message from a Receiver can be forgotten by
+	// the receiver. Sender.Send() calls Finish() when the message is sent.  A QoS
+	// 1 sender should not call Finish() until it gets an acknowledgment of
+	// receipt on the underlying transport.  For QoS 2 see ExactlyOnceMessage.
+	//
+	// Passing a non-nil err indicates sending or processing failed.
+	// A non-nil return indicates that the message was not accepted
+	// by the receivers peer.
+	Finish(error) error
+}
+
+// ExactlyOnceMessage is implemented by received Messages
+// that support QoS 2.  Only transports that support QoS 2 need to
+// implement or use this interface.
+type ExactlyOnceMessage interface {
+	Message
+
+	// Received is called by a forwarding QoS2 Sender when it gets
+	// acknowledgment of receipt (e.g. AMQP 'accept' or MQTT PUBREC)
+	//
+	// The receiver must call settle(nil) when it get's the ack-of-ack
+	// (e.g. AMQP 'settle' or MQTT PUBCOMP) or settle(err) if the
+	// transfer fails.
+	//
+	// Finally the Sender calls Finish() to indicate the message can be
+	// discarded.
+	//
+	// If sending fails, or if the sender does not support QoS 2, then
+	// Finish() may be called without any call to Received()
+	Received(settle func(error))
+}
+
+// Receiver receives messages.
+type Receiver interface {
+	// Receive blocks till a message is received or ctx expires.
+	//
+	// A non-nil error means the receiver is closed.
+	// io.EOF means it closed cleanly, any other value indicates an error.
+	Receive(ctx context.Context) (Message, error)
+}
+
+// Sender sends messages.
+type Sender interface {
+	// Send blocks till the message is sent or ctx expires.
+	//
+	// To support optimized forwading of structured-mode messages, Send()
+	// should use the encoding returned by m.Structured() if there is one.
+	// Otherwise m.Event() can be encoded as per the binding's rules.
+	Send(ctx context.Context, m Message) error
+}
+
+// Closer is the common interface for things that can be closed
+type Closer interface {
+	Close(ctx context.Context) error
+}
+
+// ReceiveCloser is a Receiver that can be closed.
+type ReceiveCloser interface {
+	Receiver
+	Closer
+}
+
+// SendCloser is a Sender that can be closed.
+type SendCloser interface {
+	Sender
+	Closer
+}
+
+// Encoder encodes events as messages.
+type Encoder interface {
+	Encode(ce.Event) (Message, error)
+}
diff --git a/pkg/binding/message.go b/pkg/binding/message.go
index 2dd36f0..c2b2f58 100644
--- a/pkg/binding/message.go
+++ b/pkg/binding/message.go
@@ -1,92 +1,60 @@
 package binding
 
 import (
-	"context"
-
-	"github.com/cloudevents/sdk-go/pkg/cloudevents"
+	"github.com/cloudevents/sdk-go/pkg/binding/format"
+	ce "github.com/cloudevents/sdk-go/pkg/cloudevents"
 )
 
-// Message is the interface to a transport-specific message containing an event.
-//
-// There are 3 reliable qualities of service for messages:
-//
-// 0/at-most-once/unreliable: messages can be dropped silently.
-//
-// 1/at-least-once: messages are not dropped without signaling an error
-// at the sender but may be duplicated.
-//
-// 2/exactly-once: messages are never dropped (without error) or
-// duplicated, as long as both ends maintain some transport-specific
-// state.
-//
-// The Message interface supports QoS 0 and 1, the ExactlyOnceMessage inteface
-// supports QoS 2
-//
-type Message interface {
-	// Event decodes and returns the contained Event.
-	Event() (cloudevents.Event, error)
-
-	// Structured optionally returns an encoded structured event if it
-	// is efficient to do so, or ("", nil) if not.
-	//
-	// Enables Senders to optimize the case where structured events are
-	// passed from transport to transport without being decoded and
-	// re-encoded.
-	//
-	// Transport Message and Sender implementations are not required to
-	// implement the optimization, they can ignore it.
-	Structured() (encodingMediaType string, encodedEvent []byte)
+// EventMessage type-converts a cloudevents.Event object to implement Message.
+// This allows local cloudevents.Event objects to be sent directly via Sender.Send()
+//     s.Send(ctx, binding.EventMessage(e))
+type EventMessage ce.Event
 
-	// Finish *must* be called when message from a Receiver can be
-	// forgotten by the receiver.
-	//
-	// A QoS 1 sender forwarding messages should not call Finish()
-	// until it gets an acknowledgment of receipt.
-	//
-	// A non-nil error indicates an error in sending or processing.
-	Finish(error)
-}
+func (m EventMessage) Event() (ce.Event, error)     { return ce.Event(m), nil }
+func (m EventMessage) Structured() (string, []byte) { return "", nil }
+func (EventMessage) Finish(error) error             { return nil }
 
-// ExactlyOnceMessage is implemented by incoming transport messages
-// that support QoS 2.  Only transports that support QoS 2 need to
-// implement or use this interface.
-type ExactlyOnceMessage interface {
-	Message
+var _ Message = EventMessage{} // Test it conforms to the interface
 
-	// Received is called by a Sender when it gets acknowledgment of receipt
-	// (e.g. AMQP ACCEPT or MQTT PUBREC)
-	//
-	// The sender passes a finish() function that the original receiver
-	// must call when it get's the ack-of-the-ack (e.g. AMQP SETTLE, MQTT
-	// PUBCOMP)
-	//
-	// If sending fails, the sender must call Finish(err) with a non-nil
-	// error instead of Received. ExactlyOnceMessage implementations
-	// must also be prepared to handle Finish(nil) if the sender does
-	// not support QoS 3.
-	Received(finish func(error))
+// StructMessage implements a structured-mode message as a simple struct.
+type StructMessage struct {
+	Format string
+	Bytes  []byte
 }
 
-// EventMessage wraps a local cloudevents.Event as a Message.
-type EventMessage cloudevents.Event
-
-// Event returns the event.
-func (m EventMessage) Event() (cloudevents.Event, error) { return cloudevents.Event(m), nil }
-
-// Structured returns ("", nil).
-func (EventMessage) Structured() (string, []byte) { return "", nil }
+// Event can only decode built-in formats supported by format.Unmarshal.
+func (m StructMessage) Event() (e ce.Event, err error) {
+	err = format.Unmarshal(m.Format, m.Bytes, &e)
+	return e, err
+}
+func (m StructMessage) Structured() (string, []byte) { return m.Format, m.Bytes }
+func (StructMessage) Finish(error) error             { return nil }
 
-// Finish does nothing.
-func (EventMessage) Finish(error) {}
+var _ Message = StructMessage{} // Test it conforms to the interface
 
-var _ Message = EventMessage{} // Test it conforms to the interface
+// StructEncoder encodes events as StructMessage using a Format.
+type StructEncoder struct{ Format format.Format }
 
-// Receiver receives messages.
-type Receiver interface {
-	Receive(ctx context.Context) (Message, error)
+func (enc StructEncoder) Encode(e ce.Event) (Message, error) {
+	b, err := enc.Format.Marshal(e)
+	return StructMessage{Format: enc.Format.MediaType(), Bytes: b}, err
 }
 
-// Sender sends messages.
-type Sender interface {
-	Send(ctx context.Context, m Message) error
+type BinaryEncoder struct{}
+
+func (BinaryEncoder) Encode(e ce.Event) (Message, error) { return EventMessage(e), nil }
+
+// Structured returns the structured encoding of a message using a format.
+// m.Structured() returns the correct format, return that.
+// Otherwise use format the message's event with f.Format().
+func Structured(m Message, f format.Format) ([]byte, error) {
+	mt, b := m.Structured()
+	if mt == f.MediaType() {
+		return b, nil
+	}
+	e, err := m.Event()
+	if err != nil {
+		return nil, err
+	}
+	return f.Marshal(e)
 }
diff --git a/pkg/binding/spec/attributes.go b/pkg/binding/spec/attributes.go
new file mode 100644
index 0000000..7951859
--- /dev/null
+++ b/pkg/binding/spec/attributes.go
@@ -0,0 +1,122 @@
+package spec
+
+import (
+	"errors"
+	"time"
+
+	ce "github.com/cloudevents/sdk-go/pkg/cloudevents"
+)
+
+// Kind is a version-independent identifier for a CloudEvent context attribute.
+type Kind uint8
+
+const (
+	// Required cloudevents attributes
+	ID Kind = iota
+	Source
+	SpecVersion
+	Type
+	// Optional cloudevents attributes
+	DataContentType
+	DataSchema
+	Subject
+	Time
+)
+const nAttrs = int(Time) + 1
+
+var kindNames = [nAttrs]string{
+	"id",
+	"source",
+	"specversion",
+	"type",
+	"datacontenttype",
+	"dataschema",
+	"subject",
+	"time",
+}
+
+// String is a human-readable string, for a valid attribute name use Attribute.Name
+func (k Kind) String() string { return kindNames[k] }
+
+// IsRequired returns true for attributes defined as "required" by the CE spec.
+func (k Kind) IsRequired() bool { return k < DataContentType }
+
+// ErrBadType returned when there is an attribute/value type mismatch.
+var ErrBadType = errors.New("bad context attribute type")
+
+// Attribute is a named attribute accessor.
+// The attribute name is specific to a Version.
+type Attribute interface {
+	Kind() Kind
+	// Name of the attribute with respect to the current spec Version()
+	Name() string
+	// Version of the spec that this attribute belongs to
+	Version() Version
+	// Get the value of this attribute from an event context
+	Get(ce.EventContextReader) interface{}
+	// Set the value of this attribute on an event context
+	Set(ce.EventContextWriter, interface{}) error
+}
+
+// accessor provides Kind, Get, Set.
+type accessor interface {
+	Kind() Kind
+	Get(ce.EventContextReader) interface{}
+	Set(ce.EventContextWriter, interface{}) error
+}
+
+var acc = [nAttrs]accessor{
+	&aStr{aKind(ID), ce.EventContextReader.GetID, ce.EventContextWriter.SetID},
+	&aStr{aKind(Source), ce.EventContextReader.GetSource, ce.EventContextWriter.SetSource},
+	&aStr{aKind(SpecVersion), ce.EventContextReader.GetSpecVersion, ce.EventContextWriter.SetSpecVersion},
+	&aStr{aKind(Type), ce.EventContextReader.GetType, ce.EventContextWriter.SetType},
+	&aStr{aKind(DataContentType), ce.EventContextReader.GetDataContentType, ce.EventContextWriter.SetDataContentType},
+	&aStr{aKind(DataSchema), ce.EventContextReader.GetDataSchema, ce.EventContextWriter.SetDataSchema},
+	&aStr{aKind(Subject), ce.EventContextReader.GetSubject, ce.EventContextWriter.SetSubject},
+	&aTime{aKind(Time), ce.EventContextReader.GetTime, ce.EventContextWriter.SetTime},
+}
+
+// aKind implements Kind()
+type aKind Kind
+
+func (kind aKind) Kind() Kind { return Kind(kind) }
+
+type aStr struct {
+	aKind
+	get func(ce.EventContextReader) string
+	set func(ce.EventContextWriter, string) error
+}
+
+func (a *aStr) Get(c ce.EventContextReader) interface{} {
+	if s := a.get(c); s != "" {
+		return s
+	}
+	return nil // Treat blank as missing
+}
+
+func (a *aStr) Set(c ce.EventContextWriter, v interface{}) error {
+	if v, ok := v.(string); ok {
+		return a.set(c, v)
+	}
+	return ErrBadType
+}
+
+type aTime struct {
+	aKind
+	get func(ce.EventContextReader) time.Time
+	set func(ce.EventContextWriter, time.Time) error
+}
+
+func (a *aTime) Get(c ce.EventContextReader) interface{} {
+	if v := a.get(c); !v.IsZero() {
+		return v
+	}
+	return nil // Treat zero time as missing.
+}
+
+func (a *aTime) Set(c ce.EventContextWriter, v interface{}) error {
+	if v, ok := v.(time.Time); ok {
+		return a.set(c, v)
+	}
+	return ErrBadType
+}
diff --git a/pkg/binding/spec/doc.go b/pkg/binding/spec/doc.go
new file mode 100644
index 0000000..38d6fdd
--- /dev/null
+++ b/pkg/binding/spec/doc.go
@@ -0,0 +1,8 @@
+/*
+Package spec provides spec-version metadata.
+
+For use by code that maps events using (prefixed) attribute name strings.
+Supports handling multiple spec versions uniformly.
+
+*/
+package spec
diff --git a/pkg/binding/spec/spec.go b/pkg/binding/spec/spec.go
new file mode 100644
index 0000000..c034725
--- /dev/null
+++ b/pkg/binding/spec/spec.go
@@ -0,0 +1,166 @@
+package spec
+
+import (
+	"fmt"
+
+	ce "github.com/cloudevents/sdk-go/pkg/cloudevents"
+)
+
+// Version provides meta-data for a single spec-version.
+type Version interface {
+	// String name of the version, e.g. "1.0"
+	String() string
+	// Prefix for attribute names.
+	Prefix() string
+	// Attribute looks up a prefixed attribute name.
+	// Returns nil if not found.
+	Attribute(name string) Attribute
+	// Attributes returns all the context attributes for this version.
+	Attributes() []Attribute
+	// NewContext returns a new context for this version.
+	NewContext() ce.EventContext
+	// Convert translates a context to this version.
+	Convert(ce.EventContextConverter) ce.EventContext
+}
+
+// Versions contains all known versions with the same attribute prefix.
+type Versions struct {
+	prefix  string
+	all     []Version
+	m       map[string]Version
+	svnames []string
+}
+
+// Versions returns the list of all known versions, most recent first.
+func (vs *Versions) Versions() []Version { return vs.all }
+
+// Version returns the named version.
+func (vs *Versions) Version(name string) (Version, error) {
+	if v := vs.m[name]; v != nil {
+		return v, nil
+	}
+	return nil, fmt.Errorf("invalid spec version %#v", name)
+}
+
+// Latest returns the latest Version
+func (vs *Versions) Latest() Version { return vs.all[0] }
+
+// SpecVersionNames returns distinct names of the specversion
+// attribute used in all versions, newest first.
+// Names are prefixed.
+func (vs *Versions) SpecVersionNames() []string { return vs.svnames }
+
+// Prefix is the attribute name prefix.
+func (vs *Versions) Prefix() string { return vs.prefix }
+
+type attribute struct {
+	accessor
+	name    string
+	version Version
+}
+
+func (a *attribute) Name() string     { return a.name }
+func (a *attribute) Version() Version { return a.version }
+
+type version struct {
+	prefix  string
+	context ce.EventContext
+	convert func(ce.EventContextConverter) ce.EventContext
+	attrMap map[string]Attribute
+	attrs   []Attribute
+}
+
+func (v *version) Attribute(name string) Attribute { return v.attrMap[name] }
+func (v *version) Attributes() []Attribute         { return v.attrs }
+func (v *version) String() string                  { return v.context.GetSpecVersion() }
+func (v *version) Prefix() string                  { return v.prefix }
+func (v *version) NewContext() ce.EventContext     { return v.context.Clone() }
+
+func (v *version) Convert(c ce.EventContextConverter) ce.EventContext { return v.convert(c) }
+
+func newVersion(
+	prefix string,
+	context ce.EventContext,
+	convert func(ce.EventContextConverter) ce.EventContext,
+	attrs ...*attribute,
+) *version {
+	v := &version{
+		prefix:  prefix,
+		context: context,
+		convert: convert,
+		attrMap: map[string]Attribute{},
+		attrs:   make([]Attribute, len(attrs)),
+	}
+	for i, a := range attrs {
+		a.name = prefix + a.name
+		a.version = v
+		v.attrs[i] = a
+		v.attrMap[a.name] = a
+	}
+	return v
+}
+
+// WithPrefix returns a set of versions with prefix added to all attribute names.
+func WithPrefix(prefix string) *Versions {
+	attr := func(name string, kind Kind) *attribute {
+		return &attribute{accessor: acc[kind], name: name}
+	}
+	vs := &Versions{
+		m: map[string]Version{},
+		svnames: []string{
+			prefix + "specversion",
+			prefix + "cloudEventsVersion",
+		},
+		all: []Version{
+			newVersion(prefix, ce.EventContextV1{}.AsV1(),
+				func(c ce.EventContextConverter) ce.EventContext { return c.AsV1() },
+				attr("id", ID),
+				attr("source", Source),
+				attr("specversion", SpecVersion),
+				attr("type", Type),
+				attr("datacontenttype", DataContentType),
+				attr("dataschema", DataSchema),
+				attr("subject", Subject),
+				attr("time", Time),
+			),
+			newVersion(prefix, ce.EventContextV03{}.AsV03(),
+				func(c ce.EventContextConverter) ce.EventContext { return c.AsV03() },
+				attr("specversion", SpecVersion),
+				attr("type", Type),
+				attr("source", Source),
+				attr("schemaurl", DataSchema),
+				attr("subject", Subject),
+				attr("id", ID),
+				attr("time", Time),
+				attr("datacontenttype", DataContentType),
+			),
+			newVersion(prefix, ce.EventContextV02{}.AsV02(),
+				func(c ce.EventContextConverter) ce.EventContext { return c.AsV02() },
+				attr("specversion", SpecVersion),
+				attr("type", Type),
+				attr("source", Source),
+				attr("schemaurl", DataSchema),
+				attr("id", ID),
+				attr("time", Time),
+				attr("contenttype", DataContentType),
+			),
+			newVersion(prefix, ce.EventContextV01{}.AsV01(),
+				func(c ce.EventContextConverter) ce.EventContext { return c.AsV01() },
+				attr("cloudEventsVersion", SpecVersion),
+				attr("eventType", Type),
+				attr("source", Source),
+				attr("schemaURL", DataSchema),
+				attr("eventID", ID),
+				attr("eventTime", Time),
+				attr("contentType", DataContentType),
+			),
+		},
+	}
+	for _, v := range vs.all {
+		vs.m[v.String()] = v
+	}
+	return vs
+}
+
+// New returns a set of versions
+func New() *Versions { return WithPrefix("") }
diff --git a/pkg/binding/transport.go b/pkg/binding/transport.go
index 81c239b..6e9fa80 100644
--- a/pkg/binding/transport.go
+++ b/pkg/binding/transport.go
@@ -2,51 +2,57 @@ package binding
 
 import (
 	"context"
+	"io"
 
-	"github.com/cloudevents/sdk-go/pkg/cloudevents"
+	ce "github.com/cloudevents/sdk-go/pkg/cloudevents"
 	"github.com/cloudevents/sdk-go/pkg/cloudevents/transport"
 )
 
-// Transport implements the transport.Transport interface using a
-// Sender and Receiver.
+// Transport implements transport.Transport using a Sender and Receiver.
 type Transport struct {
 	Sender   Sender
 	Receiver Receiver
 	handler  transport.Receiver
 }
 
-var _ transport.Transport = (*Transport)(nil) // Test it conforms to the interface
+var _ transport.Transport = (*Transport)(nil) // Conforms to the interface
 
 func NewTransport(s Sender, r Receiver) *Transport {
 	return &Transport{Sender: s, Receiver: r}
 }
 
-func (t *Transport) Send(ctx context.Context, e cloudevents.Event) (context.Context, *cloudevents.Event, error) {
+func (t *Transport) Send(ctx context.Context, e ce.Event) (context.Context, *ce.Event, error) {
 	return ctx, nil, t.Sender.Send(ctx, EventMessage(e))
 }
 
 func (t *Transport) SetReceiver(r transport.Receiver) { t.handler = r }
 
-func (t *Transport) StartReceiver(ctx context.Context) error {
+func (t *Transport) StartReceiver(ctx context.Context) (reterr error) {
 	for {
-		if m, err := t.Receiver.Receive(ctx); err != nil {
+		m, err := t.Receiver.Receive(ctx)
+		if err == io.EOF { // Normal close
+			return nil
+		} else if err != nil {
 			return err
-		} else if e, err := m.Event(); err != nil {
-			m.Finish(err)
-			return err
-		} else if err := t.handler.Receive(ctx, e, nil); err != nil {
-			m.Finish(err)
+		}
+		if err := t.handle(ctx, m); err != nil {
 			return err
-		} else {
-			m.Finish(nil)
 		}
 	}
 }
 
-func (t *Transport) SetConverter(transport.Converter) {
-	// TODO(alanconway) Can we separate Converter from the base transport interface?
+func (t *Transport) handle(ctx context.Context, m Message) (err error) {
+	defer func() {
+		if err2 := m.Finish(err); err == nil {
+			err = err2
+		}
+	}()
+	e, err := m.Event()
+	if err != nil {
+		return err
+	}
+	return t.handler.Receive(ctx, e, nil)
 }
 
-func (t *Transport) HasConverter() bool {
-	return false
-}
+func (t *Transport) SetConverter(transport.Converter) {}
+func (t *Transport) HasConverter() bool               { return false }
diff --git a/pkg/cloudevents/event.go b/pkg/cloudevents/event.go
index d0db72d..e00b9dc 100644
--- a/pkg/cloudevents/event.go
+++ b/pkg/cloudevents/event.go
@@ -11,8 +11,8 @@ import (
 type Event struct {
 	Context     EventContext
 	Data        interface{}
-	DataBase64  []byte
 	DataEncoded bool
+	DataBinary  bool
 }
 
 const (
diff --git a/pkg/cloudevents/event_data.go b/pkg/cloudevents/event_data.go
index 621b665..26ce70e 100644
--- a/pkg/cloudevents/event_data.go
+++ b/pkg/cloudevents/event_data.go
@@ -14,6 +14,38 @@ import (
 
 // SetData implements EventWriter.SetData
 func (e *Event) SetData(obj interface{}) error {
+	if e.SpecVersion() != CloudEventsVersionV1 {
+		return e.legacySetData(obj)
+	}
+
+	// Version 1.0 and above.
+
+	// TODO: we will have to be smarter about how data relates to media type.
+	//  but the issue is we can not just encode data anymore without understanding
+	//  what the encoding will be on the outbound event. Structured will use
+	//  data_base64, binary will not (if the transport supports binary mode).
+
+	// TODO: look at content encoding too.
+
+	switch obj.(type) {
+	case []byte:
+		e.Data = obj
+		e.DataEncoded = true
+		e.DataBinary = true
+	default:
+		data, err := datacodec.Encode(context.Background(), e.DataMediaType(), obj)
+		if err != nil {
+			return err
+		}
+		e.Data = data
+		e.DataEncoded = true
+		e.DataBinary = false
+	}
+
+	return nil
+}
+
+func (e *Event) legacySetData(obj interface{}) error {
 	data, err := datacodec.Encode(context.Background(), e.DataMediaType(), obj)
 	if err != nil {
 		return err
@@ -91,9 +123,13 @@ func (e Event) DataAs(data interface{}) error { // TODO: Clean this function up
 		obj = buf[:n]
 	}
 
-	mediaType, err := e.Context.GetDataMediaType()
-	if err != nil {
-		return err
+	mediaType := ""
+	if e.Context.GetDataContentType() != "" {
+		var err error
+		mediaType, err = e.Context.GetDataMediaType()
+		if err != nil {
+			return err
+		}
 	}
 	return datacodec.Decode(context.Background(), mediaType, obj, data)
 }
diff --git a/pkg/cloudevents/event_marshal.go b/pkg/cloudevents/event_marshal.go
index 1a71296..f7c864f 100644
--- a/pkg/cloudevents/event_marshal.go
+++ b/pkg/cloudevents/event_marshal.go
@@ -2,7 +2,9 @@ package cloudevents
 
 import (
 	"context"
+	"encoding/base64"
 	"encoding/json"
+	"errors"
 	"fmt"
 	"strconv"
 	"strings"
@@ -20,7 +22,17 @@ func (e Event) MarshalJSON() ([]byte, error) {
 		return nil, err
 	}
 
-	b, err := JsonEncode(e)
+	var b []byte
+	var err error
+
+	switch e.SpecVersion() {
+	case CloudEventsVersionV01, CloudEventsVersionV02, CloudEventsVersionV03:
+		b, err = JsonEncodeLegacy(e)
+	case CloudEventsVersionV1:
+		b, err = JsonEncode(e)
+	default:
+		return nil, fmt.Errorf("unnknown spec version: %q", e.SpecVersion())
+	}
 
 	// Report the observable
 	if err != nil {
@@ -92,24 +104,21 @@ func versionFromRawMessage(raw map[string]json.RawMessage) string {
 
 // JsonEncode
 func JsonEncode(e Event) ([]byte, error) {
-	if e.DataContentType() == "" {
-		e.SetDataContentType(ApplicationJSON)
+	data, err := e.DataBytes()
+	if err != nil {
+		return nil, err
 	}
+	return jsonEncode(e.Context, data, e.DataBinary)
+}
+
+// JsonEncodeLegacy
+func JsonEncodeLegacy(e Event) ([]byte, error) {
 	var data []byte
 	isBase64 := e.Context.DeprecatedGetDataContentEncoding() == Base64
-	if e.DataBase64 != nil {
-		isBase64 = true
-		var err error
-		data, err = json.Marshal(e.DataBase64)
-		if err != nil {
-			return nil, err
-		}
-	} else {
-		var err error
-		data, err = e.DataBytes()
-		if err != nil {
-			return nil, err
-		}
+	var err error
+	data, err = e.DataBytes()
+	if err != nil {
+		return nil, err
 	}
 	return jsonEncode(e.Context, data, isBase64)
 }
@@ -145,6 +154,9 @@ func jsonEncode(ctx EventContextReader, data []byte, isBase64 bool) ([]byte, err
 			var dataKey string
 			if ctx.GetSpecVersion() == CloudEventsVersionV1 {
 				dataKey = "data_base64"
+				buf := make([]byte, base64.StdEncoding.EncodedLen(len(data)))
+				base64.StdEncoding.Encode(buf, data)
+				data = buf
 			} else {
 				dataKey = "data"
 			}
@@ -312,8 +324,14 @@ func (e *Event) JsonDecodeV1(body []byte, raw map[string]json.RawMessage) error
 	}
 
 	e.Context = &ec
-	e.Data = data
-	e.DataBase64 = dataBase64
+	if data != nil && dataBase64 != nil {
+		return errors.New("parsing error: JSON decoder found both 'data', and 'data_base64' in JSON payload")
+	}
+	if data != nil {
+		e.Data = data
+	} else if dataBase64 != nil {
+		e.Data = dataBase64
+	}
 	e.DataEncoded = data != nil
 
 	return nil
diff --git a/pkg/cloudevents/eventcontext_v03.go b/pkg/cloudevents/eventcontext_v03.go
index 1ca8c37..2e714d1 100644
--- a/pkg/cloudevents/eventcontext_v03.go
+++ b/pkg/cloudevents/eventcontext_v03.go
@@ -83,9 +83,9 @@ func (ec *EventContextV03) SetExtension(name string, value interface{}) error {
 	if value == nil {
 		delete(ec.Extensions, name)
 	} else {
-		v, err := types.ValueOf(value)
+		v, err := types.Validate(value)
 		if err == nil {
-			ec.Extensions[name] = v.Interface()
+			ec.Extensions[name] = v
 		}
 		return err
 	}
diff --git a/pkg/cloudevents/transport/amqp/codec.go b/pkg/cloudevents/transport/amqp/codec.go
index e41d066..80949bf 100644
--- a/pkg/cloudevents/transport/amqp/codec.go
+++ b/pkg/cloudevents/transport/amqp/codec.go
@@ -197,10 +197,10 @@ func (c Codec) fromHeaders(h map[string]interface{}, event *cloudevents.Event) e
 	delete(h, prefix+"source")
 
 	if t, ok := h[prefix+"time"].(string); ok {
-		if timestamp := types.ParseTimestamp(t); timestamp != nil {
-			if err := ec.SetTime(timestamp.Time); err != nil {
-				return err
-			}
+		if timestamp, err := types.ParseTimestamp(t); err != nil {
+			return err
+		} else if err := ec.SetTime(timestamp.Time); err != nil {
+			return err
 		}
 	}
 	delete(h, prefix+"time")
diff --git a/pkg/cloudevents/transport/http/codec_v01.go b/pkg/cloudevents/transport/http/codec_v01.go
index cf680db..435ba93 100644
--- a/pkg/cloudevents/transport/http/codec_v01.go
+++ b/pkg/cloudevents/transport/http/codec_v01.go
@@ -119,13 +119,8 @@ func (v CodecV01) toHeaders(ec *cloudevents.EventContextV01) (http.Header, error
 	if ec.SchemaURL != nil {
 		h["CE-DataSchema"] = []string{ec.SchemaURL.String()}
 	}
-	if ec.ContentType != nil {
+	if ec.ContentType != nil && *ec.ContentType != "" {
 		h.Set("Content-Type", *ec.ContentType)
-	} else {
-		// in binary v0.1, the Content-Type header is tied to ec.ContentType
-		// This was later found to be an issue with the spec, but yolo.
-		// TODO: not sure what the default should be?
-		h.Set("Content-Type", cloudevents.ApplicationJSON)
 	}
 
 	// Regarding Extensions, v0.1 Spec says the following:
@@ -182,7 +177,11 @@ func (v CodecV01) fromHeaders(h http.Header) (cloudevents.EventContextV01, error
 	if source != nil {
 		ec.Source = *source
 	}
-	ec.EventTime = types.ParseTimestamp(h.Get("CE-EventTime"))
+	var err error
+	ec.EventTime, err = types.ParseTimestamp(h.Get("CE-EventTime"))
+	if err != nil {
+		return ec, err
+	}
 	h.Del("CE-EventTime")
 	etv := h.Get("CE-EventTypeVersion")
 	h.Del("CE-EventTypeVersion")
@@ -192,7 +191,9 @@ func (v CodecV01) fromHeaders(h http.Header) (cloudevents.EventContextV01, error
 	ec.SchemaURL = types.ParseURLRef(h.Get("CE-DataSchema"))
 	h.Del("CE-DataSchema")
 	et := h.Get("Content-Type")
-	ec.ContentType = &et
+	if et != "" {
+		ec.ContentType = &et
+	}
 
 	extensions := make(map[string]interface{})
 	for k, v := range h {
diff --git a/pkg/cloudevents/transport/http/codec_v02.go b/pkg/cloudevents/transport/http/codec_v02.go
index 8e7283b..aeb67c0 100644
--- a/pkg/cloudevents/transport/http/codec_v02.go
+++ b/pkg/cloudevents/transport/http/codec_v02.go
@@ -114,12 +114,8 @@ func (v CodecV02) toHeaders(ec *cloudevents.EventContextV02) (http.Header, error
 	if ec.SchemaURL != nil {
 		h.Set("ce-schemaurl", ec.SchemaURL.String())
 	}
-	if ec.ContentType != nil {
+	if ec.ContentType != nil && *ec.ContentType != "" {
 		h.Set("Content-Type", *ec.ContentType)
-	} else {
-		// in binary v0.2, the Content-Type header is tied to ec.ContentType
-		// This was later found to be an issue with the spec, but yolo.
-		h.Set("Content-Type", cloudevents.ApplicationJSON)
 	}
 	for k, v := range ec.Extensions {
 		// Per spec, map-valued extensions are converted to a list of headers as:
@@ -191,7 +187,11 @@ func (v CodecV02) fromHeaders(h http.Header) (cloudevents.EventContextV02, error
 	}
 	h.Del("ce-source")
 
-	ec.Time = types.ParseTimestamp(h.Get("ce-time"))
+	var err error
+	ec.Time, err = types.ParseTimestamp(h.Get("ce-time"))
+	if err != nil {
+		return ec, err
+	}
 	h.Del("ce-time")
 
 	ec.SchemaURL = types.ParseURLRef(h.Get("ce-schemaurl"))
diff --git a/pkg/cloudevents/transport/http/codec_v03.go b/pkg/cloudevents/transport/http/codec_v03.go
index f481205..b2b3c87 100644
--- a/pkg/cloudevents/transport/http/codec_v03.go
+++ b/pkg/cloudevents/transport/http/codec_v03.go
@@ -122,12 +122,8 @@ func (v CodecV03) toHeaders(ec *cloudevents.EventContextV03) (http.Header, error
 	if ec.SchemaURL != nil {
 		h.Set("ce-schemaurl", ec.SchemaURL.String())
 	}
-	if ec.DataContentType != nil {
+	if ec.DataContentType != nil && *ec.DataContentType != "" {
 		h.Set("Content-Type", *ec.DataContentType)
-	} else {
-		// in binary v0.3, the Content-Type header is tied to ec.ContentType
-		// This was later found to be an issue with the spec, but yolo.
-		h.Set("Content-Type", cloudevents.ApplicationJSON)
 	}
 	if ec.DataContentEncoding != nil {
 		h.Set("ce-datacontentencoding", *ec.DataContentEncoding)
@@ -222,7 +218,11 @@ func (v CodecV03) fromHeaders(h http.Header) (cloudevents.EventContextV03, error
 	}
 	h.Del("ce-subject")
 
-	ec.Time = types.ParseTimestamp(h.Get("ce-time"))
+	var err error
+	ec.Time, err = types.ParseTimestamp(h.Get("ce-time"))
+	if err != nil {
+		return ec, err
+	}
 	h.Del("ce-time")
 
 	ec.SchemaURL = types.ParseURLRef(h.Get("ce-schemaurl"))
diff --git a/pkg/cloudevents/transport/http/codec_v1.go b/pkg/cloudevents/transport/http/codec_v1.go
index 8e52040..4ebe742 100644
--- a/pkg/cloudevents/transport/http/codec_v1.go
+++ b/pkg/cloudevents/transport/http/codec_v1.go
@@ -121,10 +121,8 @@ func (v CodecV1) toHeaders(ec *cloudevents.EventContextV1) (http.Header, error)
 	if ec.DataSchema != nil {
 		h.Set("ce-dataschema", ec.DataSchema.String())
 	}
-	if ec.DataContentType != nil {
+	if ec.DataContentType != nil && *ec.DataContentType != "" {
 		h.Set("Content-Type", *ec.DataContentType)
-	} else {
-		h.Set("Content-Type", cloudevents.ApplicationJSON)
 	}
 
 	for k, v := range ec.Extensions {
@@ -194,7 +192,11 @@ func (v CodecV1) fromHeaders(h http.Header) (cloudevents.EventContextV1, error)
 	}
 	h.Del("ce-subject")
 
-	ec.Time = types.ParseTimestamp(h.Get("ce-time"))
+	var err error
+	ec.Time, err = types.ParseTimestamp(h.Get("ce-time"))
+	if err != nil {
+		return ec, err
+	}
 	h.Del("ce-time")
 
 	ec.DataSchema = types.ParseURI(h.Get("ce-dataschema"))
diff --git a/pkg/cloudevents/transport/pubsub/codec.go b/pkg/cloudevents/transport/pubsub/codec.go
index 71e2526..ab9249c 100644
--- a/pkg/cloudevents/transport/pubsub/codec.go
+++ b/pkg/cloudevents/transport/pubsub/codec.go
@@ -187,10 +187,10 @@ func (c Codec) fromAttributes(a map[string]string, event *cloudevents.Event) err
 	delete(a, prefix+"source")
 
 	if t := a[prefix+"time"]; t != "" {
-		if timestamp := types.ParseTimestamp(t); timestamp != nil {
-			if err := ec.SetTime(timestamp.Time); err != nil {
-				return err
-			}
+		if timestamp, err := types.ParseTimestamp(t); err != nil {
+			return err
+		} else if err := ec.SetTime(timestamp.Time); err != nil {
+			return err
 		}
 	}
 	delete(a, prefix+"time")
diff --git a/pkg/cloudevents/types/doc.go b/pkg/cloudevents/types/doc.go
index f9ea711..0e13426 100644
--- a/pkg/cloudevents/types/doc.go
+++ b/pkg/cloudevents/types/doc.go
@@ -1,12 +1,14 @@
 /*
 Package types implements the CloudEvents type system.
 
-CloudEvents defines a small set of abstract types for all event context
-attribute values, including custom extension attributes. Each CloudEvents type
-has a corresponding native Go type and a canonical string encoding.
+CloudEvents defines a set of abstract types for event context attributes. Each
+type has a corresponding native Go type and a canonical string encoding.  The
+native Go types used to represent the CloudEvents types are:
+
+    bool, int32, string, []byte, *url.URL, time.Time
 
  +----------------+----------------+-----------------------------------+
- |CloudEvents Type|Native Type     |Convertible From Types             |
+ |CloudEvents Type|Native Type     |Convertible From                   |
  +================+================+===================================+
  |Bool            |bool            |bool                               |
  +----------------+----------------+-----------------------------------+
@@ -25,16 +27,16 @@ has a corresponding native Go type and a canonical string encoding.
  |Timestamp       |time.Time       |time.Time, types.Timestamp         |
  +----------------+----------------+-----------------------------------+
 
-The native types used to represent CloudEvents types are:
-
-    bool, int32, string, []byte, *url.URL, time.Time
+Extension attributes may be stored as a native type or a canonical string.  The
+To<Type> functions will convert to the desired <Type> from any convertible type
+or from the canonical string form.
 
-This package provides Parse... and Format... functions to convert to/from
-canonical strings. Use standard url.URL.String() and url.Parse() for URL
-values, the canonical string for a string is the string itself.
+The Parse<Type> and Format<Type> functions convert native types to/from
+canonical strings.
 
-It provides To... functions to convert any extension attribute value, in native
-or string form, to the expected type.
+Note are no Parse or Format functions for URL or string. For URL use the
+standard url.Parse() and url.URL.String(). The canonical string format of a
+string is the string itself.
 
 */
 package types
diff --git a/pkg/cloudevents/types/timestamp.go b/pkg/cloudevents/types/timestamp.go
index 6534aac..3ae1c7d 100644
--- a/pkg/cloudevents/types/timestamp.go
+++ b/pkg/cloudevents/types/timestamp.go
@@ -16,15 +16,12 @@ type Timestamp struct {
 }
 
 // ParseTimestamp attempts to parse the given time assuming RFC3339 layout
-func ParseTimestamp(t string) *Timestamp {
-	if t == "" {
-		return nil
+func ParseTimestamp(s string) (*Timestamp, error) {
+	if s == "" {
+		return nil, nil
 	}
-	timestamp, err := time.Parse(time.RFC3339Nano, t)
-	if err != nil {
-		return nil
-	}
-	return &Timestamp{Time: timestamp}
+	tt, err := ParseTime(s)
+	return &Timestamp{Time: tt}, err
 }
 
 // MarshalJSON implements a custom json marshal method used when this type is
@@ -33,8 +30,7 @@ func (t *Timestamp) MarshalJSON() ([]byte, error) {
 	if t == nil || t.IsZero() {
 		return []byte(`""`), nil
 	}
-	rfc3339 := fmt.Sprintf("%q", t.UTC().Format(time.RFC3339Nano))
-	return []byte(rfc3339), nil
+	return []byte(fmt.Sprintf("%q", t)), nil
 }
 
 // UnmarshalJSON implements the json unmarshal method used when this type is
@@ -44,10 +40,9 @@ func (t *Timestamp) UnmarshalJSON(b []byte) error {
 	if err := json.Unmarshal(b, &timestamp); err != nil {
 		return err
 	}
-	if pt := ParseTimestamp(timestamp); pt != nil {
-		*t = *pt
-	}
-	return nil
+	var err error
+	t.Time, err = ParseTime(timestamp)
+	return err
 }
 
 // MarshalXML implements a custom xml marshal method used when this type is
@@ -56,8 +51,7 @@ func (t *Timestamp) MarshalXML(e *xml.Encoder, start xml.StartElement) error {
 	if t == nil || t.IsZero() {
 		return e.EncodeElement(nil, start)
 	}
-	v := t.UTC().Format(time.RFC3339Nano)
-	return e.EncodeElement(v, start)
+	return e.EncodeElement(t.String(), start)
 }
 
 // UnmarshalXML implements the xml unmarshal method used when this type is
@@ -67,17 +61,10 @@ func (t *Timestamp) UnmarshalXML(d *xml.Decoder, start xml.StartElement) error {
 	if err := d.DecodeElement(&timestamp, &start); err != nil {
 		return err
 	}
-	if pt := ParseTimestamp(timestamp); pt != nil {
-		*t = *pt
-	}
-	return nil
+	var err error
+	t.Time, err = ParseTime(timestamp)
+	return err
 }
 
-// String outputs the time using layout RFC3339.
-func (t *Timestamp) String() string {
-	if t == nil {
-		return time.Time{}.UTC().Format(time.RFC3339Nano)
-	}
-
-	return t.UTC().Format(time.RFC3339Nano)
-}
+// String outputs the time using RFC3339 format.
+func (t Timestamp) String() string { return FormatTime(t.Time) }
diff --git a/pkg/cloudevents/types/value.go b/pkg/cloudevents/types/value.go
index fa6e336..803be2f 100644
--- a/pkg/cloudevents/types/value.go
+++ b/pkg/cloudevents/types/value.go
@@ -42,15 +42,24 @@ func ParseInteger(v string) (int32, error) {
 func ParseBinary(v string) ([]byte, error) { return base64.StdEncoding.DecodeString(v) }
 
 // ParseTime parse canonical string format: RFC3339 with nanoseconds
-func ParseTime(v string) (time.Time, error) { return time.Parse(time.RFC3339Nano, v) }
+func ParseTime(v string) (time.Time, error) {
+	t, err := time.Parse(time.RFC3339Nano, v)
+	if err != nil {
+		err := convertErr(time.Time{}, v)
+		err.extra = ": not in RFC3339 format"
+		return time.Time{}, err
+	}
+	return t, nil
+}
 
 // Format returns the canonical string format of v, where v can be
 // any type that is convertible to a CloudEvents type.
 func Format(v interface{}) (string, error) {
 	v, err := Validate(v)
-	switch v := v.(type) {
-	case nil:
+	if err != nil {
 		return "", err
+	}
+	switch v := v.(type) {
 	case bool:
 		return FormatBool(v), nil
 	case int32:
@@ -125,26 +134,32 @@ func Validate(v interface{}) (interface{}, error) {
 // ToBool accepts a bool value or canonical "true"/"false" string.
 func ToBool(v interface{}) (bool, error) {
 	v, err := Validate(v)
+	if err != nil {
+		return false, err
+	}
 	switch v := v.(type) {
 	case bool:
 		return v, nil
 	case string:
 		return ParseBool(v)
 	default:
-		return false, convertErr("Bool", v, err)
+		return false, convertErr(true, v)
 	}
 }
 
 // ToInteger accepts any numeric value in int32 range, or canonical string.
 func ToInteger(v interface{}) (int32, error) {
 	v, err := Validate(v)
+	if err != nil {
+		return 0, err
+	}
 	switch v := v.(type) {
 	case int32:
 		return v, nil
 	case string:
 		return ParseInteger(v)
 	default:
-		return 0, convertErr("Integer", v, err)
+		return 0, convertErr(int32(0), v)
 	}
 }
 
@@ -154,30 +169,39 @@ func ToInteger(v interface{}) (int32, error) {
 // Format functions for that.
 func ToString(v interface{}) (string, error) {
 	v, err := Validate(v)
+	if err != nil {
+		return "", err
+	}
 	switch v := v.(type) {
 	case string:
 		return v, nil
 	default:
-		return "", convertErr("String", v, err)
+		return "", convertErr("", v)
 	}
 }
 
 // ToBinary returns a []byte value, decoding from base64 string if necessary.
 func ToBinary(v interface{}) ([]byte, error) {
 	v, err := Validate(v)
+	if err != nil {
+		return nil, err
+	}
 	switch v := v.(type) {
 	case []byte:
 		return v, nil
 	case string:
 		return base64.StdEncoding.DecodeString(v)
 	default:
-		return nil, convertErr("Binary", v, err)
+		return nil, convertErr([]byte(nil), v)
 	}
 }
 
 // ToURL returns a *url.URL value, parsing from string if necessary.
 func ToURL(v interface{}) (*url.URL, error) {
 	v, err := Validate(v)
+	if err != nil {
+		return nil, err
+	}
 	switch v := v.(type) {
 	case *url.URL:
 		return v, nil
@@ -188,13 +212,16 @@ func ToURL(v interface{}) (*url.URL, error) {
 		}
 		return u, nil
 	default:
-		return nil, convertErr("URL", v, err)
+		return nil, convertErr((*url.URL)(nil), v)
 	}
 }
 
 // ToTime returns a time.Time value, parsing from RFC3339 string if necessary.
 func ToTime(v interface{}) (time.Time, error) {
 	v, err := Validate(v)
+	if err != nil {
+		return time.Time{}, err
+	}
 	switch v := v.(type) {
 	case time.Time:
 		return v, nil
@@ -205,17 +232,29 @@ func ToTime(v interface{}) (time.Time, error) {
 		}
 		return ts, nil
 	default:
-		return time.Time{}, convertErr("Time", v, err)
+		return time.Time{}, convertErr(time.Time{}, v)
 	}
 }
 
-func convertErr(name string, v interface{}, err error) error {
-	if err != nil {
-		return err
-	}
-	return fmt.Errorf("cannot convert %T to %s", v, name)
+type ConvertErr struct {
+	// Value being converted
+	Value interface{}
+	// Type of attempted conversion
+	Type reflect.Type
+
+	extra string
+}
+
+func (e *ConvertErr) Error() string {
+	return fmt.Sprintf("cannot convert %#v to %s%s", e.Value, e.Type, e.extra)
+}
+
+func convertErr(target, v interface{}) *ConvertErr {
+	return &ConvertErr{Value: v, Type: reflect.TypeOf(target)}
 }
 
 func rangeErr(v interface{}) error {
-	return fmt.Errorf("%v is out of range for Integer", v)
+	e := convertErr(int32(0), v)
+	e.extra = ": out of range"
+	return e
 }


Test patch:
diff --git a/pkg/binding/example_implementing_test.go b/pkg/binding/example_implementing_test.go
index 2511db2..a6e4759 100644
--- a/pkg/binding/example_implementing_test.go
+++ b/pkg/binding/example_implementing_test.go
@@ -7,56 +7,68 @@ import (
 
 	"github.com/cloudevents/sdk-go/pkg/binding"
 	"github.com/cloudevents/sdk-go/pkg/cloudevents"
+	ce "github.com/cloudevents/sdk-go/pkg/cloudevents"
 	"github.com/cloudevents/sdk-go/pkg/cloudevents/transport"
 )
 
-// ExMessage is a json.RawMessage, which is just a byte slice
-// containing a JSON encoded event.
-type ExMessage struct{ json.RawMessage }
+// ExMessage is a json.RawMessage, a byte slice containing a JSON encoded event.
+// It implements binding.StructMessage
+//
+// Note: a good binding implementation should provide an easy way to convert
+// between the Message implementation and the "native" message format.
+// In this case it's as simple as:
+//
+//    native = ExMessage(impl)
+//    impl = json.RawMessage(native)
+//
+// For example in a HTTP binding it should be easy to convert between
+// the HTTP binding.Message implementation and net/http.Request and
+// Response types.  There are no interfaces for this conversion as it
+// requires the use of unknown types.
+type ExMessage json.RawMessage
 
 func (m ExMessage) Structured() (string, []byte) {
-	return cloudevents.ApplicationCloudEventsJSON, []byte(m.RawMessage)
+	return cloudevents.ApplicationCloudEventsJSON, []byte(m)
 }
 
 func (m ExMessage) Event() (e cloudevents.Event, err error) {
-	err = json.Unmarshal(m.RawMessage, &e)
+	err = json.Unmarshal(json.RawMessage(m), &e)
 	return e, err
 }
 
-func (m ExMessage) Finish(error) {}
+func (m ExMessage) Finish(error) error { return nil }
 
 // ExSender sends by writing JSON encoded events to an io.Writer
-type ExSender struct{ *json.Encoder }
+type ExSender struct{ encoder *json.Encoder }
 
 func NewExSender(w io.Writer) ExSender { return ExSender{json.NewEncoder(w)} }
 
 func (s ExSender) Send(ctx context.Context, m binding.Message) error {
-	if t, b := m.Structured(); t != "" {
-		// Fast case: if the Message is already structured JSON we can
-		// send it directly, no need to decode and re-encode. Encoding a
-		// json.RawMessage to a json.Encoder() is just a byte-buffer copy.
-		return s.Encode(json.RawMessage(b))
+	if f, b := m.Structured(); f == ce.ApplicationCloudEventsJSON {
+		// Fast case: Message is already structured JSON.
+		return s.encoder.Encode(json.RawMessage(b))
 	}
-	// Some other message encoding. Decode as a generic cloudevents.Event
-	// and then re-encode as JSON
+	// Some other message encoding. Decode as generic Event and re-encode.
 	if e, err := m.Event(); err != nil {
 		return err
-	} else if err := s.Encode(e); err != nil {
+	} else if err := s.encoder.Encode(&e); err != nil {
 		return err
 	}
 	return nil
 }
+func (s ExSender) Close(context.Context) error { return nil }
 
 // ExReceiver receives by reading JSON encoded events from an io.Reader
-type ExReceiver struct{ *json.Decoder }
+type ExReceiver struct{ decoder *json.Decoder }
 
 func NewExReceiver(r io.Reader) ExReceiver { return ExReceiver{json.NewDecoder(r)} }
 
-func (sr ExReceiver) Receive(context.Context) (binding.Message, error) {
-	var m ExMessage
-	err := sr.Decode(&m) // This is just a byte copy since m is a json.RawMessage
-	return m, err
+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
diff --git a/pkg/binding/example_using_test.go b/pkg/binding/example_using_test.go
index 6068560..90d028d 100644
--- a/pkg/binding/example_using_test.go
+++ b/pkg/binding/example_using_test.go
@@ -103,7 +103,6 @@ func Example_using() {
 	//   type: example.com/event
 	//   source: example.com/source
 	//   id: 0
-	//   contenttype: application/json
 	// Data,
 	//   "hello 0"
 	//
@@ -113,7 +112,6 @@ func Example_using() {
 	//   type: example.com/event
 	//   source: example.com/source
 	//   id: 1
-	//   contenttype: application/json
 	// Data,
 	//   "hello 1"
 	//
@@ -123,7 +121,6 @@ func Example_using() {
 	//   type: example.com/event
 	//   source: example.com/source
 	//   id: 2
-	//   contenttype: application/json
 	// Data,
 	//   "hello 2"
 }
diff --git a/pkg/binding/format/format_test.go b/pkg/binding/format/format_test.go
new file mode 100644
index 0000000..d45d667
--- /dev/null
+++ b/pkg/binding/format/format_test.go
@@ -0,0 +1,82 @@
+package format_test
+
+import (
+	"testing"
+
+	"github.com/cloudevents/sdk-go/pkg/binding/format"
+	ce "github.com/cloudevents/sdk-go/pkg/cloudevents"
+	"github.com/cloudevents/sdk-go/pkg/cloudevents/types"
+	"github.com/stretchr/testify/assert"
+)
+
+func TestJSON(t *testing.T) {
+	assert := assert.New(t)
+	e := ce.Event{
+		Context: ce.EventContextV03{
+			Type:   "type",
+			ID:     "id",
+			Source: *types.ParseURLRef("source"),
+		}.AsV03(),
+	}
+	assert.NoError(e.SetData("foo"))
+	b, err := format.JSON.Marshal(e)
+	assert.NoError(err)
+	assert.Equal(`{"data":"foo","datacontenttype":"application/json","id":"id","source":"source","specversion":"0.3","type":"type"}`, string(b))
+
+	var e2 ce.Event
+	assert.NoError(format.JSON.Unmarshal(b, &e2))
+	assert.Equal(e, e2)
+}
+
+func TestLookup(t *testing.T) {
+	assert := assert.New(t)
+	assert.Nil(format.Lookup("nosuch"))
+
+	f := format.Lookup(ce.ApplicationCloudEventsJSON)
+	assert.Equal(f.MediaType(), ce.ApplicationCloudEventsJSON)
+	assert.Equal(format.JSON, f)
+}
+
+func TestMarshalUnmarshal(t *testing.T) {
+	assert := assert.New(t)
+	e := ce.Event{
+		Context: ce.EventContextV03{
+			Type:   "type",
+			ID:     "id",
+			Source: *types.ParseURLRef("source"),
+		}.AsV03(),
+	}
+	assert.NoError(e.SetData("foo"))
+	b, err := format.Marshal(format.JSON.MediaType(), e)
+	assert.NoError(err)
+	assert.Equal(`{"data":"foo","datacontenttype":"application/json","id":"id","source":"source","specversion":"0.3","type":"type"}`, string(b))
+
+	var e2 ce.Event
+	assert.NoError(format.Unmarshal(format.JSON.MediaType(), b, &e2))
+	assert.Equal(e, e2)
+
+	_, err = format.Marshal("nosuchformat", e)
+	assert.EqualError(err, "unknown event format media-type \"nosuchformat\"")
+	err = format.Unmarshal("nosuchformat", nil, &e)
+	assert.EqualError(err, "unknown event format media-type \"nosuchformat\"")
+}
+
+type dummyFormat struct{}
+
+func (dummyFormat) MediaType() string                     { return "dummy" }
+func (dummyFormat) Marshal(ce.Event) ([]byte, error)      { return []byte("dummy!"), nil }
+func (dummyFormat) Unmarshal(b []byte, e *ce.Event) error { e.Data = "undummy!"; return nil }
+
+func TestAdd(t *testing.T) {
+	assert := assert.New(t)
+	format.Add(dummyFormat{})
+	assert.Equal(dummyFormat{}, format.Lookup("dummy"))
+
+	e := ce.Event{}
+	b, err := format.Marshal("dummy", e)
+	assert.NoError(err)
+	assert.Equal("dummy!", string(b))
+	err = format.Unmarshal("dummy", b, &e)
+	assert.NoError(err)
+	assert.Equal("undummy!", e.Data)
+}
diff --git a/pkg/binding/message_test.go b/pkg/binding/message_test.go
new file mode 100644
index 0000000..3061012
--- /dev/null
+++ b/pkg/binding/message_test.go
@@ -0,0 +1,72 @@
+package binding_test
+
+import (
+	"net/url"
+	"testing"
+
+	"github.com/cloudevents/sdk-go/pkg/binding"
+	"github.com/cloudevents/sdk-go/pkg/binding/format"
+	ce "github.com/cloudevents/sdk-go/pkg/cloudevents"
+	"github.com/cloudevents/sdk-go/pkg/cloudevents/types"
+	"github.com/stretchr/testify/assert"
+)
+
+var testEvent = ce.Event{
+	Data:    "data",
+	Context: ce.EventContextV03{Source: types.URLRef{URL: url.URL{Path: "source"}}, ID: "id", Type: "type"}.AsV03(),
+}
+
+var testJSON = `{"data":"data","datacontenttype":"application/json","id":"id","source":"source","specversion":"0.3","type":"type"}`
+
+func TestEventMessage(t *testing.T) {
+	assert := assert.New(t)
+	m, err := binding.BinaryEncoder{}.Encode(testEvent)
+	assert.NoError(err)
+	f, b := m.Structured()
+	assert.Empty(f)
+	assert.Empty(b)
+	e, err := m.Event()
+	assert.NoError(err)
+	assert.Equal(testEvent, e)
+}
+
+func TestStructMessage(t *testing.T) {
+	assert := assert.New(t)
+	m, err := binding.StructEncoder{Format: format.JSON}.Encode(testEvent)
+	assert.NoError(err)
+	f, b := m.Structured()
+	assert.Equal(ce.ApplicationCloudEventsJSON, f)
+	assert.Equal(testJSON, string(b))
+	e, err := m.Event()
+	assert.NoError(err)
+	assert.Equal(testEvent.Context, e.Context)
+	var s string
+	assert.NoError(e.DataAs(&s))
+	assert.Equal(testEvent.Data, s)
+
+	_, err = binding.StructMessage{Format: "nosuch"}.Event()
+	assert.EqualError(err, "unknown event format media-type \"nosuch\"")
+}
+
+type dummyFormat struct{}
+
+func (dummyFormat) MediaType() string                     { return "dummy" }
+func (dummyFormat) Marshal(ce.Event) ([]byte, error)      { return []byte("dummy!"), nil }
+func (dummyFormat) Unmarshal(b []byte, e *ce.Event) error { e.Data = "undummy!"; return nil }
+
+func TestStructured(t *testing.T) {
+	sm := binding.StructMessage{Format: format.JSON.MediaType(), Bytes: []byte(testJSON)}
+	b, err := binding.Structured(sm, format.JSON)
+	assert.NoError(t, err)
+	assert.Equal(t, &sm.Bytes, &b) // Not just equal but at the same address.
+
+	d := dummyFormat{}
+	b, err = binding.Structured(sm, d)
+	assert.NoError(t, err)
+	assert.Equal(t, "dummy!", string(b)) // Reformat as dummy
+
+	bm := binding.EventMessage(testEvent)
+	b, err = binding.Structured(bm, format.JSON)
+	assert.NoError(t, err)
+	assert.Equal(t, sm.Bytes, b) // Same bytes
+}
diff --git a/pkg/binding/spec/attributes_test.go b/pkg/binding/spec/attributes_test.go
new file mode 100644
index 0000000..5d908a7
--- /dev/null
+++ b/pkg/binding/spec/attributes_test.go
@@ -0,0 +1,93 @@
+package spec_test
+
+import (
+	"testing"
+	"time"
+
+	"github.com/cloudevents/sdk-go/pkg/binding/spec"
+	"github.com/google/go-cmp/cmp"
+	"github.com/stretchr/testify/assert"
+)
+
+func TestAttributes03(t *testing.T) {
+	assert := assert.New(t)
+	v, err := spec.WithPrefix("x:").Version("0.3")
+	assert.NoError(err)
+	subject := v.Attribute("x:subject")
+	assert.Equal(spec.Subject, subject.Kind())
+	assert.Equal("x:subject", subject.Name())
+
+	c := v.NewContext()
+	assert.Equal("0.3", c.GetSpecVersion())
+	assert.NoError(subject.Set(c, "foobar"))
+	got := subject.Get(c)
+	assert.Equal("foobar", got)
+	assert.Equal("foobar", c.GetSubject())
+
+	now := time.Now()
+	atime := v.Attribute("x:time")
+	assert.Equal(spec.Time, atime.Kind())
+	assert.NoError(atime.Set(c, now))
+	tm := atime.Get(c)
+	assert.Empty(cmp.Diff(now, tm))
+	assert.Empty(cmp.Diff(now, c.GetTime()))
+
+	nosuch := v.Attribute("nosuch")
+	assert.Nil(nosuch)
+
+	err = subject.Set(c, 1)
+	assert.Equal(err, spec.ErrBadType)
+	err = atime.Set(c, "foo")
+	assert.Equal(err, spec.ErrBadType)
+}
+
+func TestAttributes02(t *testing.T) {
+	assert := assert.New(t)
+	v, err := spec.WithPrefix("x:").Version("0.2")
+	assert.NoError(err)
+	c := v.NewContext()
+	id := v.Attribute("x:id")
+	assert.NoError(id.Set(c, "foobar"))
+	s := id.Get(c)
+	assert.Equal("foobar", s)
+	assert.Equal("foobar", c.GetID())
+
+	now := time.Now()
+	atime := v.Attribute("x:time")
+	assert.Equal(spec.Time, atime.Kind())
+	assert.NoError(atime.Set(c, now))
+	tm := atime.Get(c)
+	assert.Empty(cmp.Diff(now, tm))
+	assert.Empty(cmp.Diff(now, c.GetTime()))
+
+	nosuch := v.Attribute("nosuch")
+	assert.Nil(nosuch)
+
+	err = id.Set(c, 1)
+	assert.Equal(err, spec.ErrBadType)
+	err = atime.Set(c, "foo")
+	assert.Equal(err, spec.ErrBadType)
+}
+
+func TestAttributes01(t *testing.T) {
+	assert := assert.New(t)
+	v, err := spec.WithPrefix("x:").Version("0.1")
+	assert.NoError(err)
+	c := v.NewContext()
+	contentType := v.Attribute("x:contentType")
+	assert.Equal(spec.DataContentType, contentType.Kind())
+	assert.NoError(contentType.Set(c, "foobar"))
+	s := contentType.Get(c)
+	assert.Equal("foobar", s)
+	assert.Equal("foobar", c.GetDataContentType())
+
+	nosuch := v.Attribute("x:subject")
+	assert.Nil(nosuch)
+}
+
+func TestAttributesBadVersions(t *testing.T) {
+	assert := assert.New(t)
+	v, err := spec.WithPrefix("x:").Version("0.x")
+	assert.Nil(v)
+	assert.EqualError(err, `invalid spec version "0.x"`)
+}
diff --git a/pkg/binding/spec/spec_test.go b/pkg/binding/spec/spec_test.go
new file mode 100644
index 0000000..788c710
--- /dev/null
+++ b/pkg/binding/spec/spec_test.go
@@ -0,0 +1,26 @@
+package spec_test
+
+import (
+	"testing"
+
+	"github.com/cloudevents/sdk-go/pkg/binding/spec"
+	ce "github.com/cloudevents/sdk-go/pkg/cloudevents"
+	"github.com/stretchr/testify/assert"
+)
+
+func TestVersions(t *testing.T) {
+	assert := assert.New(t)
+	versions := spec.New()
+	assert.Equal([]string{"specversion", "cloudEventsVersion"}, versions.SpecVersionNames())
+
+	want := []string{"1.0", "0.3", "0.2", "0.1"}
+	all := versions.Versions()
+	assert.Equal(len(want), len(all))
+	for i, s := range want {
+		assert.Equal(s, all[i].String())
+		assert.Equal(s, all[i].NewContext().GetSpecVersion())
+		converted := all[i].Convert(ce.EventContextV01{}.AsV01())
+		assert.Equal(s, converted.GetSpecVersion(), "%v %v %v", i, s, converted)
+	}
+	assert.Equal(want[0], versions.Latest().NewContext().GetSpecVersion())
+}
diff --git a/pkg/cloudevents/client/client_test.go b/pkg/cloudevents/client/client_test.go
index 37677be..3b30ca8 100644
--- a/pkg/cloudevents/client/client_test.go
+++ b/pkg/cloudevents/client/client_test.go
@@ -96,7 +96,6 @@ func TestClientSend(t *testing.T) {
 					"ce-eventtime":          {now.UTC().Format(time.RFC3339Nano)},
 					"ce-eventtype":          {"unit.test.client"},
 					"ce-source":             {"/unit/test/client"},
-					"content-type":          {"application/json"},
 				},
 				Body: `{"msg":"hello","sq":42}`,
 			},
@@ -125,7 +124,6 @@ func TestClientSend(t *testing.T) {
 					"ce-time":        {now.UTC().Format(time.RFC3339Nano)},
 					"ce-type":        {"unit.test.client"},
 					"ce-source":      {"/unit/test/client"},
-					"content-type":   {"application/json"},
 				},
 				Body: `{"msg":"hello","sq":42}`,
 			},
@@ -154,7 +152,6 @@ func TestClientSend(t *testing.T) {
 					"ce-time":        {now.UTC().Format(time.RFC3339Nano)},
 					"ce-type":        {"unit.test.client"},
 					"ce-source":      {"/unit/test/client"},
-					"content-type":   {"application/json"},
 				},
 				Body: `{"msg":"hello","sq":42}`,
 			},
@@ -180,7 +177,7 @@ func TestClientSend(t *testing.T) {
 				Headers: map[string][]string{
 					"content-type": {"application/cloudevents+json"},
 				},
-				Body: fmt.Sprintf(`{"cloudEventsVersion":"0.1","contentType":"application/json","data":{"msg":"hello","sq":42},"eventID":"AABBCCDDEE","eventTime":%q,"eventType":"unit.test.client","source":"/unit/test/client"}`,
+				Body: fmt.Sprintf(`{"cloudEventsVersion":"0.1","data":{"msg":"hello","sq":42},"eventID":"AABBCCDDEE","eventTime":%q,"eventType":"unit.test.client","source":"/unit/test/client"}`,
 					now.UTC().Format(time.RFC3339Nano),
 				),
 			},
@@ -206,7 +203,7 @@ func TestClientSend(t *testing.T) {
 				Headers: map[string][]string{
 					"content-type": {"application/cloudevents+json"},
 				},
-				Body: fmt.Sprintf(`{"contenttype":"application/json","data":{"msg":"hello","sq":42},"id":"AABBCCDDEE","source":"/unit/test/client","specversion":"0.2","time":%q,"type":"unit.test.client"}`,
+				Body: fmt.Sprintf(`{"data":{"msg":"hello","sq":42},"id":"AABBCCDDEE","source":"/unit/test/client","specversion":"0.2","time":%q,"type":"unit.test.client"}`,
 					now.UTC().Format(time.RFC3339Nano),
 				),
 			},
@@ -232,7 +229,7 @@ func TestClientSend(t *testing.T) {
 				Headers: map[string][]string{
 					"content-type": {"application/cloudevents+json"},
 				},
-				Body: fmt.Sprintf(`{"data":{"msg":"hello","sq":42},"datacontenttype":"application/json","id":"AABBCCDDEE","source":"/unit/test/client","specversion":"0.3","time":%q,"type":"unit.test.client"}`,
+				Body: fmt.Sprintf(`{"data":{"msg":"hello","sq":42},"id":"AABBCCDDEE","source":"/unit/test/client","specversion":"0.3","time":%q,"type":"unit.test.client"}`,
 					now.UTC().Format(time.RFC3339Nano),
 				),
 			},
diff --git a/pkg/cloudevents/datacodec/codec_test.go b/pkg/cloudevents/datacodec/codec_test.go
index 0267802..fc8eece 100644
--- a/pkg/cloudevents/datacodec/codec_test.go
+++ b/pkg/cloudevents/datacodec/codec_test.go
@@ -113,7 +113,14 @@ func TestCodecEncode(t *testing.T) {
 			contentType: "unit/testing-invalid",
 			wantErr:     `[encode] unsupported content type: "unit/testing-invalid"`,
 		},
-
+		"blank": {
+			contentType: "",
+			in: map[string]string{
+				"a": "apple",
+				"b": "banana",
+			},
+			want: []byte(`{"a":"apple","b":"banana"}`),
+		},
 		"application/json": {
 			contentType: "application/json",
 			in: map[string]string{
diff --git a/pkg/cloudevents/event_test.go b/pkg/cloudevents/event_test.go
index 6cd4198..0151ea0 100644
--- a/pkg/cloudevents/event_test.go
+++ b/pkg/cloudevents/event_test.go
@@ -782,7 +782,7 @@ func TestExtensions(t *testing.T) {
 			},
 			extension:    "anothertest",
 			wantError:    true,
-			wantErrorMsg: "cannot convert int32 to String",
+			wantErrorMsg: "cannot convert 1 to string",
 		},
 		"full v02, test extension": {
 			event: ce.Event{
@@ -797,7 +797,7 @@ func TestExtensions(t *testing.T) {
 			},
 			extension:    "anothertest",
 			wantError:    true,
-			wantErrorMsg: "cannot convert int32 to String",
+			wantErrorMsg: "cannot convert 1 to string",
 		},
 		"full v03, test extension": {
 			event: ce.Event{
@@ -812,7 +812,7 @@ func TestExtensions(t *testing.T) {
 			},
 			extension:    "anothertest",
 			wantError:    true,
-			wantErrorMsg: "cannot convert int32 to String",
+			wantErrorMsg: "cannot convert 1 to string",
 		},
 		"full v1, test extension": {
 			event: ce.Event{
@@ -827,7 +827,7 @@ func TestExtensions(t *testing.T) {
 			},
 			extension:    "anothertest",
 			wantError:    true,
-			wantErrorMsg: "cannot convert int32 to String",
+			wantErrorMsg: "cannot convert 1 to string",
 		},
 	}
 	for n, tc := range testCases {
diff --git a/pkg/cloudevents/transport/amqp/codec_test.go b/pkg/cloudevents/transport/amqp/codec_test.go
index f74e5e7..4fb024e 100644
--- a/pkg/cloudevents/transport/amqp/codec_test.go
+++ b/pkg/cloudevents/transport/amqp/codec_test.go
@@ -41,7 +41,6 @@ func TestCodecEncode(t *testing.T) {
 				ContentType: "application/cloudevents+json",
 				Body: func() []byte {
 					body := map[string]interface{}{
-						"contenttype": "application/json",
 						"specversion": "0.2",
 						"id":          "ABC-123",
 						"type":        "com.example.test",
@@ -65,12 +64,11 @@ func TestCodecEncode(t *testing.T) {
 				ContentType: "application/cloudevents+json",
 				Body: func() []byte {
 					body := map[string]interface{}{
-						"datacontenttype": "application/json",
-						"specversion":     "0.3",
-						"id":              "ABC-123",
-						"type":            "com.example.test",
-						"source":          "http://example.com/source",
-						"subject":         "a-subject",
+						"specversion": "0.3",
+						"id":          "ABC-123",
+						"type":        "com.example.test",
+						"source":      "http://example.com/source",
+						"subject":     "a-subject",
 					}
 					return toBytes(body)
 				}(),
@@ -300,7 +298,6 @@ func TestCodecRoundTrip(t *testing.T) {
 						EventType:          "com.example.test",
 						Source:             *source,
 						EventID:            "ABC-123",
-						ContentType:        cloudevents.StringOfApplicationJSON(),
 					},
 					Data: map[string]interface{}{
 						"a": "apple",
@@ -328,7 +325,6 @@ func TestCodecRoundTrip(t *testing.T) {
 						EventType:          "com.example.test",
 						Source:             *source,
 						EventID:            "ABC-123",
-						ContentType:        cloudevents.StringOfApplicationJSON(),
 					},
 					Data: &DataExample{
 						AnInt:   42,
@@ -422,7 +418,6 @@ func TestCodecAsMiddleware(t *testing.T) {
 						EventType:          "com.example.test",
 						Source:             *source,
 						EventID:            "ABC-123",
-						ContentType:        cloudevents.StringOfApplicationJSON(),
 					},
 					Data: map[string]interface{}{
 						"a": "apple",
@@ -450,7 +445,6 @@ func TestCodecAsMiddleware(t *testing.T) {
 						EventType:          "com.example.test",
 						Source:             *source,
 						EventID:            "ABC-123",
-						ContentType:        cloudevents.StringOfApplicationJSON(),
 					},
 					Data: &DataExample{
 						AnInt:   42,
diff --git a/pkg/cloudevents/transport/amqp/codec_v02_test.go b/pkg/cloudevents/transport/amqp/codec_v02_test.go
index ce881ef..c1aab08 100644
--- a/pkg/cloudevents/transport/amqp/codec_v02_test.go
+++ b/pkg/cloudevents/transport/amqp/codec_v02_test.go
@@ -40,7 +40,6 @@ func TestCodecV02_Encode(t *testing.T) {
 				ContentType: cloudevents.ApplicationCloudEventsJSON,
 				Body: func() []byte {
 					body := map[string]interface{}{
-						"contenttype": "application/json",
 						"specversion": "0.2",
 						"id":          "ABC-123",
 						"type":        "com.example.test",
@@ -101,7 +100,6 @@ func TestCodecV02_Encode(t *testing.T) {
 				ContentType: cloudevents.ApplicationCloudEventsJSON,
 				Body: func() []byte {
 					body := map[string]interface{}{
-						"contenttype": "application/json",
 						"specversion": "0.2",
 						"id":          "ABC-123",
 						"type":        "com.example.test",
diff --git a/pkg/cloudevents/transport/amqp/codec_v03_test.go b/pkg/cloudevents/transport/amqp/codec_v03_test.go
index 0b896de..adb6a00 100644
--- a/pkg/cloudevents/transport/amqp/codec_v03_test.go
+++ b/pkg/cloudevents/transport/amqp/codec_v03_test.go
@@ -40,11 +40,10 @@ func TestCodecV03_Encode(t *testing.T) {
 				ContentType: "application/cloudevents+json",
 				Body: func() []byte {
 					body := map[string]interface{}{
-						"datacontenttype": "application/json",
-						"specversion":     "0.3",
-						"id":              "ABC-123",
-						"type":            "com.example.test",
-						"source":          "http://example.com/source",
+						"specversion": "0.3",
+						"id":          "ABC-123",
+						"type":        "com.example.test",
+						"source":      "http://example.com/source",
 					}
 					return toBytes(body)
 				}(),
@@ -101,11 +100,10 @@ func TestCodecV03_Encode(t *testing.T) {
 				ContentType: cloudevents.ApplicationCloudEventsJSON,
 				Body: func() []byte {
 					body := map[string]interface{}{
-						"datacontenttype": "application/json",
-						"specversion":     "0.3",
-						"id":              "ABC-123",
-						"type":            "com.example.test",
-						"source":          "http://example.com/source",
+						"specversion": "0.3",
+						"id":          "ABC-123",
+						"type":        "com.example.test",
+						"source":      "http://example.com/source",
 					}
 					return toBytes(body)
 				}(),
diff --git a/pkg/cloudevents/transport/http/codec_test.go b/pkg/cloudevents/transport/http/codec_test.go
index 3819453..7af098d 100644
--- a/pkg/cloudevents/transport/http/codec_test.go
+++ b/pkg/cloudevents/transport/http/codec_test.go
@@ -134,7 +134,6 @@ func TestCodecEncode(t *testing.T) {
 					"CE-EventID":            {"ABC-123"},
 					"CE-EventType":          {"com.example.test"},
 					"CE-Source":             {"http://example.com/source"},
-					"Content-Type":          {"application/json"},
 				},
 			},
 		},
@@ -155,7 +154,6 @@ func TestCodecEncode(t *testing.T) {
 				},
 				Body: func() []byte {
 					body := map[string]interface{}{
-						"contentType":        "application/json",
 						"cloudEventsVersion": "0.1",
 						"eventID":            "ABC-123",
 						"eventType":          "com.example.test",
@@ -182,7 +180,6 @@ func TestCodecEncode(t *testing.T) {
 					"Ce-Id":          {"ABC-123"},
 					"Ce-Type":        {"com.example.test"},
 					"Ce-Source":      {"http://example.com/source"},
-					"Content-Type":   {"application/json"},
 				},
 			},
 		},
@@ -203,7 +200,6 @@ func TestCodecEncode(t *testing.T) {
 				},
 				Body: func() []byte {
 					body := map[string]interface{}{
-						"contenttype": "application/json",
 						"specversion": "0.2",
 						"id":          "ABC-123",
 						"type":        "com.example.test",
@@ -230,7 +226,6 @@ func TestCodecEncode(t *testing.T) {
 					"Ce-Id":          {"ABC-123"},
 					"Ce-Type":        {"com.example.test"},
 					"Ce-Source":      {"http://example.com/source"},
-					"Content-Type":   {"application/json"},
 				},
 			},
 		},
@@ -251,11 +246,10 @@ func TestCodecEncode(t *testing.T) {
 				},
 				Body: func() []byte {
 					body := map[string]interface{}{
-						"datacontenttype": "application/json",
-						"specversion":     "0.3",
-						"id":              "ABC-123",
-						"type":            "com.example.test",
-						"source":          "http://example.com/source",
+						"specversion": "0.3",
+						"id":          "ABC-123",
+						"type":        "com.example.test",
+						"source":      "http://example.com/source",
 					}
 					return toBytes(body)
 				}(),
@@ -276,7 +270,6 @@ func TestCodecEncode(t *testing.T) {
 					"CE-EventID":            {"ABC-123"},
 					"CE-EventType":          {"com.example.test"},
 					"CE-Source":             {"http://example.com/source"},
-					"Content-Type":          {"application/json"},
 				},
 			},
 		},
@@ -295,7 +288,6 @@ func TestCodecEncode(t *testing.T) {
 				},
 				Body: func() []byte {
 					body := map[string]interface{}{
-						"contentType":        "application/json",
 						"cloudEventsVersion": "0.1",
 						"eventID":            "ABC-123",
 						"eventType":          "com.example.test",
@@ -320,7 +312,6 @@ func TestCodecEncode(t *testing.T) {
 					"Ce-Id":          {"ABC-123"},
 					"Ce-Type":        {"com.example.test"},
 					"Ce-Source":      {"http://example.com/source"},
-					"Content-Type":   {"application/json"},
 				},
 			},
 		},
@@ -339,7 +330,6 @@ func TestCodecEncode(t *testing.T) {
 				},
 				Body: func() []byte {
 					body := map[string]interface{}{
-						"contenttype": "application/json",
 						"specversion": "0.2",
 						"id":          "ABC-123",
 						"type":        "com.example.test",
@@ -364,7 +354,6 @@ func TestCodecEncode(t *testing.T) {
 					"Ce-Id":          {"ABC-123"},
 					"Ce-Type":        {"com.example.test"},
 					"Ce-Source":      {"http://example.com/source"},
-					"Content-Type":   {"application/json"},
 				},
 			},
 		},
@@ -383,11 +372,10 @@ func TestCodecEncode(t *testing.T) {
 				},
 				Body: func() []byte {
 					body := map[string]interface{}{
-						"datacontenttype": "application/json",
-						"specversion":     "0.3",
-						"id":              "ABC-123",
-						"type":            "com.example.test",
-						"source":          "http://example.com/source",
+						"specversion": "0.3",
+						"id":          "ABC-123",
+						"type":        "com.example.test",
+						"source":      "http://example.com/source",
 					}
 					return toBytes(body)
 				}(),
@@ -747,7 +735,6 @@ func TestCodecRoundTrip(t *testing.T) {
 						EventType:          "com.example.test",
 						Source:             *source,
 						EventID:            "ABC-123",
-						ContentType:        cloudevents.StringOfApplicationJSON(),
 					}.AsV01(),
 					Data: map[string]interface{}{
 						"a": "apple",
@@ -775,7 +762,6 @@ func TestCodecRoundTrip(t *testing.T) {
 						EventType:          "com.example.test",
 						Source:             *source,
 						EventID:            "ABC-123",
-						ContentType:        cloudevents.StringOfApplicationJSON(),
 					}.AsV01(),
 					Data: &DataExample{
 						AnInt:   42,
diff --git a/pkg/cloudevents/transport/http/codec_v01_test.go b/pkg/cloudevents/transport/http/codec_v01_test.go
index c72f3b5..f315d5b 100644
--- a/pkg/cloudevents/transport/http/codec_v01_test.go
+++ b/pkg/cloudevents/transport/http/codec_v01_test.go
@@ -42,7 +42,6 @@ func TestCodecV01_Encode(t *testing.T) {
 					"CE-EventID":            {"ABC-123"},
 					"CE-EventType":          {"com.example.test"},
 					"CE-Source":             {"http://example.com/source"},
-					"Content-Type":          {"application/json"},
 				},
 			},
 		},
@@ -95,7 +94,6 @@ func TestCodecV01_Encode(t *testing.T) {
 					"CE-EventID":            {"ABC-123"},
 					"CE-EventType":          {"com.example.test"},
 					"CE-Source":             {"http://example.com/source"},
-					"Content-Type":          {"application/json"},
 				},
 			},
 		},
@@ -148,7 +146,6 @@ func TestCodecV01_Encode(t *testing.T) {
 				},
 				Body: func() []byte {
 					body := map[string]interface{}{
-						"contentType":        "application/json",
 						"cloudEventsVersion": "0.1",
 						"eventID":            "ABC-123",
 						"eventType":          "com.example.test",
diff --git a/pkg/cloudevents/transport/http/codec_v02_test.go b/pkg/cloudevents/transport/http/codec_v02_test.go
index 86c202c..3b365db 100644
--- a/pkg/cloudevents/transport/http/codec_v02_test.go
+++ b/pkg/cloudevents/transport/http/codec_v02_test.go
@@ -42,7 +42,6 @@ func TestCodecV02_Encode(t *testing.T) {
 					"Ce-Id":          {"ABC-123"},
 					"Ce-Type":        {"com.example.test"},
 					"Ce-Source":      {"http://example.com/source"},
-					"Content-Type":   {"application/json"},
 				},
 			},
 		},
@@ -93,7 +92,6 @@ func TestCodecV02_Encode(t *testing.T) {
 					"Ce-Id":          {"ABC-123"},
 					"Ce-Type":        {"com.example.test"},
 					"Ce-Source":      {"http://example.com/source"},
-					"Content-Type":   {"application/json"},
 				},
 			},
 		},
@@ -155,7 +153,6 @@ func TestCodecV02_Encode(t *testing.T) {
 				},
 				Body: func() []byte {
 					body := map[string]interface{}{
-						"contenttype": "application/json",
 						"specversion": "0.2",
 						"id":          "ABC-123",
 						"type":        "com.example.test",
diff --git a/pkg/cloudevents/transport/http/codec_v03_test.go b/pkg/cloudevents/transport/http/codec_v03_test.go
index a1cfc2a..e28540f 100644
--- a/pkg/cloudevents/transport/http/codec_v03_test.go
+++ b/pkg/cloudevents/transport/http/codec_v03_test.go
@@ -44,7 +44,6 @@ func TestCodecV03_Encode(t *testing.T) {
 					"Ce-Id":          {"ABC-123"},
 					"Ce-Type":        {"com.example.test"},
 					"Ce-Source":      {"http://example.com/source"},
-					"Content-Type":   {"application/json"},
 				},
 			},
 		},
@@ -97,7 +96,6 @@ func TestCodecV03_Encode(t *testing.T) {
 					"Ce-Id":          {"ABC-123"},
 					"Ce-Type":        {"com.example.test"},
 					"Ce-Source":      {"http://example.com/source"},
-					"Content-Type":   {"application/json"},
 				},
 			},
 		},
@@ -208,11 +206,10 @@ func TestCodecV03_Encode(t *testing.T) {
 				},
 				Body: func() []byte {
 					body := map[string]interface{}{
-						"datacontenttype": "application/json",
-						"specversion":     "0.3",
-						"id":              "ABC-123",
-						"type":            "com.example.test",
-						"source":          "http://example.com/source",
+						"specversion": "0.3",
+						"id":          "ABC-123",
+						"type":        "com.example.test",
+						"source":      "http://example.com/source",
 					}
 					return toBytes(body)
 				}(),
diff --git a/pkg/cloudevents/transport/http/codec_v1_test.go b/pkg/cloudevents/transport/http/codec_v1_test.go
index d30e4f2..31625ad 100644
--- a/pkg/cloudevents/transport/http/codec_v1_test.go
+++ b/pkg/cloudevents/transport/http/codec_v1_test.go
@@ -44,7 +44,6 @@ func TestCodecV1_Encode(t *testing.T) {
 					"Ce-Id":          {"ABC-123"},
 					"Ce-Type":        {"com.example.test"},
 					"Ce-Source":      {"http://example.com/source"},
-					"Content-Type":   {"application/json"},
 				},
 			},
 		},
@@ -97,7 +96,6 @@ func TestCodecV1_Encode(t *testing.T) {
 					"Ce-Id":          {"ABC-123"},
 					"Ce-Type":        {"com.example.test"},
 					"Ce-Source":      {"http://example.com/source"},
-					"Content-Type":   {"application/json"},
 				},
 			},
 		},
@@ -150,11 +148,10 @@ func TestCodecV1_Encode(t *testing.T) {
 				},
 				Body: func() []byte {
 					body := map[string]interface{}{
-						"datacontenttype": "application/json",
-						"specversion":     "1.0",
-						"id":              "ABC-123",
-						"type":            "com.example.test",
-						"source":          "http://example.com/source",
+						"specversion": "1.0",
+						"id":          "ABC-123",
+						"type":        "com.example.test",
+						"source":      "http://example.com/source",
 					}
 					return toBytes(body)
 				}(),
@@ -217,7 +214,8 @@ func TestCodecV1_Encode(t *testing.T) {
 						"test": "extended",
 					},
 				}.AsV1(),
-				DataBase64:  []byte(`{"hello":"world"}`),
+				Data:        []byte(`{"hello":"world"}`),
+				DataBinary:  true,
 				DataEncoded: true,
 			},
 			want: &http.Message{
@@ -443,7 +441,7 @@ func TestCodecV1_Decode(t *testing.T) {
 						"test": "extended",
 					},
 				}.AsV1(),
-				DataBase64: []byte(`{"hello":"world"}`),
+				Data: []byte(`{"hello":"world"}`),
 			},
 		},
 		"simple v1.0 binary with short header": {
diff --git a/pkg/cloudevents/transport/nats/codec_test.go b/pkg/cloudevents/transport/nats/codec_test.go
index 5135638..e7d1a4a 100644
--- a/pkg/cloudevents/transport/nats/codec_test.go
+++ b/pkg/cloudevents/transport/nats/codec_test.go
@@ -36,7 +36,6 @@ func TestCodecEncode(t *testing.T) {
 			want: &nats.Message{
 				Body: func() []byte {
 					body := map[string]interface{}{
-						"contenttype": "application/json",
 						"specversion": "0.2",
 						"id":          "ABC-123",
 						"type":        "com.example.test",
@@ -166,7 +165,6 @@ func TestCodecRoundTrip(t *testing.T) {
 						EventType:          "com.example.test",
 						Source:             *source,
 						EventID:            "ABC-123",
-						ContentType:        cloudevents.StringOfApplicationJSON(),
 					},
 					Data: map[string]interface{}{
 						"a": "apple",
@@ -194,7 +192,6 @@ func TestCodecRoundTrip(t *testing.T) {
 						EventType:          "com.example.test",
 						Source:             *source,
 						EventID:            "ABC-123",
-						ContentType:        cloudevents.StringOfApplicationJSON(),
 					},
 					Data: &DataExample{
 						AnInt:   42,
@@ -288,7 +285,6 @@ func TestCodecAsMiddleware(t *testing.T) {
 						EventType:          "com.example.test",
 						Source:             *source,
 						EventID:            "ABC-123",
-						ContentType:        cloudevents.StringOfApplicationJSON(),
 					},
 					Data: map[string]interface{}{
 						"a": "apple",
@@ -316,7 +312,6 @@ func TestCodecAsMiddleware(t *testing.T) {
 						EventType:          "com.example.test",
 						Source:             *source,
 						EventID:            "ABC-123",
-						ContentType:        cloudevents.StringOfApplicationJSON(),
 					},
 					Data: &DataExample{
 						AnInt:   42,
diff --git a/pkg/cloudevents/transport/nats/codec_v02_test.go b/pkg/cloudevents/transport/nats/codec_v02_test.go
index 0c3145b..fd151f8 100644
--- a/pkg/cloudevents/transport/nats/codec_v02_test.go
+++ b/pkg/cloudevents/transport/nats/codec_v02_test.go
@@ -39,7 +39,6 @@ func TestCodecV02_Encode(t *testing.T) {
 			want: &nats.Message{
 				Body: func() []byte {
 					body := map[string]interface{}{
-						"contenttype": "application/json",
 						"specversion": "0.2",
 						"id":          "ABC-123",
 						"type":        "com.example.test",
@@ -98,7 +97,6 @@ func TestCodecV02_Encode(t *testing.T) {
 			want: &nats.Message{
 				Body: func() []byte {
 					body := map[string]interface{}{
-						"contenttype": "application/json",
 						"specversion": "0.2",
 						"id":          "ABC-123",
 						"type":        "com.example.test",
diff --git a/pkg/cloudevents/transport/nats/codec_v03_test.go b/pkg/cloudevents/transport/nats/codec_v03_test.go
index 7039169..2438300 100644
--- a/pkg/cloudevents/transport/nats/codec_v03_test.go
+++ b/pkg/cloudevents/transport/nats/codec_v03_test.go
@@ -39,11 +39,10 @@ func TestCodecV03_Encode(t *testing.T) {
 			want: &nats.Message{
 				Body: func() []byte {
 					body := map[string]interface{}{
-						"datacontenttype": "application/json",
-						"specversion":     "0.3",
-						"id":              "ABC-123",
-						"type":            "com.example.test",
-						"source":          "http://example.com/source",
+						"specversion": "0.3",
+						"id":          "ABC-123",
+						"type":        "com.example.test",
+						"source":      "http://example.com/source",
 					}
 					return toBytes(body)
 				}(),
@@ -98,11 +97,10 @@ func TestCodecV03_Encode(t *testing.T) {
 			want: &nats.Message{
 				Body: func() []byte {
 					body := map[string]interface{}{
-						"datacontenttype": "application/json",
-						"specversion":     "0.3",
-						"id":              "ABC-123",
-						"type":            "com.example.test",
-						"source":          "http://example.com/source",
+						"specversion": "0.3",
+						"id":          "ABC-123",
+						"type":        "com.example.test",
+						"source":      "http://example.com/source",
 					}
 					return toBytes(body)
 				}(),
diff --git a/pkg/cloudevents/transport/pubsub/codec_test.go b/pkg/cloudevents/transport/pubsub/codec_test.go
index 20acaf4..f2360b7 100644
--- a/pkg/cloudevents/transport/pubsub/codec_test.go
+++ b/pkg/cloudevents/transport/pubsub/codec_test.go
@@ -44,12 +44,11 @@ func TestCodecEncode(t *testing.T) {
 				},
 				Data: func() []byte {
 					body := map[string]interface{}{
-						"datacontenttype": "application/json",
-						"specversion":     "0.3",
-						"id":              "ABC-123",
-						"type":            "com.example.test",
-						"source":          "http://example.com/source",
-						"subject":         "a-subject",
+						"specversion": "0.3",
+						"id":          "ABC-123",
+						"type":        "com.example.test",
+						"source":      "http://example.com/source",
+						"subject":     "a-subject",
 					}
 					return toBytes(body)
 				}(),
@@ -197,11 +196,10 @@ func TestCodecAsMiddleware(t *testing.T) {
 				},
 				want: cloudevents.Event{
 					Context: &cloudevents.EventContextV03{
-						SpecVersion:     cloudevents.CloudEventsVersionV03,
-						Type:            "com.example.test",
-						Source:          *source,
-						ID:              "ABC-123",
-						DataContentType: cloudevents.StringOfApplicationJSON(),
+						SpecVersion: cloudevents.CloudEventsVersionV03,
+						Type:        "com.example.test",
+						Source:      *source,
+						ID:          "ABC-123",
 					},
 					Data: map[string]interface{}{
 						"a": "apple",
@@ -225,11 +223,10 @@ func TestCodecAsMiddleware(t *testing.T) {
 				},
 				want: cloudevents.Event{
 					Context: &cloudevents.EventContextV03{
-						SpecVersion:     cloudevents.CloudEventsVersionV03,
-						Type:            "com.example.test",
-						Source:          *source,
-						ID:              "ABC-123",
-						DataContentType: cloudevents.StringOfApplicationJSON(),
+						SpecVersion: cloudevents.CloudEventsVersionV03,
+						Type:        "com.example.test",
+						Source:      *source,
+						ID:          "ABC-123",
 					},
 					Data: &DataExample{
 						AnInt:   42,
diff --git a/pkg/cloudevents/transport/pubsub/codec_v03_test.go b/pkg/cloudevents/transport/pubsub/codec_v03_test.go
index 993c180..d1e4ed4 100644
--- a/pkg/cloudevents/transport/pubsub/codec_v03_test.go
+++ b/pkg/cloudevents/transport/pubsub/codec_v03_test.go
@@ -42,11 +42,10 @@ func TestCodecV03_Encode(t *testing.T) {
 				},
 				Data: func() []byte {
 					body := map[string]interface{}{
-						"datacontenttype": "application/json",
-						"specversion":     "0.3",
-						"id":              "ABC-123",
-						"type":            "com.example.test",
-						"source":          "http://example.com/source",
+						"specversion": "0.3",
+						"id":          "ABC-123",
+						"type":        "com.example.test",
+						"source":      "http://example.com/source",
 					}
 					return toBytes(body)
 				}(),
@@ -107,11 +106,10 @@ func TestCodecV03_Encode(t *testing.T) {
 				},
 				Data: func() []byte {
 					body := map[string]interface{}{
-						"datacontenttype": "application/json",
-						"specversion":     "0.3",
-						"id":              "ABC-123",
-						"type":            "com.example.test",
-						"source":          "http://example.com/source",
+						"specversion": "0.3",
+						"id":          "ABC-123",
+						"type":        "com.example.test",
+						"source":      "http://example.com/source",
 					}
 					return toBytes(body)
 				}(),
diff --git a/pkg/cloudevents/types/timestamp_test.go b/pkg/cloudevents/types/timestamp_test.go
index 669eeb5..8c5ae4e 100644
--- a/pkg/cloudevents/types/timestamp_test.go
+++ b/pkg/cloudevents/types/timestamp_test.go
@@ -1,294 +1,82 @@
 package types_test
 
 import (
+	"encoding/json"
 	"encoding/xml"
-	"strings"
+	"fmt"
 	"testing"
 	"time"
 
 	"github.com/cloudevents/sdk-go/pkg/cloudevents/types"
-	"github.com/google/go-cmp/cmp"
+	"github.com/stretchr/testify/assert"
 )
 
-func TestParseTimestamp(t *testing.T) {
-	testCases := map[string]struct {
-		t    string
-		want *types.Timestamp
-	}{
-		"empty": {
-			want: nil,
-		},
-		"empty string": {
-			t:    "",
-			want: nil,
-		},
-		"invalid format": {
-			t:    "2019-02-28",
-			want: nil,
-		},
-		"RFC3339 format": {
-			t: "1984-02-28T15:04:05Z",
-			want: func() *types.Timestamp {
-				t, _ := time.Parse(time.RFC3339, "1984-02-28T15:04:05Z")
-				return &types.Timestamp{Time: t}
-			}(),
-		},
-		"RFC3339Nano format": {
-			t: "1984-02-28T15:04:05.999999999Z",
-			want: func() *types.Timestamp {
-				t, _ := time.Parse(time.RFC3339Nano, "1984-02-28T15:04:05.999999999Z")
-				return &types.Timestamp{Time: t}
-			}(),
-		},
+func TestTimestampParseString(t *testing.T) {
+	ok := func(s string, want time.Time) {
+		t.Helper()
+		got, err := types.ParseTimestamp(s)
+		assert.NoError(t, err)
+		assert.Equal(t, want, got.Time)
+		assert.Equal(t, s, got.String())
 	}
-	for n, tc := range testCases {
-		tc := tc // Don't use range variable in func literal.
-		t.Run(n, func(t *testing.T) {
+	ok("1984-02-28T15:04:05Z", time.Date(1984, 02, 28, 15, 04, 05, 0, time.UTC))
+	ok("1984-02-28T15:04:05.999999999Z", time.Date(1984, 02, 28, 15, 04, 05, 999999999, time.UTC))
 
-			got := types.ParseTimestamp(tc.t)
-
-			if diff := cmp.Diff(tc.want, got); diff != "" {
-				t.Errorf("unexpected object (-want, +got) = %v", diff)
-			}
-		})
-	}
-}
-
-func TestJsonMarshalTimestamp(t *testing.T) {
-	testCases := map[string]struct {
-		t    string
-		want []byte
-	}{
-		"empty": {
-			want: []byte(`""`),
-		},
-		"empty string": {
-			t:    "",
-			want: []byte(`""`),
-		},
-		"invalid format": {
-			t:    "2019-02-28",
-			want: []byte(`""`),
-		},
-		"RFC3339 format": {
-			t:    "1984-02-28T15:04:05Z",
-			want: []byte(`"1984-02-28T15:04:05Z"`),
-		},
-		"RFC3339Nano format": {
-			t:    "1984-02-28T15:04:05.999999999Z",
-			want: []byte(`"1984-02-28T15:04:05.999999999Z"`),
-		},
-	}
-	for n, tc := range testCases {
-		tc := tc // Don't use range variable in func literal.
-		t.Run(n, func(t *testing.T) {
-
-			tt := types.ParseTimestamp(tc.t)
-			got, _ := tt.MarshalJSON()
-
-			if diff := cmp.Diff(tc.want, got); diff != "" {
-				t.Logf("got: %s", string(got))
-				t.Errorf("unexpected object (-want, +got) = %v", diff)
-			}
-		})
-	}
-}
-
-func TestXMLMarshalTimestamp(t *testing.T) {
-	testCases := map[string]struct {
-		t       string
-		want    []byte
-		wantErr string
-	}{
-		"empty": {
-			want: []byte(nil),
-		},
-		"empty string": {
-			t:    "",
-			want: []byte(nil),
-		},
-		"invalid format": {
-			t:    "2019-02-28",
-			want: []byte(nil),
-		},
-		"RFC3339 format": {
-			t:    "1984-02-28T15:04:05Z",
-			want: []byte(`<Timestamp>1984-02-28T15:04:05Z</Timestamp>`),
-		},
-		"RFC3339Nano format": {
-			t:    "1984-02-28T15:04:05.999999999Z",
-			want: []byte(`<Timestamp>1984-02-28T15:04:05.999999999Z</Timestamp>`),
-		},
-	}
-	for n, tc := range testCases {
-		tc := tc // Don't use range variable in func literal.
-		t.Run(n, func(t *testing.T) {
-
-			var got []byte
-			var err error
-			tt := types.ParseTimestamp(tc.t)
-
-			got, err = xml.Marshal(tt)
-
-			if tc.wantErr != "" {
-				if err != nil {
-					gotErr := err.Error()
-					if !strings.Contains(gotErr, tc.wantErr) {
-						t.Errorf("unexpected error, expected to contain %q, got: %q", tc.wantErr, gotErr)
-					}
-				} else {
-					t.Errorf("expected error to contain %q, got: nil", tc.wantErr)
-				}
-				return
-			}
-
-			if diff := cmp.Diff(tc.want, got); diff != "" {
-				t.Logf("got: %s", string(got))
-				t.Errorf("unexpected object (-want, +got) = %v", diff)
-			}
-		})
-	}
-}
-
-func TestJsonUnmarshalTimestamp(t *testing.T) {
-	testCases := map[string]struct {
-		b       []byte
-		want    *types.Timestamp
-		wantErr string
-	}{
-		"empty": {
-			wantErr: "unexpected end of JSON input",
-		},
-		"invalid format": {
-			b:       []byte("2019-02-28"),
-			wantErr: "invalid character '-' after top-level value",
-		},
-		"RFC3339 format": {
-			b: []byte(`"1984-02-28T15:04:05Z"`),
-			want: func() *types.Timestamp {
-				t, _ := time.Parse(time.RFC3339, "1984-02-28T15:04:05Z")
-				return &types.Timestamp{Time: t}
-			}(),
-		},
-		"RFC3339Nano format": {
-			b: []byte(`"1984-02-28T15:04:05.999999999Z"`),
-			want: func() *types.Timestamp {
-				t, _ := time.Parse(time.RFC3339Nano, "1984-02-28T15:04:05.999999999Z")
-				return &types.Timestamp{Time: t}
-			}(),
-		},
-	}
-	for n, tc := range testCases {
-		tc := tc // Don't use range variable in func literal.
-		t.Run(n, func(t *testing.T) {
-
-			got := &types.Timestamp{}
-			err := got.UnmarshalJSON(tc.b)
-
-			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
-			}
-
-			if diff := cmp.Diff(tc.want, got); diff != "" {
-				t.Errorf("unexpected object (-want, +got) = %v", diff)
-			}
-		})
+	bad := func(s, wanterr string) {
+		t.Helper()
+		_, err := types.ParseTime(s)
+		assert.EqualError(t, err, wanterr)
 	}
+	bad("", "cannot convert \"\" to time.Time: not in RFC3339 format")
+	bad("2019-02-28", "cannot convert \"2019-02-28\" to time.Time: not in RFC3339 format")
 }
 
-func TestXMLUnmarshalTimestamp(t *testing.T) {
-	testCases := map[string]struct {
-		b       []byte
-		want    *types.Timestamp
-		wantErr string
-	}{
-		"empty": {
-			wantErr: "EOF",
-		},
-		"invalid format": {
-			b:    []byte("<Timestamp>2019-02-28</Timestamp>"),
-			want: &types.Timestamp{},
-		},
-		"nil": {
-			b:    []byte("<Timestamp></Timestamp>"),
-			want: &types.Timestamp{},
-		},
-		"RFC3339 format": {
-			b: []byte(`<Timestamp>1984-02-28T15:04:05Z</Timestamp>`),
-			want: func() *types.Timestamp {
-				t, _ := time.Parse(time.RFC3339, "1984-02-28T15:04:05Z")
-				return &types.Timestamp{Time: t}
-			}(),
-		},
-		"RFC3339Nano format": {
-			b: []byte(`<Timestamp>1984-02-28T15:04:05.999999999Z</Timestamp>`),
-			want: func() *types.Timestamp {
-				t, _ := time.Parse(time.RFC3339Nano, "1984-02-28T15:04:05.999999999Z")
-				return &types.Timestamp{Time: t}
-			}(),
-		},
+func TestJsonMarshalUnmarshalTimestamp(t *testing.T) {
+	ok := func(ts string) {
+		t.Helper()
+		tt, err := types.ParseTime(ts)
+		assert.NoError(t, err)
+		got, err := json.Marshal(types.Timestamp{Time: tt})
+		assert.NoError(t, err)
+		assert.Equal(t, fmt.Sprintf(`"%s"`, ts), string(got))
 	}
-	for n, tc := range testCases {
-		tc := tc // Don't use range variable in func literal.
-		t.Run(n, func(t *testing.T) {
-
-			got := &types.Timestamp{}
-
-			err := xml.Unmarshal(tc.b, got)
-
-			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
-			}
-
-			if diff := cmp.Diff(tc.want, got); diff != "" {
-				t.Errorf("unexpected object (-want, +got) = %v", diff)
-			}
-		})
+	ok("1984-02-28T15:04:05Z")
+	ok("1984-02-28T15:04:05.999999999Z")
+
+	bad := func(s, wanterr string) {
+		t.Helper()
+		var ts types.Timestamp
+		err := json.Unmarshal([]byte(s), &ts)
+		assert.EqualError(t, err, wanterr)
 	}
+	bad("", "unexpected end of JSON input")
+	bad("2019-02-28", "invalid character '-' after top-level value")
 }
 
-func TestTimestampString(t *testing.T) {
-	testCases := map[string]struct {
-		t    string
-		want string
-	}{
-		"empty": {
-			want: "0001-01-01T00:00:00Z",
-		},
-		"RFC3339 format": {
-			t:    "1984-02-28T15:04:05Z",
-			want: `1984-02-28T15:04:05Z`,
-		},
-		"RFC3339Nano format": {
-			t:    "1984-02-28T15:04:05.999999999Z",
-			want: `1984-02-28T15:04:05.999999999Z`,
-		},
+func TestXMLMarshalUnmarshalTimestamp(t *testing.T) {
+	ok := func(tstr string) {
+		t.Helper()
+		tt, err := types.ParseTime(tstr)
+		assert.NoError(t, err)
+		got, err := xml.Marshal(types.Timestamp{Time: tt})
+		assert.NoError(t, err)
+		assert.Equal(t, fmt.Sprintf("<Timestamp>%s</Timestamp>", tstr), string(got))
+		var ts types.Timestamp
+		err = xml.Unmarshal(got, &ts)
+		assert.NoError(t, err)
+		assert.Equal(t, tt, ts.Time)
 	}
-	for n, tc := range testCases {
-		tc := tc // Don't use range variable in func literal.
-		t.Run(n, func(t *testing.T) {
-
-			tt := types.ParseTimestamp(tc.t)
-			got := tt.String()
-
-			if diff := cmp.Diff(tc.want, got); diff != "" {
-				t.Logf("got: %s", got)
-				t.Errorf("unexpected string (-want, +got) = %v", diff)
-			}
-		})
+	ok("1984-02-28T15:04:05Z")
+	ok("1984-02-28T15:04:05.999999999Z")
+
+	bad := func(s, wanterr string) {
+		t.Helper()
+		var ts types.Timestamp
+		err := xml.Unmarshal([]byte(s), &ts)
+		assert.EqualError(t, err, wanterr)
 	}
+	bad("", "EOF")
+	bad("2019-02-28", "EOF")
+	bad("<Timestamp>2019-02-28</Timestamp>", "cannot convert \"2019-02-28\" to time.Time: not in RFC3339 format")
+	bad("<Timestamp></Timestamp>", "cannot convert \"\" to time.Time: not in RFC3339 format")
 }
diff --git a/pkg/cloudevents/types/value_test.go b/pkg/cloudevents/types/value_test.go
index 668911e..0ef7289 100644
--- a/pkg/cloudevents/types/value_test.go
+++ b/pkg/cloudevents/types/value_test.go
@@ -41,7 +41,7 @@ func Example() {
 	// 123 <nil>
 	// 456 <nil>
 	// 99999 <nil>
-	// 0 2147483648 is out of range for Integer
+	// 0 cannot convert 2147483648 to int32: out of range
 	// 0 strconv.ParseFloat: parsing "not an int": invalid syntax
 }
 
@@ -106,7 +106,7 @@ func TestBool(t *testing.T) {
 	x.ok(false, false, "false")
 
 	x.err("notabool", "strconv.ParseBool: parsing \"notabool\": invalid syntax")
-	x.err(0, "cannot convert int32 to Bool")
+	x.err(0, "cannot convert 0 to bool")
 	x.err(nil, "invalid CloudEvents value: <nil>")
 }
 
@@ -141,18 +141,18 @@ func TestInteger(t *testing.T) {
 	x.str(".9", int32(0))
 	x.str("-.9", int32(0))
 
-	x.err(math.MaxInt32+1, "2147483648 is out of range for Integer")
-	x.err(uint32(math.MaxInt32+1), "2147483648 is out of range for Integer")
-	x.err(int64(math.MaxInt32+1), "2147483648 is out of range for Integer")
-	x.err(int64(math.MinInt32-1), "-2147483649 is out of range for Integer")
-	x.err(float64(math.MinInt32-1), "-2.147483649e+09 is out of range for Integer")
-	x.err(float64(math.MaxInt32+1), "2.147483648e+09 is out of range for Integer")
+	x.err(math.MaxInt32+1, "cannot convert 2147483648 to int32: out of range")
+	x.err(uint32(math.MaxInt32+1), "cannot convert 0x80000000 to int32: out of range")
+	x.err(int64(math.MaxInt32+1), "cannot convert 2147483648 to int32: out of range")
+	x.err(int64(math.MinInt32-1), "cannot convert -2147483649 to int32: out of range")
+	x.err(float64(math.MinInt32-1), "cannot convert -2.147483649e+09 to int32: out of range")
+	x.err(float64(math.MaxInt32+1), "cannot convert 2.147483648e+09 to int32: out of range")
 	// Float32 doesn't keep all the bits of an int32 so we need to exaggerate fof range error.
-	x.err(float64(2*math.MinInt32), "-4.294967296e+09 is out of range for Integer")
-	x.err(float64(-2*math.MaxInt32), "-4.294967294e+09 is out of range for Integer")
+	x.err(float64(2*math.MinInt32), "cannot convert -4.294967296e+09 to int32: out of range")
+	x.err(float64(-2*math.MaxInt32), "cannot convert -4.294967294e+09 to int32: out of range")
 
 	x.err("X", "strconv.ParseFloat: parsing \"X\": invalid syntax")
-	x.err(true, "cannot convert bool to Integer")
+	x.err(true, "cannot convert true to int32")
 	x.err(nil, "invalid CloudEvents value: <nil>")
 }
 
@@ -206,7 +206,7 @@ func TestTime(t *testing.T) {
 	x.str(timeStr, someTime)
 
 	x.err(nil, "invalid CloudEvents value: <nil>")
-	x.err(5, "cannot convert int32 to Time")
+	x.err(5, "cannot convert 5 to time.Time")
 	x.err((*time.Time)(nil), "invalid CloudEvents value: (*time.Time)(nil)")
 	x.err((*types.Timestamp)(nil), "invalid CloudEvents value: (*types.Timestamp)(nil)")
 	x.err("not a time", "parsing time \"not a time\" as \"2006-01-02T15:04:05.999999999Z07:00\": cannot parse \"not a time\" as \"2006\"")
diff --git a/test/http/conversion.go b/test/http/conversion.go
index 35b6fc4..0682d88 100644
--- a/test/http/conversion.go
+++ b/test/http/conversion.go
@@ -43,7 +43,6 @@ func UnitTestConvert(ctx context.Context, m transport.Message, err error) (*clou
 		event.SetSource("github.com/cloudevents/test/http/conversion")
 		event.SetType(fmt.Sprintf("io.cloudevents.conversion.http.%s", strings.ToLower(tx.Method)))
 		event.SetID("321-CBA")
-		event.SetExtension(unitTestIDKey, msg.Header.Get(unitTestIDKey))
 		event.Data = msg.Body
 
 		return &event, nil
@@ -82,8 +81,8 @@ func ClientConversion(t *testing.T, tc ConversionTest, topts ...cehttp.Option) {
 
 	recvCtx, recvCancel := context.WithCancel(context.Background())
 	go func() {
-		t.Log(ce.StartReceiver(recvCtx, func(got *cloudevents.Event) {
-			assertEventEquality(t, "got event", tc.want, got)
+		t.Log(ce.StartReceiver(recvCtx, func(got cloudevents.Event) {
+			assertEventEquality(t, "got event", tc.want, &got)
 		}))
 	}()
 
diff --git a/test/http/conversion_v02_test.go b/test/http/conversion_v02_test.go
index acde175..076ca7e 100644
--- a/test/http/conversion_v02_test.go
+++ b/test/http/conversion_v02_test.go
@@ -22,7 +22,7 @@ func TestClientConversion_v02(t *testing.T) {
 					Type:   "io.cloudevents.conversion.http.post",
 					Source: *cloudevents.ParseURLRef("github.com/cloudevents/test/http/conversion"),
 				}.AsV02(),
-				Data: map[string]string{"unittest": "response"},
+				Data: map[string]string{"hello": "unittest"},
 			},
 			asSent: &TapValidation{
 				Method: "POST",
diff --git a/test/http/direct.go b/test/http/direct.go
new file mode 100644
index 0000000..2645197
--- /dev/null
+++ b/test/http/direct.go
@@ -0,0 +1,87 @@
+package http
+
+import (
+	"context"
+	"net/http/httptest"
+	"testing"
+	"time"
+
+	"github.com/google/uuid"
+
+	cloudevents "github.com/cloudevents/sdk-go"
+	cehttp "github.com/cloudevents/sdk-go/pkg/cloudevents/transport/http"
+)
+
+// Direct Test:
+
+//         Obj -> Send -> Wire Format -> Receive -> Got
+// Given:   ^                 ^                      ^==Want
+// Obj is an event of a version.
+// Client is a set to binary or
+
+type DirectTapTest struct {
+	now    time.Time
+	event  *cloudevents.Event
+	want   *cloudevents.Event
+	asSent *TapValidation
+}
+
+type DirectTapTestCases map[string]DirectTapTest
+
+func ClientDirect(t *testing.T, tc DirectTapTest, topts ...cehttp.Option) {
+	tap := NewTap()
+	server := httptest.NewServer(tap)
+	defer server.Close()
+
+	if len(topts) == 0 {
+		topts = append(topts, cloudevents.WithBinaryEncoding())
+	}
+	topts = append(topts, cloudevents.WithTarget(server.URL))
+	topts = append(topts, cloudevents.WithPort(0)) // random port
+	transport, err := cloudevents.NewHTTPTransport(
+		topts...,
+	)
+	if err != nil {
+		t.Fatal(err)
+	}
+
+	tap.handler = transport
+
+	ce, err := cloudevents.NewClient(
+		transport,
+		cloudevents.WithEventDefaulter(AlwaysThen(tc.now)),
+	)
+	if err != nil {
+		t.Fatal(err)
+	}
+
+	testID := uuid.New().String()
+	ctx := cloudevents.ContextWithHeader(context.Background(), unitTestIDKey, testID)
+
+	recvCtx, recvCancel := context.WithTimeout(ctx, time.Second*5)
+	defer recvCancel()
+
+	var got *cloudevents.Event
+	go func() {
+		if err := ce.StartReceiver(recvCtx, func(event cloudevents.Event) {
+			got = &event
+			recvCancel()
+		}); err != nil {
+			t.Log(err)
+		}
+	}()
+
+	_, _, err = ce.Send(ctx, *tc.event)
+	if err != nil {
+		t.Fatal(err)
+	}
+
+	// Wait until the receiver is done.
+	<-recvCtx.Done()
+
+	assertEventEqualityExact(t, "event", tc.want, got)
+
+	if req, ok := tap.req[testID]; ok {
+		assertTappedEquality(t, "http request", tc.asSent, &req)
+	}
+}
diff --git a/test/http/direct_v1_test.go b/test/http/direct_v1_test.go
new file mode 100644
index 0000000..27a8953
--- /dev/null
+++ b/test/http/direct_v1_test.go
@@ -0,0 +1,146 @@
+package http
+
+import (
+	"fmt"
+	"testing"
+	"time"
+
+	"github.com/cloudevents/sdk-go"
+)
+
+func TestSenderReceiver_binary_v01(t *testing.T) {
+	now := time.Now()
+
+	testCases := DirectTapTestCases{
+		"Binary v1.0": {
+			now: now,
+			event: &cloudevents.Event{
+				Context: cloudevents.EventContextV1{
+					ID:      "ABC-123",
+					Type:    "unit.test.client.sent",
+					Source:  *cloudevents.ParseURIRef("/unit/test/client"),
+					Subject: strptr("resource"),
+				}.AsV1(),
+				Data: map[string]string{"hello": "unittest"},
+			},
+			want: &cloudevents.Event{
+				Context: cloudevents.EventContextV1{
+					ID:      "ABC-123",
+					Type:    "unit.test.client.sent",
+					Time:    &cloudevents.Timestamp{Time: now},
+					Source:  *cloudevents.ParseURIRef("/unit/test/client"),
+					Subject: strptr("resource"),
+				}.AsV1(),
+				Data: toBytes(map[string]interface{}{"hello": "unittest"}),
+			},
+			asSent: &TapValidation{
+				Method: "POST",
+				URI:    "/",
+				Header: map[string][]string{
+					"ce-specversion": {"1.0"},
+					"ce-id":          {"ABC-123"},
+					"ce-time":        {now.UTC().Format(time.RFC3339Nano)},
+					"ce-type":        {"unit.test.client.sent"},
+					"ce-source":      {"/unit/test/client"},
+					"ce-subject":     {"resource"},
+				},
+				Body:          `{"hello":"unittest"}`,
+				ContentLength: 20,
+			},
+		},
+	}
+
+	for n, tc := range testCases {
+		t.Run(n, func(t *testing.T) {
+			ClientDirect(t, tc)
+		})
+	}
+}
+
+func TestSenderReceiver_structured_v01(t *testing.T) {
+	now := time.Now()
+
+	testCases := DirectTapTestCases{
+		"Structured v1.0": {
+			now: now,
+			event: func() *cloudevents.Event {
+				event := cloudevents.NewEvent(cloudevents.VersionV1)
+				event.SetID("ABC-123")
+				event.SetType("unit.test.client.sent")
+				event.SetSource("/unit/test/client")
+				event.SetSubject("resource")
+				_ = event.SetData(map[string]string{"hello": "unittest"})
+				return &event
+			}(),
+			want: &cloudevents.Event{
+				Context: cloudevents.EventContextV1{
+					ID:      "ABC-123",
+					Type:    "unit.test.client.sent",
+					Time:    &cloudevents.Timestamp{Time: now},
+					Source:  *cloudevents.ParseURIRef("/unit/test/client"),
+					Subject: strptr("resource"),
+				}.AsV1(),
+				Data: toBytes(map[string]interface{}{"hello": "unittest"}),
+			},
+			asSent: &TapValidation{
+				Method: "POST",
+				URI:    "/",
+				Header: map[string][]string{
+					"content-type": {"application/cloudevents+json"},
+				},
+				Body:          fmt.Sprintf(`{"data":{"hello":"unittest"},"id":"ABC-123","source":"/unit/test/client","specversion":"1.0","subject":"resource","time":%q,"type":"unit.test.client.sent"}`, now.UTC().Format(time.RFC3339Nano)),
+				ContentLength: 182,
+			},
+		},
+	}
+
+	for n, tc := range testCases {
+		t.Run(n, func(t *testing.T) {
+			ClientDirect(t, tc, cloudevents.WithStructuredEncoding())
+		})
+	}
+}
+
+func TestSenderReceiver_data_base64_v01(t *testing.T) {
+	now := time.Now()
+
+	testCases := DirectTapTestCases{
+		"Structured v1.0": {
+			now: now,
+			event: func() *cloudevents.Event {
+				event := cloudevents.NewEvent(cloudevents.VersionV1)
+				event.SetID("ABC-123")
+				event.SetType("unit.test.client.sent")
+				event.SetSource("/unit/test/client")
+				event.SetSubject("resource")
+				_ = event.SetData([]byte("hello: unittest"))
+				return &event
+			}(),
+			want: &cloudevents.Event{
+				Context: cloudevents.EventContextV1{
+					ID:      "ABC-123",
+					Type:    "unit.test.client.sent",
+					Time:    &cloudevents.Timestamp{Time: now},
+					Source:  *cloudevents.ParseURIRef("/unit/test/client"),
+					Subject: strptr("resource"),
+				}.AsV1(),
+				Data: []byte("hello: unittest"),
+			},
+			asSent: &TapValidation{
+				Method: "POST",
+				URI:    "/",
+				Header: map[string][]string{
+					"content-type": {"application/cloudevents+json"},
+				},
+				Body:          fmt.Sprintf(`{"data_base64":"aGVsbG86IHVuaXR0ZXN0","id":"ABC-123","source":"/unit/test/client","specversion":"1.0","subject":"resource","time":%q,"type":"unit.test.client.sent"}`, now.UTC().Format(time.RFC3339Nano)),
+				ContentLength: 191,
+			},
+		},
+	}
+
+	for n, tc := range testCases {
+		t.Run(n, func(t *testing.T) {
+			ClientDirect(t, tc, cloudevents.WithStructuredEncoding())
+		})
+	}
+}
diff --git a/test/http/loopback.go b/test/http/loopback.go
index 557fe66..5684fd3 100644
--- a/test/http/loopback.go
+++ b/test/http/loopback.go
@@ -73,11 +73,13 @@ func ClientLoopback(t *testing.T, tc TapTest, topts ...cehttp.Option) {
 	recvCtx, recvCancel := context.WithCancel(context.Background())
 
 	go func() {
-		t.Log(ce.StartReceiver(recvCtx, func(resp *cloudevents.EventResponse) {
+		if err := ce.StartReceiver(recvCtx, func(resp *cloudevents.EventResponse) {
 			if tc.resp != nil {
 				resp.RespondWith(200, tc.resp)
 			}
-		}))
+		}); err != nil {
+			t.Log(err)
+		}
 	}()
 
 	_, got, err := ce.Send(ctx, *tc.event)
@@ -97,48 +99,3 @@ func ClientLoopback(t *testing.T, tc TapTest, topts ...cehttp.Option) {
 		assertTappedEquality(t, "http response", tc.asRecv, &resp)
 	}
 }
-
-// To help with debug, if needed.
-func printTap(t *testing.T, tap *tapHandler, testID string) {
-	if r, ok := tap.req[testID]; ok {
-		t.Log("tap request ", r.URI, r.Method)
-		if r.ContentLength > 0 {
-			t.Log(" .body: ", r.Body)
-		} else {
-			t.Log("tap request had no body.")
-		}
-
-		if len(r.Header) > 0 {
-			for h, vs := range r.Header {
-				for _, v := range vs {
-					t.Logf(" .header %s: %s", h, v)
-				}
-			}
-		} else {
-			t.Log("tap request had no headers.")
-		}
-	}
-
-	if r, ok := tap.resp[testID]; ok {
-		t.Log("tap response.status: ", r.Status)
-		if r.ContentLength > 0 {
-			t.Log(" .body: ", r.Body)
-		} else {
-			t.Log("tap response had no body.")
-		}
-
-		if len(r.Header) > 0 {
-			for h, vs := range r.Header {
-				for _, v := range vs {
-					t.Logf(" .header %s: %s", h, v)
-				}
-			}
-		} else {
-			t.Log("tap response had no headers.")
-		}
-	}
-}
-
-// Don't consider printTap as dead code even if not currently in use.
-// We want to keep it for possible future debugging.
-var _ = printTap
diff --git a/test/http/loopback_setters_test.go b/test/http/loopback_setters_test.go
index 0af2ef6..9f3a934 100644
--- a/test/http/loopback_setters_test.go
+++ b/test/http/loopback_setters_test.go
@@ -27,6 +27,7 @@ func TestClientLoopback_setters_binary_json(t *testing.T) {
 				event.SetID("ABC-123")
 				event.SetType("unit.test.client.sent")
 				event.SetSource("/unit/test/client")
+				event.SetDataContentType(cloudevents.ApplicationJSON)
 				if err := event.SetData(map[string]string{"hello": "unittest"}); err != nil {
 					t.Fatal(err)
 				}
@@ -37,6 +38,7 @@ func TestClientLoopback_setters_binary_json(t *testing.T) {
 				event.SetID("321-CBA")
 				event.SetType("unit.test.client.response")
 				event.SetSource("/unit/test/client")
+				event.SetDataContentType(cloudevents.ApplicationJSON)
 				if err := event.SetData(map[string]string{"unittest": "response"}); err != nil {
 					t.Fatal(err)
 				}
@@ -198,6 +200,7 @@ func TestClientLoopback_setters_binary_json_noBody(t *testing.T) {
 				event.SetID("ABC-123")
 				event.SetType("unit.test.client.sent")
 				event.SetSource("/unit/test/client")
+				event.SetDataContentType(cloudevents.ApplicationJSON)
 				return &event
 			},
 			resp: func(version string) *cloudevents.Event {
@@ -205,6 +208,7 @@ func TestClientLoopback_setters_binary_json_noBody(t *testing.T) {
 				event.SetID("321-CBA")
 				event.SetType("unit.test.client.response")
 				event.SetSource("/unit/test/client")
+				event.SetDataContentType(cloudevents.ApplicationJSON)
 				return &event
 			},
 			want: map[string]*cloudevents.Event{
@@ -357,6 +361,7 @@ func TestClientLoopback_setters_structured_json(t *testing.T) {
 				event.SetID("ABC-123")
 				event.SetType("unit.test.client.sent")
 				event.SetSource("/unit/test/client")
+				event.SetDataContentType(cloudevents.ApplicationJSON)
 				if err := event.SetData(map[string]string{"hello": "unittest"}); err != nil {
 					t.Fatal(err)
 				}
@@ -367,6 +372,7 @@ func TestClientLoopback_setters_structured_json(t *testing.T) {
 				event.SetID("321-CBA")
 				event.SetType("unit.test.client.response")
 				event.SetSource("/unit/test/client")
+				event.SetDataContentType(cloudevents.ApplicationJSON)
 				if err := event.SetData(map[string]string{"unittest": "response"}); err != nil {
 					t.Fatal(err)
 				}
@@ -496,6 +502,7 @@ func TestClientLoopback_setters_structured_json_base64(t *testing.T) {
 				event.SetID("ABC-123")
 				event.SetType("unit.test.client.sent")
 				event.SetSource("/unit/test/client")
+				event.SetDataContentType(cloudevents.ApplicationJSON)
 				event.SetDataContentEncoding(cloudevents.Base64)
 				if err := event.SetData(map[string]string{"hello": "unittest"}); err != nil {
 					t.Fatal(err)
@@ -507,6 +514,7 @@ func TestClientLoopback_setters_structured_json_base64(t *testing.T) {
 				event.SetID("321-CBA")
 				event.SetType("unit.test.client.response")
 				event.SetSource("/unit/test/client")
+				event.SetDataContentType(cloudevents.ApplicationJSON)
 				event.SetDataContentEncoding(cloudevents.Base64)
 				if err := event.SetData(map[string]string{"unittest": "response"}); err != nil {
 					t.Fatal(err)
diff --git a/test/http/loopback_v01_test.go b/test/http/loopback_v01_test.go
index 7c552fc..6fba1b2 100644
--- a/test/http/loopback_v01_test.go
+++ b/test/http/loopback_v01_test.go
@@ -15,17 +15,19 @@ func TestClientLoopback_binary_v01tov01(t *testing.T) {
 			now: now,
 			event: &cloudevents.Event{
 				Context: cloudevents.EventContextV02{
-					ID:     "ABC-123",
-					Type:   "unit.test.client.sent",
-					Source: *cloudevents.ParseURLRef("/unit/test/client"),
+					ID:          "ABC-123",
+					Type:        "unit.test.client.sent",
+					Source:      *cloudevents.ParseURLRef("/unit/test/client"),
+					ContentType: cloudevents.StringOfApplicationJSON(),
 				}.AsV01(),
 				Data: map[string]string{"hello": "unittest"},
 			},
 			resp: &cloudevents.Event{
 				Context: cloudevents.EventContextV01{
-					EventID:   "321-CBA",
-					EventType: "unit.test.client.response",
-					Source:    *cloudevents.ParseURLRef("/unit/test/client"),
+					EventID:     "321-CBA",
+					EventType:   "unit.test.client.response",
+					Source:      *cloudevents.ParseURLRef("/unit/test/client"),
+					ContentType: cloudevents.StringOfApplicationJSON(),
 				}.AsV01(),
 				Data: map[string]string{"unittest": "response"},
 			},
@@ -84,17 +86,19 @@ func TestClientLoopback_binary_v01tov02(t *testing.T) {
 			now: now,
 			event: &cloudevents.Event{
 				Context: cloudevents.EventContextV02{
-					ID:     "ABC-123",
-					Type:   "unit.test.client.sent",
-					Source: *cloudevents.ParseURLRef("/unit/test/client"),
+					ID:          "ABC-123",
+					Type:        "unit.test.client.sent",
+					Source:      *cloudevents.ParseURLRef("/unit/test/client"),
+					ContentType: cloudevents.StringOfApplicationJSON(),
 				}.AsV01(),
 				Data: map[string]string{"hello": "unittest"},
 			},
 			resp: &cloudevents.Event{
 				Context: cloudevents.EventContextV02{
-					ID:     "321-CBA",
-					Type:   "unit.test.client.response",
-					Source: *cloudevents.ParseURLRef("/unit/test/client"),
+					ID:          "321-CBA",
+					Type:        "unit.test.client.response",
+					Source:      *cloudevents.ParseURLRef("/unit/test/client"),
+					ContentType: cloudevents.StringOfApplicationJSON(),
 				}.AsV02(),
 				Data: map[string]string{"unittest": "response"},
 			},
@@ -153,17 +157,19 @@ func TestClientLoopback_binary_v01tov03(t *testing.T) {
 			now: now,
 			event: &cloudevents.Event{
 				Context: cloudevents.EventContextV02{
-					ID:     "ABC-123",
-					Type:   "unit.test.client.sent",
-					Source: *cloudevents.ParseURLRef("/unit/test/client"),
+					ID:          "ABC-123",
+					Type:        "unit.test.client.sent",
+					Source:      *cloudevents.ParseURLRef("/unit/test/client"),
+					ContentType: cloudevents.StringOfApplicationJSON(),
 				}.AsV01(),
 				Data: map[string]string{"hello": "unittest"},
 			},
 			resp: &cloudevents.Event{
 				Context: cloudevents.EventContextV03{
-					ID:     "321-CBA",
-					Type:   "unit.test.client.response",
-					Source: *cloudevents.ParseURLRef("/unit/test/client"),
+					ID:              "321-CBA",
+					Type:            "unit.test.client.response",
+					Source:          *cloudevents.ParseURLRef("/unit/test/client"),
+					DataContentType: cloudevents.StringOfApplicationJSON(),
 				}.AsV03(),
 				Data: map[string]string{"unittest": "response"},
 			},
@@ -222,17 +228,19 @@ func TestClientLoopback_binary_v01tov1(t *testing.T) {
 			now: now,
 			event: &cloudevents.Event{
 				Context: cloudevents.EventContextV02{
-					ID:     "ABC-123",
-					Type:   "unit.test.client.sent",
-					Source: *cloudevents.ParseURLRef("/unit/test/client"),
+					ID:          "ABC-123",
+					Type:        "unit.test.client.sent",
+					Source:      *cloudevents.ParseURLRef("/unit/test/client"),
+					ContentType: cloudevents.StringOfApplicationJSON(),
 				}.AsV01(),
 				Data: map[string]string{"hello": "unittest"},
 			},
 			resp: &cloudevents.Event{
 				Context: cloudevents.EventContextV03{
-					ID:     "321-CBA",
-					Type:   "unit.test.client.response",
-					Source: *cloudevents.ParseURLRef("/unit/test/client"),
+					ID:              "321-CBA",
+					Type:            "unit.test.client.response",
+					Source:          *cloudevents.ParseURLRef("/unit/test/client"),
+					DataContentType: cloudevents.StringOfApplicationJSON(),
 				}.AsV03(),
 				Data: map[string]string{"unittest": "response"},
 			},
diff --git a/test/http/loopback_v02_test.go b/test/http/loopback_v02_test.go
index bd97826..250515b 100644
--- a/test/http/loopback_v02_test.go
+++ b/test/http/loopback_v02_test.go
@@ -16,17 +16,19 @@ func TestClientLoopback_binary_v02tov01(t *testing.T) {
 			now: now,
 			event: &cloudevents.Event{
 				Context: cloudevents.EventContextV02{
-					ID:     "ABC-123",
-					Type:   "unit.test.client.sent",
-					Source: *cloudevents.ParseURLRef("/unit/test/client"),
+					ID:          "ABC-123",
+					Type:        "unit.test.client.sent",
+					Source:      *cloudevents.ParseURLRef("/unit/test/client"),
+					ContentType: cloudevents.StringOfApplicationJSON(),
 				}.AsV02(),
 				Data: map[string]string{"hello": "unittest"},
 			},
 			resp: &cloudevents.Event{
 				Context: cloudevents.EventContextV01{
-					EventID:   "321-CBA",
-					EventType: "unit.test.client.response",
-					Source:    *cloudevents.ParseURLRef("/unit/test/client"),
+					EventID:     "321-CBA",
+					EventType:   "unit.test.client.response",
+					Source:      *cloudevents.ParseURLRef("/unit/test/client"),
+					ContentType: cloudevents.StringOfApplicationJSON(),
 				}.AsV01(),
 				Data: map[string]string{"unittest": "response"},
 			},
@@ -85,17 +87,19 @@ func TestClientLoopback_binary_v02tov02(t *testing.T) {
 			now: now,
 			event: &cloudevents.Event{
 				Context: cloudevents.EventContextV02{
-					ID:     "ABC-123",
-					Type:   "unit.test.client.sent",
-					Source: *cloudevents.ParseURLRef("/unit/test/client"),
+					ID:          "ABC-123",
+					Type:        "unit.test.client.sent",
+					Source:      *cloudevents.ParseURLRef("/unit/test/client"),
+					ContentType: cloudevents.StringOfApplicationJSON(),
 				}.AsV02(),
 				Data: map[string]string{"hello": "unittest"},
 			},
 			resp: &cloudevents.Event{
 				Context: cloudevents.EventContextV02{
-					ID:     "321-CBA",
-					Type:   "unit.test.client.response",
-					Source: *cloudevents.ParseURLRef("/unit/test/client"),
+					ID:          "321-CBA",
+					Type:        "unit.test.client.response",
+					Source:      *cloudevents.ParseURLRef("/unit/test/client"),
+					ContentType: cloudevents.StringOfApplicationJSON(),
 				}.AsV02(),
 				Data: map[string]string{"unittest": "response"},
 			},
@@ -154,17 +158,19 @@ func TestClientLoopback_structured_v02tov02(t *testing.T) {
 			now: now,
 			event: &cloudevents.Event{
 				Context: cloudevents.EventContextV02{
-					ID:     "ABC-123",
-					Type:   "unit.test.client.sent",
-					Source: *cloudevents.ParseURLRef("/unit/test/client"),
+					ID:          "ABC-123",
+					Type:        "unit.test.client.sent",
+					Source:      *cloudevents.ParseURLRef("/unit/test/client"),
+					ContentType: cloudevents.StringOfApplicationJSON(),
 				}.AsV02(),
 				Data: map[string]string{"hello": "unittest"},
 			},
 			resp: &cloudevents.Event{
 				Context: cloudevents.EventContextV02{
-					ID:     "321-CBA",
-					Type:   "unit.test.client.response",
-					Source: *cloudevents.ParseURLRef("/unit/test/client"),
+					ID:          "321-CBA",
+					Type:        "unit.test.client.response",
+					Source:      *cloudevents.ParseURLRef("/unit/test/client"),
+					ContentType: cloudevents.StringOfApplicationJSON(),
 				}.AsV02(),
 				Data: map[string]string{"unittest": "response"},
 			},
@@ -215,17 +221,19 @@ func TestClientLoopback_binary_v02tov03(t *testing.T) {
 			now: now,
 			event: &cloudevents.Event{
 				Context: cloudevents.EventContextV02{
-					ID:     "ABC-123",
-					Type:   "unit.test.client.sent",
-					Source: *cloudevents.ParseURLRef("/unit/test/client"),
+					ID:          "ABC-123",
+					Type:        "unit.test.client.sent",
+					Source:      *cloudevents.ParseURLRef("/unit/test/client"),
+					ContentType: cloudevents.StringOfApplicationJSON(),
 				}.AsV02(),
 				Data: map[string]string{"hello": "unittest"},
 			},
 			resp: &cloudevents.Event{
 				Context: cloudevents.EventContextV03{
-					ID:     "321-CBA",
-					Type:   "unit.test.client.response",
-					Source: *cloudevents.ParseURLRef("/unit/test/client"),
+					ID:              "321-CBA",
+					Type:            "unit.test.client.response",
+					Source:          *cloudevents.ParseURLRef("/unit/test/client"),
+					DataContentType: cloudevents.StringOfApplicationJSON(),
 				}.AsV03(),
 				Data: map[string]string{"unittest": "response"},
 			},
diff --git a/test/http/loopback_v03_test.go b/test/http/loopback_v03_test.go
index 7df4fee..de84475 100644
--- a/test/http/loopback_v03_test.go
+++ b/test/http/loopback_v03_test.go
@@ -16,18 +16,20 @@ func TestClientLoopback_binary_v03tov01(t *testing.T) {
 			now: now,
 			event: &cloudevents.Event{
 				Context: cloudevents.EventContextV03{
-					ID:      "ABC-123",
-					Type:    "unit.test.client.sent",
-					Source:  *cloudevents.ParseURLRef("/unit/test/client"),
-					Subject: strptr("resource"),
+					ID:              "ABC-123",
+					Type:            "unit.test.client.sent",
+					Source:          *cloudevents.ParseURLRef("/unit/test/client"),
+					Subject:         strptr("resource"),
+					DataContentType: cloudevents.StringOfApplicationJSON(),
 				}.AsV03(),
 				Data: map[string]string{"hello": "unittest"},
 			},
 			resp: &cloudevents.Event{
 				Context: cloudevents.EventContextV01{
-					EventID:   "321-CBA",
-					EventType: "unit.test.client.response",
-					Source:    *cloudevents.ParseURLRef("/unit/test/client"),
+					EventID:     "321-CBA",
+					EventType:   "unit.test.client.response",
+					Source:      *cloudevents.ParseURLRef("/unit/test/client"),
+					ContentType: cloudevents.StringOfApplicationJSON(),
 				}.AsV01(),
 				Data: map[string]string{"unittest": "response"},
 			},
@@ -87,18 +89,20 @@ func TestClientLoopback_binary_v03tov02(t *testing.T) {
 			now: now,
 			event: &cloudevents.Event{
 				Context: cloudevents.EventContextV03{
-					ID:      "ABC-123",
-					Type:    "unit.test.client.sent",
-					Source:  *cloudevents.ParseURLRef("/unit/test/client"),
-					Subject: strptr("resource"),
+					ID:              "ABC-123",
+					Type:            "unit.test.client.sent",
+					Source:          *cloudevents.ParseURLRef("/unit/test/client"),
+					Subject:         strptr("resource"),
+					DataContentType: cloudevents.StringOfApplicationJSON(),
 				}.AsV03(),
 				Data: map[string]string{"hello": "unittest"},
 			},
 			resp: &cloudevents.Event{
 				Context: cloudevents.EventContextV02{
-					ID:     "321-CBA",
-					Type:   "unit.test.client.response",
-					Source: *cloudevents.ParseURLRef("/unit/test/client"),
+					ID:          "321-CBA",
+					Type:        "unit.test.client.response",
+					Source:      *cloudevents.ParseURLRef("/unit/test/client"),
+					ContentType: cloudevents.StringOfApplicationJSON(),
 				}.AsV02(),
 				Data: map[string]string{"unittest": "response"},
 			},
@@ -158,18 +162,20 @@ func TestClientLoopback_binary_v03tov03(t *testing.T) {
 			now: now,
 			event: &cloudevents.Event{
 				Context: cloudevents.EventContextV03{
-					ID:      "ABC-123",
-					Type:    "unit.test.client.sent",
-					Source:  *cloudevents.ParseURLRef("/unit/test/client"),
-					Subject: strptr("resource"),
+					ID:              "ABC-123",
+					Type:            "unit.test.client.sent",
+					Source:          *cloudevents.ParseURLRef("/unit/test/client"),
+					Subject:         strptr("resource"),
+					DataContentType: cloudevents.StringOfApplicationJSON(),
 				}.AsV03(),
 				Data: map[string]string{"hello": "unittest"},
 			},
 			resp: &cloudevents.Event{
 				Context: cloudevents.EventContextV03{
-					ID:     "321-CBA",
-					Type:   "unit.test.client.response",
-					Source: *cloudevents.ParseURLRef("/unit/test/client"),
+					ID:              "321-CBA",
+					Type:            "unit.test.client.response",
+					Source:          *cloudevents.ParseURLRef("/unit/test/client"),
+					DataContentType: cloudevents.StringOfApplicationJSON(),
 				}.AsV03(),
 				Data: map[string]string{"unittest": "response"},
 			},
@@ -234,6 +240,7 @@ func TestClientLoopback_binary_base64_v03tov03(t *testing.T) {
 					Source:              *cloudevents.ParseURLRef("/unit/test/client"),
 					Subject:             strptr("resource"),
 					DataContentEncoding: cloudevents.StringOfBase64(),
+					DataContentType:     cloudevents.StringOfApplicationJSON(),
 				}.AsV03(),
 				Data: map[string]string{"hello": "unittest"},
 			},
@@ -243,6 +250,7 @@ func TestClientLoopback_binary_base64_v03tov03(t *testing.T) {
 					Type:                "unit.test.client.response",
 					Source:              *cloudevents.ParseURLRef("/unit/test/client"),
 					DataContentEncoding: cloudevents.StringOfBase64(),
+					DataContentType:     cloudevents.StringOfApplicationJSON(),
 				}.AsV03(),
 				Data: map[string]string{"unittest": "response"},
 			},
@@ -309,6 +317,7 @@ func TestClientLoopback_structured_base64_v03tov03(t *testing.T) {
 					Type:                "unit.test.client.sent",
 					Source:              *cloudevents.ParseURLRef("/unit/test/client"),
 					DataContentEncoding: cloudevents.StringOfBase64(),
+					DataContentType:     cloudevents.StringOfApplicationJSON(),
 				}.AsV03(),
 				Data: map[string]string{"hello": "unittest"},
 			},
@@ -318,6 +327,7 @@ func TestClientLoopback_structured_base64_v03tov03(t *testing.T) {
 					Type:                "unit.test.client.response",
 					Source:              *cloudevents.ParseURLRef("/unit/test/client"),
 					DataContentEncoding: cloudevents.StringOfBase64(),
+					DataContentType:     cloudevents.StringOfApplicationJSON(),
 				}.AsV03(),
 				Data: map[string]string{"unittest": "response"},
 			},
diff --git a/test/http/loopback_v1_test.go b/test/http/loopback_v1_test.go
index 3ab895d..89d0df9 100644
--- a/test/http/loopback_v1_test.go
+++ b/test/http/loopback_v1_test.go
@@ -15,18 +15,20 @@ func TestClientLoopback_binary_v1tov01(t *testing.T) {
 			now: now,
 			event: &cloudevents.Event{
 				Context: cloudevents.EventContextV1{
-					ID:      "ABC-123",
-					Type:    "unit.test.client.sent",
-					Source:  *cloudevents.ParseURIRef("/unit/test/client"),
-					Subject: strptr("resource"),
+					ID:              "ABC-123",
+					Type:            "unit.test.client.sent",
+					Source:          *cloudevents.ParseURIRef("/unit/test/client"),
+					Subject:         strptr("resource"),
+					DataContentType: cloudevents.StringOfApplicationJSON(),
 				}.AsV1(),
 				Data: map[string]string{"hello": "unittest"},
 			},
 			resp: &cloudevents.Event{
 				Context: cloudevents.EventContextV01{
-					EventID:   "321-CBA",
-					EventType: "unit.test.client.response",
-					Source:    *cloudevents.ParseURLRef("/unit/test/client"),
+					EventID:     "321-CBA",
+					EventType:   "unit.test.client.response",
+					Source:      *cloudevents.ParseURLRef("/unit/test/client"),
+					ContentType: cloudevents.StringOfApplicationJSON(),
 				}.AsV01(),
 				Data: map[string]string{"unittest": "response"},
 			},
@@ -86,18 +88,20 @@ func TestClientLoopback_binary_v1tov02(t *testing.T) {
 			now: now,
 			event: &cloudevents.Event{
 				Context: cloudevents.EventContextV1{
-					ID:      "ABC-123",
-					Type:    "unit.test.client.sent",
-					Source:  *cloudevents.ParseURIRef("/unit/test/client"),
-					Subject: strptr("resource"),
+					ID:              "ABC-123",
+					Type:            "unit.test.client.sent",
+					Source:          *cloudevents.ParseURIRef("/unit/test/client"),
+					Subject:         strptr("resource"),
+					DataContentType: cloudevents.StringOfApplicationJSON(),
 				}.AsV1(),
 				Data: map[string]string{"hello": "unittest"},
 			},
 			resp: &cloudevents.Event{
 				Context: cloudevents.EventContextV02{
-					ID:     "321-CBA",
-					Type:   "unit.test.client.response",
-					Source: *cloudevents.ParseURLRef("/unit/test/client"),
+					ID:          "321-CBA",
+					Type:        "unit.test.client.response",
+					Source:      *cloudevents.ParseURLRef("/unit/test/client"),
+					ContentType: cloudevents.StringOfApplicationJSON(),
 				}.AsV02(),
 				Data: map[string]string{"unittest": "response"},
 			},
@@ -157,18 +161,20 @@ func TestClientLoopback_binary_v1tov03(t *testing.T) {
 			now: now,
 			event: &cloudevents.Event{
 				Context: cloudevents.EventContextV1{
-					ID:      "ABC-123",
-					Type:    "unit.test.client.sent",
-					Source:  *cloudevents.ParseURIRef("/unit/test/client"),
-					Subject: strptr("resource"),
+					ID:              "ABC-123",
+					Type:            "unit.test.client.sent",
+					Source:          *cloudevents.ParseURIRef("/unit/test/client"),
+					Subject:         strptr("resource"),
+					DataContentType: cloudevents.StringOfApplicationJSON(),
 				}.AsV1(),
 				Data: map[string]string{"hello": "unittest"},
 			},
 			resp: &cloudevents.Event{
 				Context: cloudevents.EventContextV03{
-					ID:     "321-CBA",
-					Type:   "unit.test.client.response",
-					Source: *cloudevents.ParseURLRef("/unit/test/client"),
+					ID:              "321-CBA",
+					Type:            "unit.test.client.response",
+					Source:          *cloudevents.ParseURLRef("/unit/test/client"),
+					DataContentType: cloudevents.StringOfApplicationJSON(),
 				}.AsV03(),
 				Data: map[string]string{"unittest": "response"},
 			},
@@ -224,22 +230,24 @@ func TestClientLoopback_binary_v1tov1(t *testing.T) {
 	now := time.Now()
 
 	testCases := TapTestCases{
-		"Loopback v1.0 -> v0.3": {
+		"Loopback v1.0 -> v1.0": {
 			now: now,
 			event: &cloudevents.Event{
 				Context: cloudevents.EventContextV1{
-					ID:      "ABC-123",
-					Type:    "unit.test.client.sent",
-					Source:  *cloudevents.ParseURIRef("/unit/test/client"),
-					Subject: strptr("resource"),
+					ID:              "ABC-123",
+					Type:            "unit.test.client.sent",
+					Source:          *cloudevents.ParseURIRef("/unit/test/client"),
+					Subject:         strptr("resource"),
+					DataContentType: cloudevents.StringOfApplicationJSON(),
 				}.AsV1(),
 				Data: map[string]string{"hello": "unittest"},
 			},
 			resp: &cloudevents.Event{
 				Context: cloudevents.EventContextV1{
-					ID:     "321-CBA",
-					Type:   "unit.test.client.response",
-					Source: *cloudevents.ParseURIRef("/unit/test/client"),
+					ID:              "321-CBA",
+					Type:            "unit.test.client.response",
+					Source:          *cloudevents.ParseURIRef("/unit/test/client"),
+					DataContentType: cloudevents.StringOfApplicationJSON(),
 				}.AsV1(),
 				Data: map[string]string{"unittest": "response"},
 			},
diff --git a/test/http/tap_handler.go b/test/http/tap_handler.go
index f5cdf2e..1a13728 100644
--- a/test/http/tap_handler.go
+++ b/test/http/tap_handler.go
@@ -6,6 +6,7 @@ import (
 	"io/ioutil"
 	"net/http"
 	"net/http/httptest"
+	"testing"
 )
 
 type TapValidation struct {
@@ -31,6 +32,51 @@ func NewTap() *tapHandler {
 	}
 }
 
+// To help with debug, if needed.
+func printTap(t *testing.T, tap *tapHandler, testID string) {
+	if r, ok := tap.req[testID]; ok {
+		t.Log("tap request ", r.URI, r.Method)
+		if r.ContentLength > 0 {
+			t.Log(" .body: ", r.Body)
+		} else {
+			t.Log("tap request had no body.")
+		}
+
+		if len(r.Header) > 0 {
+			for h, vs := range r.Header {
+				for _, v := range vs {
+					t.Logf(" .header %s: %s", h, v)
+				}
+			}
+		} else {
+			t.Log("tap request had no headers.")
+		}
+	}
+
+	if r, ok := tap.resp[testID]; ok {
+		t.Log("tap response.status: ", r.Status)
+		if r.ContentLength > 0 {
+			t.Log(" .body: ", r.Body)
+		} else {
+			t.Log("tap response had no body.")
+		}
+
+		if len(r.Header) > 0 {
+			for h, vs := range r.Header {
+				for _, v := range vs {
+					t.Logf(" .header %s: %s", h, v)
+				}
+			}
+		} else {
+			t.Log("tap response had no headers.")
+		}
+	}
+}
+
+// Don't consider printTap as dead code even if not currently in use.
+// We want to keep it for possible future debugging.
+var _ = printTap
+
 const (
 	unitTestIDKey = "Test-Ce-Id"
 )
diff --git a/test/http/validation.go b/test/http/validation.go
index 51c8830..573c1b5 100644
--- a/test/http/validation.go
+++ b/test/http/validation.go
@@ -1,6 +1,8 @@
 package http
 
 import (
+	"encoding/json"
+	"fmt"
 	"strings"
 	"testing"
 
@@ -20,8 +22,28 @@ var (
 	}
 )
 
+func toBytes(body map[string]interface{}) []byte {
+	b, err := json.Marshal(body)
+	if err != nil {
+		return []byte(fmt.Sprintf(`{"error":%q}`, err.Error()))
+	}
+	return b
+}
+
+func assertEventEqualityExact(t *testing.T, ctx string, expected, actual *cloudevents.Event) {
+	if diff := cmp.Diff(expected, actual, cmpopts.IgnoreFields(cloudevents.Event{}, "Data", "DataEncoded", "DataBinary")); diff != "" {
+		t.Errorf("Unexpected difference in %s (-want, +got): %v", ctx, diff)
+	}
+	if expected == nil || actual == nil {
+		return
+	}
+	if diff := cmp.Diff(expected.Data, actual.Data); diff != "" {
+		t.Errorf("Unexpected data difference in %s (-want, +got): %v", ctx, diff)
+	}
+}
+
 func assertEventEquality(t *testing.T, ctx string, expected, actual *cloudevents.Event) {
-	if diff := cmp.Diff(expected, actual, cmpopts.IgnoreFields(cloudevents.Event{}, "Data", "DataEncoded")); diff != "" {
+	if diff := cmp.Diff(expected, actual, cmpopts.IgnoreFields(cloudevents.Event{}, "Data", "DataEncoded", "DataBinary")); diff != "" {
 		t.Errorf("Unexpected difference in %s (-want, +got): %v", ctx, diff)
 	}
 	if expected == nil || actual == nil {
@@ -39,7 +61,7 @@ func assertEventEquality(t *testing.T, ctx string, expected, actual *cloudevents
 
 func assertTappedEquality(t *testing.T, ctx string, expected, actual *TapValidation) {
 	canonicalizeHeaders(expected, actual)
-	if diff := cmp.Diff(expected, actual); diff != "" {
+	if diff := cmp.Diff(expected, actual, cmpopts.IgnoreFields(TapValidation{}, "ContentLength")); diff != "" {
 		t.Errorf("Unexpected difference in %s (-want, +got): %v", ctx, diff)
 	}
 }