ProcessCSV now uses parallel processing with workers
(ctx context.Context, uploadID, source string, reader io.Reader, store StoreFunc)
| 227 | |
| 228 | // ProcessCSV now uses parallel processing with workers |
| 229 | func ProcessCSV(ctx context.Context, uploadID, source string, reader io.Reader, store StoreFunc) (int, error) { |
| 230 | bufferedReader := bufio.NewReaderSize(reader, DefaultBufferSize) |
| 231 | csvReader := csv.NewReader(bufferedReader) |
| 232 | csvReader.ReuseRecord = true // Reuse slice to reduce allocations |
| 233 | |
| 234 | headers, err := csvReader.Read() |
| 235 | if err != nil { |
| 236 | return 0, fmt.Errorf("error reading CSV headers: %w", err) |
| 237 | } |
| 238 | |
| 239 | // Make a copy of headers since ReuseRecord is enabled |
| 240 | headersCopy := make([]string, len(headers)) |
| 241 | copy(headersCopy, headers) |
| 242 | |
| 243 | columnMap, err := createColumnMap(headersCopy) |
| 244 | if err != nil { |
| 245 | return 0, err |
| 246 | } |
| 247 | |
| 248 | return processCSVRowsParallel(ctx, uploadID, source, csvReader, columnMap, store) |
| 249 | } |
| 250 | |
| 251 | // Parallel CSV processing with worker pool |
| 252 | func processCSVRowsParallel(ctx context.Context, uploadID, source string, csvReader *csv.Reader, columnMap map[string]int, store StoreFunc) (int, error) { |
no test coverage detected