func (as *Avro) Decode(buffer []byte, schemaName string) (msg gbus.Message, err error) {
	schemaID := binary.BigEndian.Uint32(buffer[1:5])
	rel, ok := as.registeredObjects[int(schemaID)]
	if !ok {
		err = fmt.Errorf("could not find avroRelation")
		logrus.WithError(err).Error("no avroRelation for obj in registeredObjects")
		return
	}
	var buf bytes.Buffer
	buf.Write(buffer[5:])
	o, err := rel.Deserializer(&buf)
	if err != nil {
		return
	}
	obj, ok := o.(gbus.Message)
	if !ok {
		return nil, fmt.Errorf("could not cast %v to gbus.Message", o)
	}
	return obj, nil
}
Decode decodes a byte array into an object
package serialization
import (
	"bufio"
	"bytes"
	"encoding/binary"
	"fmt"
	"io"
	"io/ioutil"
	"reflect"
	"sync"
	kafka "github.com/dangkaka/go-kafka-avro"
	"github.com/linkedin/goavro"
	"github.com/sirupsen/logrus"
	"github.com/wework/grabbit/gbus"
)
var _ gbus.Serializer = &Avro{}
type avroDeserializer func(r io.Reader) (interface{}, error)
type avroRelation struct {
	SchemaID     int
	Schema       string
	Codec        *goavro.Codec
	ObjType      reflect.Type
	Deserializer avroDeserializer
}
type AvroMessageGenerated interface {
	Schema() string
	SchemaName() string
	Serialize(w io.Writer) error
}
type Avro struct {
	lock                 *sync.Mutex
	registeredSchemas    map[string]*avroRelation
	registeredObjects    map[int]*avroRelation
	schemaRegistryUrls   []string
	schemaRegistryClient *kafka.CachedSchemaRegistryClient
}
func NewMessageEncoding(schemaRegistryUrls ...string) gbus.Serializer {
	return NewAvroSerializer(schemaRegistryUrls...)
}
func NewAvroSerializer(schemaRegistryUrls ...string) *Avro {
	return &Avro{
		schemaRegistryUrls:   schemaRegistryUrls,
		schemaRegistryClient: kafka.NewCachedSchemaRegistryClient(schemaRegistryUrls),
		registeredSchemas:    make(map[string]*avroRelation),
		registeredObjects:    make(map[int]*avroRelation),
		lock:                 &sync.Mutex{},
	}
}
func (as *Avro) Name() string {
	return "avro/binary"
}
func (as *Avro) Encode(obj gbus.Message) (msg []byte, err error) {
	name := obj.SchemaName()
	rel, ok := as.registeredSchemas[name]
	if !ok {
		err := fmt.Errorf("not a registered object :(")
		logrus.WithError(err).WithField("name", name).Error("not a registered type")
		return nil, err
	}
	binarySchemaID := make([]byte, 4)
	binary.BigEndian.PutUint32(binarySchemaID, uint32(rel.SchemaID))
	msg = make([]byte, 0)
	msg = append(msg, byte(0))
	msg = append(msg, binarySchemaID...)
	tobj, ok := obj.(AvroMessageGenerated)
	if !ok {
		err := fmt.Errorf("could not convert obj to AvroMessageGenerated")
		logrus.WithError(err).Error("could not convert object")
		return nil, err
	}
	var buf bytes.Buffer
	bufWriter := bufio.NewWriter(&buf)
	err = tobj.Serialize(bufWriter)
	if err != nil {
		logrus.
			WithError(err).
			WithField("name", name).
			Error("could not convert obj to bindata")
		return nil, err
	}
	err = bufWriter.Flush()
	if err != nil {
		logrus.WithError(err).Error("could not flush to buffer")
		return nil, err
	}
	binaryValue := buf.Bytes()
	msg = append(msg, binaryValue...)
	return msg, nil
}