New functions to create PreviousGtids events, and to update event checksum

go/mysql/binlog_event_make.go

@@ -119,6 +119,19 @@ func packetize(f BinlogFormat, typ byte, flags uint16, data []byte, m BinlogEven
 	return result
 }
 
+// UpdateChecksum updates the checksum for the specified |event|. The BinlogFormat, |f|, indicates
+// if checksums are enabled. If checksums are not enabled, then no change is made to |event|.
+func UpdateChecksum(f BinlogFormat, event BinlogEvent) {
+	result := event.Bytes()
+	length := len(result)
+
+	switch f.ChecksumAlgorithm {
+	case BinlogChecksumAlgCRC32:
+		checksum := crc32.ChecksumIEEE(result[0 : length-4])
+		binary.LittleEndian.PutUint32(result[length-4:], checksum)
+	}
+}
+
 // NewInvalidEvent returns an invalid event (its size is <19).
 func NewInvalidEvent() BinlogEvent {
 	return NewMysql56BinlogEvent([]byte{0})
@@ -146,6 +159,16 @@ func NewFormatDescriptionEvent(f BinlogFormat, m BinlogEventMetadata) BinlogEven
 	return NewMysql56BinlogEvent(ev)
 }
 
+// NewPreviousGtidsEvent creates a new Previous GTIDs BinlogEvent. The BinlogFormat, |f|, indicates if checksums are
+// enabled, the BinlogEventMeatadata, |m|, specifies the unique server ID, and |gtids| is the MySQL 5.6 GTID set
+// to include in the event, indicating the events that have been previously executed by the server.
+func NewPreviousGtidsEvent(f BinlogFormat, m BinlogEventMetadata, gtids Mysql56GTIDSet) BinlogEvent {
+	data := gtids.SIDBlock()
+
+	ev := packetize(f, ePreviousGTIDsEvent, 0, data, m)
+	return NewMysql56BinlogEvent(ev)
+}
+
 // NewInvalidFormatDescriptionEvent returns an invalid FormatDescriptionEvent.
 // The binlog version is set to 3. It IsValid() though.
 func NewInvalidFormatDescriptionEvent(f BinlogFormat, m BinlogEventMetadata) BinlogEvent {

go/mysql/binlog_event_make_test.go

@@ -17,9 +17,12 @@ limitations under the License.
 package mysql
 
 import (
+	"encoding/binary"
 	"reflect"
 	"testing"
 
+	"github.com/stretchr/testify/require"
+
 	binlogdatapb "github.com/dolthub/vitess/go/vt/proto/binlogdata"
 )
 
@@ -141,6 +144,52 @@ func TestIntVarEvent(t *testing.T) {
 	}
 }
 
+func TestUpdateChecksum(t *testing.T) {
+	f := NewMySQL56BinlogFormat()
+	m := NewTestBinlogMetadata()
+
+	q := Query{
+		Database: "my database",
+		SQL:      "my query",
+		Charset: &binlogdatapb.Charset{
+			Client: 0x1234,
+			Conn:   0x5678,
+			Server: 0x9abc,
+		},
+	}
+	event := NewQueryEvent(f, m, q)
+	bytes := event.Bytes()
+
+	// Calling UpdateChecksum without changing the event should not change the checksum
+	oldChecksum := append([]byte{}, bytes[len(bytes)-4:]...)
+	UpdateChecksum(f, event)
+	newChecksum := append([]byte{}, bytes[len(bytes)-4:]...)
+	require.Equal(t, oldChecksum, newChecksum)
+	require.Equal(t, []byte{0x65, 0xaa, 0x33, 0x0e}, newChecksum)
+
+	// Calling UpdateChecksum after changing the event should generate a new checksum
+	binary.LittleEndian.PutUint32(bytes[13:13+4], uint32(420))
+	UpdateChecksum(f, event)
+	newChecksum = append([]byte{}, bytes[len(bytes)-4:]...)
+	require.NotEqual(t, oldChecksum, newChecksum)
+	require.Equal(t, []byte{0x26, 0xD0, 0xa4, 0x05}, newChecksum)
+}
+
+func TestPreviousGtidsEvent(t *testing.T) {
+	f := NewMySQL56BinlogFormat()
+	m := NewTestBinlogMetadata()
+
+	gtidSetString := "32a5b8c9-4716-40f5-9a9b-3d7be0cb33d7:1-42"
+	gtidSet, err := ParseMysql56GTIDSet(gtidSetString)
+	require.NoError(t, err)
+
+	event := NewPreviousGtidsEvent(f, m, gtidSet.(Mysql56GTIDSet))
+	require.True(t, event.IsPreviousGTIDs())
+	position, err := event.PreviousGTIDs(f)
+	require.NoError(t, err)
+	require.Equal(t, gtidSetString, position.String())
+}
+
 func TestInvalidEvents(t *testing.T) {
 	f := NewMySQL56BinlogFormat()
 	m := NewTestBinlogMetadata()