Add String

duringnone · Jun 27, 2018 · dc05c0a · dc05c0a
1 parent 5391202
commit dc05c0a
Show file tree

Hide file tree

Showing 2 changed files with 59 additions and 0 deletions.
diff --git a/message.go b/message.go
@@ -25,6 +25,21 @@ const (
 	CompressionLZ4
 )
 
+func (c CompressionCodec) String() string {
+	switch c {
+	case CompressionNone:
+		return "none"
+	case CompressionGZIP:
+		return "gzip"
+	case CompressionSnappy:
+		return "snappy"
+	case CompressionLZ4:
+		return "lz4"
+	default:
+		return "unknown"
+	}
+}
+
 // Message is a data structure representing kafka messages.
 type Message struct {
 	// Topic is reads only and MUST NOT be set when writing messages
@@ -40,7 +55,10 @@ type Message struct {
 	// writing the message.
 	Time time.Time
 
+	// Compression codec used to encode the message value
 	CompressionCodec CompressionCodec
+
+	// Compression level for the codec if supported (only gzip)
 	CompressionLevel int
 }
 

diff --git a/message_test.go b/message_test.go
@@ -0,0 +1,41 @@
+package kafka
+
+import (
+	"testing"
+)
+
+func TestCompression(t *testing.T) {
+	msg := Message{
+		Value: []byte("message"),
+	}
+
+	testEncodeDecode(t, msg, CompressionNone)
+	testEncodeDecode(t, msg, CompressionGZIP)
+	testEncodeDecode(t, msg, CompressionSnappy)
+}
+
+func testEncodeDecode(t *testing.T, m Message, codec CompressionCodec) {
+	var encoded Message
+
+	t.Run("encode with "+codec.String(), func(t *testing.T) {
+		var err error
+
+		m.CompressionCodec = codec
+		encoded, err = m.encode()
+		if err != nil {
+			t.Error(err)
+		}
+	})
+
+	t.Run("decode with "+codec.String(), func(t *testing.T) {
+		decoded, err := encoded.decode()
+		if err != nil {
+			t.Error(err)
+		}
+		if string(decoded.Value) != "message" {
+			t.Error("bad message")
+			t.Log("got: ", string(decoded.Value))
+			t.Log("expected: message")
+		}
+	})
+}