Decode remaining parts of fetch responses

fetch_response.go

@@ -23,7 +23,18 @@ func (pr *fetchResponsePartitionBlock) decode(pd packetDecoder) (err error) {
 		return err
 	}
 
-	return nil
+	msgSetSize, err := pd.getInt32()
+	if err != nil {
+		return err
+	}
+
+	msgSetDecoder, err := pd.getSubset(int(msgSetSize))
+	if err != nil {
+		return err
+	}
+	err = (&pr.msgSet).decode(msgSetDecoder)
+
+	return err
 }
 
 type fetchResponseTopicBlock struct {

message_set.go

@@ -47,7 +47,7 @@ func (ms *messageSet) encode(pe packetEncoder) {
 }
 
 func (ms *messageSet) decode(pd packetDecoder) (err error) {
-	ms.msgs = make([]*messageSetBlock, 0)
+	ms.msgs = nil
 
 	for pd.remaining() > 0 {
 		msb := new(messageSetBlock)

packet_decoder.go

@@ -17,6 +17,7 @@ type packetDecoder interface {
 	getError() (KError, error)
 	getString() (*string, error)
 	getBytes() (*[]byte, error)
+	getSubset(length int) (packetDecoder, error)
 
 	// stackable
 	push(in pushDecoder) error

real_decoder.go

@@ -146,6 +146,14 @@ func (rd *realDecoder) getBytes() (*[]byte, error) {
 	}
 }
 
+func (rd *realDecoder) getSubset(length int) (packetDecoder, error) {
+	if rd.remaining() < length {
+		return nil, DecodingError{"Not enough data for subset."}
+	}
+
+	return &realDecoder{raw: rd.raw[rd.off : rd.off+length]}, nil
+}
+
 // stackable
 
 func (rd *realDecoder) push(in pushDecoder) error {