Function AttachToCdcStream

flow/pua/stream_adapter.go:41–85

(
	ctx context.Context,
	ls *lua.LState,
	lfn *lua.LFunction,
	stream *model.CDCStream[model.RecordItems],
	onErr context.CancelCauseFunc,
)

}

func AttachToCdcStream(
	ctx context.Context,
	ls *lua.LState,
	lfn *lua.LFunction,
	stream *model.CDCStream[model.RecordItems],
	onErr context.CancelCauseFunc,
) *model.CDCStream[model.RecordItems] {
	outstream := model.NewCDCStream[model.RecordItems](0)

	handleErr := func(err error) {
		onErr(err)
		<-ctx.Done()
		for range stream.GetRecords() {
			// still read records to make sure input closes first
		}
	}

	go func() {
		if stream.WaitAndCheckEmpty() {
			outstream.SignalAsEmpty()
			<-stream.GetRecords() // needed because empty signal comes before Close
		} else {
			outstream.SignalAsNotEmpty()
			for record := range stream.GetRecords() {
				ls.Push(lfn)
				ls.Push(LuaRecord.New(ls, record))
				if err := ls.PCall(1, 0, nil); err != nil {
					handleErr(err)
					break
				}
				err := outstream.AddRecord(ctx, record)
				if err != nil {
					handleErr(err)
					break
				}
			}
		}
		outstream.SchemaDeltas = stream.SchemaDeltas
		lastCP := stream.GetLastCheckpoint()
		outstream.UpdateLatestCheckpointID(lastCP.ID)
		outstream.UpdateLatestCheckpointText(lastCP.Text)
		outstream.Close()
	}()
	return outstream
}