Method TestStreamEvents

go/logic/streamer_test.go:66–141

()

}

func (suite *EventsStreamerTestSuite) TestStreamEvents() {
	ctx := context.Background()

	_, err := suite.db.ExecContext(ctx, fmt.Sprintf("CREATE TABLE %s (id INT PRIMARY KEY, name VARCHAR(255))", getTestTableName()))
	suite.Require().NoError(err)

	connectionConfig, err := getTestConnectionConfig(ctx, suite.mysqlContainer)
	suite.Require().NoError(err)

	migrationContext := newTestMigrationContext()
	migrationContext.ApplierConnectionConfig = connectionConfig
	migrationContext.InspectorConnectionConfig = connectionConfig
	migrationContext.SetConnectionConfig("innodb")

	streamer := NewEventsStreamer(migrationContext)

	err = streamer.InitDBConnections()
	suite.Require().NoError(err)
	defer streamer.Close()
	defer streamer.Teardown()

	streamCtx, cancel := context.WithCancel(context.Background())

	dmlEvents := make([]*binlog.BinlogDMLEvent, 0)
	err = streamer.AddListener(false, testMysqlDatabase, testMysqlTableName, func(event *binlog.BinlogEntry) error {
		dmlEvents = append(dmlEvents, event.DmlEvent)

		// Stop once we've collected three events
		if len(dmlEvents) == 3 {
			cancel()
		}

		return nil
	})
	suite.Require().NoError(err)

	group := errgroup.Group{}
	group.Go(func() error {
		//nolint:contextcheck
		return streamer.StreamEvents(func() bool {
			return streamCtx.Err() != nil
		})
	})

	group.Go(func() error {
		var err error

		_, err = suite.db.ExecContext(ctx, fmt.Sprintf("INSERT INTO %s (id, name) VALUES (1, 'foo')", getTestTableName()))
		if err != nil {
			return err
		}

		_, err = suite.db.ExecContext(ctx, fmt.Sprintf("INSERT INTO %s (id, name) VALUES (2, 'bar')", getTestTableName()))
		if err != nil {
			return err
		}

		_, err = suite.db.ExecContext(ctx, fmt.Sprintf("INSERT INTO %s (id, name) VALUES (3, 'baz')", getTestTableName()))