Repository: hamba/avro
Pull request #8 (2019-04-03 07:21:59)

OCF


Patch:
diff --git a/.travis.yml b/.travis.yml
index c2430a2..2cd0b5c 100644
--- a/.travis.yml
+++ b/.travis.yml
@@ -8,6 +8,10 @@ go:
 env:
   - GO111MODULE=on
 
+cache:
+  directories:
+    - $GOPATH/pkg/mod
+
 before_install:
   - go mod download
   - go get golang.org/x/lint/golint
diff --git a/README.md b/README.md
index 432641a..27eadae 100644
--- a/README.md
+++ b/README.md
@@ -116,5 +116,4 @@ Always benchmark with your own workload. The result depends heavily on the data
     * Refactor parsing to be cleaner
     * Better schema validation
     * Aliases
-* Avro container file codecs
 * Avro Textual Form?
diff --git a/go.mod b/go.mod
index cf6fd7c..1604a42 100644
--- a/go.mod
+++ b/go.mod
@@ -1,6 +1,7 @@
 module github.com/hamba/avro
 
 require (
+	github.com/golang/snappy v0.0.1
 	github.com/json-iterator/go v1.1.5
 	github.com/modern-go/concurrent v0.0.0-20180306012644-bacd9c7ef1dd
 	github.com/modern-go/reflect2 v1.0.1
diff --git a/go.sum b/go.sum
index eed0add..46143d1 100644
--- a/go.sum
+++ b/go.sum
@@ -1,5 +1,7 @@
 github.com/davecgh/go-spew v1.1.0 h1:ZDRjVQ15GmhC3fiQ8ni8+OwkZQO4DARzQgrnXU1Liz8=
 github.com/davecgh/go-spew v1.1.0/go.mod h1:J7Y8YcW2NihsgmVo/mv3lAwl/skON4iLHjSsI+c5H38=
+github.com/golang/snappy v0.0.1 h1:Qgr9rKW7uDUkrbSmQeiDsGa8SjGyCOGtuasMWwvp2P4=
+github.com/golang/snappy v0.0.1/go.mod h1:/XxbfmMg8lxefKM7IXC3fBNl/7bRcc72aCRzEWrmP2Q=
 github.com/json-iterator/go v1.1.5 h1:gL2yXlmiIo4+t+y32d4WGwOjKGYcGOuyrg46vadswDE=
 github.com/json-iterator/go v1.1.5/go.mod h1:+SdeFBvtyEkXs7REEP0seUULqWtbJapLOCVDaaPEHmU=
 github.com/modern-go/concurrent v0.0.0-20180306012644-bacd9c7ef1dd h1:TRLaZ9cD/w8PVh93nsPXa1VrQ6jlwL5oN8l14QlcNfg=
diff --git a/ocf/codec.go b/ocf/codec.go
new file mode 100644
index 0000000..b080059
--- /dev/null
+++ b/ocf/codec.go
@@ -0,0 +1,106 @@
+package ocf
+
+import (
+	"bytes"
+	"compress/flate"
+	"encoding/binary"
+	"errors"
+	"fmt"
+	"hash/crc32"
+	"io/ioutil"
+
+	"github.com/golang/snappy"
+)
+
+type CodecName string
+
+const (
+	Null    CodecName = "null"
+	Deflate CodecName = "deflate"
+	Snappy  CodecName = "snappy"
+)
+
+func resolveCodec(name CodecName) (Codec, error) {
+	switch name {
+	case Null, "":
+		return &NullCodec{}, nil
+
+	case Deflate:
+		return &DeflateCodec{}, nil
+
+	case Snappy:
+		return &SnappyCodec{}, nil
+
+	default:
+		return nil, fmt.Errorf("unknown codec %s", name)
+	}
+}
+
+type Codec interface {
+	Decode([]byte) ([]byte, error)
+	Encode([]byte) []byte
+}
+
+type NullCodec struct{}
+
+func (*NullCodec) Decode(b []byte) ([]byte, error) {
+	return b, nil
+}
+
+func (*NullCodec) Encode(b []byte) []byte {
+	return b
+}
+
+type DeflateCodec struct{}
+
+func (*DeflateCodec) Decode(b []byte) ([]byte, error) {
+	r := flate.NewReader(bytes.NewBuffer(b))
+	data, err := ioutil.ReadAll(r)
+	if err != nil {
+		_ = r.Close()
+		return nil, err
+	}
+	_ = r.Close()
+
+	return data, nil
+}
+
+func (*DeflateCodec) Encode(b []byte) []byte {
+	data := bytes.NewBuffer(make([]byte, 0, len(b)))
+
+	w, _ := flate.NewWriter(data, flate.DefaultCompression)
+	_, _ = w.Write(b)
+	_ = w.Close()
+
+	return data.Bytes()
+}
+
+type SnappyCodec struct{}
+
+func (*SnappyCodec) Decode(b []byte) ([]byte, error) {
+	l := len(b)
+	if l < 5 {
+		return nil, errors.New("block does not contain snappy checksum")
+	}
+
+	dst, err := snappy.Decode(nil, b[:l-4])
+	if err != nil {
+		return nil, err
+	}
+
+	crc := binary.BigEndian.Uint32(b[l-4:])
+	if crc32.ChecksumIEEE(dst) != crc {
+		return nil, errors.New("snappy checksum mismatch")
+	}
+
+	return dst, nil
+}
+
+func (*SnappyCodec) Encode(b []byte) []byte {
+	dst := snappy.Encode(nil, b)
+
+	dst = append(dst, 0, 0, 0, 0)
+	binary.BigEndian.PutUint32(dst[len(dst)-4:], crc32.ChecksumIEEE(b))
+
+	return dst
+}
diff --git a/container/container.go b/ocf/ocf.go
similarity index 80%
rename from container/container.go
rename to ocf/ocf.go
index c172bea..4da1586 100644
--- a/container/container.go
+++ b/ocf/ocf.go
@@ -4,7 +4,7 @@ Package container implements encoding and decoding of Avro Object Container File
 See the Avro specification for an understanding of Avro: http://avro.apache.org/docs/current/
 
 */
-package container
+package ocf
 
 import (
 	"bytes"
@@ -49,6 +49,8 @@ type Decoder struct {
 	decoder     *avro.Decoder
 	sync        [16]byte
 
+	codec Codec
+
 	count int64
 }
 
@@ -70,19 +72,19 @@ func NewDecoder(r io.Reader) (*Decoder, error) {
 		return nil, err
 	}
 
-	decReader := bytesx.NewResetReader([]byte{})
+	codec, err := resolveCodec(CodecName(h.Meta[codecKey]))
+	if err != nil {
+		return nil, err
+	}
 
-	// TODO: File Codecs
-	// codec, ok := codecs[string(h.Meta[codecKey])]
-	//if codec, ok := codecs[string(h.Meta[codecKey])]; !ok {
-	//	return nil, fmt.Errorf("file: unknown codec %s", string(h.Meta[codecKey]))
-	//}
+	decReader := bytesx.NewResetReader([]byte{})
 
 	return &Decoder{
 		reader:      reader,
 		resetReader: decReader,
 		decoder:     avro.NewDecoderForSchema(schema, decReader),
 		sync:        h.Sync,
+		codec:       codec,
 	}, nil
 }
 
@@ -93,6 +95,10 @@ func (d *Decoder) HasNext() bool {
 		d.count = count
 	}
 
+	if d.reader.Error != nil {
+		return false
+	}
+
 	return d.count > 0
 }
 
@@ -120,9 +126,17 @@ func (d *Decoder) readBlock() int64 {
 	count := d.reader.ReadLong()
 	size := d.reader.ReadLong()
 
-	data := make([]byte, size)
-	d.reader.Read(data)
-	d.resetReader.Reset(data)
+	if count > 0 {
+		data := make([]byte, size)
+		d.reader.Read(data)
+
+		data, err := d.codec.Decode(data)
+		if err != nil {
+			d.reader.Error = err
+		}
+
+		d.resetReader.Reset(data)
+	}
 
 	var sync [16]byte
 	d.reader.Read(sync[:])
@@ -133,13 +147,25 @@ func (d *Decoder) readBlock() int64 {
 	return count
 }
 
+type encoderConfig struct {
+	BlockLength int
+	CodecName   CodecName
+}
+
 // EncoderFunc represents an configuration function for Encoder
-type EncoderFunc func(e *Encoder)
+type EncoderFunc func(cfg *encoderConfig)
 
 // WithBlockLength sets the block length on the encoder.
 func WithBlockLength(length int) EncoderFunc {
-	return func(e *Encoder) {
-		e.blockLength = length
+	return func(cfg *encoderConfig) {
+		cfg.BlockLength = length
+	}
+}
+
+// WithCodec sets the compression codec on the encoder.
+func WithCodec(codec CodecName) EncoderFunc {
+	return func(cfg *encoderConfig) {
+		cfg.CodecName = codec
 	}
 }
 
@@ -150,6 +176,8 @@ type Encoder struct {
 	encoder *avro.Encoder
 	sync    [16]byte
 
+	codec Codec
+
 	blockLength int
 	count       int
 }
@@ -161,17 +189,31 @@ func NewEncoder(s string, w io.Writer, opts ...EncoderFunc) (*Encoder, error) {
 		return nil, err
 	}
 
+	cfg := encoderConfig{
+		BlockLength: 100,
+		CodecName:   Null,
+	}
+	for _, opt := range opts {
+		opt(&cfg)
+	}
+
 	writer := avro.NewWriter(w, 512)
 
 	header := Header{
 		Magic: magicBytes,
 		Meta: map[string][]byte{
 			schemaKey: []byte(schema.String()),
+			codecKey:  []byte(cfg.CodecName),
 		},
 	}
 	_, _ = rand.Read(header.Sync[:])
 	writer.WriteVal(HeaderSchema, header)
 
+	codec, err := resolveCodec(cfg.CodecName)
+	if err != nil {
+		return nil, err
+	}
+
 	buf := &bytes.Buffer{}
 
 	e := &Encoder{
@@ -179,11 +221,8 @@ func NewEncoder(s string, w io.Writer, opts ...EncoderFunc) (*Encoder, error) {
 		buf:         buf,
 		encoder:     avro.NewEncoderForSchema(schema, buf),
 		sync:        header.Sync,
-		blockLength: 100,
-	}
-
-	for _, opt := range opts {
-		opt(e)
+		codec:       codec,
+		blockLength: cfg.BlockLength,
 	}
 
 	return e, nil
@@ -220,9 +259,14 @@ func (e *Encoder) Close() error {
 
 func (e *Encoder) writerBlock() error {
 	e.writer.WriteLong(int64(e.count))
-	e.writer.WriteLong(int64(e.buf.Len()))
-	e.writer.Write(e.buf.Bytes())
+
+	b := e.codec.Encode(e.buf.Bytes())
+
+	e.writer.WriteLong(int64(len(b)))
+	e.writer.Write(b)
+
 	e.writer.Write(e.sync[:])
+
 	e.count = 0
 	e.buf.Reset()
 	return e.writer.Flush()


Test patch:
diff --git a/container/container_test.go b/container/container_test.go
deleted file mode 100644
index ecd387b..0000000
--- a/container/container_test.go
+++ /dev/null
@@ -1,273 +0,0 @@
-package container_test
-
-import (
-	"bytes"
-	"errors"
-	"os"
-	"testing"
-
-	"github.com/hamba/avro/container"
-	"github.com/stretchr/testify/assert"
-)
-
-var schema = `{
-	"type":"record",
-	"name":"FullRecord",
-	"namespace":"org.hamba.avro",
-	"fields":[
-		{"name":"strings","type":{"type":"array","items":"string"}},
-		{"name":"longs","type":{"type":"array","items":"long"}},
-		{"name":"enum","type":{"type":"enum","name":"foo","symbols":["A","B","C","D"]}},
-		{"name":"map","type":{"type":"map","values":"int"}},
-		{"name":"nullable","type":["null","string"]},
-		{"name":"fixed","type":{"type":"fixed","name":"md5","size":16}},
-		{"name":"record","type":{
-			"type":"record",
-			"name":"TestRecord",
-			"fields":[
-				{"name":"long","type":"long"},
-				{"name":"string","type":"string"},
-				{"name":"int","type":"int"},
-				{"name":"float","type":"float"},
-				{"name":"double","type":"double"},
-				{"name":"bool","type":"boolean"}
-			]
-		}}
-	]
-}`
-
-type FullRecord struct {
-	Strings  []string       `avro:"strings"`
-	Longs    []int64        `avro:"longs"`
-	Enum     string         `avro:"enum"`
-	Map      map[string]int `avro:"map"`
-	Nullable *string        `avro:"nullable"`
-	Fixed    [16]byte       `avro:"fixed"`
-	Record   *TestRecord    `avro:"record"`
-}
-
-type TestRecord struct {
-	Long   int64   `avro:"long"`
-	String string  `avro:"string"`
-	Int    int32   `avro:"int"`
-	Float  float32 `avro:"float"`
-	Double float64 `avro:"double"`
-	Bool   bool    `avro:"bool"`
-}
-
-func TestNewDecoder_InvalidHeader(t *testing.T) {
-	data := []byte{'O', 'b', 'j'}
-
-	_, err := container.NewDecoder(bytes.NewReader(data))
-
-	assert.Error(t, err)
-}
-
-func TestNewDecoder_InvalidMagic(t *testing.T) {
-	data := []byte{'f', 'o', 'o', 0x01, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00}
-
-	_, err := container.NewDecoder(bytes.NewReader(data))
-
-	assert.Error(t, err)
-}
-
-func TestNewDecoder_InvalidSchema(t *testing.T) {
-	data := []byte{'O', 'b', 'j', 0x01, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00}
-
-	_, err := container.NewDecoder(bytes.NewReader(data))
-
-	assert.Error(t, err)
-}
-
-func TestDecoder(t *testing.T) {
-	unionStr := "union value"
-	want := FullRecord{
-		Strings: []string{"string1", "string2", "string3", "string4", "string5"},
-		Longs:   []int64{1, 2, 3, 4, 5},
-		Enum:    "C",
-		Map: map[string]int{
-			"key1": 1,
-			"key2": 2,
-			"key3": 3,
-			"key4": 4,
-			"key5": 5,
-		},
-		Nullable: &unionStr,
-		Fixed:    [16]byte{0x01, 0x02, 0x03, 0x04, 0x01, 0x02, 0x03, 0x04, 0x01, 0x02, 0x03, 0x04, 0x01, 0x02, 0x03, 0x04},
-		Record: &TestRecord{
-			Long:   1925639126735,
-			String: "I am a test record",
-			Int:    666,
-			Float:  7171.17,
-			Double: 916734926348163.01973408746523,
-			Bool:   true,
-		},
-	}
-
-	f, err := os.Open("../testdata/full.avro")
-	if err != nil {
-		t.Error(err)
-		return
-	}
-	defer f.Close()
-
-	dec, err := container.NewDecoder(f)
-	if err != nil {
-		t.Error(err)
-		return
-	}
-
-	var count int
-	for dec.HasNext() {
-		count++
-		var got FullRecord
-		err = dec.Decode(&got)
-
-		assert.NoError(t, err)
-		assert.Equal(t, want, got)
-	}
-
-	assert.NoError(t, dec.Error())
-	assert.Equal(t, 1, count)
-}
-
-func TestDecoder_DecodeAvroError(t *testing.T) {
-	data := []byte{'O', 'b', 'j', 0x01, 0x01, 0x26, 0x16, 'a', 'v', 'r', 'o', '.', 's', 'c', 'h', 'e', 'm', 'a',
-		0x0c, '"', 'l', 'o', 'n', 'g', '"', 0x00, 0xfb, 0x2b, 0x0f, 0x1a, 0xdd, 0xfd, 0x90, 0x7d, 0x87, 0x12,
-		0x15, 0x29, 0xd7, 0x1d, 0x1c, 0xdd, 0x02, 0x16, 0xe2, 0xa2, 0xf3, 0xad, 0xad, 0xad, 0xe2, 0xa2, 0xf3,
-		0xad, 0xad, 0xfb, 0x2b, 0x0f, 0x1a, 0xdd, 0xfd, 0x90, 0x7d, 0x87, 0x12, 0x15, 0x29, 0xd7, 0x1d, 0x1c, 0xdd,
-	}
-
-	dec, _ := container.NewDecoder(bytes.NewReader(data))
-	_ = dec.HasNext()
-
-	var l int64
-	err := dec.Decode(&l)
-
-	assert.Error(t, err)
-}
-
-func TestDecoder_DecodeMustCallHasNext(t *testing.T) {
-	data := []byte{'O', 'b', 'j', 0x01, 0x01, 0x26, 0x16, 'a', 'v', 'r', 'o', '.', 's', 'c', 'h', 'e', 'm', 'a',
-		0x0c, '"', 'l', 'o', 'n', 'g', '"', 0x00, 0xfb, 0x2b, 0x0f, 0x1a, 0xdd, 0xfd, 0x90, 0x7d, 0x87, 0x12,
-		0x15, 0x29, 0xd7, 0x1d, 0x1c, 0xdd, 0x02, 0x02, 0x02, 0xfb, 0x2b, 0x0f, 0x1a, 0xdd, 0xfd, 0x90, 0x7d,
-		0x87, 0x12, 0x15, 0x29, 0xd7, 0x1d, 0x1c, 0xdd,
-	}
-
-	dec, _ := container.NewDecoder(bytes.NewReader(data))
-
-	var l int64
-	err := dec.Decode(&l)
-
-	assert.Error(t, err)
-}
-
-func TestDecoder_InvalidBlock(t *testing.T) {
-	data := []byte{'O', 'b', 'j', 0x01, 0x01, 0x26, 0x16, 'a', 'v', 'r', 'o', '.', 's', 'c', 'h', 'e', 'm', 'a',
-		0x0c, '"', 'l', 'o', 'n', 'g', '"', 0x00, 0xfa, 0x2b, 0x0f, 0x1a, 0xdd, 0xfd, 0x90, 0x7d, 0x87, 0x12,
-		0x15, 0x29, 0xd7, 0x1d, 0x1c, 0xdd, 0x02, 0x02, 0x02, 0xfb, 0x2b, 0x0f, 0x1a, 0xdd, 0xfd, 0x90, 0x7d,
-		0x87, 0x12, 0x15, 0x29, 0xd7, 0x1d, 0x1c, 0xdd,
-	}
-
-	dec, _ := container.NewDecoder(bytes.NewReader(data))
-
-	dec.HasNext()
-
-	assert.Error(t, dec.Error())
-}
-
-func TestNewEncoder_InvalidSchema(t *testing.T) {
-	buf := &bytes.Buffer{}
-
-	_, err := container.NewEncoder(``, buf)
-
-	assert.Error(t, err)
-}
-
-func TestEncoder(t *testing.T) {
-	unionStr := "union value"
-	record := FullRecord{
-		Strings: []string{"string1", "string2", "string3", "string4", "string5"},
-		Longs:   []int64{1, 2, 3, 4, 5},
-		Enum:    "C",
-		Map: map[string]int{
-			"key1": 1,
-			"key2": 2,
-			"key3": 3,
-			"key4": 4,
-			"key5": 5,
-		},
-		Nullable: &unionStr,
-		Fixed:    [16]byte{0x01, 0x02, 0x03, 0x04, 0x01, 0x02, 0x03, 0x04, 0x01, 0x02, 0x03, 0x04, 0x01, 0x02, 0x03, 0x04},
-		Record: &TestRecord{
-			Long:   1925639126735,
-			String: "I am a test record",
-			Int:    666,
-			Float:  7171.17,
-			Double: 916734926348163.01973408746523,
-			Bool:   true,
-		},
-	}
-
-	buf := &bytes.Buffer{}
-	enc, err := container.NewEncoder(schema, buf)
-	if err != nil {
-		t.Error(err)
-		return
-	}
-
-	err = enc.Encode(record)
-
-	if err := enc.Close(); err != nil {
-		t.Error(err)
-		return
-	}
-
-	assert.NoError(t, err)
-}
-
-func TestEncoder_EncodeError(t *testing.T) {
-	buf := &bytes.Buffer{}
-	enc, _ := container.NewEncoder(`"long"`, buf)
-
-	err := enc.Encode("test")
-
-	assert.Error(t, err)
-}
-
-func TestEncoder_EncodeWritesBlocks(t *testing.T) {
-	buf := &bytes.Buffer{}
-	enc, _ := container.NewEncoder(`"long"`, buf, container.WithBlockLength(1))
-	defer enc.Close()
-
-	err := enc.Encode(int64(1))
-
-	assert.NoError(t, err)
-	assert.Equal(t, 61, buf.Len())
-}
-
-func TestEncoder_EncodeHandlesWriteBlockError(t *testing.T) {
-	w := &errorWriter{}
-	enc, _ := container.NewEncoder(`"long"`, w, container.WithBlockLength(1))
-	defer enc.Close()
-
-	err := enc.Encode(int64(1))
-
-	assert.Error(t, err)
-}
-
-func TestEncoder_CloseHandlesWriteBlockError(t *testing.T) {
-	w := &errorWriter{}
-	enc, _ := container.NewEncoder(`"long"`, w)
-	_ = enc.Encode(int64(1))
-
-	err := enc.Close()
-
-	assert.Error(t, err)
-}
-
-type errorWriter struct{}
-
-func (*errorWriter) Write(p []byte) (n int, err error) {
-	return 0, errors.New("test")
-}
diff --git a/container/example_test.go b/ocf/example_test.go
similarity index 88%
rename from container/example_test.go
rename to ocf/example_test.go
index bf6c835..e606887 100644
--- a/container/example_test.go
+++ b/ocf/example_test.go
@@ -1,10 +1,10 @@
-package container_test
+package ocf_test
 
 import (
 	"log"
 	"os"
 
-	"github.com/hamba/avro/container"
+	"github.com/hamba/avro/ocf"
 )
 
 func ExampleNewDecoder() {
@@ -19,7 +19,7 @@ func ExampleNewDecoder() {
 	}
 	defer f.Close()
 
-	dec, err := container.NewDecoder(f)
+	dec, err := ocf.NewDecoder(f)
 	if err != nil {
 		log.Fatal(err)
 	}
@@ -61,7 +61,7 @@ func ExampleNewEncoder() {
 	}
 	defer f.Close()
 
-	enc, err := container.NewEncoder(schema, f)
+	enc, err := ocf.NewEncoder(schema, f)
 	if err != nil {
 		log.Fatal(err)
 	}
diff --git a/ocf/ocf_test.go b/ocf/ocf_test.go
new file mode 100644
index 0000000..26fa728
--- /dev/null
+++ b/ocf/ocf_test.go
@@ -0,0 +1,545 @@
+package ocf_test
+
+import (
+	"bytes"
+	"errors"
+	"os"
+	"testing"
+
+	"github.com/hamba/avro/ocf"
+	"github.com/stretchr/testify/assert"
+)
+
+var schema = `{
+	"type":"record",
+	"name":"FullRecord",
+	"namespace":"org.hamba.avro",
+	"fields":[
+		{"name":"strings","type":{"type":"array","items":"string"}},
+		{"name":"longs","type":{"type":"array","items":"long"}},
+		{"name":"enum","type":{"type":"enum","name":"foo","symbols":["A","B","C","D"]}},
+		{"name":"map","type":{"type":"map","values":"int"}},
+		{"name":"nullable","type":["null","string"]},
+		{"name":"fixed","type":{"type":"fixed","name":"md5","size":16}},
+		{"name":"record","type":{
+			"type":"record",
+			"name":"TestRecord",
+			"fields":[
+				{"name":"long","type":"long"},
+				{"name":"string","type":"string"},
+				{"name":"int","type":"int"},
+				{"name":"float","type":"float"},
+				{"name":"double","type":"double"},
+				{"name":"bool","type":"boolean"}
+			]
+		}}
+	]
+}`
+
+type FullRecord struct {
+	Strings  []string       `avro:"strings"`
+	Longs    []int64        `avro:"longs"`
+	Enum     string         `avro:"enum"`
+	Map      map[string]int `avro:"map"`
+	Nullable *string        `avro:"nullable"`
+	Fixed    [16]byte       `avro:"fixed"`
+	Record   *TestRecord    `avro:"record"`
+}
+
+type TestRecord struct {
+	Long   int64   `avro:"long"`
+	String string  `avro:"string"`
+	Int    int32   `avro:"int"`
+	Float  float32 `avro:"float"`
+	Double float64 `avro:"double"`
+	Bool   bool    `avro:"bool"`
+}
+
+func TestNewDecoder_InvalidHeader(t *testing.T) {
+	data := []byte{'O', 'b', 'j'}
+
+	_, err := ocf.NewDecoder(bytes.NewReader(data))
+
+	assert.Error(t, err)
+}
+
+func TestNewDecoder_InvalidMagic(t *testing.T) {
+	data := []byte{'f', 'o', 'o', 0x01, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00}
+
+	_, err := ocf.NewDecoder(bytes.NewReader(data))
+
+	assert.Error(t, err)
+}
+
+func TestNewDecoder_InvalidSchema(t *testing.T) {
+	data := []byte{'O', 'b', 'j', 0x01, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00}
+
+	_, err := ocf.NewDecoder(bytes.NewReader(data))
+
+	assert.Error(t, err)
+}
+
+func TestNewDecoder_InvalidCodec(t *testing.T) {
+	data := []byte{'O', 'b', 'j', 0x1, 0x3, 0x4c, 0x16, 'a', 'v', 'r', 'o', '.', 's', 'c', 'h', 'e', 'm', 'a', 0xc, 0x22, 'l', 'o', 'n', 'g',
+		0x22, 0x14, 'a', 'v', 'r', 'o', 0x2e, 'c', 'o', 'd', 'e', 'c', 0xe, 'd', 'e', 'a', 'l', 'a', 't', 'e', 0x0,
+		0x72, 0xce, 0x78, 0x7, 0x35, 0x81, 0xb0, 0x80, 0x77, 0x59, 0xa9, 0x83, 0xaf, 0x90, 0x3e, 0xaf,
+	}
+
+	_, err := ocf.NewDecoder(bytes.NewReader(data))
+
+	assert.Error(t, err)
+}
+
+func TestDecoder(t *testing.T) {
+	unionStr := "union value"
+	want := FullRecord{
+		Strings: []string{"string1", "string2", "string3", "string4", "string5"},
+		Longs:   []int64{1, 2, 3, 4, 5},
+		Enum:    "C",
+		Map: map[string]int{
+			"key1": 1,
+			"key2": 2,
+			"key3": 3,
+			"key4": 4,
+			"key5": 5,
+		},
+		Nullable: &unionStr,
+		Fixed:    [16]byte{0x01, 0x02, 0x03, 0x04, 0x01, 0x02, 0x03, 0x04, 0x01, 0x02, 0x03, 0x04, 0x01, 0x02, 0x03, 0x04},
+		Record: &TestRecord{
+			Long:   1925639126735,
+			String: "I am a test record",
+			Int:    666,
+			Float:  7171.17,
+			Double: 916734926348163.01973408746523,
+			Bool:   true,
+		},
+	}
+
+	f, err := os.Open("../testdata/full.avro")
+	if err != nil {
+		t.Error(err)
+		return
+	}
+	defer f.Close()
+
+	dec, err := ocf.NewDecoder(f)
+	if err != nil {
+		t.Error(err)
+		return
+	}
+
+	var count int
+	for dec.HasNext() {
+		count++
+		var got FullRecord
+		err = dec.Decode(&got)
+
+		assert.NoError(t, err)
+		assert.Equal(t, want, got)
+	}
+
+	assert.NoError(t, dec.Error())
+	assert.Equal(t, 1, count)
+}
+
+func TestDecoderDeflate(t *testing.T) {
+	unionStr := "union value"
+	want := FullRecord{
+		Strings: []string{"string1", "string2", "string3", "string4", "string5"},
+		Longs:   []int64{1, 2, 3, 4, 5},
+		Enum:    "C",
+		Map: map[string]int{
+			"key1": 1,
+			"key2": 2,
+			"key3": 3,
+			"key4": 4,
+			"key5": 5,
+		},
+		Nullable: &unionStr,
+		Fixed:    [16]byte{0x01, 0x02, 0x03, 0x04, 0x01, 0x02, 0x03, 0x04, 0x01, 0x02, 0x03, 0x04, 0x01, 0x02, 0x03, 0x04},
+		Record: &TestRecord{
+			Long:   1925639126735,
+			String: "I am a test record",
+			Int:    666,
+			Float:  7171.17,
+			Double: 916734926348163.01973408746523,
+			Bool:   true,
+		},
+	}
+
+	f, err := os.Open("../testdata/full-deflate.avro")
+	if err != nil {
+		t.Error(err)
+		return
+	}
+	defer f.Close()
+
+	dec, err := ocf.NewDecoder(f)
+	if err != nil {
+		t.Error(err)
+		return
+	}
+
+	var count int
+	for dec.HasNext() {
+		count++
+		var got FullRecord
+		err = dec.Decode(&got)
+
+		assert.NoError(t, err)
+		assert.Equal(t, want, got)
+	}
+
+	assert.NoError(t, dec.Error())
+	assert.Equal(t, 1, count)
+}
+
+func TestDecoderDeflate_InvalidData(t *testing.T) {
+	f, err := os.Open("../testdata/deflate-invalid-data.avro")
+	if err != nil {
+		t.Error(err)
+		return
+	}
+	defer f.Close()
+
+	dec, err := ocf.NewDecoder(f)
+	if err != nil {
+		t.Error(err)
+		return
+	}
+
+	dec.HasNext()
+
+	assert.Error(t, dec.Error())
+}
+
+func TestDecoderSnappy(t *testing.T) {
+	unionStr := "union value"
+	want := FullRecord{
+		Strings: []string{"string1", "string2", "string3", "string4", "string5"},
+		Longs:   []int64{1, 2, 3, 4, 5},
+		Enum:    "C",
+		Map: map[string]int{
+			"key1": 1,
+			"key2": 2,
+			"key3": 3,
+			"key4": 4,
+			"key5": 5,
+		},
+		Nullable: &unionStr,
+		Fixed:    [16]byte{0x01, 0x02, 0x03, 0x04, 0x01, 0x02, 0x03, 0x04, 0x01, 0x02, 0x03, 0x04, 0x01, 0x02, 0x03, 0x04},
+		Record: &TestRecord{
+			Long:   1925639126735,
+			String: "I am a test record",
+			Int:    666,
+			Float:  7171.17,
+			Double: 916734926348163.01973408746523,
+			Bool:   true,
+		},
+	}
+
+	f, err := os.Open("../testdata/full-snappy.avro")
+	if err != nil {
+		t.Error(err)
+		return
+	}
+	defer f.Close()
+
+	dec, err := ocf.NewDecoder(f)
+	if err != nil {
+		t.Error(err)
+		return
+	}
+
+	var count int
+	for dec.HasNext() {
+		count++
+		var got FullRecord
+		err = dec.Decode(&got)
+
+		assert.NoError(t, err)
+		assert.Equal(t, want, got)
+	}
+
+	assert.NoError(t, dec.Error())
+	assert.Equal(t, 1, count)
+}
+
+func TestDecoderSnappy_InvalidData(t *testing.T) {
+	f, err := os.Open("../testdata/snappy-invalid-data.avro")
+	if err != nil {
+		t.Error(err)
+		return
+	}
+	defer f.Close()
+
+	dec, err := ocf.NewDecoder(f)
+	if err != nil {
+		t.Error(err)
+		return
+	}
+
+	dec.HasNext()
+
+	assert.Error(t, dec.Error())
+}
+
+func TestDecoderSnappy_ShortCRC(t *testing.T) {
+	f, err := os.Open("../testdata/snappy-short-crc.avro")
+	if err != nil {
+		t.Error(err)
+		return
+	}
+	defer f.Close()
+
+	dec, err := ocf.NewDecoder(f)
+	if err != nil {
+		t.Error(err)
+		return
+	}
+
+	dec.HasNext()
+
+	assert.Error(t, dec.Error())
+}
+
+func TestDecoderSnappy_InvalidCRC(t *testing.T) {
+	f, err := os.Open("../testdata/snappy-invalid-crc.avro")
+	if err != nil {
+		t.Error(err)
+		return
+	}
+	defer f.Close()
+
+	dec, err := ocf.NewDecoder(f)
+	if err != nil {
+		t.Error(err)
+		return
+	}
+
+	dec.HasNext()
+
+	assert.Error(t, dec.Error())
+}
+
+func TestDecoder_DecodeAvroError(t *testing.T) {
+	data := []byte{'O', 'b', 'j', 0x01, 0x01, 0x26, 0x16, 'a', 'v', 'r', 'o', '.', 's', 'c', 'h', 'e', 'm', 'a',
+		0x0c, '"', 'l', 'o', 'n', 'g', '"', 0x00, 0xfb, 0x2b, 0x0f, 0x1a, 0xdd, 0xfd, 0x90, 0x7d, 0x87, 0x12,
+		0x15, 0x29, 0xd7, 0x1d, 0x1c, 0xdd, 0x02, 0x16, 0xe2, 0xa2, 0xf3, 0xad, 0xad, 0xad, 0xe2, 0xa2, 0xf3,
+		0xad, 0xad, 0xfb, 0x2b, 0x0f, 0x1a, 0xdd, 0xfd, 0x90, 0x7d, 0x87, 0x12, 0x15, 0x29, 0xd7, 0x1d, 0x1c, 0xdd,
+	}
+
+	dec, _ := ocf.NewDecoder(bytes.NewReader(data))
+	_ = dec.HasNext()
+
+	var l int64
+	err := dec.Decode(&l)
+
+	assert.Error(t, err)
+}
+
+func TestDecoder_DecodeMustCallHasNext(t *testing.T) {
+	data := []byte{'O', 'b', 'j', 0x01, 0x01, 0x26, 0x16, 'a', 'v', 'r', 'o', '.', 's', 'c', 'h', 'e', 'm', 'a',
+		0x0c, '"', 'l', 'o', 'n', 'g', '"', 0x00, 0xfb, 0x2b, 0x0f, 0x1a, 0xdd, 0xfd, 0x90, 0x7d, 0x87, 0x12,
+		0x15, 0x29, 0xd7, 0x1d, 0x1c, 0xdd, 0x02, 0x02, 0x02, 0xfb, 0x2b, 0x0f, 0x1a, 0xdd, 0xfd, 0x90, 0x7d,
+		0x87, 0x12, 0x15, 0x29, 0xd7, 0x1d, 0x1c, 0xdd,
+	}
+
+	dec, _ := ocf.NewDecoder(bytes.NewReader(data))
+
+	var l int64
+	err := dec.Decode(&l)
+
+	assert.Error(t, err)
+}
+
+func TestDecoder_InvalidBlock(t *testing.T) {
+	data := []byte{'O', 'b', 'j', 0x01, 0x01, 0x26, 0x16, 'a', 'v', 'r', 'o', '.', 's', 'c', 'h', 'e', 'm', 'a',
+		0x0c, '"', 'l', 'o', 'n', 'g', '"', 0x00, 0xfa, 0x2b, 0x0f, 0x1a, 0xdd, 0xfd, 0x90, 0x7d, 0x87, 0x12,
+		0x15, 0x29, 0xd7, 0x1d, 0x1c, 0xdd, 0x02, 0x02, 0x02, 0xfb, 0x2b, 0x0f, 0x1a, 0xdd, 0xfd, 0x90, 0x7d,
+		0x87, 0x12, 0x15, 0x29, 0xd7, 0x1d, 0x1c, 0xdd,
+	}
+
+	dec, _ := ocf.NewDecoder(bytes.NewReader(data))
+
+	got := dec.HasNext()
+
+	assert.False(t, got)
+	assert.Error(t, dec.Error())
+}
+
+func TestNewEncoder_InvalidSchema(t *testing.T) {
+	buf := &bytes.Buffer{}
+
+	_, err := ocf.NewEncoder(``, buf)
+
+	assert.Error(t, err)
+}
+
+func TestNewEncoder_InvalidCodec(t *testing.T) {
+	buf := &bytes.Buffer{}
+
+	_, err := ocf.NewEncoder(`"long"`, buf, ocf.WithCodec(ocf.CodecName("test")))
+
+	assert.Error(t, err)
+}
+
+func TestEncoder(t *testing.T) {
+	unionStr := "union value"
+	record := FullRecord{
+		Strings: []string{"string1", "string2", "string3", "string4", "string5"},
+		Longs:   []int64{1, 2, 3, 4, 5},
+		Enum:    "C",
+		Map: map[string]int{
+			"key1": 1,
+			"key2": 2,
+			"key3": 3,
+			"key4": 4,
+			"key5": 5,
+		},
+		Nullable: &unionStr,
+		Fixed:    [16]byte{0x01, 0x02, 0x03, 0x04, 0x01, 0x02, 0x03, 0x04, 0x01, 0x02, 0x03, 0x04, 0x01, 0x02, 0x03, 0x04},
+		Record: &TestRecord{
+			Long:   1925639126735,
+			String: "I am a test record",
+			Int:    666,
+			Float:  7171.17,
+			Double: 916734926348163.01973408746523,
+			Bool:   true,
+		},
+	}
+
+	buf := &bytes.Buffer{}
+	enc, err := ocf.NewEncoder(schema, buf)
+	if err != nil {
+		t.Error(err)
+		return
+	}
+
+	err = enc.Encode(record)
+	assert.NoError(t, err)
+
+	err = enc.Close()
+	assert.NoError(t, err)
+}
+
+func TestEncoder_EncodeCompressesDeflate(t *testing.T) {
+	unionStr := "union value"
+	record := FullRecord{
+		Strings: []string{"string1", "string2", "string3", "string4", "string5"},
+		Longs:   []int64{1, 2, 3, 4, 5},
+		Enum:    "C",
+		Map: map[string]int{
+			"key1": 1,
+			"key2": 2,
+			"key3": 3,
+			"key4": 4,
+			"key5": 5,
+		},
+		Nullable: &unionStr,
+		Fixed:    [16]byte{0x01, 0x02, 0x03, 0x04, 0x01, 0x02, 0x03, 0x04, 0x01, 0x02, 0x03, 0x04, 0x01, 0x02, 0x03, 0x04},
+		Record: &TestRecord{
+			Long:   1925639126735,
+			String: "I am a test record",
+			Int:    666,
+			Float:  7171.17,
+			Double: 916734926348163.01973408746523,
+			Bool:   true,
+		},
+	}
+
+	buf := &bytes.Buffer{}
+	enc, _ := ocf.NewEncoder(schema, buf, ocf.WithCodec(ocf.Deflate))
+	defer enc.Close()
+
+	err := enc.Encode(record)
+	assert.NoError(t, err)
+
+	err = enc.Close()
+	assert.NoError(t, err)
+
+	assert.Equal(t, 926, buf.Len())
+}
+
+func TestEncoder_EncodeCompressesSnappy(t *testing.T) {
+	unionStr := "union value"
+	record := FullRecord{
+		Strings: []string{"string1", "string2", "string3", "string4", "string5"},
+		Longs:   []int64{1, 2, 3, 4, 5},
+		Enum:    "C",
+		Map: map[string]int{
+			"key1": 1,
+			"key2": 2,
+			"key3": 3,
+			"key4": 4,
+			"key5": 5,
+		},
+		Nullable: &unionStr,
+		Fixed:    [16]byte{0x01, 0x02, 0x03, 0x04, 0x01, 0x02, 0x03, 0x04, 0x01, 0x02, 0x03, 0x04, 0x01, 0x02, 0x03, 0x04},
+		Record: &TestRecord{
+			Long:   1925639126735,
+			String: "I am a test record",
+			Int:    666,
+			Float:  7171.17,
+			Double: 916734926348163.01973408746523,
+			Bool:   true,
+		},
+	}
+
+	buf := &bytes.Buffer{}
+	enc, _ := ocf.NewEncoder(schema, buf, ocf.WithBlockLength(1), ocf.WithCodec(ocf.Snappy))
+
+	err := enc.Encode(record)
+	assert.NoError(t, err)
+
+	err = enc.Close()
+	assert.NoError(t, err)
+
+	assert.Equal(t, 938, buf.Len())
+}
+
+func TestEncoder_EncodeError(t *testing.T) {
+	buf := &bytes.Buffer{}
+	enc, _ := ocf.NewEncoder(`"long"`, buf)
+
+	err := enc.Encode("test")
+
+	assert.Error(t, err)
+}
+
+func TestEncoder_EncodeWritesBlocks(t *testing.T) {
+	buf := &bytes.Buffer{}
+	enc, _ := ocf.NewEncoder(`"long"`, buf, ocf.WithBlockLength(1))
+	defer enc.Close()
+
+	err := enc.Encode(int64(1))
+
+	assert.NoError(t, err)
+	assert.Equal(t, 77, buf.Len())
+}
+
+func TestEncoder_EncodeHandlesWriteBlockError(t *testing.T) {
+	w := &errorWriter{}
+	enc, _ := ocf.NewEncoder(`"long"`, w, ocf.WithBlockLength(1))
+	defer enc.Close()
+
+	err := enc.Encode(int64(1))
+
+	assert.Error(t, err)
+}
+
+func TestEncoder_CloseHandlesWriteBlockError(t *testing.T) {
+	w := &errorWriter{}
+	enc, _ := ocf.NewEncoder(`"long"`, w)
+	_ = enc.Encode(int64(1))
+
+	err := enc.Close()
+
+	assert.Error(t, err)
+}
+
+type errorWriter struct{}
+
+func (*errorWriter) Write(p []byte) (n int, err error) {
+	return 0, errors.New("test")
+}