Part 1 - CSV Transformation

snippet_csv_parser.go

// Read reads from the csv file
func (c *CSVParser) Read(wg *sync.WaitGroup, record chan<- []string, done chan<- bool) {
	start := time.Now()

	defer func() {
		close(done)
		close(record)
		c.reporter.AddDuration(time.Since(start).Seconds())

		wg.Done()
	}()

	// open file for reading
	f, err := os.Open(c.reporter.GetFilename())
	if err != nil {
		panic(err)
	}
	defer f.Close()

	// set the headers
	c.reporter.SetHeaders(utils.GetHeaders())

	// set the file name
	c.reporter.SetFilename(filepath.Base(c.reporter.GetFilename()))

	// create csv reader
	reader := csv.NewReader(f)

	// parse headers detected in file
	if err := c.parseHeaders(reader); err != nil {
		c.reporter.AddError(err)
	}

	// read from file
	for {
		row, err := reader.Read()
		if err != nil {
			if err != io.EOF {
				c.reporter.AddError(err)
				c.reporter.RecordFailed()
				continue
			}
			break
		}

		c.reporter.RecordProcessed()
		record <- row
	}

	done <- true
}