Added support for snappy stream decoding

message.go

@@ -2,7 +2,6 @@ package sarama
 
 import (
 	"bytes"
-	"code.google.com/p/snappy-go/snappy"
 	"compress/gzip"
 	"io/ioutil"
 )
@@ -61,7 +60,7 @@ func (m *Message) encode(pe packetEncoder) error {
 			m.compressedCache = buf.Bytes()
 			payload = m.compressedCache
 		case CompressionSnappy:
-			tmp, err := snappy.Encode(nil, m.Value)
+			tmp, err := SnappyEncode(m.Value)
 			if err != nil {
 				return err
 			}
@@ -129,7 +128,7 @@ func (m *Message) decode(pd packetDecoder) (err error) {
 		if m.Value == nil {
 			return DecodingError{Info: "Snappy compression specified, but no data to uncompress"}
 		}
-		m.Value, err = snappy.Decode(nil, m.Value)
+		m.Value, err = SnappyDecode(m.Value)
 		if err != nil {
 			return err
 		}

snappy.go

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