*: add Reader, Transformer for relay log

pkg/binlog/event/event.go

@@ -159,6 +159,33 @@ func GenFormatDescriptionEvent(header *replication.EventHeader, latestPos uint32
 	return ev, errors.Trace(err)
 }
 
+// GenRotateEvent generates a RotateEvent.
+// ref: https://dev.mysql.com/doc/internals/en/rotate-event.html
+func GenRotateEvent(header *replication.EventHeader, latestPos uint32, nextLogName []byte, position uint64) (*replication.BinlogEvent, error) {
+	if len(nextLogName) == 0 {
+		return nil, errors.NotValidf("empty next binlog name")
+	}
+
+	// Post-header
+	postHeader := new(bytes.Buffer)
+	err := binary.Write(postHeader, binary.LittleEndian, position)
+	if err != nil {
+		return nil, errors.Annotatef(err, "write position %d", position)
+	}
+
+	// Payload
+	payload := new(bytes.Buffer)
+	err = binary.Write(payload, binary.LittleEndian, nextLogName)
+	if err != nil {
+		return nil, errors.Annotatef(err, "write next binlog name % X", nextLogName)
+	}
+
+	buf := new(bytes.Buffer)
+	event := &replication.RotateEvent{}
+	ev, err := assembleEvent(buf, event, false, *header, replication.ROTATE_EVENT, latestPos, postHeader.Bytes(), payload.Bytes())
+	return ev, errors.Trace(err)
+}
+
 // GenPreviousGTIDsEvent generates a PreviousGTIDsEvent.
 // go-mysql has no PreviousGTIDsEvent struct defined, so return the event's raw data instead.
 // MySQL has no internal doc for PREVIOUS_GTIDS_EVENT.

pkg/binlog/event/event_test.go

@@ -115,6 +115,40 @@ func (t *testEventSuite) TestGenFormatDescriptionEvent(c *C) {
 	c.Assert(err, IsNil)
 }
 
+func (t *testEventSuite) TestGenRotateEvent(c *C) {
+	var (
+		header = &replication.EventHeader{
+			Timestamp: uint32(time.Now().Unix()),
+			ServerID:  11,
+			Flags:     0x01,
+		}
+		latestPos   uint32 = 4
+		nextLogName []byte // nil
+		position    uint64 = 123
+	)
+
+	// empty nextLogName, invalid
+	rotateEv, err := GenRotateEvent(header, latestPos, nextLogName, position)
+	c.Assert(err, NotNil)
+	c.Assert(rotateEv, IsNil)
+
+	// valid nextLogName
+	nextLogName = []byte("mysql-bin.000010")
+	rotateEv, err = GenRotateEvent(header, latestPos, nextLogName, position)
+	c.Assert(err, IsNil)
+	c.Assert(rotateEv, NotNil)
+
+	// verify the header
+	verifyHeader(c, rotateEv.Header, header, replication.ROTATE_EVENT, latestPos, uint32(len(rotateEv.RawData)))
+
+	// verify the body
+	rotateEvBody, ok := rotateEv.Event.(*replication.RotateEvent)
+	c.Assert(ok, IsTrue)
+	c.Assert(rotateEvBody, NotNil)
+	c.Assert(rotateEvBody.NextLogName, DeepEquals, nextLogName)
+	c.Assert(rotateEvBody.Position, Equals, position)
+}
+
 func (t *testEventSuite) TestGenPreviousGTIDsEvent(c *C) {
 	var (
 		header = &replication.EventHeader{

pkg/binlog/reader/file.go

@@ -0,0 +1,181 @@
+// Copyright 2019 PingCAP, Inc.
+//
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+//     http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+// binlog events generator for MySQL used to generate some binlog events for tests.
+// Readability takes precedence over performance.
+
+package reader
+
+import (
+	"context"
+	"encoding/json"
+	"sync"
+	"time"
+
+	"github.com/pingcap/errors"
+	gmysql "github.com/siddontang/go-mysql/mysql"
+	"github.com/siddontang/go-mysql/replication"
+	"github.com/siddontang/go/sync2"
+
+	"github.com/pingcap/dm/pkg/gtid"
+	"github.com/pingcap/dm/pkg/log"
+)
+
+// FileReader is a binlog event reader which reads binlog events from a file.
+type FileReader struct {
+	mu sync.RWMutex
+	wg sync.WaitGroup
+
+	stage      readerStage
+	readOffset sync2.AtomicUint32
+	sendOffset sync2.AtomicUint32
+
+	parser *replication.BinlogParser
+	ch     chan *replication.BinlogEvent
+	ech    chan error
+
+	ctx    context.Context
+	cancel context.CancelFunc
+}
+
+// FileReaderConfig is the configuration used by a FileReader.
+type FileReaderConfig struct {
+	EnableRawMode bool
+	Timezone      *time.Location
+	ChBufferSize  int // event channel's buffer size
+	EchBufferSize int // error channel's buffer size
+}
+
+// FileReaderStatus represents the status of a FileReader.
+type FileReaderStatus struct {
+	Stage      string `json:"stage"`
+	ReadOffset uint32 `json:"read-offset"` // read event's offset in the file
+	SendOffset uint32 `json:"send-offset"` // sent event's offset in the file
+}
+
+// String implements Stringer.String.
+func (s *FileReaderStatus) String() string {
+	data, err := json.Marshal(s)
+	if err != nil {
+		log.Errorf("[FileReaderStatus] marshal status to json error %v", err)
+	}
+	return string(data)
+}
+
+// NewFileReader creates a FileReader instance.
+func NewFileReader(cfg *FileReaderConfig) Reader {
+	parser := replication.NewBinlogParser()
+	parser.SetVerifyChecksum(true)
+	parser.SetUseDecimal(true)
+	parser.SetRawMode(cfg.EnableRawMode)
+	if cfg.Timezone != nil {
+		parser.SetTimestampStringLocation(cfg.Timezone)
+	}
+	return &FileReader{
+		parser: parser,
+		ch:     make(chan *replication.BinlogEvent, cfg.ChBufferSize),
+		ech:    make(chan error, cfg.EchBufferSize),
+	}
+}
+
+// StartSyncByPos implements Reader.StartSyncByPos.
+func (r *FileReader) StartSyncByPos(pos gmysql.Position) error {
+	r.mu.Lock()
+	defer r.mu.Unlock()
+
+	if r.stage != stageNew {
+		return errors.Errorf("stage %s, expect %s, already started", r.stage, stageNew)
+	}
+
+	r.ctx, r.cancel = context.WithCancel(context.Background())
+	r.wg.Add(1)
+	go func() {
+		defer r.wg.Done()
+		err := r.parser.ParseFile(pos.Name, int64(pos.Pos), r.onEvent)
+		if err != nil {
+			log.Errorf("[file reader] parse binlog file with error %s", errors.ErrorStack(err))
+			select {
+			case r.ech <- err:
+			case <-r.ctx.Done():
+			}
+		}
+	}()
+
+	r.stage = stagePrepared
+	return nil
+}
+
+// StartSyncByGTID implements Reader.StartSyncByGTID.
+func (r *FileReader) StartSyncByGTID(gSet gtid.Set) error {
+	// NOTE: may be supported later.
+	return errors.NotSupportedf("read from file by GTID")
+}
+
+// Close implements Reader.Close.
+func (r *FileReader) Close() error {
+	r.mu.Lock()
+	defer r.mu.Unlock()
+
+	if r.stage == stageClosed {
+		return errors.New("already closed")
+	}
+
+	r.parser.Stop()
+	r.cancel()
+	r.wg.Wait()
+	r.stage = stageClosed
+	return nil
+}
+
+// GetEvent implements Reader.GetEvent.
+func (r *FileReader) GetEvent(ctx context.Context) (*replication.BinlogEvent, error) {
+	r.mu.RLock()
+	defer r.mu.RUnlock()
+
+	if r.stage != stagePrepared {
+		return nil, errors.Errorf("stage %s, expect %s, please start sync first", r.stage, stagePrepared)
+	}
+
+	select {
+	case ev := <-r.ch:
+		r.sendOffset.Set(ev.Header.LogPos)
+		return ev, nil
+	case err := <-r.ech:
+		return nil, err
+	case <-ctx.Done():
+		return nil, ctx.Err()
+	}
+}
+
+// Status implements Reader.Status.
+func (r *FileReader) Status() interface{} {
+	r.mu.RLock()
+	stage := r.stage
+	r.mu.RUnlock()
+
+	return &FileReaderStatus{
+		Stage:      stage.String(),
+		ReadOffset: r.readOffset.Get(),
+		SendOffset: r.sendOffset.Get(),
+	}
+}
+
+func (r *FileReader) onEvent(ev *replication.BinlogEvent) error {
+	select {
+	case r.ch <- ev:
+		r.readOffset.Set(ev.Header.LogPos)
+		return nil
+	case <-r.ctx.Done():
+		return r.ctx.Err()
+	}
+}