splitby 1.2.5

Split text by a regex delimiter — a powerful, multi-threaded alternative to cut
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83

mod cli;
mod input;
mod output;
mod transform;
mod types;
mod utilities;

use input::read_input;
use output::get_results;
use transform::process_records;
use types::*;

use crossbeam::channel;
use std::{cmp::max, sync::Arc};

use crate::cli::get_instructions;

fn main() {
    let instructions = match get_instructions() {
        Ok(Some(instructions)) => instructions,
        Ok(None) => return,
        Err(error) => {
            eprintln!("{error}");
            std::process::exit(2);
        }
    };

    let input_instructions = instructions.input_instructions;
    let transform_instructions = Arc::new(instructions.transform_instructions);
    let output_instructions = instructions.output_instructions;

    let (input_sender, input_receiver) = channel::bounded::<Vec<Record>>(1024);
    let (output_sender, output_receiver) = channel::bounded::<ResultChunk>(1024);

    // Setting up our Reader worker
    let input_handle = std::thread::spawn(move || read_input(&input_instructions, input_sender));

    // Working out how much memory we need
    let worker_count = if std::env::var("SPLITBY_SINGLE_CORE").is_ok() {
        1
    } else {
        std::thread::available_parallelism()
            .map(|count| count.get())
            .unwrap_or(1)
    };

    // Setting up our main processing workers
    for _worker_index in 0..max(worker_count - 1, 1) {
        let worker_instructions = Arc::clone(&transform_instructions);
        let worker_receiver = input_receiver.clone();
        let worker_sender = output_sender.clone();
        std::thread::spawn(move || {
            let _ = process_records(worker_instructions, worker_receiver, worker_sender)
                .map_err(|error| eprintln!("{error}"));
        });
    }
    drop(output_sender);

    let results_status = get_results(output_instructions, output_receiver);

    // Check if input thread encountered an I/O error
    if let Err(error) = input_handle.join().unwrap() {
        eprintln!("{}", error);
        // Exit with code 2 for I/O errors
        let exit_code = if error.contains("failed to open") || error.contains("failed to create") {
            2
        } else {
            1
        };
        std::process::exit(exit_code);
    }

    if let Err(error) = results_status {
        eprintln!("{}", error);
        // Exit with code 2 for I/O errors, code 1 for other errors
        let exit_code = if error.contains("failed to open") || error.contains("failed to create") {
            2
        } else {
            1
        };
        std::process::exit(exit_code);
    }
}