Method TestInitialLoad

flow/e2e/pubsub_test.go:257–329

()

}

func (s PubSubSuite) TestInitialLoad() {
	srcTableName := AttachSchema(s, "psinitial")

	_, err := s.Conn().Exec(s.t.Context(), fmt.Sprintf(`
		CREATE TABLE IF NOT EXISTS %s (
			id SERIAL PRIMARY KEY,
			val text
		);
	`, srcTableName))
	require.NoError(s.t, err)

	sa, err := ServiceAccount()
	require.NoError(s.t, err)

	_, err = s.Conn().Exec(s.t.Context(), `insert into public.scripts (name, lang, source) values
	('e2e_psinitial', 'lua', 'function onRecord(r) return r.row and r.row.val end') on conflict do nothing`)
	require.NoError(s.t, err)

	flowName := AddSuffix(s, "e2epsinitial")
	connectionGen := FlowConnectionGenerationConfig{
		FlowJobName:      flowName,
		TableNameMapping: map[string]string{srcTableName: flowName},
		Destination:      s.Peer(sa).Name,
	}
	flowConnConfig := connectionGen.GenerateFlowConnectionConfigs(s)
	flowConnConfig.Script = "e2e_psinitial"
	flowConnConfig.DoInitialSnapshot = true

	psclient, err := sa.CreatePubSubClient(s.t.Context())
	require.NoError(s.t, err)
	defer psclient.Close()
	topicName := fmt.Sprintf("projects/%s/topics/%s", psclient.Project(), flowName)
	_, err = psclient.TopicAdminClient.CreateTopic(s.t.Context(), &pubsubpb.Topic{
		Name: topicName,
	})
	require.NoError(s.t, err)
	sub, err := psclient.SubscriptionAdminClient.CreateSubscription(s.t.Context(), &pubsubpb.Subscription{
		Name:                     fmt.Sprintf("projects/%s/subscriptions/%s", psclient.Project(), flowName),
		Topic:                    topicName,
		MessageRetentionDuration: durationpb.New(10 * time.Minute),
		ExpirationPolicy:         &pubsubpb.ExpirationPolicy{Ttl: durationpb.New(24 * time.Hour)},
	})
	require.NoError(s.t, err)
	_, err = s.Conn().Exec(s.t.Context(), fmt.Sprintf(
		`INSERT INTO %s (id, val) VALUES (1, 'testval')`, srcTableName))
	require.NoError(s.t, err)

	tc := NewTemporalClient(s.t)
	env := ExecutePeerflow(s.t, tc, flowConnConfig)
	SetupCDCFlowStatusQuery(s.t, env, flowConnConfig)

	ctx, cancel := context.WithTimeout(s.t.Context(), 3*time.Minute)
	defer cancel()

	msgs := make(chan *pubsub.Message)
	go func() {
		subclient := psclient.Subscriber(sub.Name)
		_ = subclient.Receive(ctx, func(_ context.Context, m *pubsub.Message) {