finish basic message encoding/decoding

message.go

@@ -1,11 +1,71 @@
 package kafka
 
+type compressionCodec int
+const (
+	COMPRESSION_NONE compressionCodec = 0
+	COMPRESSION_GZIP compressionCodec = 1
+	COMPRESSION_SNAPPY compressionCodec = 2
+)
+
+// The spec just says: "This is a version id used to allow backwards compatible evolution of the message
+// binary format." but it doesn't say what the current value is, so presumably 0...
+const MESSAGE_FORMAT int8 = 0
+
 type message struct {
+	codec compressionCodec
+	key *[]byte
+	value *[]byte
 }
 
 func (m *message) encode(pe packetEncoder) {
+	pe.pushCRC32()
+
+	pe.putInt8(MESSAGE_FORMAT)
+
+	var attributes int8 = 0
+	attributes |= int8(m.codec & 0x07)
+	pe.putInt8(attributes)
+
+	pe.putBytes(m.key)
+	pe.putBytes(m.value)
+
+	pe.pop()
 }
 
 func (m *message) decode(pd packetDecoder) (err error) {
+	err = pd.pushCRC32()
+	if err != nil {
+		return err
+	}
+
+	format, err := pd.getInt8()
+	if err != nil {
+		return err
+	}
+	if format != MESSAGE_FORMAT {
+		return DecodingError{}
+	}
+
+	attribute, err := pd.getInt8()
+	if err != nil {
+		return err
+	}
+	m.codec = compressionCodec(attribute & 0x07)
+
+	m.key, err = pd.getBytes()
+	if err != nil {
+		return err
+	}
+
+	m.value, err = pd.getBytes()
+	if err != nil {
+		return err
+	}
+
+	err = pd.pop()
+	if err != nil {
+		return err
+	}
+
 	return nil
 }

packet_decoder.go

@@ -3,6 +3,7 @@ package kafka
 type packetDecoder interface {
 	remaining() int
 
+	getInt8() (int8, error)
 	getInt16() (int16, error)
 	getInt32() (int32, error)
 	getInt64() (int64, error)

packet_encoder.go

@@ -1,6 +1,7 @@
 package kafka
 
 type packetEncoder interface {
+	putInt8(in int8)
 	putInt16(in int16)
 	putInt32(in int32)
 	putInt64(in int64)

prep_encoder.go

@@ -7,6 +7,10 @@ type prepEncoder struct {
 	err    bool
 }
 
+func (pe *prepEncoder) putInt8(in int8) {
+	pe.length += 1
+}
+
 func (pe *prepEncoder) putInt16(in int16) {
 	pe.length += 2
 }

real_decoder.go

@@ -15,6 +15,15 @@ func (rd *realDecoder) remaining() int {
 	return len(rd.raw) - rd.off
 }
 
+func (rd *realDecoder) getInt8() (int8, error) {
+	if rd.remaining() < 1 {
+		return -1, DecodingError{}
+	}
+	tmp := int8(rd.raw[rd.off])
+	rd.off += 1
+	return tmp, nil
+}
+
 func (rd *realDecoder) getInt16() (int16, error) {
 	if rd.remaining() < 2 {
 		return -1, DecodingError{}

real_encoder.go

@@ -8,6 +8,11 @@ type realEncoder struct {
 	stack []pushEncoder
 }
 
+func (re *realEncoder) putInt8(in int8) {
+	re.raw[re.off] = byte(in)
+	re.off += 1
+}
+
 func (re *realEncoder) putInt16(in int16) {
 	binary.BigEndian.PutUint16(re.raw[re.off:], uint16(in))
 	re.off += 2