bocadillo/reader/reader.go

package reader

import (
	"context"

	"github.com/juju/errors"
	"github.com/localhots/bocadillo/binlog"
	"github.com/localhots/bocadillo/mysql/driver"
)

// Reader is a binary log reader.
type Reader struct {
	conn     *driver.Conn
	state    binlog.Position
	format   binlog.FormatDescription
	tableMap map[uint64]binlog.TableDescription
}

// Event contains binlog event details.
type Event struct {
	Format binlog.FormatDescription
	Header binlog.EventHeader
	Buffer []byte
	Offset uint64

	// Table is not empty for rows events
	Table *binlog.TableDescription
}

var (
	// ErrUnknownTableID is returned when a table ID from a rows event is
	// missing in the table map index.
	ErrUnknownTableID = errors.New("Unknown table ID")
)

// New creates a new binary log reader.
func New(dsn string, sc driver.Config) (*Reader, error) {
	conn, err := driver.Connect(dsn, sc)
	if err != nil {
		return nil, errors.Annotate(err, "establish connection")
	}

	r := &Reader{
		conn: conn,
		state: binlog.Position{
			File:   sc.File,
			Offset: uint64(sc.Offset),
		},
	}
	r.initTableMap()

	if err := conn.DisableChecksum(); err != nil {
		return nil, errors.Annotate(err, "disable binlog checksum")
	}
	if err := conn.RegisterSlave(); err != nil {
		return nil, errors.Annotate(err, "register replica server")
	}
	if err := conn.StartBinlogDump(); err != nil {
		return nil, errors.Annotate(err, "start binlog dump")
	}

	return r, nil
}

// ReadEvent reads next event from the binary log.
func (r *Reader) ReadEvent(ctx context.Context) (*Event, error) {
	connBuff, err := r.conn.ReadPacket(ctx)
	if err != nil {
		return nil, errors.Annotate(err, "read next event")
	}

	evt := Event{Format: r.format, Offset: r.state.Offset}
	if err := evt.Header.Decode(connBuff, r.format); err != nil {
		return nil, errors.Annotate(err, "decode event header")
	}
	if evt.Header.NextOffset > 0 {
		r.state.Offset = uint64(evt.Header.NextOffset)
	}

	evt.Buffer = connBuff[r.format.HeaderLen():]
	csa := r.format.ServerDetails.ChecksumAlgorithm
	if evt.Header.Type != binlog.EventTypeFormatDescription && csa == binlog.ChecksumAlgorithmCRC32 {
		// Remove trailing CRC32 checksum, we're not going to verify it
		evt.Buffer = evt.Buffer[:len(evt.Buffer)-4]
	}

	switch evt.Header.Type {
	case binlog.EventTypeFormatDescription:
		var fde binlog.FormatDescriptionEvent
		if err := fde.Decode(evt.Buffer); err != nil {
			return nil, errors.Annotate(err, "decode format description event")
		}
		r.format = fde.FormatDescription
		evt.Format = fde.FormatDescription

	case binlog.EventTypeRotate:
		var re binlog.RotateEvent
		if err := re.Decode(evt.Buffer, r.format); err != nil {
			return nil, errors.Annotate(err, "decode rotate event")
		}
		r.state = re.NextFile

	case binlog.EventTypeTableMap:
		var tme binlog.TableMapEvent
		if err := tme.Decode(evt.Buffer, r.format); err != nil {
			return nil, errors.Annotate(err, "decode table map event")
		}
		r.tableMap[tme.TableID] = tme.TableDescription

	case binlog.EventTypeWriteRowsV0,
		binlog.EventTypeWriteRowsV1,
		binlog.EventTypeWriteRowsV2,
		binlog.EventTypeUpdateRowsV0,
		binlog.EventTypeUpdateRowsV1,
		binlog.EventTypeUpdateRowsV2,
		binlog.EventTypeDeleteRowsV0,
		binlog.EventTypeDeleteRowsV1,
		binlog.EventTypeDeleteRowsV2:

		re := binlog.RowsEvent{Type: evt.Header.Type}
		tableID, flags := re.PeekTableIDAndFlags(evt.Buffer, r.format)
		td, ok := r.tableMap[tableID]
		if !ok {
			return nil, ErrUnknownTableID
		}
		evt.Table = &td

		// Throttle table map clearing. This flag could be part of every single
		// rows event
		if binlog.RowsFlagEndOfStatement&flags > 0 && len(r.tableMap) > 100 {
			// Clear table map
			r.initTableMap()
		}
	case binlog.EventTypeQuery:
		// Can be decoded by the receiver
	case binlog.EventTypeXID:
		// Can be decoded by the receiver
	case binlog.EventTypeGTID:
		// TODO: Add support
	}

	return &evt, err
}

// State returns current position in the binary log.
func (r *Reader) State() binlog.Position {
	return r.state
}

// Close underlying database connection.
func (r *Reader) Close() error {
	return r.conn.Close()
}

func (r *Reader) initTableMap() {
	r.tableMap = make(map[uint64]binlog.TableDescription)
}

// DecodeRows decodes buffer into a rows event.
func (e Event) DecodeRows() (binlog.RowsEvent, error) {
	re := binlog.RowsEvent{Type: e.Header.Type}
	if binlog.RowsEventVersion(e.Header.Type) < 0 {
		return re, errors.New("invalid rows event")
	}
	err := re.Decode(e.Buffer, e.Format, *e.Table)
	return re, err
}