replication: support parsing event with header only and no checksum (#431)

* replication: support parsing event with header only and no checksum

Author: amyangfei

replication/parser.go

@@ -120,7 +120,7 @@ func (p *BinlogParser) parseSingleEvent(r io.Reader, onEvent OnEventFunc) (bool,
 		return false, errors.Trace(err)
 	}
 
-	if h.EventSize <= uint32(EventHeaderSize) {
+	if h.EventSize < uint32(EventHeaderSize) {
 		return false, errors.Errorf("invalid event header, event size is %d, too small", h.EventSize)
 	}
 	if n, err = io.CopyN(&buf, r, int64(h.EventSize-EventHeaderSize)); err != nil {

replication/parser_test.go

@@ -1,6 +1,8 @@
 package replication
 
 import (
+	"bytes"
+
 	. "github.com/pingcap/check"
 )
 
@@ -40,3 +42,37 @@ func (t *testSyncerSuite) TestIndexOutOfRange(c *C) {
 
 	c.Assert(err, IsNil)
 }
+
+func (t *testSyncerSuite) TestParseEvent(c *C) {
+	parser := NewBinlogParser()
+	parser.format = &FormatDescriptionEvent{
+		Version:                0x4,
+		ServerVersion:          []uint8{0x35, 0x2e, 0x36, 0x2e, 0x32, 0x30, 0x2d, 0x6c, 0x6f, 0x67, 0x0, 0x0, 0x0, 0x0, 0x0, 0x0, 0x0, 0x0, 0x0, 0x0, 0x0, 0x0, 0x0, 0x0, 0x0, 0x0, 0x0, 0x0, 0x0, 0x0, 0x0, 0x0, 0x0, 0x0, 0x0, 0x0, 0x0, 0x0, 0x0, 0x0, 0x0, 0x0, 0x0, 0x0, 0x0, 0x0, 0x0, 0x0, 0x0, 0x0},
+		CreateTimestamp:        0x0,
+		EventHeaderLength:      0x13,
+		EventTypeHeaderLengths: []uint8{0x38, 0xd, 0x0, 0x8, 0x0, 0x12, 0x0, 0x4, 0x4, 0x4, 0x4, 0x12, 0x0, 0x0, 0x5c, 0x0, 0x4, 0x1a, 0x8, 0x0, 0x0, 0x0, 0x8, 0x8, 0x8, 0x2, 0x0, 0x0, 0x0, 0xa, 0xa, 0xa, 0x19, 0x19, 0x0},
+		ChecksumAlgorithm:      0x0,
+	}
+	testCases := []struct {
+		byteData  []byte
+		eventSize uint32
+	}{
+		{[]byte{0x86, 0x4c, 0x9c, 0x5d, 0x03, 0x65, 0x00, 0x00, 0x00, 0x13, 0x00, 0x00, 0x00, 0x4d, 0x01, 0x00, 0x00, 0x00, 0x00}, uint32(19)},
+		{[]byte{0x15, 0x50, 0x9c, 0x5d, 0x03, 0x65, 0x00, 0x00, 0x00, 0x17, 0x00, 0x00, 0x00, 0x59, 0x01, 0x00, 0x00, 0x00, 0x00, 0x7d, 0x82, 0xa8, 0x50}, uint32(23)},
+	}
+
+	for _, tc := range testCases {
+		r := bytes.NewReader(tc.byteData)
+		_, err := parser.ParseSingleEvent(r, func(e *BinlogEvent) error {
+			c.Assert(e.Header.EventType, Equals, STOP_EVENT)
+			c.Assert(e.Header.EventSize, Equals, tc.eventSize)
+			return nil
+		})
+		c.Assert(err, IsNil)
+
+		e, err2 := parser.Parse(tc.byteData)
+		c.Assert(e.Header.EventType, Equals, STOP_EVENT)
+		c.Assert(e.Header.EventSize, Equals, tc.eventSize)
+		c.Assert(err2, IsNil)
+	}
+}