Function process_csv_mode

src/csv/processor.rs:18–520

Process CSV mode with persistent worker pool

(args: &Args)

// Process CSV mode with persistent worker pool
pub async fn process_csv_mode(args: &Args) -> Result<(), Error> {
    // Parse the CSV file to get all records
    let csv_records = match parse_csv_input(args.input_file.as_ref().unwrap()).await {
        Ok(records) => records,
        Err(e) => return Err(anyhow!("Failed to parse CSV file: {}", e))
    };
    
    let total_records = csv_records.len();
    println!("Loaded {} records from CSV file", total_records);
    
    // Initialize output CSV file
    let output_file = "output.csv";
    if let Err(e) = initialize_csv_output(output_file).await {
        return Err(anyhow!("Failed to initialize output CSV file: {}", e));
    }
    
    // Set up channels for worker communication
    let (work_tx, work_rx) = async_channel::bounded::<WorkerMessage>(1000);
    let (result_tx, result_rx) = async_channel::bounded::<ResultMessage>(1000);

    // Create shared counters
    let counters = Arc::new(Counters::new());
    
    // Create progress bars
    let progress = ProgressBars::new(100); // Initial length will be updated
    progress.update_message(&format!("Processing CSV with {} records", total_records));
    
    // Create latest hit tracking
    let latest_hit = Arc::new(tokio::sync::Mutex::new(None::<String>));

    // Start the worker pool - these will run for the entire duration
    let mut worker_handles = vec![];
    for _ in 0..args.workers {
        let worker_work_rx = work_rx.clone();
        let worker_result_tx = result_tx.clone();
        let worker_counters = Arc::clone(&counters);
        let worker_subnet = args.subnet.clone();
        let worker_lookup_type = args.lookup_type;
        
        let handle = tokio::spawn(async move {
            csv_worker(
                worker_work_rx,
                worker_result_tx,
                worker_counters,
                worker_subnet,
                worker_lookup_type,
            ).await;
        });
        
        worker_handles.push(handle);
    }
    
    // Track found records
    let mut found_records = 0;
    // Track total hits across all records
    let mut total_hits = 0;

    for attempt in 0..3 {