Merge pull request #670 from Shopify/split-snappy

Use external snappy package

message.go

@@ -5,6 +5,8 @@ import (
 	"compress/gzip"
 	"fmt"
 	"io/ioutil"
+
+	"github.com/eapache/go-xerial-snappy"
 )
 
 // CompressionCodec represents the various compression codecs recognized by Kafka in messages.
@@ -66,7 +68,7 @@ func (m *Message) encode(pe packetEncoder) error {
 			m.compressedCache = buf.Bytes()
 			payload = m.compressedCache
 		case CompressionSnappy:
-			tmp := snappyEncode(m.Value)
+			tmp := snappy.Encode(m.Value)
 			m.compressedCache = tmp
 			payload = m.compressedCache
 		default:
@@ -132,7 +134,7 @@ func (m *Message) decode(pd packetDecoder) (err error) {
 		if m.Value == nil {
 			break
 		}
-		if m.Value, err = snappyDecode(m.Value); err != nil {
+		if m.Value, err = snappy.Decode(m.Value); err != nil {
 			return err
 		}
 		if err := m.decodeSet(); err != nil {

snappy.go

@@ -1,41 +0,0 @@
-package sarama
-
-import (
-	"bytes"
-	"encoding/binary"
-
-	"github.com/golang/snappy"
-)
-
-var snappyMagic = []byte{130, 83, 78, 65, 80, 80, 89, 0}
-
-// SnappyEncode encodes binary data
-func snappyEncode(src []byte) []byte {
-	return snappy.Encode(nil, src)
-}
-
-// SnappyDecode decodes snappy data
-func snappyDecode(src []byte) ([]byte, error) {
-	if bytes.Equal(src[:8], snappyMagic) {
-		var (
-			pos   = uint32(16)
-			max   = uint32(len(src))
-			dst   = make([]byte, 0, len(src))
-			chunk []byte
-			err   error
-		)
-		for pos < max {
-			size := binary.BigEndian.Uint32(src[pos : pos+4])
-			pos += 4
-
-			chunk, err = snappy.Decode(chunk, src[pos:pos+size])
-			if err != nil {
-				return nil, err
-			}
-			pos += size
-			dst = append(dst, chunk...)
-		}
-		return dst, nil
-	}
-	return snappy.Decode(nil, src)
-}