func TestBinlogStreamerParseEventsRollback(t *testing.T) {
	input := []proto.BinlogEvent{
		rotateEvent{},
		formatEvent{},
		queryEvent{query: proto.Query{Database: "vt_test_keyspace", Sql: "BEGIN"}},
		queryEvent{query: proto.Query{Database: "vt_test_keyspace", Sql: "insert into vt_a(eid, id) values (1, 1) /* _stream vt_a (eid id ) (1 1 ); */"}},
		queryEvent{query: proto.Query{Database: "vt_test_keyspace", Sql: "insert into vt_a(eid, id) values (1, 1) /* _stream vt_a (eid id ) (1 1 ); */"}},
		queryEvent{query: proto.Query{Database: "vt_test_keyspace", Sql: "ROLLBACK"}},
		queryEvent{query: proto.Query{Database: "vt_test_keyspace", Sql: "BEGIN"}},
		queryEvent{query: proto.Query{Database: "vt_test_keyspace", Sql: "insert into vt_a(eid, id) values (1, 1) /* _stream vt_a (eid id ) (1 1 ); */"}},
		xidEvent{},
	}

	events := make(chan proto.BinlogEvent)

	want := []proto.BinlogTransaction{
		proto.BinlogTransaction{
			Statements: nil,
			Timestamp:  1407805592,
			TransactionID: myproto.EncodeGTID(myproto.MariadbGTID{
				Domain:   0,
				Server:   62344,
				Sequence: 0x0d,
			}),
		},
		proto.BinlogTransaction{
			Statements: []proto.Statement{
				proto.Statement{Category: proto.BL_SET, Sql: "SET TIMESTAMP=1407805592"},
				proto.Statement{Category: proto.BL_DML, Sql: "insert into vt_a(eid, id) values (1, 1) /* _stream vt_a (eid id ) (1 1 ); */"},
			},
			Timestamp: 1407805592,
			TransactionID: myproto.EncodeGTID(myproto.MariadbGTID{
				Domain:   0,
				Server:   62344,
				Sequence: 0x0d,
			}),
		},
	}
	var got []proto.BinlogTransaction
	sendTransaction := func(trans *proto.BinlogTransaction) error {
		got = append(got, *trans)
		return nil
	}
	bls := NewBinlogStreamer("vt_test_keyspace", nil, nil, myproto.ReplicationPosition{}, sendTransaction)

	go sendTestEvents(events, input)
	svm := &sync2.ServiceManager{}
	svm.Go(func(ctx *sync2.ServiceContext) error {
		_, err := bls.parseEvents(ctx, events)
		return err
	})
	if err := svm.Join(); err != ErrServerEOF {
		t.Errorf("unexpected error: %v", err)
	}

	if !reflect.DeepEqual(got, want) {
		t.Errorf("binlogConnStreamer.parseEvents(): got %v, want %v", got, want)
	}
}

// UpdateBlpCheckpoint returns a statement to update a value in the
// _vt.blp_checkpoint table.
func UpdateBlpCheckpoint(uid uint32, gtid myproto.GTID, timeUpdated int64, txTimestamp int64) string {
	if txTimestamp != 0 {
		return fmt.Sprintf(
			"UPDATE _vt.blp_checkpoint "+
				"SET gtid='%v', time_updated=%v, transaction_timestamp=%v "+
				"WHERE source_shard_uid=%v",
			myproto.EncodeGTID(gtid), timeUpdated, txTimestamp, uid)
	} else {
		return fmt.Sprintf(
			"UPDATE _vt.blp_checkpoint "+
				"SET gtid='%v', time_updated=%v "+
				"WHERE source_shard_uid=%v",
			myproto.EncodeGTID(gtid), timeUpdated, uid)
	}
}

// RegisterBinlogPlayerMap registers the varz for the players
func RegisterBinlogPlayerMap(blm *BinlogPlayerMap) {
	stats.Publish("BinlogPlayerMapSize", stats.IntFunc(blm.size))
	stats.Publish("BinlogPlayerSecondsBehindMaster", stats.IntFunc(func() int64 {
		sbm := int64(0)
		blm.mu.Lock()
		for _, bpc := range blm.players {
			psbm := bpc.binlogPlayerStats.SecondsBehindMaster.Get()
			if psbm > sbm {
				sbm = psbm
			}
		}
		blm.mu.Unlock()
		return sbm
	}))
	stats.Publish("BinlogPlayerSecondsBehindMasterMap", stats.CountersFunc(func() map[string]int64 {
		blm.mu.Lock()
		result := make(map[string]int64, len(blm.players))
		for i, bpc := range blm.players {
			sbm := bpc.binlogPlayerStats.SecondsBehindMaster.Get()
			result[fmt.Sprintf("%v", i)] = sbm
		}
		blm.mu.Unlock()
		return result
	}))
	stats.Publish("BinlogPlayerGTIDMap", stats.StringMapFunc(func() map[string]string {
		blm.mu.Lock()
		result := make(map[string]string, len(blm.players))
		for i, bpc := range blm.players {
			lgtid := bpc.binlogPlayerStats.GetLastGTID()
			result[fmt.Sprintf("%v", i)] = myproto.EncodeGTID(lgtid)
		}
		blm.mu.Unlock()
		return result
	}))
	stats.Publish("BinlogPlayerSourceShardNameMap", stats.StringMapFunc(func() map[string]string {
		blm.mu.Lock()
		result := make(map[string]string, len(blm.players))
		for i, bpc := range blm.players {
			name := bpc.sourceShard.Keyspace + "/" + bpc.sourceShard.Shard
			result[fmt.Sprintf("%v", i)] = name
		}
		blm.mu.Unlock()
		return result
	}))
	stats.Publish("BinlogPlayerSourceTabletAliasMap", stats.StringMapFunc(func() map[string]string {
		blm.mu.Lock()
		result := make(map[string]string, len(blm.players))
		for i, bpc := range blm.players {
			bpc.playerMutex.Lock()
			result[fmt.Sprintf("%v", i)] = bpc.sourceTablet.String()
			bpc.playerMutex.Unlock()
		}
		blm.mu.Unlock()
		return result
	}))
}

// BinlogTransactionToProto converts a BinlogTransaction to a proto3
func BinlogTransactionToProto(bt *BinlogTransaction) *pb.BinlogTransaction {
	result := &pb.BinlogTransaction{
		Timestamp: bt.Timestamp,
		Gtid:      myproto.EncodeGTID(bt.GTIDField.Value),
	}
	if len(bt.Statements) > 0 {
		result.Statements = make([]*pb.BinlogTransaction_Statement, len(bt.Statements))
		for i, s := range bt.Statements {
			result.Statements[i] = StatementToProto(&s)
		}
	}

	return result
}

func TestBinlogStreamerParseEventsMariadbBeginGTID(t *testing.T) {
	input := []proto.BinlogEvent{
		mariadbRotateEvent,
		mariadbFormatEvent,
		mariadbBeginGTIDEvent,
		mariadbInsertEvent,
		mariadbXidEvent,
	}

	events := make(chan proto.BinlogEvent)

	want := []proto.BinlogTransaction{
		proto.BinlogTransaction{
			Statements: []proto.Statement{
				proto.Statement{Category: proto.BL_SET, Charset: charset, Sql: "SET TIMESTAMP=1409892744"},
				proto.Statement{Category: proto.BL_DML, Charset: charset, Sql: "insert into vt_insert_test(msg) values ('test 0') /* _stream vt_insert_test (id ) (null ); */"},
			},
			Timestamp: 1409892744,
			TransactionID: myproto.EncodeGTID(myproto.MariadbGTID{
				Domain:   0,
				Server:   62344,
				Sequence: 10,
			}),
		},
	}
	var got []proto.BinlogTransaction
	sendTransaction := func(trans *proto.BinlogTransaction) error {
		got = append(got, *trans)
		return nil
	}
	bls := NewBinlogStreamer("vt_test_keyspace", nil, nil, myproto.ReplicationPosition{}, sendTransaction)

	go sendTestEvents(events, input)
	svm := &sync2.ServiceManager{}
	svm.Go(func(ctx *sync2.ServiceContext) error {
		_, err := bls.parseEvents(ctx, events)
		return err
	})
	if err := svm.Join(); err != ErrServerEOF {
		t.Errorf("unexpected error: %v", err)
	}

	if !reflect.DeepEqual(got, want) {
		t.Errorf("binlogConnStreamer.parseEvents(): got %v, want %v", got, want)
	}
}

func TestBinlogStreamerParseEventsMariadbStandaloneGTID(t *testing.T) {
	input := []proto.BinlogEvent{
		mariadbRotateEvent,
		mariadbFormatEvent,
		mariadbStandaloneGTIDEvent,
		mariadbCreateEvent,
	}

	events := make(chan proto.BinlogEvent)

	want := []proto.BinlogTransaction{
		proto.BinlogTransaction{
			Statements: []proto.Statement{
				proto.Statement{Category: proto.BL_SET, Charset: &mproto.Charset{Client: 8, Conn: 8, Server: 33}, Sql: "SET TIMESTAMP=1409892744"},
				proto.Statement{Category: proto.BL_DDL, Charset: &mproto.Charset{Client: 8, Conn: 8, Server: 33}, Sql: "create table if not exists vt_insert_test (\nid bigint auto_increment,\nmsg varchar(64),\nprimary key (id)\n) Engine=InnoDB"},
			},
			Timestamp: 1409892744,
			TransactionID: myproto.EncodeGTID(myproto.MariadbGTID{
				Domain:   0,
				Server:   62344,
				Sequence: 9,
			}),
		},
	}
	var got []proto.BinlogTransaction
	sendTransaction := func(trans *proto.BinlogTransaction) error {
		got = append(got, *trans)
		return nil
	}
	bls := NewBinlogStreamer("vt_test_keyspace", nil, nil, myproto.ReplicationPosition{}, sendTransaction)

	go sendTestEvents(events, input)
	svm := &sync2.ServiceManager{}
	svm.Go(func(ctx *sync2.ServiceContext) error {
		_, err := bls.parseEvents(ctx, events)
		return err
	})
	if err := svm.Join(); err != ErrServerEOF {
		t.Errorf("unexpected error: %v", err)
	}

	if !reflect.DeepEqual(got, want) {
		t.Errorf("binlogConnStreamer.parseEvents(): got %v, want %v", got, want)
	}
}

// StreamEventToProto converts a StreamEvent to a proto3
func StreamEventToProto(s *StreamEvent) *pb.StreamEvent {
	result := &pb.StreamEvent{
		TableName:        s.TableName,
		PrimaryKeyFields: mproto.FieldsToProto3(s.PrimaryKeyFields),
		PrimaryKeyValues: mproto.RowsToProto3(s.PrimaryKeyValues),
		Sql:              s.Sql,
		Timestamp:        s.Timestamp,
		Gtid:             myproto.EncodeGTID(s.GTIDField.Value),
	}
	switch s.Category {
	case "DML":
		result.Category = pb.StreamEvent_SE_DML
	case "DDL":
		result.Category = pb.StreamEvent_SE_DDL
	case "POS":
		result.Category = pb.StreamEvent_SE_POS
	default:
		result.Category = pb.StreamEvent_SE_ERR
	}
	return result
}

// parseEvents processes the raw binlog dump stream from the server, one event
// at a time, and groups them into transactions. It is called from within the
// service function launched by Stream().
//
// If the sendTransaction func returns io.EOF, parseEvents returns ErrClientEOF.
// If the events channel is closed, parseEvents returns ErrServerEOF.
func (bls *BinlogStreamer) parseEvents(ctx *sync2.ServiceContext, events <-chan proto.BinlogEvent) (myproto.ReplicationPosition, error) {
	var statements []proto.Statement
	var format proto.BinlogFormat
	var gtid myproto.GTID
	var pos = bls.startPos
	var autocommit = true
	var err error

	// A begin can be triggered either by a BEGIN query, or by a GTID_EVENT.
	begin := func() {
		if statements != nil {
			// If this happened, it would be a legitimate error.
			log.Errorf("BEGIN in binlog stream while still in another transaction; dropping %d statements: %v", len(statements), statements)
			binlogStreamerErrors.Add("ParseEvents", 1)
		}
		statements = make([]proto.Statement, 0, 10)
		autocommit = false
	}
	// A commit can be triggered either by a COMMIT query, or by an XID_EVENT.
	// Statements that aren't wrapped in BEGIN/COMMIT are committed immediately.
	commit := func(timestamp uint32) error {
		trans := &proto.BinlogTransaction{
			Statements:    statements,
			Timestamp:     int64(timestamp),
			TransactionID: myproto.EncodeGTID(gtid),
		}
		if err = bls.sendTransaction(trans); err != nil {
			if err == io.EOF {
				return ErrClientEOF
			}
			return fmt.Errorf("send reply error: %v", err)
		}
		statements = nil
		autocommit = true
		return nil
	}

	// Parse events.
	for ctx.IsRunning() {
		var ev proto.BinlogEvent
		var ok bool

		select {
		case ev, ok = <-events:
			if !ok {
				// events channel has been closed, which means the connection died.
				log.Infof("reached end of binlog event stream")
				return pos, ErrServerEOF
			}
		case <-ctx.ShuttingDown:
			log.Infof("stopping early due to BinlogStreamer service shutdown")
			return pos, nil
		}

		// Validate the buffer before reading fields from it.
		if !ev.IsValid() {
			return pos, fmt.Errorf("can't parse binlog event, invalid data: %#v", ev)
		}

		// We need to keep checking for FORMAT_DESCRIPTION_EVENT even after we've
		// seen one, because another one might come along (e.g. on log rotate due to
		// binlog settings change) that changes the format.
		if ev.IsFormatDescription() {
			format, err = ev.Format()
			if err != nil {
				return pos, fmt.Errorf("can't parse FORMAT_DESCRIPTION_EVENT: %v, event data: %#v", err, ev)
			}
			continue
		}

		// We can't parse anything until we get a FORMAT_DESCRIPTION_EVENT that
		// tells us the size of the event header.
		if format.IsZero() {
			// The only thing that should come before the FORMAT_DESCRIPTION_EVENT
			// is a fake ROTATE_EVENT, which the master sends to tell us the name
			// of the current log file.
			if ev.IsRotate() {
				continue
			}
			return pos, fmt.Errorf("got a real event before FORMAT_DESCRIPTION_EVENT: %#v", ev)
		}

		// Strip the checksum, if any. We don't actually verify the checksum, so discard it.
		ev, _, err = ev.StripChecksum(format)
		if err != nil {
			return pos, fmt.Errorf("can't strip checksum from binlog event: %v, event data: %#v", err, ev)
		}

		// Update the GTID if the event has one. The actual event type could be
		// something special like GTID_EVENT (MariaDB, MySQL 5.6), or it could be
		// an arbitrary event with a GTID in the header (Google MySQL).
		if ev.HasGTID(format) {
			gtid, err = ev.GTID(format)
			if err != nil {
//.........这里部分代码省略.........

// PopulateBlpCheckpoint returns a statement to populate the first value into
// the _vt.blp_checkpoint table.
func PopulateBlpCheckpoint(index uint32, gtid myproto.GTID, timeUpdated int64, flags string) string {
	return fmt.Sprintf("INSERT INTO _vt.blp_checkpoint "+
		"(source_shard_uid, gtid, time_updated, transaction_timestamp, flags) "+
		"VALUES (%v, '%v', %v, 0, '%v')",
		index, myproto.EncodeGTID(gtid), timeUpdated, flags)
}