FormatDescriptionEvent should be readed by binlog file always (go-mysql-org#224)

Author: svart-ravn

replication/parser.go

@@ -50,6 +50,13 @@ func (p *BinlogParser) ParseFile(name string, offset int64, onEvent OnEventFunc)
 
 	if offset < 4 {
 		offset = 4
+	} else if offset > 4 {
+		//  FORMAT_DESCRIPTION event should be read by default always (despite that fact passed offset may be higher than 4)
+		if _, err = f.Seek(4, os.SEEK_SET); err != nil {
+			return errors.Errorf("seek %s to %d error %v", name, offset, err)
+		}
+
+		p.getFormatDescriptionEvent(f, onEvent)
 	}
 
 	if _, err = f.Seek(offset, os.SEEK_SET); err != nil {
@@ -59,55 +66,82 @@ func (p *BinlogParser) ParseFile(name string, offset int64, onEvent OnEventFunc)
 	return p.ParseReader(f, onEvent)
 }
 
-func (p *BinlogParser) ParseReader(r io.Reader, onEvent OnEventFunc) error {
+
+
+func (p *BinlogParser) getFormatDescriptionEvent(r io.Reader, onEvent OnEventFunc) error {
+	_, err := p.parseSingleEvent(&r, onEvent)
+	return err
+}
+
+
+
+func (p *BinlogParser) parseSingleEvent(r *io.Reader, onEvent OnEventFunc) (bool, error) {
 	var err error
 	var n int64
 
-	for {
-		headBuf := make([]byte, EventHeaderSize)
+	headBuf := make([]byte, EventHeaderSize)
 
-		if _, err = io.ReadFull(r, headBuf); err == io.EOF {
-			return nil
-		} else if err != nil {
-			return errors.Trace(err)
-		}
+	if _, err = io.ReadFull(*r, headBuf); err == io.EOF {
+		return true, nil
+	} else if err != nil {
+		return false, errors.Trace(err)
+	}
 
-		var h *EventHeader
-		h, err = p.parseHeader(headBuf)
-		if err != nil {
-			return errors.Trace(err)
-		}
+	var h *EventHeader
+	h, err = p.parseHeader(headBuf)
+	if err != nil {
+		return false, errors.Trace(err)
+	}
 
-		if h.EventSize <= uint32(EventHeaderSize) {
-			return errors.Errorf("invalid event header, event size is %d, too small", h.EventSize)
+	if h.EventSize <= uint32(EventHeaderSize) {
+		return false, errors.Errorf("invalid event header, event size is %d, too small", h.EventSize)
+	}
 
-		}
+	var buf bytes.Buffer
+	if n, err = io.CopyN(&buf, *r, int64(h.EventSize)-int64(EventHeaderSize)); err != nil {
+		return false, errors.Errorf("get event body err %v, need %d - %d, but got %d", err, h.EventSize, EventHeaderSize, n)
+	}
 
-		var buf bytes.Buffer
-		if n, err = io.CopyN(&buf, r, int64(h.EventSize)-int64(EventHeaderSize)); err != nil {
-			return errors.Errorf("get event body err %v, need %d - %d, but got %d", err, h.EventSize, EventHeaderSize, n)
-		}
+	data := buf.Bytes()
+	rawData := data
 
-		data := buf.Bytes()
-		rawData := data
+	eventLen := int(h.EventSize) - EventHeaderSize
 
-		eventLen := int(h.EventSize) - EventHeaderSize
+	if len(data) != eventLen {
+		return false, errors.Errorf("invalid data size %d in event %s, less event length %d", len(data), h.EventType, eventLen)
+	}
 
-		if len(data) != eventLen {
-			return errors.Errorf("invalid data size %d in event %s, less event length %d", len(data), h.EventType, eventLen)
+	var e Event
+	e, err = p.parseEvent(h, data)
+	if err != nil {
+		if _, ok := err.(errMissingTableMapEvent); ok {
+			return false, nil
 		}
+		return false, errors.Trace(err)
+	}
 
-		var e Event
-		e, err = p.parseEvent(h, data)
+	if err = onEvent(&BinlogEvent{rawData, h, e}); err != nil {
+		return false, errors.Trace(err)
+	}
+
+
+	return false, nil
+}
+
+
+func (p *BinlogParser) ParseReader(r io.Reader, onEvent OnEventFunc) error {
+
+	for {
+		done, err := p.parseSingleEvent(&r, onEvent); 
 		if err != nil {
 			if _, ok := err.(errMissingTableMapEvent); ok {
 				continue
 			}
 			return errors.Trace(err)
 		}
 
-		if err = onEvent(&BinlogEvent{rawData, h, e}); err != nil {
-			return errors.Trace(err)
+		if done {
+			break
 		}
 	}