Repository: hamba/avro
Pull request #15 (2019-05-13 06:59:00)

IPC Prep


Patch:
diff --git a/README.md b/README.md
index 1c461ba..48354ec 100644
--- a/README.md
+++ b/README.md
@@ -91,8 +91,11 @@ type name, or scheam full name in the case of a named schema (enum, fixed or rec
 * ***T:** This is allowed in a "nullable" union. A nullable union is defined as a two schema union, 
 with the first being `null` (ie. `["null", "string"]`), in this case a `*T` is allowed, 
 with `T` matching the conversion table above.
-* **avro.UnionType:** A `struct` in implementing `avro.UnionType` can be provided, allowing for
-strong type encoding. An example can be found in the [godoc](https://godoc.org/github.com/hamba/avro).
+* **interface{}:** An `interface` can be provided and the type or name resolved. Primitive types
+are pre-registered, but named types, maps and slices will need to be registered with the `Register` function. In the 
+case of arrays and maps the enclosed schema type or name is postfix to the type
+with a `:` separator, e.g `"map:string"`. If any type cannot be resolved the map type above is used unless
+`Config.UnionResolutionError` is set to `true` in which case an error is returned.
 
 ## Benchmark
 
diff --git a/avro.go b/avro.go
deleted file mode 100644
index 8c4d82a..0000000
--- a/avro.go
+++ /dev/null
@@ -1,9 +0,0 @@
-package avro
-
-// UnionType is the interface implemented by types that can
-// resolve schema types/names to and from objects.
-type UnionType interface {
-	Value() *interface{}
-	SetType(string) error
-	GetType() (string, error)
-}
diff --git a/codec.go b/codec.go
index 2039771..a9e4d86 100644
--- a/codec.go
+++ b/codec.go
@@ -8,6 +8,8 @@ import (
 	"github.com/modern-go/reflect2"
 )
 
+type null struct{}
+
 // ValDecoder represents an internal value decoder.
 //
 // You should never use ValDecoder directly.
@@ -72,8 +74,8 @@ func (c *frozenConfig) DecoderOf(schema Schema, typ reflect2.Type) ValDecoder {
 }
 
 func decoderOfType(cfg *frozenConfig, schema Schema, typ reflect2.Type) ValDecoder {
-	// Handle eface case
-	if typ.Kind() == reflect.Interface {
+	// Handle eface case when it isnt a union
+	if typ.Kind() == reflect.Interface && schema.Type() != Union {
 		if _, ok := typ.(*reflect2.UnsafeIFaceType); !ok {
 			return &efaceDecoder{schema: schema}
 		}
@@ -111,6 +113,10 @@ func decoderOfType(cfg *frozenConfig, schema Schema, typ reflect2.Type) ValDecod
 }
 
 func (c *frozenConfig) EncoderOf(schema Schema, typ reflect2.Type) ValEncoder {
+	if typ == nil {
+		typ = reflect2.TypeOf((*null)(nil))
+	}
+
 	rtype := typ.RType()
 	encoder := c.getEncoderFromCache(schema.Fingerprint(), rtype)
 	if encoder != nil {
@@ -139,7 +145,7 @@ func encoderOfType(cfg *frozenConfig, schema Schema, typ reflect2.Type) ValEncod
 	}
 
 	switch schema.Type() {
-	case String, Bytes, Int, Long, Float, Double, Boolean:
+	case String, Bytes, Int, Long, Float, Double, Boolean, Null:
 		return createEncoderOfNative(schema, typ)
 
 	case Record:
diff --git a/codec_native.go b/codec_native.go
index 9e237d1..81a075c 100644
--- a/codec_native.go
+++ b/codec_native.go
@@ -137,9 +137,17 @@ func createEncoderOfNative(schema Schema, typ reflect2.Type) ValEncoder {
 		return &bytesCodec{sliceType: typ.(*reflect2.UnsafeSliceType)}
 	}
 
+	if schema.Type() == Null {
+		return &nullCodec{}
+	}
+
 	return &errorEncoder{err: fmt.Errorf("avro: %s is unsupported for Avro %s", typ.String(), schema.Type())}
 }
 
+type nullCodec struct{}
+
+func (*nullCodec) Encode(ptr unsafe.Pointer, w *Writer) {}
+
 type boolCodec struct{}
 
 func (*boolCodec) Decode(ptr unsafe.Pointer, r *Reader) {
diff --git a/codec_record.go b/codec_record.go
index fb9a72a..4aa024a 100644
--- a/codec_record.go
+++ b/codec_record.go
@@ -119,16 +119,16 @@ func encoderOfStruct(cfg *frozenConfig, schema Schema, typ reflect2.Type) ValEnc
 		sf := structDesc.Fields.Get(field.Name())
 
 		if sf == nil {
-			if field.Default() == nil {
-				if field.Type().Type() == Null {
-					// We write nothing in a Null case, just skip it
-					continue
-				}
-
+			if !field.HasDefault() {
 				// In all other cases, this is a required field
 				return &errorEncoder{err: fmt.Errorf("avro: record %s is missing required field %s", rec.FullName(), field.Name())}
 			}
 
+			if field.Default() == nil {
+				// We write nothing in a Null case, just skip it
+				continue
+			}
+
 			defaultType := reflect2.TypeOf(field.Default())
 			fields = append(fields, &structFieldEncoder{
 				defaultPtr: reflect2.PtrOf(field.Default()),
@@ -234,7 +234,7 @@ func encoderOfRecord(cfg *frozenConfig, schema Schema, typ reflect2.Type) ValEnc
 	for i, field := range rec.Fields() {
 		fields[i] = mapEncoderField{
 			name:    field.Name(),
-			hasDef:  field.Default() != nil || field.Type().Type() == Null,
+			hasDef:  field.HasDefault(),
 			def:     field.Default(),
 			encoder: encoderOfType(cfg, field.Type(), mapType.Elem()),
 		}
diff --git a/codec_skip.go b/codec_skip.go
index 2f4557e..f9e34d7 100644
--- a/codec_skip.go
+++ b/codec_skip.go
@@ -203,7 +203,7 @@ type unionSkipDecoder struct {
 }
 
 func (d *unionSkipDecoder) Decode(ptr unsafe.Pointer, r *Reader) {
-	resSchema := getUnionSchema(d.schema, r)
+	_, resSchema := getUnionSchema(d.schema, r)
 	if resSchema == nil {
 		return
 	}
diff --git a/codec_union.go b/codec_union.go
index 11b7f6e..6d14556 100644
--- a/codec_union.go
+++ b/codec_union.go
@@ -4,13 +4,12 @@ import (
 	"errors"
 	"fmt"
 	"reflect"
+	"strings"
 	"unsafe"
 
 	"github.com/modern-go/reflect2"
 )
 
-var unionType = reflect2.TypeOfPtr((*UnionType)(nil)).Elem()
-
 func createDecoderOfUnion(cfg *frozenConfig, schema Schema, typ reflect2.Type) ValDecoder {
 	switch typ.Kind() {
 	case reflect.Map:
@@ -21,13 +20,15 @@ func createDecoderOfUnion(cfg *frozenConfig, schema Schema, typ reflect2.Type) V
 		return decoderOfMapUnion(cfg, schema, typ)
 
 	case reflect.Ptr:
-		if typ.Implements(unionType) {
-			return decoderOfTypedUnion(schema, typ)
-		}
 		if !schema.(*UnionSchema).Nullable() {
 			break
 		}
 		return decoderOfPtrUnion(cfg, schema, typ)
+
+	case reflect.Interface:
+		if _, ok := typ.(*reflect2.UnsafeIFaceType); !ok {
+			return decoderOfResolvedUnion(cfg, schema)
+		}
 	}
 
 	return &errorDecoder{err: fmt.Errorf("avro: %s is unsupported for Avro %s", typ.String(), schema.Type())}
@@ -43,16 +44,13 @@ func createEncoderOfUnion(cfg *frozenConfig, schema Schema, typ reflect2.Type) V
 		return encoderOfMapUnion(cfg, schema, typ)
 
 	case reflect.Ptr:
-		if typ.Implements(unionType) {
-			return encoderOfTypedUnion(schema, typ)
-		}
 		if !schema.(*UnionSchema).Nullable() {
 			break
 		}
 		return encoderOfPtrUnion(cfg, schema, typ)
 	}
 
-	return &errorEncoder{err: fmt.Errorf("avro: %s is unsupported for Avro %s", typ.String(), schema.Type())}
+	return encoderOfResolverUnion(cfg, schema, typ)
 }
 
 func decoderOfMapUnion(cfg *frozenConfig, schema Schema, typ reflect2.Type) ValDecoder {
@@ -75,7 +73,7 @@ type mapUnionDecoder struct {
 }
 
 func (d *mapUnionDecoder) Decode(ptr unsafe.Pointer, r *Reader) {
-	resSchema := getUnionSchema(d.schema, r)
+	_, resSchema := getUnionSchema(d.schema, r)
 	if resSchema == nil {
 		return
 	}
@@ -89,10 +87,7 @@ func (d *mapUnionDecoder) Decode(ptr unsafe.Pointer, r *Reader) {
 		d.mapType.UnsafeSet(ptr, d.mapType.UnsafeMakeMap(0))
 	}
 
-	key := string(resSchema.Type())
-	if n, ok := resSchema.(NamedSchema); ok {
-		key = n.FullName()
-	}
+	key := schemaTypeName(resSchema)
 	keyPtr := reflect2.PtrOf(key)
 
 	elemPtr := d.elemType.UnsafeNew()
@@ -148,96 +143,6 @@ func (e *mapUnionEncoder) Encode(ptr unsafe.Pointer, w *Writer) {
 	encoderOfType(e.cfg, schema, elemType).Encode(elemPtr, w)
 }
 
-func decoderOfTypedUnion(schema Schema, typ reflect2.Type) ValDecoder {
-	union := schema.(*UnionSchema)
-	ptrType := typ.(*reflect2.UnsafePtrType)
-	elemType := ptrType.Elem()
-
-	return &unionTypedDecoder{
-		schema: union,
-		typ:    ptrType,
-		elem:   elemType,
-	}
-}
-
-type unionTypedDecoder struct {
-	schema *UnionSchema
-	typ    reflect2.Type
-	elem   reflect2.Type
-}
-
-func (d *unionTypedDecoder) Decode(ptr unsafe.Pointer, r *Reader) {
-	if *((*unsafe.Pointer)(ptr)) == nil {
-		newPtr := d.elem.UnsafeNew()
-		*((*unsafe.Pointer)(ptr)) = newPtr
-	}
-	union := d.typ.UnsafeIndirect(ptr).(UnionType)
-
-	schema := getUnionSchema(d.schema, r)
-	if schema == nil {
-		return
-	}
-
-	key := string(schema.Type())
-	if n, ok := schema.(NamedSchema); ok {
-		key = n.FullName()
-	}
-
-	if err := union.SetType(key); err != nil {
-		r.Error = err
-		return
-	}
-
-	// In a null case, just return
-	if schema.Type() == Null {
-		return
-	}
-
-	if *union.Value() == nil {
-		r.ReportError("decode union type", "can not read into nil pointer")
-		return
-	}
-	r.ReadVal(schema, union.Value())
-}
-
-func encoderOfTypedUnion(schema Schema, typ reflect2.Type) ValEncoder {
-	union := schema.(*UnionSchema)
-
-	return &unionTypedEncoder{
-		schema: union,
-		typ:    typ,
-	}
-}
-
-type unionTypedEncoder struct {
-	schema *UnionSchema
-	typ    reflect2.Type
-}
-
-func (e *unionTypedEncoder) Encode(ptr unsafe.Pointer, w *Writer) {
-	union := e.typ.UnsafeIndirect(ptr).(UnionType)
-	name, err := union.GetType()
-	if err != nil {
-		w.Error = err
-		return
-	}
-
-	schema, pos := e.schema.Types().Get(name)
-	if schema == nil {
-		w.Error = fmt.Errorf("avro: unknown union type %s", name)
-		return
-	}
-
-	w.WriteLong(int64(pos))
-
-	val := *union.Value()
-	if schema.Type() == Null && val == nil {
-		return
-	}
-
-	w.WriteVal(schema, val)
-}
-
 func decoderOfPtrUnion(cfg *frozenConfig, schema Schema, typ reflect2.Type) ValDecoder {
 	union := schema.(*UnionSchema)
 	ptrType := typ.(*reflect2.UnsafePtrType)
@@ -257,7 +162,7 @@ type unionPtrDecoder struct {
 }
 
 func (d *unionPtrDecoder) Decode(ptr unsafe.Pointer, r *Reader) {
-	schema := getUnionSchema(d.schema, r)
+	_, schema := getUnionSchema(d.schema, r)
 	if schema == nil {
 		return
 	}
@@ -300,14 +205,156 @@ func (e *unionPtrEncoder) Encode(ptr unsafe.Pointer, w *Writer) {
 	e.encoder.Encode(*((*unsafe.Pointer)(ptr)), w)
 }
 
-func getUnionSchema(schema *UnionSchema, r *Reader) Schema {
+func decoderOfResolvedUnion(cfg *frozenConfig, schema Schema) ValDecoder {
+	union := schema.(*UnionSchema)
+
+	types := make([]reflect2.Type, len(union.Types()))
+	decoders := make([]ValDecoder, len(union.Types()))
+	for i, schema := range union.Types() {
+		name := unionResolutionName(schema)
+		if typ, err := cfg.resolver.Type(name); err == nil {
+			decoder := decoderOfType(cfg, schema, typ)
+			decoders[i] = decoder
+			types[i] = typ
+			continue
+		}
+
+		decoders = []ValDecoder{}
+		types = []reflect2.Type{}
+		break
+	}
+
+	return &unionResolvedDecoder{
+		cfg:      cfg,
+		schema:   union,
+		types:    types,
+		decoders: decoders,
+	}
+}
+
+type unionResolvedDecoder struct {
+	cfg      *frozenConfig
+	schema   *UnionSchema
+	types    []reflect2.Type
+	decoders []ValDecoder
+}
+
+func (d *unionResolvedDecoder) Decode(ptr unsafe.Pointer, r *Reader) {
+	i, schema := getUnionSchema(d.schema, r)
+	if schema == nil {
+		return
+	}
+
+	pObj := (*interface{})(ptr)
+
+	if schema.Type() == Null {
+		*pObj = nil
+		return
+	}
+
+	if i >= len(d.decoders) {
+		if d.cfg.config.UnionResolutionError {
+			r.ReportError("decode union type", "unknown union type")
+			return
+		}
+
+		// We cannot resolve this, set it to the map type
+		name := schemaTypeName(schema)
+		obj := map[string]interface{}{}
+		obj[name] = r.ReadNext(schema)
+
+		*pObj = obj
+		return
+	}
+
+	typ := d.types[i]
+	var newPtr unsafe.Pointer
+	switch typ.Kind() {
+	case reflect.Map:
+		mapType := typ.(*reflect2.UnsafeMapType)
+		newPtr = mapType.UnsafeMakeMap(1)
+
+	case reflect.Slice:
+		mapType := typ.(*reflect2.UnsafeSliceType)
+		newPtr = mapType.UnsafeMakeSlice(1, 1)
+
+	case reflect.Ptr:
+		elemType := typ.(*reflect2.UnsafePtrType).Elem()
+		newPtr = elemType.UnsafeNew()
+
+	default:
+		newPtr = typ.UnsafeNew()
+	}
+
+	d.decoders[i].Decode(newPtr, r)
+	*pObj = typ.UnsafeIndirect(newPtr)
+}
+
+func unionResolutionName(schema Schema) string {
+	name := schemaTypeName(schema)
+	switch schema.Type() {
+	case Map:
+		name += ":"
+		valSchema := schema.(*MapSchema).Values()
+		valName := schemaTypeName(valSchema)
+
+		name += valName
+
+	case Array:
+		name += ":"
+		itemSchema := schema.(*ArraySchema).Items()
+		itemName := schemaTypeName(itemSchema)
+
+		name += itemName
+	}
+
+	return name
+}
+
+func encoderOfResolverUnion(cfg *frozenConfig, schema Schema, typ reflect2.Type) ValEncoder {
+	union := schema.(*UnionSchema)
+
+	name, err := cfg.resolver.Name(typ)
+	if err != nil {
+		return &errorEncoder{err: err}
+	}
+
+	if idx := strings.Index(name, ":"); idx > 0 {
+		name = name[:idx]
+	}
+
+	schema, pos := union.Types().Get(name)
+	if schema == nil {
+		return &errorEncoder{err: fmt.Errorf("avro: unknown union type %s", name)}
+	}
+
+	encoder := encoderOfType(cfg, schema, typ)
+
+	return &unionResolverEncoder{
+		pos:     pos,
+		encoder: encoder,
+	}
+}
+
+type unionResolverEncoder struct {
+	pos     int
+	encoder ValEncoder
+}
+
+func (e *unionResolverEncoder) Encode(ptr unsafe.Pointer, w *Writer) {
+	w.WriteLong(int64(e.pos))
+
+	e.encoder.Encode(ptr, w)
+}
+
+func getUnionSchema(schema *UnionSchema, r *Reader) (int, Schema) {
 	types := schema.Types()
 
 	idx := int(r.ReadLong())
 	if idx < 0 || idx > len(types)-1 {
 		r.ReportError("decode union type", "unknown union type")
-		return nil
+		return 0, nil
 	}
 
-	return types[idx]
+	return idx, types[idx]
 }
diff --git a/config.go b/config.go
index 3946fc3..d5ca54b 100644
--- a/config.go
+++ b/config.go
@@ -20,6 +20,10 @@ type Config struct {
 	// BlockLength is the length of blocks for maps and arrays.
 	// This defaults to 100.
 	BlockLength int
+
+	// UnionResolutionError determines if an error will be returned
+	// when a type cannot be resolved while decoding a union.
+	UnionResolutionError bool
 }
 
 // Freeze makes the configuration immutable.
@@ -28,6 +32,7 @@ func (c Config) Freeze() API {
 		config:       c,
 		decoderCache: concurrent.NewMap(),
 		encoderCache: concurrent.NewMap(),
+		resolver:     NewTypeResolver(),
 	}
 
 	api.readerPool = &sync.Pool{
@@ -75,6 +80,9 @@ type API interface {
 
 	// EncoderOf returns the value encoder for a given schema and type.
 	EncoderOf(schema Schema, tpy reflect2.Type) ValEncoder
+
+	// Register registers names to their types for resolution. All primitive types are pre-registered.
+	Register(name string, obj interface{})
 }
 
 type frozenConfig struct {
@@ -85,6 +93,8 @@ type frozenConfig struct {
 
 	readerPool *sync.Pool
 	writerPool *sync.Pool
+
+	resolver *TypeResolver
 }
 
 func (c *frozenConfig) Marshal(schema Schema, v interface{}) ([]byte, error) {
@@ -159,6 +169,10 @@ func (c *frozenConfig) NewDecoder(schema Schema, r io.Reader) *Decoder {
 	}
 }
 
+func (c *frozenConfig) Register(name string, obj interface{}) {
+	c.resolver.Register(name, obj)
+}
+
 type cacheKey struct {
 	fingerprint [32]byte
 	rtype       uintptr
diff --git a/protocol.go b/protocol.go
new file mode 100644
index 0000000..a8db627
--- /dev/null
+++ b/protocol.go
@@ -0,0 +1,322 @@
+package avro
+
+import (
+	"crypto/md5"
+	"encoding/hex"
+	"errors"
+	"io/ioutil"
+
+	jsoniter "github.com/json-iterator/go"
+)
+
+var (
+	protocolReserved = []string{"doc", "types", "messages", "protocol", "namespace"}
+	messageReserved  = []string{"doc", "response", "request", "errors", "one-way"}
+)
+
+// Protocol is an Avro protocol.
+type Protocol struct {
+	name
+	properties
+
+	types    []NamedSchema
+	messages map[string]*Message
+
+	hash string
+}
+
+// NewProtocol creates a protocol instance.
+func NewProtocol(name, space string, types []NamedSchema, messages map[string]*Message) (*Protocol, error) {
+	n, err := newName(name, space)
+	if err != nil {
+		return nil, err
+	}
+
+	p := &Protocol{
+		name:       n,
+		properties: properties{reserved: protocolReserved},
+		types:      types,
+		messages:   messages,
+	}
+
+	b := md5.Sum([]byte(p.String()))
+	p.hash = hex.EncodeToString(b[:])
+
+	return p, nil
+}
+
+// Message returns a message with the given name or nil.
+func (p *Protocol) Message(name string) *Message {
+	return p.messages[name]
+}
+
+// Hash returns the MD5 hash of the protocol.
+func (p *Protocol) Hash() string {
+	return p.hash
+}
+
+// String returns the canonical form of the protocol.
+func (p *Protocol) String() string {
+	types := ""
+	for _, f := range p.types {
+		types += f.String() + ","
+	}
+	if len(types) > 0 {
+		types = types[:len(types)-1]
+	}
+
+	messages := ""
+	for k, m := range p.messages {
+		messages += `"` + k + `":` + m.String() + ","
+	}
+	if len(messages) > 0 {
+		messages = messages[:len(messages)-1]
+	}
+
+	return `{"protocol":"` + p.Name() + `","namespace":"` + p.Namespace() + `","types":[` + types + `],"messages":{` + messages + `}}`
+}
+
+// Message is an Avro protocol message.
+type Message struct {
+	properties
+
+	req    *RecordSchema
+	resp   Schema
+	errs   *UnionSchema
+	oneWay bool
+}
+
+// NewMessage creates a protocol message instance.
+func NewMessage(req *RecordSchema, resp Schema, errors *UnionSchema, oneWay bool) *Message {
+	return &Message{
+		properties: properties{reserved: messageReserved},
+		req:        req,
+		resp:       resp,
+		errs:       errors,
+		oneWay:     oneWay,
+	}
+}
+
+// Request returns the message request schema.
+func (m *Message) Request() *RecordSchema {
+	return m.req
+}
+
+// Response returns the message response schema.
+func (m *Message) Response() Schema {
+	return m.resp
+}
+
+// Errors returns the message errors union schema.
+func (m *Message) Errors() *UnionSchema {
+	return m.errs
+}
+
+// OneWay determines of the message is a one way message.
+func (m *Message) OneWay() bool {
+	return m.oneWay
+}
+
+// String returns the canonical form of the message.
+func (m *Message) String() string {
+	fields := ""
+	for _, f := range m.req.fields {
+		fields += f.String() + ","
+	}
+	if len(fields) > 0 {
+		fields = fields[:len(fields)-1]
+	}
+
+	str := `{"request":[` + fields + `]`
+
+	if m.resp != nil {
+		str += `,"response":` + m.resp.String()
+	}
+
+	if m.errs != nil && len(m.errs.Types()) > 1 {
+		errs, _ := NewUnionSchema(m.errs.Types()[1:])
+		str += `,"errors":` + errs.String()
+	}
+
+	str += "}"
+	return str
+}
+
+// ParseProtocolFile parses an Avro protocol from a file.
+func ParseProtocolFile(path string) (*Protocol, error) {
+	s, err := ioutil.ReadFile(path)
+	if err != nil {
+		return nil, err
+	}
+
+	return ParseProtocol(string(s))
+}
+
+// MustParseProtocol parses an Avro protocol, panicing if there is an error.
+func MustParseProtocol(protocol string) *Protocol {
+	parsed, err := ParseProtocol(protocol)
+	if err != nil {
+		panic(err)
+	}
+
+	return parsed
+}
+
+// ParseProtocol parses an Avro protocol.
+func ParseProtocol(protocol string) (*Protocol, error) {
+	cache := &SchemaCache{}
+
+	var m map[string]interface{}
+	if err := jsoniter.Unmarshal([]byte(protocol), &m); err != nil {
+		return nil, err
+	}
+
+	name, err := resolveProtocolName(m)
+	if err != nil {
+		return nil, err
+	}
+
+	var types []NamedSchema
+	if ts, ok := m["types"].([]interface{}); ok {
+		types, err = parseProtocolTypes(name.space, ts, cache)
+		if err != nil {
+			return nil, err
+		}
+	}
+
+	messages := map[string]*Message{}
+	if msgs, ok := m["messages"].(map[string]interface{}); ok {
+		for k, msg := range msgs {
+			m, ok := msg.(map[string]interface{})
+			if !ok {
+				return nil, errors.New("avro: message must be an object")
+			}
+
+			message, err := parseMessage(name.space, m, cache)
+			if err != nil {
+				return nil, err
+			}
+
+			messages[k] = message
+		}
+	}
+
+	proto, _ := NewProtocol(name.name, name.space, types, messages)
+
+	for k, v := range m {
+		proto.AddProp(k, v)
+	}
+
+	return proto, nil
+}
+
+func parseProtocolTypes(namespace string, types []interface{}, cache *SchemaCache) ([]NamedSchema, error) {
+	ts := make([]NamedSchema, len(types))
+	for i, typ := range types {
+		schema, err := parseType(namespace, typ, cache)
+		if err != nil {
+			return nil, err
+		}
+
+		namedSchema, ok := schema.(NamedSchema)
+		if !ok {
+			return nil, errors.New("avro: protocol types must be named schemas")
+		}
+
+		ts[i] = namedSchema
+	}
+
+	return ts, nil
+}
+
+func parseMessage(namespace string, m map[string]interface{}, cache *SchemaCache) (*Message, error) {
+	req, ok := m["request"].([]interface{})
+	if !ok {
+		return nil, errors.New("avro: request must have an array of fields")
+	}
+
+	fields := make([]*Field, len(req))
+	for i, f := range req {
+		field, err := parseField(namespace, f, cache)
+		if err != nil {
+			return nil, err
+		}
+
+		fields[i] = field
+	}
+	request := &RecordSchema{
+		name:       name{},
+		properties: properties{reserved: schemaReserved},
+		fields:     fields,
+	}
+
+	var response Schema
+	if res, ok := m["response"]; ok {
+		schema, err := parseType(namespace, res, cache)
+		if err != nil {
+			return nil, err
+		}
+
+		if schema.Type() != Null {
+			response = schema
+		}
+	}
+
+	types := []Schema{NewPrimitiveSchema(String)}
+	if errs, ok := m["errors"].([]interface{}); ok {
+		for _, e := range errs {
+			schema, err := parseType(namespace, e, cache)
+			if err != nil {
+				return nil, err
+			}
+
+			if rec, ok := schema.(*RecordSchema); ok && !rec.IsError() {
+				return nil, errors.New("avro: errors record schema must be of type error")
+			}
+
+			types = append(types, schema)
+		}
+	}
+	errs, err := NewUnionSchema(types)
+	if err != nil {
+		return nil, err
+	}
+
+	oneWay := false
+	if o, ok := m["one-way"].(bool); ok {
+		oneWay = o
+		if oneWay && (len(errs.Types()) > 1 || response != nil) {
+			return nil, errors.New("avro: one-way messages cannot not have a response or errors")
+		}
+	}
+
+	if !oneWay && len(errs.Types()) <= 1 && response == nil {
+		oneWay = true
+	}
+
+	msg := NewMessage(request, response, errs, oneWay)
+
+	for k, v := range m {
+		msg.AddProp(k, v)
+	}
+
+	return msg, nil
+}
+
+func resolveProtocolName(m map[string]interface{}) (name, error) {
+	proto, ok := m["protocol"].(string)
+	if !ok {
+		return name{}, errors.New("avro: protocol key required")
+	}
+
+	space := ""
+	if namespace, ok := m["namespace"].(string); ok {
+		if namespace == "" {
+			return name{}, errors.New("avro: namespace key must be non-empty or omitted")
+		}
+
+		space = namespace
+	}
+
+	return newName(proto, space)
+}
diff --git a/reader_generic.go b/reader_generic.go
index 08b7d75..98405c7 100644
--- a/reader_generic.go
+++ b/reader_generic.go
@@ -79,11 +79,7 @@ func (r *Reader) ReadNext(schema Schema) interface{} {
 			return nil
 		}
 
-		key := string(schema.Type())
-		if n, ok := schema.(NamedSchema); ok {
-			key = n.FullName()
-		}
-
+		key := schemaTypeName(schema)
 		obj := map[string]interface{}{}
 		obj[key] = r.ReadNext(types[idx])
 
diff --git a/resolver.go b/resolver.go
new file mode 100644
index 0000000..e3441ee
--- /dev/null
+++ b/resolver.go
@@ -0,0 +1,73 @@
+package avro
+
+import (
+	"fmt"
+
+	"github.com/modern-go/concurrent"
+	"github.com/modern-go/reflect2"
+)
+
+// TypeResolver resolves types by name.
+type TypeResolver struct {
+	names *concurrent.Map // map[string]reflect2.Type
+	types *concurrent.Map // map[int]string
+}
+
+// NewTypeResolver creates a new type resolver with all primitive types
+// registered.
+func NewTypeResolver() *TypeResolver {
+	r := &TypeResolver{
+		names: concurrent.NewMap(),
+		types: concurrent.NewMap(),
+	}
+
+	// Register basic types
+	r.Register(string(Null), &null{})
+	r.Register(string(Int), int8(0))
+	r.Register(string(Int), int16(0))
+	r.Register(string(Int), int32(0))
+	r.Register(string(Int), int(0))
+	r.Register(string(Long), int64(0))
+	r.Register(string(Float), float32(0))
+	r.Register(string(Double), float64(0))
+	r.Register(string(String), "")
+	r.Register(string(Bytes), []byte{})
+
+	return r
+}
+
+// Register registers names to their types for resolution.
+func (r *TypeResolver) Register(name string, obj interface{}) {
+	typ := reflect2.TypeOf(obj)
+	rtype := typ.RType()
+
+	r.names.Store(name, typ)
+	r.types.Store(rtype, name)
+}
+
+// Name gets the name for a type, or an error.
+func (r *TypeResolver) Name(typ reflect2.Type) (string, error) {
+	rtype := typ.RType()
+
+	name, ok := r.types.Load(rtype)
+	if !ok {
+		return "", fmt.Errorf("avro: unable to resolve type %s", typ.String())
+	}
+
+	return name.(string), nil
+}
+
+// Type gets the type for a name, or an error.
+func (r *TypeResolver) Type(name string) (reflect2.Type, error) {
+	typ, ok := r.names.Load(name)
+	if !ok {
+		return nil, fmt.Errorf("avro: unable to resolve type with name %s", name)
+	}
+
+	return typ.(reflect2.Type), nil
+}
+
+// Register registers names to their types for resolution. All primitive types are pre-registered.
+func Register(name string, obj interface{}) {
+	DefaultConfig.Register(name, obj)
+}
diff --git a/schema.go b/schema.go
index 7318325..4780d11 100644
--- a/schema.go
+++ b/schema.go
@@ -11,6 +11,7 @@ import (
 )
 
 var emptyFgpt = [32]byte{}
+var nullDefault = struct{}{}
 
 // Type is a schema type
 type Type string
@@ -18,6 +19,7 @@ type Type string
 // Schema type constants.
 const (
 	Record  Type = "record"
+	Error   Type = "error"
 	Ref     Type = "<ref>"
 	Enum    Type = "enum"
 	Array   Type = "array"
@@ -34,25 +36,25 @@ const (
 	Null    Type = "null"
 )
 
-type frozenSchemaConfig struct {
-	schemaCache concurrent.Map // map[string]Schema
+// SchemaCache is a cache of schemas.
+type SchemaCache struct {
+	cache concurrent.Map // map[string]Schema
 }
 
-func (c *frozenSchemaConfig) addSchemaToCache(name string, schema Schema) {
-	c.schemaCache.Store(name, schema)
+// Add adds a schema to the cache with the given name.
+func (c *SchemaCache) Add(name string, schema Schema) {
+	c.cache.Store(name, schema)
 }
 
-// getSchemaFromCache returns the Schema if it exists.
-func (c *frozenSchemaConfig) getSchemaFromCache(name string) Schema {
-	if v, ok := c.schemaCache.Load(name); ok {
+// Get returns the Schema if it exists.
+func (c *SchemaCache) Get(name string) Schema {
+	if v, ok := c.cache.Load(name); ok {
 		return v.(Schema)
 	}
 
 	return nil
 }
 
-var schemaConfig = frozenSchemaConfig{}
-
 // Schemas is a slice of Schemas.
 type Schemas []Schema
 
@@ -241,11 +243,12 @@ type RecordSchema struct {
 	properties
 	fingerprinter
 
-	fields []*Field
+	isError bool
+	fields  []*Field
 }
 
 // NewRecordSchema creates a new record schema instance.
-func NewRecordSchema(name, space string) (*RecordSchema, error) {
+func NewRecordSchema(name, space string, fields []*Field) (*RecordSchema, error) {
 	n, err := newName(name, space)
 	if err != nil {
 		return nil, err
@@ -254,6 +257,22 @@ func NewRecordSchema(name, space string) (*RecordSchema, error) {
 	return &RecordSchema{
 		name:       n,
 		properties: properties{reserved: schemaReserved},
+		fields:     fields,
+	}, nil
+}
+
+// NewErrorRecordSchema creates a new error record schema instance.
+func NewErrorRecordSchema(name, space string, fields []*Field) (*RecordSchema, error) {
+	n, err := newName(name, space)
+	if err != nil {
+		return nil, err
+	}
+
+	return &RecordSchema{
+		name:       n,
+		properties: properties{reserved: schemaReserved},
+		isError:    true,
+		fields:     fields,
 	}, nil
 }
 
@@ -262,12 +281,9 @@ func (s *RecordSchema) Type() Type {
 	return Record
 }
 
-// AddField adds a field to the record.
-func (s *RecordSchema) AddField(field *Field) {
-	// Clear the fingerprint cache
-	s.fingerprint = emptyFgpt
-
-	s.fields = append(s.fields, field)
+// IsError determines is this is an error record.
+func (s *RecordSchema) IsError() bool {
+	return s.isError
 }
 
 // Fields returns the fields of a record.
@@ -277,6 +293,11 @@ func (s *RecordSchema) Fields() []*Field {
 
 // String returns the canonical form of the schema.
 func (s *RecordSchema) String() string {
+	typ := "record"
+	if s.isError {
+		typ = "error"
+	}
+
 	fields := ""
 	for _, f := range s.fields {
 		fields += f.String() + ","
@@ -285,7 +306,7 @@ func (s *RecordSchema) String() string {
 		fields = fields[:len(fields)-1]
 	}
 
-	return `{"name":"` + s.FullName() + `","type":"record","fields":[` + fields + `]}`
+	return `{"name":"` + s.FullName() + `","type":"` + typ + `","fields":[` + fields + `]}`
 }
 
 // Fingerprint returns the SHA256 fingerprint of the schema.
@@ -331,10 +352,19 @@ func (s *Field) Type() Schema {
 	return s.typ
 }
 
+// HasDefault determines if the field has a default value.
+func (s *Field) HasDefault() bool {
+	return s.def != nil
+}
+
 // Default returns the default of a field or nil.
 //
 // The only time a nil default is valid is for a Null Type.
 func (s *Field) Default() interface{} {
+	if s.def == nullDefault {
+		return nil
+	}
+
 	return s.def
 }
 
@@ -490,10 +520,7 @@ func NewUnionSchema(types []Schema) (*UnionSchema, error) {
 			return nil, errors.New("avro: union type cannot be a union")
 		}
 
-		strType := string(schema.Type())
-		if named, ok := schema.(NamedSchema); ok {
-			strType = named.FullName()
-		}
+		strType := schemaTypeName(schema)
 
 		if seen[strType] {
 			return nil, errors.New("avro: union type must be unique")
@@ -662,7 +689,10 @@ func validateName(name string) error {
 
 func validateDefault(name string, schema Schema, def interface{}) (interface{}, error) {
 	if def == nil {
-		return nil, nil
+		if schema.Type() != Null && !(schema.Type() == Union && schema.(*UnionSchema).Nullable()) {
+			// This is an empty default value.
+			return nil, nil
+		}
 	}
 
 	def, ok := isValidDefault(schema, def)
@@ -676,7 +706,7 @@ func validateDefault(name string, schema Schema, def interface{}) (interface{},
 func isValidDefault(schema Schema, def interface{}) (interface{}, bool) {
 	switch schema.Type() {
 	case Null:
-		return nil, def == nil
+		return nullDefault, def == nil
 
 	case String, Bytes, Enum, Fixed:
 		if _, ok := def.(string); ok {
@@ -791,3 +821,11 @@ func isValidDefault(schema Schema, def interface{}) (interface{}, bool) {
 
 	return nil, false
 }
+
+func schemaTypeName(schema Schema) string {
+	if n, ok := schema.(NamedSchema); ok {
+		return n.FullName()
+	}
+
+	return string(schema.Type())
+}
diff --git a/schema_compatibility.go b/schema_compatibility.go
new file mode 100644
index 0000000..7464a01
--- /dev/null
+++ b/schema_compatibility.go
@@ -0,0 +1,244 @@
+package avro
+
+import (
+	"errors"
+	"fmt"
+
+	"github.com/modern-go/concurrent"
+)
+
+type recursionError struct{}
+
+func (e recursionError) Error() string {
+	return ""
+}
+
+type compatKey struct {
+	reader [32]byte
+	writer [32]byte
+}
+
+// SchemaCompatibility determines the compatibility of schemas.
+type SchemaCompatibility struct {
+	cache *concurrent.Map // map[compatKey]error
+}
+
+// NewSchemaCompatibility creates a new schema compatibility instance.
+func NewSchemaCompatibility() *SchemaCompatibility {
+	return &SchemaCompatibility{
+		cache: concurrent.NewMap(),
+	}
+}
+
+// Compatible determines the compatibility if the reader and writer schemas.
+func (c *SchemaCompatibility) Compatible(reader, writer Schema) error {
+	return c.compatible(reader, writer)
+}
+
+func (c *SchemaCompatibility) compatible(reader, writer Schema) error {
+	key := compatKey{reader: reader.Fingerprint(), writer: writer.Fingerprint()}
+	if err, ok := c.cache.Load(key); ok {
+		if _, ok := err.(recursionError); ok {
+			// Break the recursion here.
+			return nil
+		}
+
+		if err == nil {
+			return nil
+		}
+
+		return err.(error)
+	}
+
+	c.cache.Store(key, recursionError{})
+	err := c.match(reader, writer)
+	if err != nil {
+		// We dont want to pay the cost of fmt.Errorf every time
+		err = errors.New(err.Error())
+	}
+	c.cache.Store(key, err)
+	return err
+}
+
+func (c *SchemaCompatibility) match(reader, writer Schema) error {
+	// If the schema is a reference, get the actual schema
+	if reader.Type() == Ref {
+		reader = reader.(*RefSchema).Schema()
+	}
+	if writer.Type() == Ref {
+		writer = writer.(*RefSchema).Schema()
+	}
+
+	if reader.Type() != writer.Type() {
+		if writer.Type() == Union {
+			// Reader must be compatible with all types in writer
+			for _, schema := range writer.(*UnionSchema).Types() {
+				if err := c.compatible(reader, schema); err != nil {
+					return err
+				}
+			}
+
+			return nil
+		}
+
+		if reader.Type() == Union {
+			// Writer must be compatible with at least one reader schema
+			var err error
+			for _, schema := range reader.(*UnionSchema).Types() {
+				err = c.compatible(schema, writer)
+				if err == nil {
+					return nil
+				}
+			}
+
+			return fmt.Errorf("reader union lacking writer schema %s", writer.Type())
+		}
+
+		switch writer.Type() {
+		case Int:
+			if reader.Type() == Long || reader.Type() == Float || reader.Type() == Double {
+				return nil
+			}
+
+		case Long:
+			if reader.Type() == Float || reader.Type() == Double {
+				return nil
+			}
+
+		case Float:
+			if reader.Type() == Double {
+				return nil
+			}
+
+		case String:
+			if reader.Type() == Bytes {
+				return nil
+			}
+
+		case Bytes:
+			if reader.Type() == String {
+				return nil
+			}
+		}
+
+		return fmt.Errorf("reader schema %s not compatible with writer schema %s", reader.Type(), writer.Type())
+	}
+
+	switch reader.Type() {
+	case Array:
+		return c.compatible(reader.(*ArraySchema).Items(), writer.(*ArraySchema).Items())
+
+	case Map:
+		return c.compatible(reader.(*MapSchema).Values(), writer.(*MapSchema).Values())
+
+	case Fixed:
+		r := reader.(*FixedSchema)
+		w := writer.(*FixedSchema)
+
+		if err := c.checkSchemaName(r, w); err != nil {
+			return err
+		}
+
+		if err := c.checkFixedSize(r, w); err != nil {
+			return err
+		}
+
+	case Enum:
+		r := reader.(*EnumSchema)
+		w := writer.(*EnumSchema)
+
+		if err := c.checkSchemaName(r, w); err != nil {
+			return err
+		}
+
+		if err := c.checkEnumSymbols(r, w); err != nil {
+			return err
+		}
+
+	case Record:
+		r := reader.(*RecordSchema)
+		w := writer.(*RecordSchema)
+
+		if err := c.checkSchemaName(r, w); err != nil {
+			return err
+		}
+
+		if err := c.checkRecordFields(r, w); err != nil {
+			return err
+		}
+
+	case Union:
+		for _, schema := range writer.(*UnionSchema).Types() {
+			if err := c.compatible(reader, schema); err != nil {
+				return err
+			}
+		}
+	}
+
+	return nil
+}
+
+func (c *SchemaCompatibility) checkSchemaName(reader, writer NamedSchema) error {
+	if reader.FullName() != writer.FullName() {
+		return fmt.Errorf("reader schema %s and writer schema %s  names do match", reader.FullName(), writer.FullName())
+	}
+
+	return nil
+}
+
+func (c *SchemaCompatibility) checkFixedSize(reader, writer *FixedSchema) error {
+	if reader.Size() != writer.Size() {
+		return fmt.Errorf("%s reader and writer fixed sizes do not match", reader.FullName())
+	}
+
+	return nil
+}
+
+func (c *SchemaCompatibility) checkEnumSymbols(reader, writer *EnumSchema) error {
+	for _, symbol := range writer.Symbols() {
+		if !c.contains(reader.Symbols(), symbol) {
+			return fmt.Errorf("reader %s is missing symbol %s", reader.FullName(), symbol)
+		}
+	}
+
+	return nil
+}
+
+func (c *SchemaCompatibility) checkRecordFields(reader, writer *RecordSchema) error {
+	for _, field := range reader.Fields() {
+		f, ok := c.getField(writer.Fields(), field)
+		if !ok {
+			if field.HasDefault() {
+				continue
+			}
+
+			return fmt.Errorf("reader field %s is missing in writer schema and has no default", field.Name())
+		}
+
+		if err := c.compatible(field.Type(), f.Type()); err != nil {
+			return err
+		}
+	}
+
+	return nil
+}
+
+func (c *SchemaCompatibility) contains(a []string, s string) bool {
+	for _, str := range a {
+		if str == s {
+			return true
+		}
+	}
+
+	return false
+}
+
+func (c *SchemaCompatibility) getField(a []*Field, f *Field) (*Field, bool) {
+	for _, field := range a {
+		if field.Name() == f.Name() {
+			return field, true
+		}
+	}
+
+	return nil, false
+}
diff --git a/schema_parse.go b/schema_parse.go
index 8e2a622..775fe35 100644
--- a/schema_parse.go
+++ b/schema_parse.go
@@ -14,14 +14,22 @@ var (
 	fieldReserved  = []string{"default", "doc", "name", "order", "type", "aliases"}
 )
 
+// DefaultSchemaCache is the default cache for schemas.
+var DefaultSchemaCache = &SchemaCache{}
+
 // Parse parses a schema string.
 func Parse(schema string) (Schema, error) {
+	return ParseWithCache(schema, "", DefaultSchemaCache)
+}
+
+// ParseWithCache parses a schema string using the given namespace and  schema cache.
+func ParseWithCache(schema, namespace string, cache *SchemaCache) (Schema, error) {
 	var json interface{}
 	if err := jsoniter.Unmarshal([]byte(schema), &json); err != nil {
 		json = schema
 	}
 
-	return parseType("", json)
+	return parseType(namespace, json, cache)
 }
 
 // MustParse parses a schema string, panicing if there is an error.
@@ -54,25 +62,25 @@ func ParseFiles(paths ...string) (Schema, error) {
 	return schema, nil
 }
 
-func parseType(namespace string, v interface{}) (Schema, error) {
+func parseType(namespace string, v interface{}, cache *SchemaCache) (Schema, error) {
 	switch val := v.(type) {
 	case nil:
 		return &NullSchema{}, nil
 
 	case string:
-		return parsePrimitive(namespace, val)
+		return parsePrimitive(namespace, val, cache)
 
 	case map[string]interface{}:
-		return parseComplex(namespace, val)
+		return parseComplex(namespace, val, cache)
 
 	case []interface{}:
-		return parseUnion(namespace, val)
+		return parseUnion(namespace, val, cache)
 	}
 
 	return nil, fmt.Errorf("avro: unknown type: %v", v)
 }
 
-func parsePrimitive(namespace, s string) (Schema, error) {
+func parsePrimitive(namespace, s string, cache *SchemaCache) (Schema, error) {
 	typ := Type(s)
 	switch typ {
 	case Null:
@@ -82,7 +90,7 @@ func parsePrimitive(namespace, s string) (Schema, error) {
 		return NewPrimitiveSchema(typ), nil
 
 	default:
-		schema := schemaConfig.getSchemaFromCache(fullName(namespace, s))
+		schema := cache.Get(fullName(namespace, s))
 		if schema != nil {
 			return schema, nil
 		}
@@ -91,9 +99,9 @@ func parsePrimitive(namespace, s string) (Schema, error) {
 	}
 }
 
-func parseComplex(namespace string, m map[string]interface{}) (Schema, error) {
+func parseComplex(namespace string, m map[string]interface{}, cache *SchemaCache) (Schema, error) {
 	if val, ok := m["type"].([]interface{}); ok {
-		return parseUnion(namespace, val)
+		return parseUnion(namespace, val, cache)
 	}
 
 	str, ok := m["type"].(string)
@@ -109,27 +117,27 @@ func parseComplex(namespace string, m map[string]interface{}) (Schema, error) {
 	case String, Bytes, Int, Long, Float, Double, Boolean:
 		return NewPrimitiveSchema(typ), nil
 
-	case Record:
-		return parseRecord(namespace, m)
+	case Record, Error:
+		return parseRecord(typ, namespace, m, cache)
 
 	case Enum:
-		return parseEnum(namespace, m)
+		return parseEnum(namespace, m, cache)
 
 	case Array:
-		return parseArray(namespace, m)
+		return parseArray(namespace, m, cache)
 
 	case Map:
-		return parseMap(namespace, m)
+		return parseMap(namespace, m, cache)
 
 	case Fixed:
-		return parseFixed(namespace, m)
+		return parseFixed(namespace, m, cache)
 
 	default:
-		return parseType(namespace, string(typ))
+		return parseType(namespace, string(typ), cache)
 	}
 }
 
-func parseRecord(namespace string, m map[string]interface{}) (Schema, error) {
+func parseRecord(typ Type, namespace string, m map[string]interface{}, cache *SchemaCache) (Schema, error) {
 	name, newNamespace, err := resolveFullName(m)
 	if err != nil {
 		return nil, err
@@ -138,35 +146,43 @@ func parseRecord(namespace string, m map[string]interface{}) (Schema, error) {
 		namespace = newNamespace
 	}
 
-	fields, ok := m["fields"].([]interface{})
+	fs, ok := m["fields"].([]interface{})
 	if !ok {
 		return nil, errors.New("avro: record must have an array of fields")
 	}
+	fields := make([]*Field, len(fs))
 
-	rec, err := NewRecordSchema(name, namespace)
+	var rec *RecordSchema
+	switch typ {
+	case Record:
+		rec, err = NewRecordSchema(name, namespace, fields)
+
+	case Error:
+		rec, err = NewErrorRecordSchema(name, namespace, fields)
+	}
 	if err != nil {
 		return nil, err
 	}
 
-	schemaConfig.addSchemaToCache(rec.FullName(), NewRefSchema(rec))
+	cache.Add(rec.FullName(), NewRefSchema(rec))
 
 	for k, v := range m {
 		rec.AddProp(k, v)
 	}
 
-	for _, f := range fields {
-		field, err := parseField(namespace, f)
+	for i, f := range fs {
+		field, err := parseField(namespace, f, cache)
 		if err != nil {
 			return nil, err
 		}
 
-		rec.AddField(field)
+		fields[i] = field
 	}
 
 	return rec, nil
 }
 
-func parseField(namespace string, v interface{}) (*Field, error) {
+func parseField(namespace string, v interface{}, cache *SchemaCache) (*Field, error) {
 	m, ok := v.(map[string]interface{})
 	if !ok {
 		return nil, fmt.Errorf("avro: invalid field: %+v", v)
@@ -181,7 +197,7 @@ func parseField(namespace string, v interface{}) (*Field, error) {
 		return nil, errors.New("avro: field requires a type")
 	}
 
-	typ, err := parseType(namespace, m["type"])
+	typ, err := parseType(namespace, m["type"], cache)
 	if err != nil {
 		return nil, err
 	}
@@ -198,7 +214,7 @@ func parseField(namespace string, v interface{}) (*Field, error) {
 	return field, nil
 }
 
-func parseEnum(namespace string, m map[string]interface{}) (Schema, error) {
+func parseEnum(namespace string, m map[string]interface{}, cache *SchemaCache) (Schema, error) {
 	name, newNamespace, err := resolveFullName(m)
 	if err != nil {
 		return nil, err
@@ -227,7 +243,7 @@ func parseEnum(namespace string, m map[string]interface{}) (Schema, error) {
 		return nil, err
 	}
 
-	schemaConfig.addSchemaToCache(enum.FullName(), enum)
+	cache.Add(enum.FullName(), enum)
 
 	for k, v := range m {
 		enum.AddProp(k, v)
@@ -236,13 +252,13 @@ func parseEnum(namespace string, m map[string]interface{}) (Schema, error) {
 	return enum, nil
 }
 
-func parseArray(namespace string, m map[string]interface{}) (Schema, error) {
+func parseArray(namespace string, m map[string]interface{}, cache *SchemaCache) (Schema, error) {
 	items, ok := m["items"]
 	if !ok {
 		return nil, errors.New("avro: array must have an items key")
 	}
 
-	schema, err := parseType(namespace, items)
+	schema, err := parseType(namespace, items, cache)
 	if err != nil {
 		return nil, err
 	}
@@ -256,13 +272,13 @@ func parseArray(namespace string, m map[string]interface{}) (Schema, error) {
 	return arr, nil
 }
 
-func parseMap(namespace string, m map[string]interface{}) (Schema, error) {
+func parseMap(namespace string, m map[string]interface{}, cache *SchemaCache) (Schema, error) {
 	values, ok := m["values"]
 	if !ok {
 		return nil, errors.New("avro: map must have an values key")
 	}
 
-	schema, err := parseType(namespace, values)
+	schema, err := parseType(namespace, values, cache)
 	if err != nil {
 		return nil, err
 	}
@@ -276,11 +292,11 @@ func parseMap(namespace string, m map[string]interface{}) (Schema, error) {
 	return ms, nil
 }
 
-func parseUnion(namespace string, v []interface{}) (Schema, error) {
+func parseUnion(namespace string, v []interface{}, cache *SchemaCache) (Schema, error) {
 	var err error
 	types := make([]Schema, len(v))
 	for i := range v {
-		types[i], err = parseType(namespace, v[i])
+		types[i], err = parseType(namespace, v[i], cache)
 		if err != nil {
 			return nil, err
 		}
@@ -289,7 +305,7 @@ func parseUnion(namespace string, v []interface{}) (Schema, error) {
 	return NewUnionSchema(types)
 }
 
-func parseFixed(namespace string, m map[string]interface{}) (Schema, error) {
+func parseFixed(namespace string, m map[string]interface{}, cache *SchemaCache) (Schema, error) {
 	name, newNamespace, err := resolveFullName(m)
 	if err != nil {
 		return nil, err
@@ -308,7 +324,7 @@ func parseFixed(namespace string, m map[string]interface{}) (Schema, error) {
 		return nil, err
 	}
 
-	schemaConfig.addSchemaToCache(fixed.FullName(), fixed)
+	cache.Add(fixed.FullName(), fixed)
 
 	for k, v := range m {
 		fixed.AddProp(k, v)


Test patch:
diff --git a/bench_test.go b/bench_test.go
index ae275d5..f60900d 100644
--- a/bench_test.go
+++ b/bench_test.go
@@ -112,7 +112,7 @@ func BenchmarkSuperheroGenericDecode(b *testing.B) {
 	b.ReportAllocs()
 	b.ResetTimer()
 	for i := 0; i < b.N; i++ {
-		m := map[string]interface{}{}
+		var m interface{}
 		_ = avro.Unmarshal(schema, data, &m)
 	}
 }
diff --git a/decoder_union_test.go b/decoder_union_test.go
index a5c030b..444a5d7 100644
--- a/decoder_union_test.go
+++ b/decoder_union_test.go
@@ -90,21 +90,76 @@ func TestDecoder_UnionMapInvalidMap(t *testing.T) {
 	assert.Error(t, err)
 }
 
-func TestDecoder_UnionTyped(t *testing.T) {
+func TestDecoder_UnionPtr(t *testing.T) {
+	defer ConfigTeardown()
+
+	data := []byte{0x02, 0x06, 0x66, 0x6F, 0x6F}
+	schema := `["null", "string"]`
+	dec, _ := avro.NewDecoder(schema, bytes.NewReader(data))
+
+	var got *string
+	err := dec.Decode(&got)
+
+	want := "foo"
+	assert.NoError(t, err)
+	assert.Equal(t, &want, got)
+}
+
+func TestDecoder_UnionPtrNull(t *testing.T) {
+	defer ConfigTeardown()
+
+	data := []byte{0x00}
+	schema := `["null", "string"]`
+	dec, _ := avro.NewDecoder(schema, bytes.NewReader(data))
+
+	var got *string
+	err := dec.Decode(&got)
+
+	assert.NoError(t, err)
+	assert.Nil(t, got)
+}
+
+func TestDecoder_UnionPtrInvalidSchema(t *testing.T) {
+	defer ConfigTeardown()
+
+	data := []byte{0x04}
+	schema := `["null", "string"]`
+	dec, _ := avro.NewDecoder(schema, bytes.NewReader(data))
+
+	var got *string
+	err := dec.Decode(&got)
+
+	assert.Error(t, err)
+}
+
+func TestDecoder_UnionPtrNotNullable(t *testing.T) {
+	defer ConfigTeardown()
+
+	data := []byte{0x02, 0x06, 0x66, 0x6F, 0x6F}
+	schema := `["null", "string", "int"]`
+	dec, _ := avro.NewDecoder(schema, bytes.NewReader(data))
+
+	var got *string
+	err := dec.Decode(&got)
+
+	assert.Error(t, err)
+}
+
+func TestDecoder_UnionInterface(t *testing.T) {
 	defer ConfigTeardown()
 
 	data := []byte{0x02, 0x36}
 	schema := `["null", "int"]`
 	dec, _ := avro.NewDecoder(schema, bytes.NewReader(data))
 
-	got := &TestUnionType{}
+	var got interface{}
 	err := dec.Decode(&got)
 
 	assert.NoError(t, err)
-	assert.Equal(t, 27, got.Val)
+	assert.Equal(t, 27, got)
 }
 
-func TestDecoder_UnionTypedInRecord(t *testing.T) {
+func TestDecoder_UnionInterfaceInRecord(t *testing.T) {
 	defer ConfigTeardown()
 
 	data := []byte{0x02, 0x36}
@@ -117,143 +172,202 @@ func TestDecoder_UnionTypedInRecord(t *testing.T) {
 }`
 	dec, _ := avro.NewDecoder(schema, bytes.NewReader(data))
 
-	got := &TestUnionRecord{}
+	got := &TestUnion{}
 	err := dec.Decode(&got)
 
 	assert.NoError(t, err)
-	assert.Equal(t, 27, got.A.Val)
+	assert.Equal(t, 27, got.A)
 }
 
-func TestDecoder_UnionTypedNull(t *testing.T) {
+func TestDecoder_UnionInterfaceInMap(t *testing.T) {
 	defer ConfigTeardown()
 
-	data := []byte{0x00}
-	schema := `["null", "string"]`
+	avro.Register("map:int", map[string]int{})
+
+	data := []byte{0x01, 0x0c, 0x06, 0x66, 0x6f, 0x6f, 0x00, 0x36, 0x00}
+	schema := `{"type": "map", "values": ["int", "string"]}`
 	dec, _ := avro.NewDecoder(schema, bytes.NewReader(data))
 
-	got := &TestUnionType{}
+	var got map[string]interface{}
 	err := dec.Decode(&got)
 
 	assert.NoError(t, err)
-	assert.Equal(t, nil, got.Val)
+	assert.Equal(t, map[string]interface{}{"foo": 27}, got)
 }
 
-func TestDecoder_UnionTypedNamed(t *testing.T) {
+func TestDecoder_UnionInterfaceMap(t *testing.T) {
 	defer ConfigTeardown()
 
-	data := []byte{0x02, 0x02}
-	schema := `["null", {"type":"enum", "name": "test", "symbols": ["A", "B"]}]`
+	avro.Register("map:int", map[string]int{})
+
+	data := []byte{0x02, 0x01, 0x0a, 0x06, 0x66, 0x6f, 0x6f, 0x36, 0x00}
+	schema := `["int", {"type": "map", "values": "int"}]`
 	dec, _ := avro.NewDecoder(schema, bytes.NewReader(data))
 
-	got := &TestUnionType{}
+	var got interface{}
 	err := dec.Decode(&got)
 
 	assert.NoError(t, err)
-	assert.Equal(t, "B", got.Val)
+	assert.Equal(t, map[string]int{"foo": 27}, got)
 }
 
-func TestDecoder_UnionTypedNilPointer(t *testing.T) {
+func TestDecoder_UnionInterfaceMapNamed(t *testing.T) {
 	defer ConfigTeardown()
 
-	data := []byte{0x02}
-	schema := `["null", "long"]`
+	avro.Register("map:test", map[string]string{})
+
+	data := []byte{0x02, 0x01, 0x0a, 0x06, 0x66, 0x6f, 0x6f, 0x02, 0x00}
+	schema := `["int", {"type": "map", "values": {"type":"enum", "name": "test", "symbols": ["A", "B"]}}]`
 	dec, _ := avro.NewDecoder(schema, bytes.NewReader(data))
 
-	got := &TestUnionType{}
+	var got interface{}
 	err := dec.Decode(&got)
 
-	assert.Error(t, err)
+	assert.NoError(t, err)
+	assert.Equal(t, map[string]string{"foo": "B"}, got)
 }
 
-func TestDecoder_UnionTypedSetTypeError(t *testing.T) {
+func TestDecoder_UnionInterfaceArray(t *testing.T) {
 	defer ConfigTeardown()
 
-	data := []byte{0x02}
-	schema := `["null", "boolean"]`
+	avro.Register("array:int", []int{})
+
+	data := []byte{0x02, 0x01, 0x02, 0x36, 0x00}
+	schema := `["int", {"type": "array", "items": "int"}]`
 	dec, _ := avro.NewDecoder(schema, bytes.NewReader(data))
 
-	got := &TestUnionType{}
+	var got interface{}
 	err := dec.Decode(&got)
 
-	assert.Error(t, err)
+	assert.NoError(t, err)
+	assert.Equal(t, []int{27}, got)
 }
 
-func TestDecoder_UnionTypedSetTypeNilPtr(t *testing.T) {
+func TestDecoder_UnionInterfaceArrayNamed(t *testing.T) {
 	defer ConfigTeardown()
 
-	data := []byte{0x02, 0x06, 0x66, 0x6F, 0x6F}
-	schema := `["null", "string"]`
+	avro.Register("array:test", []string{})
+
+	data := []byte{0x02, 0x01, 0x02, 0x02, 0x00}
+	schema := `["int", {"type": "array", "items": {"type":"enum", "name": "test", "symbols": ["A", "B"]}}]`
 	dec, _ := avro.NewDecoder(schema, bytes.NewReader(data))
 
-	got := &TestUnionType{}
+	var got interface{}
 	err := dec.Decode(&got)
 
-	assert.Error(t, err)
+	assert.NoError(t, err)
+	assert.Equal(t, []string{"B"}, got)
 }
 
-func TestDecoder_UnionTypedInvalidSchema(t *testing.T) {
+func TestDecoder_UnionInterfaceNull(t *testing.T) {
 	defer ConfigTeardown()
 
-	data := []byte{0x04}
+	data := []byte{0x00}
 	schema := `["null", "string"]`
 	dec, _ := avro.NewDecoder(schema, bytes.NewReader(data))
 
-	got := &TestUnionType{}
+	var got interface{}
 	err := dec.Decode(&got)
 
-	assert.Error(t, err)
+	assert.NoError(t, err)
+	assert.Equal(t, nil, got)
 }
 
-func TestDecoder_UnionPtr(t *testing.T) {
+func TestDecoder_UnionInterfaceNamed(t *testing.T) {
 	defer ConfigTeardown()
 
-	data := []byte{0x02, 0x06, 0x66, 0x6F, 0x6F}
-	schema := `["null", "string"]`
+	avro.Register("test", "")
+
+	data := []byte{0x02, 0x02}
+	schema := `["null", {"type":"enum", "name": "test", "symbols": ["A", "B"]}]`
 	dec, _ := avro.NewDecoder(schema, bytes.NewReader(data))
 
-	var got *string
+	var got interface{}
 	err := dec.Decode(&got)
 
-	want := "foo"
 	assert.NoError(t, err)
-	assert.Equal(t, &want, got)
+	assert.Equal(t, "B", got)
 }
 
-func TestDecoder_UnionPtrNull(t *testing.T) {
+func TestDecoder_UnionInterfaceRecord(t *testing.T) {
 	defer ConfigTeardown()
 
-	data := []byte{0x00}
-	schema := `["null", "string"]`
+	avro.Register("test", &TestRecord{})
+
+	data := []byte{0x02, 0x36, 0x06, 0x66, 0x6F, 0x6F}
+	schema := `["int", {"type": "record", "name": "test", "fields" : [{"name": "a", "type": "long"}, {"name": "b", "type": "string"}]}]`
 	dec, _ := avro.NewDecoder(schema, bytes.NewReader(data))
 
-	var got *string
+	var got interface{}
 	err := dec.Decode(&got)
 
 	assert.NoError(t, err)
-	assert.Nil(t, got)
+	assert.IsType(t, &TestRecord{}, got)
+	rec := got.(*TestRecord)
+	assert.Equal(t, int64(27), rec.A)
+	assert.Equal(t, "foo", rec.B)
 }
 
-func TestDecoder_UnionPtrInvalidSchema(t *testing.T) {
+func TestDecoder_UnionInterfaceRecordNotReused(t *testing.T) {
 	defer ConfigTeardown()
 
-	data := []byte{0x04}
-	schema := `["null", "string"]`
+	avro.Register("test", &TestRecord{})
+
+	data := []byte{0x02, 0x36, 0x06, 0x66, 0x6F, 0x6F}
+	schema := `["int", {"type": "record", "name": "test", "fields" : [{"name": "a", "type": "long"}, {"name": "b", "type": "string"}]}]`
 	dec, _ := avro.NewDecoder(schema, bytes.NewReader(data))
 
-	var got *string
+	var got interface{} = ""
+	err := dec.Decode(&got)
+
+	assert.NoError(t, err)
+	assert.IsType(t, &TestRecord{}, got)
+	rec := got.(*TestRecord)
+	assert.Equal(t, int64(27), rec.A)
+	assert.Equal(t, "foo", rec.B)
+}
+
+func TestDecoder_UnionInterfaceUnresolvableType(t *testing.T) {
+	defer ConfigTeardown()
+
+	data := []byte{0x02, 0x36, 0x06, 0x66, 0x6F, 0x6F}
+	schema := `["int", {"type": "record", "name": "test", "fields" : [{"name": "a", "type": "long"}, {"name": "b", "type": "string"}]}]`
+	dec, _ := avro.NewDecoder(schema, bytes.NewReader(data))
+
+	var got interface{}
+	err := dec.Decode(&got)
+
+	assert.NoError(t, err)
+	assert.IsType(t, map[string]interface{}{}, got)
+	m := got.(map[string]interface{})
+	assert.IsType(t, map[string]interface{}{}, m["test"])
+	assert.Equal(t, int64(27), m["test"].(map[string]interface{})["a"])
+	assert.Equal(t, "foo", m["test"].(map[string]interface{})["b"])
+}
+
+func TestDecoder_UnionInterfaceUnresolvableTypeWithError(t *testing.T) {
+	defer ConfigTeardown()
+
+	avro.DefaultConfig = avro.Config{UnionResolutionError: true}.Freeze()
+
+	data := []byte{0x02, 0x36, 0x06, 0x66, 0x6F, 0x6F}
+	schema := `["int", {"type": "record", "name": "test", "fields" : [{"name": "a", "type": "long"}, {"name": "b", "type": "string"}]}]`
+	dec, _ := avro.NewDecoder(schema, bytes.NewReader(data))
+
+	var got interface{}
 	err := dec.Decode(&got)
 
 	assert.Error(t, err)
 }
 
-func TestDecoder_UnionPtrNotNullable(t *testing.T) {
+func TestDecoder_UnionInterfaceInvalidSchema(t *testing.T) {
 	defer ConfigTeardown()
 
-	data := []byte{0x02, 0x06, 0x66, 0x6F, 0x6F}
-	schema := `["null", "string", "int"]`
+	data := []byte{0x04}
+	schema := `["null", "int"]`
 	dec, _ := avro.NewDecoder(schema, bytes.NewReader(data))
 
-	var got *string
+	var got interface{}
 	err := dec.Decode(&got)
 
 	assert.Error(t, err)
diff --git a/encoder_record_test.go b/encoder_record_test.go
index 024bbe9..19815b9 100644
--- a/encoder_record_test.go
+++ b/encoder_record_test.go
@@ -181,6 +181,27 @@ func TestEncoder_RecordMap(t *testing.T) {
 	assert.Equal(t, []byte{0x36, 0x06, 0x66, 0x6f, 0x6f}, buf.Bytes())
 }
 
+func TestEncoder_RecordMapNilValue(t *testing.T) {
+	defer ConfigTeardown()
+
+	schema := `{
+	"type": "record",
+	"name": "test",
+	"fields" : [
+		{"name": "a", "type": "long"},
+	    {"name": "b", "type": "string"}
+	]
+}`
+	obj := map[string]interface{}{"a": int64(27), "b": nil}
+	buf := &bytes.Buffer{}
+	enc, err := avro.NewEncoder(schema, buf)
+	assert.NoError(t, err)
+
+	err = enc.Encode(obj)
+
+	assert.Error(t, err)
+}
+
 func TestEncoder_RecordMapMissingRequiredField(t *testing.T) {
 	defer ConfigTeardown()
 
diff --git a/encoder_union_test.go b/encoder_union_test.go
index da0efc3..b541229 100644
--- a/encoder_union_test.go
+++ b/encoder_union_test.go
@@ -8,19 +8,6 @@ import (
 	"github.com/stretchr/testify/assert"
 )
 
-func TestEncoder_UnionInvalidType(t *testing.T) {
-	defer ConfigTeardown()
-
-	schema := `["null", "string"]`
-	buf := bytes.NewBuffer([]byte{})
-	enc, err := avro.NewEncoder(schema, buf)
-	assert.NoError(t, err)
-
-	err = enc.Encode("foo")
-
-	assert.Error(t, err)
-}
-
 func TestEncoder_UnionMap(t *testing.T) {
 	defer ConfigTeardown()
 
@@ -103,21 +90,22 @@ func TestEncoder_UnionMapInvalidMap(t *testing.T) {
 	assert.Error(t, err)
 }
 
-func TestEncoder_UnionTyped(t *testing.T) {
+func TestEncoder_UnionPtr(t *testing.T) {
 	defer ConfigTeardown()
 
-	schema := `["null", "int"]`
+	schema := `["null", "string"]`
 	buf := bytes.NewBuffer([]byte{})
 	enc, err := avro.NewEncoder(schema, buf)
 	assert.NoError(t, err)
 
-	err = enc.Encode(&TestUnionType{Val: 27})
+	str := "foo"
+	err = enc.Encode(&str)
 
 	assert.NoError(t, err)
-	assert.Equal(t, []byte{0x02, 0x36}, buf.Bytes())
+	assert.Equal(t, []byte{0x02, 0x06, 0x66, 0x6F, 0x6F}, buf.Bytes())
 }
 
-func TestEncoder_UnionTypedNull(t *testing.T) {
+func TestEncoder_UnionPtrNull(t *testing.T) {
 	defer ConfigTeardown()
 
 	schema := `["null", "string"]`
@@ -125,105 +113,167 @@ func TestEncoder_UnionTypedNull(t *testing.T) {
 	enc, err := avro.NewEncoder(schema, buf)
 	assert.NoError(t, err)
 
-	err = enc.Encode(&TestUnionType{Val: nil})
+	var str *string
+	err = enc.Encode(str)
 
 	assert.NoError(t, err)
 	assert.Equal(t, []byte{0x00}, buf.Bytes())
 }
 
-func TestEncoder_UnionTypedNamed(t *testing.T) {
+func TestEncoder_UnionPtrNotNullable(t *testing.T) {
 	defer ConfigTeardown()
 
-	schema := `["null", {"type":"enum", "name": "test", "symbols": ["A", "B"]}]`
+	schema := `["null", "string", "int"]`
 	buf := bytes.NewBuffer([]byte{})
 	enc, err := avro.NewEncoder(schema, buf)
 	assert.NoError(t, err)
 
-	err = enc.Encode(&TestUnionType{Val: "B"})
+	str := "test"
+	err = enc.Encode(&str)
+
+	assert.Error(t, err)
+}
 
+func TestEncoder_UnionInterface(t *testing.T) {
+	defer ConfigTeardown()
+
+	schema := `["int", "string"]`
+	buf := bytes.NewBuffer([]byte{})
+	enc, err := avro.NewEncoder(schema, buf)
 	assert.NoError(t, err)
-	assert.Equal(t, []byte{0x02, 0x02}, buf.Bytes())
+
+	var val interface{} = "foo"
+	err = enc.Encode(val)
+
+	assert.NoError(t, err)
+	assert.Equal(t, []byte{0x02, 0x06, 0x66, 0x6F, 0x6F}, buf.Bytes())
 }
 
-func TestEncoder_UnionTypedNilPointer(t *testing.T) {
+func TestEncoder_UnionInterfaceRecord(t *testing.T) {
 	defer ConfigTeardown()
 
-	schema := `["null", "long"]`
+	avro.Register("test", &TestRecord{})
+
+	schema := `["int", {"type": "record", "name": "test", "fields" : [{"name": "a", "type": "long"}, {"name": "b", "type": "string"}]}]`
 	buf := bytes.NewBuffer([]byte{})
 	enc, err := avro.NewEncoder(schema, buf)
 	assert.NoError(t, err)
 
-	err = enc.Encode(&TestUnionType{Val: int64(1)})
+	var val interface{} = &TestRecord{A: 27, B: "foo"}
+	err = enc.Encode(val)
 
-	assert.Error(t, err)
+	assert.NoError(t, err)
+	assert.Equal(t, []byte{0x02, 0x36, 0x06, 0x66, 0x6F, 0x6F}, buf.Bytes())
 }
 
-func TestEncoder_UnionTypedGetTypeError(t *testing.T) {
+func TestEncoder_UnionInterfaceRecordNonPtr(t *testing.T) {
 	defer ConfigTeardown()
 
-	schema := `["null", "long"]`
+	avro.Register("test", TestRecord{})
+
+	schema := `["int", {"type": "record", "name": "test", "fields" : [{"name": "a", "type": "long"}, {"name": "b", "type": "string"}]}]`
 	buf := bytes.NewBuffer([]byte{})
 	enc, err := avro.NewEncoder(schema, buf)
 	assert.NoError(t, err)
 
-	err = enc.Encode(&TestUnionType{Val: int64(1)})
+	var val interface{} = TestRecord{A: 27, B: "foo"}
+	err = enc.Encode(val)
 
-	assert.Error(t, err)
+	assert.NoError(t, err)
+	assert.Equal(t, []byte{0x02, 0x36, 0x06, 0x66, 0x6F, 0x6F}, buf.Bytes())
 }
 
-func TestEncoder_UnionTypedInvalidSchema(t *testing.T) {
+func TestEncoder_UnionInterfaceMap(t *testing.T) {
 	defer ConfigTeardown()
 
-	schema := `["null", "long"]`
+	avro.Register("map:int", map[string]int{})
+
+	schema := `["int", {"type": "map", "values": "int"}]`
 	buf := bytes.NewBuffer([]byte{})
 	enc, err := avro.NewEncoder(schema, buf)
 	assert.NoError(t, err)
 
-	err = enc.Encode(&TestUnionType{Val: 1})
+	var val interface{} = map[string]int{"foo": 27}
+	err = enc.Encode(val)
 
-	assert.Error(t, err)
+	assert.NoError(t, err)
+	assert.Equal(t, []byte{0x02, 0x01, 0x0a, 0x06, 0x66, 0x6f, 0x6f, 0x36, 0x00}, buf.Bytes())
 }
 
-func TestEncoder_UnionPtr(t *testing.T) {
+func TestEncoder_UnionInterfaceArray(t *testing.T) {
 	defer ConfigTeardown()
 
-	schema := `["null", "string"]`
+	avro.Register("array:int", []int{})
+
+	schema := `["int", {"type": "array", "items": "int"}]`
 	buf := bytes.NewBuffer([]byte{})
 	enc, err := avro.NewEncoder(schema, buf)
 	assert.NoError(t, err)
 
-	str := "foo"
-	err = enc.Encode(&str)
+	var val interface{} = []int{27}
+	err = enc.Encode(val)
 
 	assert.NoError(t, err)
-	assert.Equal(t, []byte{0x02, 0x06, 0x66, 0x6F, 0x6F}, buf.Bytes())
+	assert.Equal(t, []byte{0x02, 0x01, 0x02, 0x36, 0x00}, buf.Bytes())
 }
 
-func TestEncoder_UnionPtrNull(t *testing.T) {
+func TestEncoder_UnionInterfaceNull(t *testing.T) {
 	defer ConfigTeardown()
 
-	schema := `["null", "string"]`
+	schema := `{"type": "record", "name": "test", "fields" : [{"name": "a", "type": ["null", "string", "int"]}]}`
 	buf := bytes.NewBuffer([]byte{})
 	enc, err := avro.NewEncoder(schema, buf)
 	assert.NoError(t, err)
 
-	var str *string
-	err = enc.Encode(str)
+	err = enc.Encode(&TestUnion{A: nil})
 
 	assert.NoError(t, err)
 	assert.Equal(t, []byte{0x00}, buf.Bytes())
 }
 
-func TestEncoder_UnionPtrNotNullable(t *testing.T) {
+func TestEncoder_UnionInterfaceNamed(t *testing.T) {
 	defer ConfigTeardown()
 
-	schema := `["null", "string", "int"]`
+	avro.Register("test", "")
+
+	schema := `["null", {"type":"enum", "name": "test", "symbols": ["A", "B"]}]`
 	buf := bytes.NewBuffer([]byte{})
 	enc, err := avro.NewEncoder(schema, buf)
 	assert.NoError(t, err)
 
-	str := "test"
-	err = enc.Encode(&str)
+	var val interface{} = "B"
+	err = enc.Encode(val)
+
+	assert.NoError(t, err)
+	assert.Equal(t, []byte{0x02, 0x02}, buf.Bytes())
+}
+
+func TestEncoder_UnionInterfaceUnregisteredType(t *testing.T) {
+	defer ConfigTeardown()
+
+	schema := `["int", {"type": "record", "name": "test", "fields" : [{"name": "a", "type": "long"}, {"name": "b", "type": "string"}]}]`
+	buf := bytes.NewBuffer([]byte{})
+	enc, err := avro.NewEncoder(schema, buf)
+	assert.NoError(t, err)
+
+	var val interface{} = &TestRecord{}
+	err = enc.Encode(val)
+
+	assert.Error(t, err)
+}
+
+func TestEncoder_UnionInterfaceNotInSchema(t *testing.T) {
+	defer ConfigTeardown()
+
+	avro.Register("test", &TestRecord{})
+
+	schema := `["int", "string"]`
+	buf := bytes.NewBuffer([]byte{})
+	enc, err := avro.NewEncoder(schema, buf)
+	assert.NoError(t, err)
+
+	var val interface{} = &TestRecord{}
+	err = enc.Encode(val)
 
 	assert.Error(t, err)
 }
diff --git a/example_resolver_test.go b/example_resolver_test.go
new file mode 100644
index 0000000..29809f9
--- /dev/null
+++ b/example_resolver_test.go
@@ -0,0 +1,25 @@
+package avro_test
+
+import (
+	"fmt"
+	"log"
+
+	"github.com/hamba/avro"
+)
+
+func ExampleRegister() {
+	data := []byte{0x02, 0x02} // Your Avro data here
+	schema := avro.MustParse(`["null", {"type":"enum", "name": "test", "symbols": ["A", "B"]}]`)
+
+	avro.Register("test", "") // Register the name test as a string type
+
+	var result interface{}
+	err := avro.Unmarshal(schema, data, &result)
+	if err != nil {
+		log.Fatal(err)
+	}
+
+	fmt.Println(result)
+
+	//Output: B
+}
diff --git a/example_simple_test.go b/example_simple_test.go
index eb69036..3086edb 100644
--- a/example_simple_test.go
+++ b/example_simple_test.go
@@ -36,7 +36,6 @@ func Example_usage() {
 	}
 
 	fmt.Printf("%+v\n", data)
-	// Outputs: [54 6 102 111 111]
 
 	out := SimpleRecord{}
 	err = avro.Unmarshal(schema, data, &out)
@@ -45,5 +44,7 @@ func Example_usage() {
 	}
 
 	fmt.Printf("%+v\n", out)
-	// Outputs: {A:27 B:foo}
+
+	//Output: [54 6 102 111 111]
+	// {A:27 B:foo}
 }
diff --git a/example_test.go b/example_test.go
index 5edf5fc..56c0857 100644
--- a/example_test.go
+++ b/example_test.go
@@ -4,7 +4,6 @@ import (
 	"bytes"
 	"fmt"
 	"log"
-	"os"
 
 	"github.com/hamba/avro"
 )
@@ -73,7 +72,7 @@ func ExampleNewDecoderForSchema() {
 		B string `avro:"b"`
 	}
 
-	r := bytes.NewReader([]byte{}) // Your reader goes here
+	r := bytes.NewReader([]byte{0x36, 0x06, 0x66, 0x6F, 0x6F}) // Your reader goes here
 	decoder := avro.NewDecoderForSchema(schema, r)
 
 	simple := SimpleRecord{}
@@ -82,6 +81,8 @@ func ExampleNewDecoderForSchema() {
 	}
 
 	fmt.Printf("%+v", simple)
+
+	//Output: {A:27 B:foo}
 }
 
 func ExampleUnmarshal() {
@@ -100,13 +101,15 @@ func ExampleUnmarshal() {
 		B string `avro:"b"`
 	}
 
-	data := []byte{} // Your Avro data here
+	data := []byte{0x36, 0x06, 0x66, 0x6F, 0x6F} // Your Avro data here
 	simple := SimpleRecord{}
 	if err := avro.Unmarshal(schema, data, &simple); err != nil {
 		fmt.Println("error:", err)
 	}
 
 	fmt.Printf("%+v", simple)
+
+	//Output: {A:27 B:foo}
 }
 
 func ExampleNewEncoder() {
@@ -131,12 +134,14 @@ func ExampleNewEncoder() {
 		fmt.Println("error:", err)
 	}
 
-	simple := SimpleRecord{}
+	simple := SimpleRecord{A: 27, B: "foo"}
 	if err := encoder.Encode(simple); err != nil {
 		fmt.Println("error:", err)
 	}
 
-	os.Stdout.Write(w.Bytes())
+	fmt.Println(w.Bytes())
+
+	//Output: [54 6 102 111 111]
 }
 
 func ExampleNewEncoderForSchema() {
@@ -158,12 +163,14 @@ func ExampleNewEncoderForSchema() {
 	w := &bytes.Buffer{}
 	encoder := avro.NewEncoderForSchema(schema, w)
 
-	simple := SimpleRecord{}
+	simple := SimpleRecord{A: 27, B: "foo"}
 	if err := encoder.Encode(simple); err != nil {
 		fmt.Println("error:", err)
 	}
 
-	os.Stdout.Write(w.Bytes())
+	fmt.Println(w.Bytes())
+
+	//Output: [54 6 102 111 111]
 }
 
 func ExampleMarshal() {
@@ -182,11 +189,13 @@ func ExampleMarshal() {
 		B string `avro:"b"`
 	}
 
-	simple := SimpleRecord{}
+	simple := SimpleRecord{A: 27, B: "foo"}
 	b, err := avro.Marshal(schema, simple)
 	if err != nil {
 		fmt.Println("error:", err)
 	}
 
-	os.Stdout.Write(b)
+	fmt.Println(b)
+
+	//Output: [54 6 102 111 111]
 }
diff --git a/example_uniontype_test.go b/example_uniontype_test.go
deleted file mode 100644
index 829f349..0000000
--- a/example_uniontype_test.go
+++ /dev/null
@@ -1,66 +0,0 @@
-package avro_test
-
-import (
-	"fmt"
-	"os"
-
-	"github.com/hamba/avro"
-)
-
-type SimpleUnionType struct {
-	Val interface{}
-}
-
-func (u *SimpleUnionType) Value() *interface{} {
-	return &u.Val
-}
-
-func (u *SimpleUnionType) SetType(typ string) error {
-	switch typ {
-	case string(avro.String):
-		u.Val = ""
-
-	case string(avro.Int):
-		u.Val = int(0)
-
-	default:
-		return fmt.Errorf("unknown type %s", typ)
-	}
-
-	return nil
-}
-
-func (u *SimpleUnionType) GetType() (string, error) {
-	switch u.Val.(type) {
-	case string:
-		return string(avro.String), nil
-
-	case int:
-		return string(avro.Int), nil
-	}
-
-	return "", fmt.Errorf("unknown type %T", u.Val)
-}
-
-type UnionRecord struct {
-	Items map[string]*SimpleUnionType `avro:"items"`
-}
-
-func ExampleUnionType() {
-	schema := avro.MustParse(`{
-	    "type": "record",
-	    "name": "union",
-	    "namespace": "org.hamba.avro",
-	    "fields" : [
-	        {"name": "items", "type": {"type": "map", "values": ["string", "int"]}}
-	    ]
-	}`)
-
-	record := UnionRecord{}
-	b, err := avro.Marshal(schema, record)
-	if err != nil {
-		fmt.Println("error:", err)
-	}
-
-	os.Stdout.Write(b)
-}
diff --git a/protocol_test.go b/protocol_test.go
new file mode 100644
index 0000000..f47c715
--- /dev/null
+++ b/protocol_test.go
@@ -0,0 +1,182 @@
+package avro_test
+
+import (
+	"testing"
+
+	"github.com/hamba/avro"
+	"github.com/stretchr/testify/assert"
+)
+
+func TestMustParseProtocol(t *testing.T) {
+	proto := avro.MustParseProtocol(`{"protocol":"test", "namespace": "org.hamba.avro", "doc": "docs"}`)
+
+	assert.IsType(t, &avro.Protocol{}, proto)
+}
+
+func TestMustParseProtocol_PanicsOnError(t *testing.T) {
+	assert.Panics(t, func() {
+		avro.MustParseProtocol("123")
+	})
+}
+
+func TestNewProtocol_ValidatesName(t *testing.T) {
+	_, err := avro.NewProtocol("0test", "", nil, nil)
+
+	assert.Error(t, err)
+}
+
+func TestNewMessage(t *testing.T) {
+	field, _ := avro.NewField("test", avro.NewPrimitiveSchema(avro.String), nil)
+	fields := []*avro.Field{field}
+	req, _ := avro.NewRecordSchema("test", "", fields)
+	resp := avro.NewPrimitiveSchema(avro.String)
+	types := []avro.Schema{avro.NewPrimitiveSchema(avro.String)}
+	errs, _ := avro.NewUnionSchema(types)
+
+	msg := avro.NewMessage(req, resp, errs, false)
+
+	assert.Equal(t, req, msg.Request())
+	assert.Equal(t, resp, msg.Response())
+	assert.Equal(t, errs, msg.Errors())
+	assert.False(t, msg.OneWay())
+}
+
+func TestParseProtocol(t *testing.T) {
+	tests := []struct {
+		name    string
+		schema  string
+		wantErr bool
+	}{
+		{
+			name:    "Valid",
+			schema:  `{"protocol":"test", "namespace": "org.hamba.avro", "doc": "docs"}`,
+			wantErr: false,
+		},
+		{
+			name:    "Invalid Json",
+			schema:  `{`,
+			wantErr: true,
+		},
+		{
+			name:    "Invalid Name First Char",
+			schema:  `{"protocol":"0test", "namespace": "org.hamba.avro"}`,
+			wantErr: true,
+		},
+		{
+			name:    "Invalid Name Other Char",
+			schema:  `{"protocol":"test+", "namespace": "org.hamba.avro"}`,
+			wantErr: true,
+		},
+		{
+			name:    "Empty Name",
+			schema:  `{"protocol":"", "namespace": "org.hamba.avro"}`,
+			wantErr: true,
+		},
+		{
+			name:    "No Name",
+			schema:  `{"namespace": "org.hamba.avro"}`,
+			wantErr: true,
+		},
+		{
+			name:    "Invalid Namespace",
+			schema:  `{"protocol":"test", "namespace": "org.hamba.avro+"}`,
+			wantErr: true,
+		},
+		{
+			name:    "Empty Namespace",
+			schema:  `{"protocol":"test", "namespace": ""}`,
+			wantErr: true,
+		},
+		{
+			name:    "Invalid Type Schema",
+			schema:  `{"protocol":"test", "namespace": "org.hamba.avro", "types":["test"]}`,
+			wantErr: true,
+		},
+		{
+			name:    "Type Not Named Schema",
+			schema:  `{"protocol":"test", "namespace": "org.hamba.avro", "types":["string"]}`,
+			wantErr: true,
+		},
+		{
+			name:    "Message Not Object",
+			schema:  `{"protocol":"test", "namespace": "org.hamba.avro", "messages":{"test":["test"]}}`,
+			wantErr: true,
+		},
+		{
+			name:    "Message Request Invalid Request Json",
+			schema:  `{"protocol":"test", "namespace": "org.hamba.avro", "messages":{"test":{"request": "test"}}}`,
+			wantErr: true,
+		},
+		{
+			name:    "Message Request Invalid Field",
+			schema:  `{"protocol":"test", "namespace": "org.hamba.avro", "messages":{"test":{"request": [{"name": "foobar"}]}}}`,
+			wantErr: true,
+		},
+		{
+			name:    "Message Response Invalid Schema",
+			schema:  `{"protocol":"test", "namespace": "org.hamba.avro", "messages":{"test":{"request": [{"name": "foobar", "type": "string"}], "response": "test"}}}`,
+			wantErr: true,
+		},
+		{
+			name:    "Message Errors Invalid Schema",
+			schema:  `{"protocol":"test", "namespace": "org.hamba.avro", "messages":{"test":{"request": [{"name": "foobar", "type": "string"}], "errors": ["test"]}}}`,
+			wantErr: true,
+		},
+		{
+			name:    "Message Errors Record Not Error Schema",
+			schema:  `{"protocol":"test", "namespace": "org.hamba.avro", "messages":{"test":{"request": [{"name": "foobar", "type": "string"}], "errors": [{"type":"record", "name":"test", "fields":[{"name": "field", "type": "int"}]}]}}}`,
+			wantErr: true,
+		},
+		{
+			name:    "Message Errors Duplicate Schema",
+			schema:  `{"protocol":"test", "namespace": "org.hamba.avro", "messages":{"test":{"request": [{"name": "foobar", "type": "string"}], "errors": ["string"]}}}`,
+			wantErr: true,
+		},
+		{
+			name:    "Message One Way Invalid",
+			schema:  `{"protocol":"test", "namespace": "org.hamba.avro", "messages":{"test":{"request": [{"name": "foobar", "type": "string"}], "errors": ["int"], "one-way": true}}}`,
+			wantErr: true,
+		},
+	}
+
+	for _, tt := range tests {
+		t.Run(tt.name, func(t *testing.T) {
+			_, err := avro.ParseProtocol(tt.schema)
+
+			if tt.wantErr {
+				assert.Error(t, err)
+				return
+			}
+
+			assert.NoError(t, err)
+		})
+	}
+}
+
+func TestParseProtocol_DeterminesOneWayMessage(t *testing.T) {
+	schema := `{"protocol":"test", "namespace": "org.hamba.avro", "messages":{"test":{"request": [{"name": "foobar", "type": "string"}]}}}`
+
+	proto, err := avro.ParseProtocol(schema)
+
+	assert.NoError(t, err)
+
+	msg := proto.Message("test")
+	assert.NotNil(t, msg)
+	assert.True(t, msg.OneWay())
+}
+
+func TestParseProtocolFile(t *testing.T) {
+	protocol, err := avro.ParseProtocolFile("testdata/echo.avpr")
+
+	want := `{"protocol":"Echo","namespace":"org.hamba.avro","types":[{"name":"org.hamba.avro.Ping","type":"record","fields":[{"name":"timestamp","type":"long"},{"name":"text","type":"string"}]},{"name":"org.hamba.avro.Pong","type":"record","fields":[{"name":"timestamp","type":"long"},{"name":"ping","type":"org.hamba.avro.Ping"}]},{"name":"org.hamba.avro.PongError","type":"error","fields":[{"name":"timestamp","type":"long"},{"name":"reason","type":"string"}]}],"messages":{"ping":{"request":[{"name":"ping","type":"org.hamba.avro.Ping"}],"response":"org.hamba.avro.Pong","errors":["org.hamba.avro.PongError"]}}}`
+	wantMD5 := "5bc594ae86fc8c209f553ce3bc4291a5"
+	assert.NoError(t, err)
+	assert.Equal(t, want, protocol.String())
+	assert.Equal(t, wantMD5, protocol.Hash())
+}
+
+func TestParseProtocolFile_InvalidPath(t *testing.T) {
+	_, err := avro.ParseProtocolFile("test.avpr")
+
+	assert.Error(t, err)
+}
diff --git a/schema_canonical_test.go b/schema_canonical_test.go
index 5459362..3da3e0a 100644
--- a/schema_canonical_test.go
+++ b/schema_canonical_test.go
@@ -91,6 +91,10 @@ func TestSchema_Canonical(t *testing.T) {
 			input:     `[ "int" , {"type":"boolean"} ]`,
 			canonical: `["int","boolean"]`,
 		},
+		{
+			input:     `{"fields":[], "type":"error", "name":"foo"}`,
+			canonical: `{"name":"foo","type":"error","fields":[]}`,
+		},
 		{
 			input:     `{"fields":[], "type":"record", "name":"foo"}`,
 			canonical: `{"name":"foo","type":"record","fields":[]}`,
@@ -162,7 +166,7 @@ func TestSchema_Canonical(t *testing.T) {
 		{
 			input: `{
 				"type":"record",
-				"namespace": "org.apache.avro",
+				"namespace": "org.hamba.avro",
 				"name":"X",
   				"fields":[
 					{"name":"value", "type":{
@@ -174,12 +178,12 @@ func TestSchema_Canonical(t *testing.T) {
 					}}
 				]
 			}`,
-			canonical: `{"name":"org.apache.avro.X","type":"record","fields":[{"name":"value","type":{"name":"org.apache.avro.Y","type":"record","fields":[{"name":"value","type":"string"}]}}]}`,
+			canonical: `{"name":"org.hamba.avro.X","type":"record","fields":[{"name":"value","type":{"name":"org.hamba.avro.Y","type":"record","fields":[{"name":"value","type":"string"}]}}]}`,
 		},
 		{
 			input: `{
 				"type":"record",
-				"namespace": "org.apache.avro",
+				"namespace": "org.hamba.avro",
 				"name":"X",
   				"fields":[
 					{"name":"value", "type":{
@@ -189,12 +193,12 @@ func TestSchema_Canonical(t *testing.T) {
 					}}
 				]
 			}`,
-			canonical: `{"name":"org.apache.avro.X","type":"record","fields":[{"name":"value","type":{"name":"org.apache.avro.Y","type":"enum","symbols":["TEST"]}}]}`,
+			canonical: `{"name":"org.hamba.avro.X","type":"record","fields":[{"name":"value","type":{"name":"org.hamba.avro.Y","type":"enum","symbols":["TEST"]}}]}`,
 		},
 		{
 			input: `{
 				"type":"record",
-				"namespace": "org.apache.avro",
+				"namespace": "org.hamba.avro",
 				"name":"X",
   				"fields":[
 					{"name":"value", "type":{
@@ -204,7 +208,7 @@ func TestSchema_Canonical(t *testing.T) {
 					}}
 				]
 			}`,
-			canonical: `{"name":"org.apache.avro.X","type":"record","fields":[{"name":"value","type":{"name":"org.apache.avro.Y","type":"fixed","size":15}}]}`,
+			canonical: `{"name":"org.hamba.avro.X","type":"record","fields":[{"name":"value","type":{"name":"org.hamba.avro.Y","type":"fixed","size":15}}]}`,
 		},
 	}
 
diff --git a/schema_compatibility_test.go b/schema_compatibility_test.go
new file mode 100644
index 0000000..b5ebd07
--- /dev/null
+++ b/schema_compatibility_test.go
@@ -0,0 +1,281 @@
+package avro_test
+
+import (
+	"testing"
+
+	"github.com/hamba/avro"
+	"github.com/stretchr/testify/assert"
+)
+
+func TestNewSchemaCompatibility(t *testing.T) {
+	sc := avro.NewSchemaCompatibility()
+
+	assert.IsType(t, &avro.SchemaCompatibility{}, sc)
+}
+
+func TestSchemaCompatibility_Compatible(t *testing.T) {
+	tests := []struct {
+		name    string
+		reader  string
+		writer  string
+		wantErr bool
+	}{
+		{
+			name:    "Primitive Matching",
+			reader:  `"int"`,
+			writer:  `"int"`,
+			wantErr: false,
+		},
+		{
+			name:    "Int Promote Long",
+			reader:  `"long"`,
+			writer:  `"int"`,
+			wantErr: false,
+		},
+		{
+			name:    "Int Promote Float",
+			reader:  `"float"`,
+			writer:  `"int"`,
+			wantErr: false,
+		},
+		{
+			name:    "Int Promote Double",
+			reader:  `"double"`,
+			writer:  `"int"`,
+			wantErr: false,
+		},
+		{
+			name:    "Long Promote Float",
+			reader:  `"float"`,
+			writer:  `"long"`,
+			wantErr: false,
+		},
+		{
+			name:    "Long Promote Double",
+			reader:  `"double"`,
+			writer:  `"long"`,
+			wantErr: false,
+		},
+		{
+			name:    "Float Promote Double",
+			reader:  `"double"`,
+			writer:  `"float"`,
+			wantErr: false,
+		},
+		{
+			name:    "String Promote Bytes",
+			reader:  `"bytes"`,
+			writer:  `"string"`,
+			wantErr: false,
+		},
+		{
+			name:    "Bytes Promote String",
+			reader:  `"string"`,
+			writer:  `"bytes"`,
+			wantErr: false,
+		},
+		{
+			name:    "Union Match",
+			reader:  `["int", "long", "string"]`,
+			writer:  `["string", "int", "long"]`,
+			wantErr: false,
+		},
+		{
+			name:    "Union Reader Missing Schema",
+			reader:  `["int", "string"]`,
+			writer:  `["string", "int", "long"]`,
+			wantErr: true,
+		},
+		{
+			name:    "Union Writer Missing Schema",
+			reader:  `["int", "long", "string"]`,
+			writer:  `["string", "int"]`,
+			wantErr: false,
+		},
+		{
+			name:    "Union Writer Not Union",
+			reader:  `["int", "long", "string"]`,
+			writer:  `"int"`,
+			wantErr: false,
+		},
+		{
+			name:    "Union Writer Not Union With Error",
+			reader:  `["string"]`,
+			writer:  `"int"`,
+			wantErr: true,
+		},
+		{
+			name:    "Union Reader Not Union",
+			reader:  `"int"`,
+			writer:  `["int"]`,
+			wantErr: false,
+		},
+		{
+			name:    "Union Reader Not Union With Error",
+			reader:  `"int"`,
+			writer:  `["string", "int", "long"]`,
+			wantErr: true,
+		},
+		{
+			name:    "Array Match",
+			reader:  `{"type":"array", "items": "int"}`,
+			writer:  `{"type":"array", "items": "int"}`,
+			wantErr: false,
+		},
+		{
+			name:    "Array Items Mismatch",
+			reader:  `{"type":"array", "items": "int"}`,
+			writer:  `{"type":"array", "items": "string"}`,
+			wantErr: true,
+		},
+		{
+			name:    "Map Match",
+			reader:  `{"type":"map", "values": "int"}`,
+			writer:  `{"type":"map", "values": "int"}`,
+			wantErr: false,
+		},
+		{
+			name:    "Map Items Mismatch",
+			reader:  `{"type":"map", "values": "int"}`,
+			writer:  `{"type":"map", "values": "string"}`,
+			wantErr: true,
+		},
+		{
+			name:    "Fixed Match",
+			reader:  `{"type":"fixed", "name":"test", "namespace": "org.hamba.avro", "size": 12}`,
+			writer:  `{"type":"fixed", "name":"test", "namespace": "org.hamba.avro", "size": 12}`,
+			wantErr: false,
+		},
+		{
+			name:    "Fixed Name Mismatch",
+			reader:  `{"type":"fixed", "name":"test1", "namespace": "org.hamba.avro", "size": 12}`,
+			writer:  `{"type":"fixed", "name":"test", "namespace": "org.hamba.avro", "size": 12}`,
+			wantErr: true,
+		},
+		{
+			name:    "Fixed Size Mismatch",
+			reader:  `{"type":"fixed", "name":"test", "namespace": "org.hamba.avro", "size": 13}`,
+			writer:  `{"type":"fixed", "name":"test", "namespace": "org.hamba.avro", "size": 12}`,
+			wantErr: true,
+		},
+		{
+			name:    "Enum Match",
+			reader:  `{"type":"enum", "name":"test", "namespace": "org.hamba.avro", "symbols":["TEST1", "TEST2"]}`,
+			writer:  `{"type":"enum", "name":"test", "namespace": "org.hamba.avro", "symbols":["TEST1", "TEST2"]}`,
+			wantErr: false,
+		},
+		{
+			name:    "Enum Name Mismatch",
+			reader:  `{"type":"enum", "name":"test1", "namespace": "org.hamba.avro", "symbols":["TEST1", "TEST2"]}`,
+			writer:  `{"type":"enum", "name":"test", "namespace": "org.hamba.avro", "symbols":["TEST1", "TEST2"]}`,
+			wantErr: true,
+		},
+		{
+			name:    "Enum Reader Missing Symbol",
+			reader:  `{"type":"enum", "name":"test", "namespace": "org.hamba.avro", "symbols":["TEST1"]}`,
+			writer:  `{"type":"enum", "name":"test", "namespace": "org.hamba.avro", "symbols":["TEST1", "TEST2"]}`,
+			wantErr: true,
+		},
+		{
+			name:    "Enum Writer Missing Symbol",
+			reader:  `{"type":"enum", "name":"test", "namespace": "org.hamba.avro", "symbols":["TEST1", "TEST2"]}`,
+			writer:  `{"type":"enum", "name":"test", "namespace": "org.hamba.avro", "symbols":["TEST1"]}`,
+			wantErr: false,
+		},
+		{
+			name:    "Record Match",
+			reader:  `{"type":"record", "name":"test", "namespace": "org.hamba.avro", "fields":[{"name": "a", "type": "int"}, {"name": "b", "type": "string"}]}`,
+			writer:  `{"type":"record", "name":"test", "namespace": "org.hamba.avro", "fields":[{"name": "b", "type": "string"}, {"name": "a", "type": "int"}]}`,
+			wantErr: false,
+		},
+		{
+			name:    "Record Name Mismatch",
+			reader:  `{"type":"record", "name":"test1", "namespace": "org.hamba.avro", "fields":[{"name": "a", "type": "int", "default": 1}, {"name": "b", "type": "string"}]}`,
+			writer:  `{"type":"record", "name":"test", "namespace": "org.hamba.avro", "fields":[{"name": "b", "type": "string", "default": "b"}, {"name": "a", "type": "int"}]}`,
+			wantErr: true,
+		},
+		{
+			name:    "Record Schema Mismatch",
+			reader:  `{"type":"record", "name":"test", "namespace": "org.hamba.avro", "fields":[{"name": "a", "type": "string"}, {"name": "b", "type": "string"}]}`,
+			writer:  `{"type":"record", "name":"test", "namespace": "org.hamba.avro", "fields":[{"name": "b", "type": "string"}, {"name": "a", "type": "int"}]}`,
+			wantErr: true,
+		},
+		{
+			name:    "Record Reader Field Missing",
+			reader:  `{"type":"record", "name":"test", "namespace": "org.hamba.avro", "fields":[{"name": "a", "type": "int"}]}`,
+			writer:  `{"type":"record", "name":"test", "namespace": "org.hamba.avro", "fields":[{"name": "b", "type": "string"}, {"name": "a", "type": "int"}]}`,
+			wantErr: false,
+		},
+		{
+			name:    "Record Writer Field Missing With Default",
+			reader:  `{"type":"record", "name":"test", "namespace": "org.hamba.avro", "fields":[{"name": "a", "type": "int"}, {"name": "b", "type": "string", "default": "test"}]}`,
+			writer:  `{"type":"record", "name":"test", "namespace": "org.hamba.avro", "fields":[{"name": "a", "type": "int"}]}`,
+			wantErr: false,
+		},
+		{
+			name:    "Record Writer Field Missing Without Default",
+			reader:  `{"type":"record", "name":"test", "namespace": "org.hamba.avro", "fields":[{"name": "a", "type": "int"}, {"name": "b", "type": "string"}]}`,
+			writer:  `{"type":"record", "name":"test", "namespace": "org.hamba.avro", "fields":[{"name": "a", "type": "int"}]}`,
+			wantErr: true,
+		},
+		{
+			name:    "Ref Dereference",
+			reader:  `{"type":"record", "name":"test", "namespace": "org.hamba.avro", "fields":[{"name": "a", "type": {"type":"record", "name":"test1", "namespace": "org.hamba.avro", "fields":[{"name": "b", "type": "int"}]}}, {"name": "b", "type": "test1"}]}`,
+			writer:  `{"type":"record", "name":"test", "namespace": "org.hamba.avro", "fields":[{"name": "a", "type": {"type":"record", "name":"test1", "namespace": "org.hamba.avro", "fields":[{"name": "b", "type": "int"}]}}, {"name": "b", "type": "test"}]}`,
+			wantErr: true,
+		},
+		{
+			name:    "Breaks Recursion",
+			reader:  `{"type":"record", "name":"test", "namespace": "org.hamba.avro", "fields":[{"name": "a", "type": "test"}]}`,
+			writer:  `{"type":"record", "name":"test", "namespace": "org.hamba.avro", "fields":[{"name": "a", "type": "test"}]}`,
+			wantErr: false,
+		},
+	}
+
+	for _, tt := range tests {
+		t.Run(tt.name, func(t *testing.T) {
+			r := avro.MustParse(tt.reader)
+			w := avro.MustParse(tt.writer)
+			sc := avro.NewSchemaCompatibility()
+
+			err := sc.Compatible(r, w)
+
+			if tt.wantErr {
+				assert.Error(t, err)
+				return
+			}
+
+			assert.NoError(t, err)
+		})
+	}
+}
+
+func TestSchemaCompatibility_CompatibleUsesCacheWithNoError(t *testing.T) {
+	reader := `"int"`
+	writer := `"int"`
+
+	r := avro.MustParse(reader)
+	w := avro.MustParse(writer)
+	sc := avro.NewSchemaCompatibility()
+
+	_ = sc.Compatible(r, w)
+
+	err := sc.Compatible(r, w)
+
+	assert.NoError(t, err)
+}
+
+func TestSchemaCompatibility_CompatibleUsesCacheWithError(t *testing.T) {
+	reader := `"int"`
+	writer := `"string"`
+
+	r := avro.MustParse(reader)
+	w := avro.MustParse(writer)
+	sc := avro.NewSchemaCompatibility()
+
+	_ = sc.Compatible(r, w)
+
+	err := sc.Compatible(r, w)
+
+	assert.Error(t, err)
+}
diff --git a/schema_internal_test.go b/schema_internal_test.go
index bf95bef..8ffbe41 100644
--- a/schema_internal_test.go
+++ b/schema_internal_test.go
@@ -84,7 +84,7 @@ func TestIsValidDefault(t *testing.T) {
 				return &NullSchema{}
 			},
 			def:    nil,
-			want:   nil,
+			want:   nullDefault,
 			wantOk: true,
 		},
 		{
@@ -319,3 +319,9 @@ func TestIsValidDefault(t *testing.T) {
 		})
 	}
 }
+
+func TestRecursionError_Error(t *testing.T) {
+	err := recursionError{}
+
+	assert.Equal(t, "", err.Error())
+}
diff --git a/schema_test.go b/schema_test.go
index e043af1..3b74c68 100644
--- a/schema_test.go
+++ b/schema_test.go
@@ -164,32 +164,32 @@ func TestRecordSchema(t *testing.T) {
 	}{
 		{
 			name:    "Valid",
-			schema:  `{"type":"record", "name":"test", "namespace": "org.apache.avro", "doc": "docs", "fields":[{"name": "field", "type": "int"}]}`,
+			schema:  `{"type":"record", "name":"test", "namespace": "org.hamba.avro", "doc": "docs", "fields":[{"name": "field", "type": "int"}]}`,
 			wantErr: false,
 		},
 		{
 			name:    "Invalid Name First Char",
-			schema:  `{"type":"record", "name":"0test", "namespace": "org.apache.avro", "fields":[{"name": "field", "type": "int"}]}`,
+			schema:  `{"type":"record", "name":"0test", "namespace": "org.hamba.avro", "fields":[{"name": "field", "type": "int"}]}`,
 			wantErr: true,
 		},
 		{
 			name:    "Invalid Name Other Char",
-			schema:  `{"type":"record", "name":"test+", "namespace": "org.apache.avro", "fields":[{"name": "field", "type": "int"}]}`,
+			schema:  `{"type":"record", "name":"test+", "namespace": "org.hamba.avro", "fields":[{"name": "field", "type": "int"}]}`,
 			wantErr: true,
 		},
 		{
 			name:    "Empty Name",
-			schema:  `{"type":"record", "name":"", "namespace": "org.apache.avro", "fields":[{"name": "field", "type": "int"}]}`,
+			schema:  `{"type":"record", "name":"", "namespace": "org.hamba.avro", "fields":[{"name": "field", "type": "int"}]}`,
 			wantErr: true,
 		},
 		{
 			name:    "No Name",
-			schema:  `{"type":"record", "namespace": "org.apache.avro", "fields":[{"name": "intField", "type": "int"}]}`,
+			schema:  `{"type":"record", "namespace": "org.hamba.avro", "fields":[{"name": "intField", "type": "int"}]}`,
 			wantErr: true,
 		},
 		{
 			name:    "Invalid Namespace",
-			schema:  `{"type":"record", "name":"test", "namespace": "org.apache.avro+", "fields":[{"name": "field", "type": "int"}]}`,
+			schema:  `{"type":"record", "name":"test", "namespace": "org.hamba.avro+", "fields":[{"name": "field", "type": "int"}]}`,
 			wantErr: true,
 		},
 		{
@@ -199,32 +199,32 @@ func TestRecordSchema(t *testing.T) {
 		},
 		{
 			name:    "No Fields",
-			schema:  `{"type":"record", "name":"test", "namespace": "org.apache.avro"}`,
+			schema:  `{"type":"record", "name":"test", "namespace": "org.hamba.avro"}`,
 			wantErr: true,
 		},
 		{
 			name:    "Invalid Field Type",
-			schema:  `{"type":"record", "name":"test", "namespace": "org.apache.avro", "fields":["test"]}`,
+			schema:  `{"type":"record", "name":"test", "namespace": "org.hamba.avro", "fields":["test"]}`,
 			wantErr: true,
 		},
 		{
 			name:    "No Field Name",
-			schema:  `{"type":"record", "name":"test", "namespace": "org.apache.avro", "fields":[{"type": "int"}]}`,
+			schema:  `{"type":"record", "name":"test", "namespace": "org.hamba.avro", "fields":[{"type": "int"}]}`,
 			wantErr: true,
 		},
 		{
 			name:    "Invalid Field Name",
-			schema:  `{"type":"record", "name":"test", "namespace": "org.apache.avro", "fields":[{"name": "field+", "type": "int"}]}`,
+			schema:  `{"type":"record", "name":"test", "namespace": "org.hamba.avro", "fields":[{"name": "field+", "type": "int"}]}`,
 			wantErr: true,
 		},
 		{
 			name:    "No Field Type",
-			schema:  `{"type":"record", "name":"test", "namespace": "org.apache.avro", "fields":[{"name": "field"}]}`,
+			schema:  `{"type":"record", "name":"test", "namespace": "org.hamba.avro", "fields":[{"name": "field"}]}`,
 			wantErr: true,
 		},
 		{
 			name:    "Invalid Field Type",
-			schema:  `{"type":"record", "name":"test", "namespace": "org.apache.avro", "fields":[{"name": "field", "type": "blah"}]}`,
+			schema:  `{"type":"record", "name":"test", "namespace": "org.hamba.avro", "fields":[{"name": "field", "type": "blah"}]}`,
 			wantErr: true,
 		},
 	}
@@ -244,6 +244,67 @@ func TestRecordSchema(t *testing.T) {
 	}
 }
 
+func TestErrorRecordSchema(t *testing.T) {
+	tests := []struct {
+		name    string
+		schema  string
+		wantErr bool
+	}{
+		{
+			name:    "Valid",
+			schema:  `{"type":"error", "name":"test", "namespace": "org.hamba.avro", "doc": "docs", "fields":[{"name": "field", "type": "int"}]}`,
+			wantErr: false,
+		},
+		{
+			name:    "Invalid Name First Char",
+			schema:  `{"type":"error", "name":"0test", "namespace": "org.hamba.avro", "fields":[{"name": "field", "type": "int"}]}`,
+			wantErr: true,
+		},
+		{
+			name:    "Invalid Name Other Char",
+			schema:  `{"type":"error", "name":"test+", "namespace": "org.hamba.avro", "fields":[{"name": "field", "type": "int"}]}`,
+			wantErr: true,
+		},
+		{
+			name:    "Empty Name",
+			schema:  `{"type":"error", "name":"", "namespace": "org.hamba.avro", "fields":[{"name": "field", "type": "int"}]}`,
+			wantErr: true,
+		},
+		{
+			name:    "No Name",
+			schema:  `{"type":"error", "namespace": "org.hamba.avro", "fields":[{"name": "intField", "type": "int"}]}`,
+			wantErr: true,
+		},
+		{
+			name:    "Invalid Namespace",
+			schema:  `{"type":"error", "name":"test", "namespace": "org.hamba.avro+", "fields":[{"name": "field", "type": "int"}]}`,
+			wantErr: true,
+		},
+		{
+			name:    "Empty Namespace",
+			schema:  `{"type":"error", "name":"test", "namespace": "", "fields":[{"name": "intField", "type": "int"}]}`,
+			wantErr: true,
+		},
+	}
+
+	for _, tt := range tests {
+		t.Run(tt.name, func(t *testing.T) {
+
+			s, err := avro.Parse(tt.schema)
+
+			if tt.wantErr {
+				assert.Error(t, err)
+				return
+			}
+
+			assert.NoError(t, err)
+			assert.Equal(t, avro.Record, s.Type())
+			recSchema := s.(*avro.RecordSchema)
+			assert.True(t, recSchema.IsError())
+		})
+	}
+}
+
 func TestRecordSchema_ValidatesDefault(t *testing.T) {
 	tests := []struct {
 		name    string
@@ -252,87 +313,87 @@ func TestRecordSchema_ValidatesDefault(t *testing.T) {
 	}{
 		{
 			name:    "String",
-			schema:  `{"type":"record", "name":"test", "namespace": "org.apache.avro", "fields":[{"name": "a", "type": "string", "default": "test"}]}`,
+			schema:  `{"type":"record", "name":"test", "namespace": "org.hamba.avro", "fields":[{"name": "a", "type": "string", "default": "test"}]}`,
 			wantErr: false,
 		},
 		{
 			name:    "Int",
-			schema:  `{"type":"record", "name":"test", "namespace": "org.apache.avro", "fields":[{"name": "a", "type": "int", "default": 1}]}`,
+			schema:  `{"type":"record", "name":"test", "namespace": "org.hamba.avro", "fields":[{"name": "a", "type": "int", "default": 1}]}`,
 			wantErr: false,
 		},
 		{
 			name:    "Long",
-			schema:  `{"type":"record", "name":"test", "namespace": "org.apache.avro", "fields":[{"name": "a", "type": "long", "default": 1}]}`,
+			schema:  `{"type":"record", "name":"test", "namespace": "org.hamba.avro", "fields":[{"name": "a", "type": "long", "default": 1}]}`,
 			wantErr: false,
 		},
 		{
 			name:    "Float",
-			schema:  `{"type":"record", "name":"test", "namespace": "org.apache.avro", "fields":[{"name": "a", "type": "float", "default": 1}]}`,
+			schema:  `{"type":"record", "name":"test", "namespace": "org.hamba.avro", "fields":[{"name": "a", "type": "float", "default": 1}]}`,
 			wantErr: false,
 		},
 		{
 			name:    "Double",
-			schema:  `{"type":"record", "name":"test", "namespace": "org.apache.avro", "fields":[{"name": "a", "type": "double", "default": 1}]}`,
+			schema:  `{"type":"record", "name":"test", "namespace": "org.hamba.avro", "fields":[{"name": "a", "type": "double", "default": 1}]}`,
 			wantErr: false,
 		},
 		{
 			name:    "Array",
-			schema:  `{"type":"record", "name":"test", "namespace": "org.apache.avro", "fields":[{"name": "a", "type": {"type":"array", "items": "int"}, "default": [1,2]}]}`,
+			schema:  `{"type":"record", "name":"test", "namespace": "org.hamba.avro", "fields":[{"name": "a", "type": {"type":"array", "items": "int"}, "default": [1,2]}]}`,
 			wantErr: false,
 		},
 		{
 			name:    "Array Not Array",
-			schema:  `{"type":"record", "name":"test", "namespace": "org.apache.avro", "fields":[{"name": "a", "type": {"type":"array", "items": "int"}, "default": "test"}]}`,
+			schema:  `{"type":"record", "name":"test", "namespace": "org.hamba.avro", "fields":[{"name": "a", "type": {"type":"array", "items": "int"}, "default": "test"}]}`,
 			wantErr: true,
 		},
 		{
 			name:    "Array Invalid Type",
-			schema:  `{"type":"record", "name":"test", "namespace": "org.apache.avro", "fields":[{"name": "a", "type": {"type":"array", "items": "int"}, "default": ["test"]}]}`,
+			schema:  `{"type":"record", "name":"test", "namespace": "org.hamba.avro", "fields":[{"name": "a", "type": {"type":"array", "items": "int"}, "default": ["test"]}]}`,
 			wantErr: true,
 		},
 		{
 			name:    "Map",
-			schema:  `{"type":"record", "name":"test", "namespace": "org.apache.avro", "fields":[{"name": "a", "type": {"type":"map", "values": "int"}, "default": {"b": 1}}]}`,
+			schema:  `{"type":"record", "name":"test", "namespace": "org.hamba.avro", "fields":[{"name": "a", "type": {"type":"map", "values": "int"}, "default": {"b": 1}}]}`,
 			wantErr: false,
 		},
 		{
 			name:    "Map Not Map",
-			schema:  `{"type":"record", "name":"test", "namespace": "org.apache.avro", "fields":[{"name": "a", "type": {"type":"map", "values": "int"}, "default": "test"}]}`,
+			schema:  `{"type":"record", "name":"test", "namespace": "org.hamba.avro", "fields":[{"name": "a", "type": {"type":"map", "values": "int"}, "default": "test"}]}`,
 			wantErr: true,
 		},
 		{
 			name:    "Map Invalid Type",
-			schema:  `{"type":"record", "name":"test", "namespace": "org.apache.avro", "fields":[{"name": "a", "type": {"type":"map", "values": "int"}, "default": {"b": "test"}}]}`,
+			schema:  `{"type":"record", "name":"test", "namespace": "org.hamba.avro", "fields":[{"name": "a", "type": {"type":"map", "values": "int"}, "default": {"b": "test"}}]}`,
 			wantErr: true,
 		},
 		{
 			name:    "Union",
-			schema:  `{"type":"record", "name":"test", "namespace": "org.apache.avro", "fields":[{"name": "a", "type": ["null", "string"], "default": null}]}`,
+			schema:  `{"type":"record", "name":"test", "namespace": "org.hamba.avro", "fields":[{"name": "a", "type": ["null", "string"], "default": null}]}`,
 			wantErr: false,
 		},
 		{
 			name:    "Union Invalid Type",
-			schema:  `{"type":"record", "name":"test", "namespace": "org.apache.avro", "fields":[{"name": "a", "type": ["null", "string"], "default": "string"}]}`,
+			schema:  `{"type":"record", "name":"test", "namespace": "org.hamba.avro", "fields":[{"name": "a", "type": ["null", "string"], "default": "string"}]}`,
 			wantErr: true,
 		},
 		{
 			name:    "Record",
-			schema:  `{"type":"record", "name":"test", "namespace": "org.apache.avro", "fields":[{"name": "a", "type": {"type":"record", "name": "test2", "fields":[{"name": "b", "type": "int"},{"name": "c", "type": "int", "default": 1}]}, "default": {"b": 1}}]}`,
+			schema:  `{"type":"record", "name":"test", "namespace": "org.hamba.avro", "fields":[{"name": "a", "type": {"type":"record", "name": "test2", "fields":[{"name": "b", "type": "int"},{"name": "c", "type": "int", "default": 1}]}, "default": {"b": 1}}]}`,
 			wantErr: false,
 		},
 		{
 			name:    "Record Not Map",
-			schema:  `{"type":"record", "name":"test", "namespace": "org.apache.avro", "fields":[{"name": "a", "type": {"type":"record", "name": "test2", "fields":[{"name": "b", "type": "int"},{"name": "c", "type": "int", "default": 1}]}, "default": "test"}]}`,
+			schema:  `{"type":"record", "name":"test", "namespace": "org.hamba.avro", "fields":[{"name": "a", "type": {"type":"record", "name": "test2", "fields":[{"name": "b", "type": "int"},{"name": "c", "type": "int", "default": 1}]}, "default": "test"}]}`,
 			wantErr: true,
 		},
 		{
 			name:    "Record Invalid Type",
-			schema:  `{"type":"record", "name":"test", "namespace": "org.apache.avro", "fields":[{"name": "a", "type": {"type":"record", "name": "test2", "fields":[{"name": "b", "type": "int"},{"name": "c", "type": "int", "default": 1}]}, "default": {"b": "test"}}]}`,
+			schema:  `{"type":"record", "name":"test", "namespace": "org.hamba.avro", "fields":[{"name": "a", "type": {"type":"record", "name": "test2", "fields":[{"name": "b", "type": "int"},{"name": "c", "type": "int", "default": 1}]}, "default": {"b": "test"}}]}`,
 			wantErr: true,
 		},
 		{
 			name:    "Record Invalid Field Type",
-			schema:  `{"type":"record", "name":"test", "namespace": "org.apache.avro", "fields":[{"name": "a", "type": {"type":"record", "name": "test2", "fields":[{"name": "b", "type": "int"},{"name": "c", "type": "int", "default": "test"}]}, "default": {"b": 1}}]}`,
+			schema:  `{"type":"record", "name":"test", "namespace": "org.hamba.avro", "fields":[{"name": "a", "type": {"type":"record", "name": "test2", "fields":[{"name": "b", "type": "int"},{"name": "c", "type": "int", "default": "test"}]}, "default": {"b": 1}}]}`,
 			wantErr: true,
 		},
 	}
@@ -356,7 +417,7 @@ func TestRecordSchema_HandlesProps(t *testing.T) {
 {
    "type": "record",
    "name": "valid_name",
-   "namespace": "org.apache.avro",
+   "namespace": "org.hamba.avro",
    "foo": "bar1",
    "fields": [
        {"name": "intField", "type": "int", "foo": "bar2"}
@@ -377,7 +438,7 @@ func TestRecordSchema_WithReference(t *testing.T) {
 {
    "type": "record",
    "name": "valid_name",
-   "namespace": "org.apache.avro",
+   "namespace": "org.hamba.avro",
    "fields": [
        {"name": "intField", "type": "int"},
        {"name": "Ref", "type": "valid_name"}
@@ -402,28 +463,28 @@ func TestEnumSchema(t *testing.T) {
 	}{
 		{
 			name:     "Valid",
-			schema:   `{"type":"enum", "name":"test", "namespace": "org.apache.avro", "symbols":["TEST"]}`,
-			wantName: "org.apache.avro.test",
+			schema:   `{"type":"enum", "name":"test", "namespace": "org.hamba.avro", "symbols":["TEST"]}`,
+			wantName: "org.hamba.avro.test",
 			wantErr:  false,
 		},
 		{
 			name:    "Invalid Name",
-			schema:  `{"type":"enum", "name":"test+", "namespace": "org.apache.avro", "symbols":["TEST"]}`,
+			schema:  `{"type":"enum", "name":"test+", "namespace": "org.hamba.avro", "symbols":["TEST"]}`,
 			wantErr: true,
 		},
 		{
 			name:    "Empty Name",
-			schema:  `{"type":"enum", "name":"", "namespace": "org.apache.avro", "symbols":["TEST"]}`,
+			schema:  `{"type":"enum", "name":"", "namespace": "org.hamba.avro", "symbols":["TEST"]}`,
 			wantErr: true,
 		},
 		{
 			name:    "No Name",
-			schema:  `{"type":"enum", "namespace": "org.apache.avro", "symbols":["TEST"]}`,
+			schema:  `{"type":"enum", "namespace": "org.hamba.avro", "symbols":["TEST"]}`,
 			wantErr: true,
 		},
 		{
 			name:    "Invalid Namespace",
-			schema:  `{"type":"enum", "name":"test", "namespace": "org.apache.avro+", "symbols":["TEST"]}`,
+			schema:  `{"type":"enum", "name":"test", "namespace": "org.hamba.avro+", "symbols":["TEST"]}`,
 			wantErr: true,
 		},
 		{
@@ -433,22 +494,22 @@ func TestEnumSchema(t *testing.T) {
 		},
 		{
 			name:    "No Symbols",
-			schema:  `{"type":"enum", "name":"test", "namespace": "org.apache.avro"}`,
+			schema:  `{"type":"enum", "name":"test", "namespace": "org.hamba.avro"}`,
 			wantErr: true,
 		},
 		{
 			name:    "Empty Symbols",
-			schema:  `{"type":"enum", "name":"test", "namespace": "org.apache.avro", "symbols":[]}`,
+			schema:  `{"type":"enum", "name":"test", "namespace": "org.hamba.avro", "symbols":[]}`,
 			wantErr: true,
 		},
 		{
 			name:    "Invalid Symbol",
-			schema:  `{"type":"enum", "name":"test", "namespace": "org.apache.avro", "symbols":["TEST+"]}`,
+			schema:  `{"type":"enum", "name":"test", "namespace": "org.hamba.avro", "symbols":["TEST+"]}`,
 			wantErr: true,
 		},
 		{
 			name:    "Invalid Symbol Type",
-			schema:  `{"type":"enum", "name":"test", "namespace": "org.apache.avro", "symbols":[1]}`,
+			schema:  `{"type":"enum", "name":"test", "namespace": "org.hamba.avro", "symbols":[1]}`,
 			wantErr: true,
 		},
 	}
@@ -471,7 +532,7 @@ func TestEnumSchema(t *testing.T) {
 }
 
 func TestEnumSchema_HandlesProps(t *testing.T) {
-	schm := `{"type":"enum", "name":"test", "namespace": "org.apache.avro", "symbols":["TEST"], "foo":"bar"}`
+	schm := `{"type":"enum", "name":"test", "namespace": "org.hamba.avro", "symbols":["TEST"], "foo":"bar"}`
 
 	s, err := avro.Parse(schm)
 
@@ -643,29 +704,29 @@ func TestFixedSchema(t *testing.T) {
 	}{
 		{
 			name:            "Valid",
-			schema:          `{"type":"fixed", "name":"test", "namespace": "org.apache.avro", "size": 12}`,
-			wantName:        "org.apache.avro.test",
-			wantFingerprint: [32]byte{0xa8, 0x13, 0xfa, 0xb4, 0xf, 0xd7, 0xe3, 0xc9, 0x3a, 0x98, 0x77, 0x24, 0xaf, 0xa9, 0x36, 0xe6, 0xe9, 0x53, 0xa9, 0x1c, 0x10, 0x70, 0xfe, 0x4e, 0x13, 0x2a, 0x7c, 0x51, 0x6, 0x5f, 0xa4, 0xbc},
+			schema:          `{"type":"fixed", "name":"test", "namespace": "org.hamba.avro", "size": 12}`,
+			wantName:        "org.hamba.avro.test",
+			wantFingerprint: [32]uint8{0x8c, 0x9e, 0xcb, 0x4, 0x83, 0x2f, 0x3b, 0xa7, 0x58, 0x85, 0x9, 0x99, 0x41, 0xe, 0xbf, 0xd4, 0x7, 0xc7, 0x87, 0x4f, 0x8a, 0x12, 0xf4, 0xd0, 0x7f, 0x45, 0xdd, 0xaa, 0x10, 0x6b, 0x2f, 0xb3},
 			wantErr:         false,
 		},
 		{
 			name:    "Invalid Name",
-			schema:  `{"type":"fixed", "name":"test+", "namespace": "org.apache.avro", "size": 12}`,
+			schema:  `{"type":"fixed", "name":"test+", "namespace": "org.hamba.avro", "size": 12}`,
 			wantErr: true,
 		},
 		{
 			name:    "Empty Name",
-			schema:  `{"type":"fixed", "name":"", "namespace": "org.apache.avro", "size": 12}`,
+			schema:  `{"type":"fixed", "name":"", "namespace": "org.hamba.avro", "size": 12}`,
 			wantErr: true,
 		},
 		{
 			name:    "No Name",
-			schema:  `{"type":"fixed", "namespace": "org.apache.avro", "size": 12}`,
+			schema:  `{"type":"fixed", "namespace": "org.hamba.avro", "size": 12}`,
 			wantErr: true,
 		},
 		{
 			name:    "Invalid Namespace",
-			schema:  `{"type":"fixed", "name":"test", "namespace": "org.apache.avro+", "size": 12}`,
+			schema:  `{"type":"fixed", "name":"test", "namespace": "org.hamba.avro+", "size": 12}`,
 			wantErr: true,
 		},
 		{
@@ -675,12 +736,12 @@ func TestFixedSchema(t *testing.T) {
 		},
 		{
 			name:    "No Size",
-			schema:  `{"type":"fixed", "name":"test", "namespace": "org.apache.avro"}`,
+			schema:  `{"type":"fixed", "name":"test", "namespace": "org.hamba.avro"}`,
 			wantErr: true,
 		},
 		{
 			name:    "Invalid Size Type",
-			schema:  `{"type":"fixed", "name":"test", "namespace": "org.apache.avro", "size": "test"}`,
+			schema:  `{"type":"fixed", "name":"test", "namespace": "org.hamba.avro", "size": "test"}`,
 			wantErr: true,
 		},
 	}
@@ -704,7 +765,7 @@ func TestFixedSchema(t *testing.T) {
 }
 
 func TestFixedSchema_HandlesProps(t *testing.T) {
-	schm := `{"type":"fixed", "name":"test", "namespace": "org.apache.avro", "size": 12, "foo":"bar"}`
+	schm := `{"type":"fixed", "name":"test", "namespace": "org.hamba.avro", "size": 12, "foo":"bar"}`
 
 	s, err := avro.Parse(schm)
 
@@ -718,7 +779,7 @@ func TestSchema_Interop(t *testing.T) {
 {
    "type": "record",
    "name": "Interop",
-   "namespace": "org.apache.avro",
+   "namespace": "org.hamba.avro",
    "fields": [
        {
            "name": "intField",
@@ -818,7 +879,7 @@ func TestSchema_Interop(t *testing.T) {
                    },
                    {
                        "name": "child",
-                       "type": {"type": "org.apache.avro.Node"}
+                       "type": {"type": "org.hamba.avro.Node"}
                    },
                    {
                        "name": "children",
diff --git a/testdata/echo.avpr b/testdata/echo.avpr
new file mode 100644
index 0000000..5368c20
--- /dev/null
+++ b/testdata/echo.avpr
@@ -0,0 +1,26 @@
+{
+  "protocol": "Echo",
+  "namespace": "org.hamba.avro",
+  "doc": "Simple echo protocol",
+  "types": [
+    {"name":  "Ping", "type": "record", "fields": [
+      {"name": "timestamp", "type": "long", "default": -1},
+      {"name": "text", "type": "string", "default": ""}
+    ]},
+    {"name":  "Pong", "type": "record", "fields": [
+      {"name": "timestamp", "type": "long", "default": -1},
+      {"name": "ping", "type": "Ping"}
+    ]},
+    {"name":  "PongError", "type": "error", "fields": [
+      {"name": "timestamp", "type": "long", "default": -1},
+      {"name": "reason", "type": "string"}
+    ]}
+  ],
+  "messages": {
+    "ping": {
+      "request": [{"name": "ping", "type": "Ping"}],
+      "response": "Pong",
+      "errors": ["PongError"]
+    }
+  }
+}
\ No newline at end of file
diff --git a/types_test.go b/types_test.go
index 1252bb2..717a1e2 100644
--- a/types_test.go
+++ b/types_test.go
@@ -1,11 +1,5 @@
 package avro_test
 
-import (
-	"fmt"
-
-	"github.com/hamba/avro"
-)
-
 type TestInterface interface {
 	SomeFunc() int
 }
@@ -28,50 +22,6 @@ type TestNestedRecord struct {
 	B TestRecord `avro:"b"`
 }
 
-type TestUnionRecord struct {
-	A *TestUnionType `avro:"a"`
-}
-
-type TestUnionType struct {
-	Val interface{}
-}
-
-func (u *TestUnionType) Value() *interface{} {
-	return &u.Val
-}
-
-func (u *TestUnionType) SetType(typ string) error {
-	switch typ {
-	case string(avro.Null):
-		u.Val = nil
-
-	case "test":
-		u.Val = ""
-
-	case string(avro.Int):
-		u.Val = int(0)
-
-	case string(avro.String):
-		u.Val = nil
-
-	default:
-		return fmt.Errorf("unknown type %s", typ)
-	}
-
-	return nil
-}
-
-func (u *TestUnionType) GetType() (string, error) {
-	switch u.Val.(type) {
-	case nil:
-		return string(avro.Null), nil
-
-	case string:
-		return "test", nil
-
-	case int:
-		return string(avro.Int), nil
-	}
-
-	return "", fmt.Errorf("unknown type %T", u.Val)
+type TestUnion struct {
+	A interface{} `avro:"a"`
 }