add api to get each msg from msg set

commitlog/message.go

@@ -34,3 +34,7 @@ func (m Message) Size() int32 {
 func (m Message) PLen() int32 {
 	return int32(big.Uint32(m[pLenPos : pLenPos+4]))
 }
+
+func (m Message) Payload() []byte {
+	return m[payloadPos:]
+}

commitlog/message_set.go

@@ -25,5 +25,23 @@ func (ms MessageSet) Offset() int64 {
 }
 
 func (ms MessageSet) Size() int32 {
-	return int32(big.Uint32(ms[sizePos : sizePos+4]))
+	return int32(big.Uint32(ms[sizePos:sizePos+4]) + msgSetHeaderLen)
+}
+
+func (ms MessageSet) Payload() []byte {
+	return ms[msgSetHeaderLen:]
+}
+
+func (ms MessageSet) Messages() (msgs []Message) {
+	p := ms.Payload()
+	for {
+		header := Message(p[:msgHeaderLen])
+		msg := Message(p[:header.Size()])
+		msgs = append(msgs, msg)
+		p = p[msg.Size():]
+		if len(p) == 0 {
+			break
+		}
+	}
+	return msgs
 }

commitlog/message_set_test.go

@@ -7,11 +7,16 @@ import (
 )
 
 func TestMessageSet(t *testing.T) {
-	msg := NewMessage([]byte("hello"))
+	msg0 := NewMessage([]byte("hello"))
+	msg1 := NewMessage([]byte("world"))
 	msgs := []Message{
-		msg,
+		msg0,
+		msg1,
 	}
 	ms := NewMessageSet(msgs, 3)
 	assert.Equal(t, int64(3), ms.Offset())
-	assert.Equal(t, msg.Size(), ms.Size())
+	assert.Equal(t, msg0.Size()+msg1.Size()+msgSetHeaderLen, ms.Size())
+	for i, m := range ms.Messages() {
+		assert.Equal(t, msgs[i], m)
+	}
 }