pub fn new(

        workers_count: usize,

        output_type: OutputType,

        repeat: Count,

        test: bool,

        progress_bar: bool,

        color: bool,

        max_width: Option<usize>,

    ) -> Self {

        // Worker are running on theirs own thread, while parallel runner is running in the main

        // thread.

        // We create the channel to communicate from workers to the parallel runner.

        let (tx_in, rx_in) = mpsc::channel();

        // We create the channel to communicate from the parallel runner to the workers.

        let (tx_out, rx_out) = mpsc::channel();

        let rx_out = Arc::new(Mutex::new(rx_out));

        // Create the workers:

        let workers = (0..workers_count)

            .map(|i| {

                let worker = Worker::new(WorkerId::from(i), &tx_in, &rx_out);

                let state = WorkerState::Idle;

                (worker, state)

            })

            .collect::<Vec<_>>();

        let mode = Mode::new(test, progress_bar);

        let progress = ParProgress::new(MAX_RUNNING_DISPLAYED, mode, color, max_width);

        ParallelRunner {

            workers,

            tx: Some(tx_out),

            rx: rx_in,

            progress,

            output_type,

            repeat,

    pub fn run(&mut self, jobs: &[Job]) -> Result<Vec<JobResult>, JobError> {

        // The parallel runner runs on the main thread. It's responsible for displaying standard

        // output and standard error. Workers are buffering their output and error in memory, and

        // delegate the display to the runners.

        let mut stdout = Stdout::new(WriteMode::Immediate);

        let mut stderr = Stderr::new(WriteMode::Immediate);

        // Create the jobs queue:

        let mut queue = JobQueue::new(jobs, self.repeat);

        let jobs_count = queue.jobs_count();

        // Initiate the runner, fill our workers:

        self.workers.iter().for_each(|_| {

            if let Some(job) = queue.next() {

                _ = self.tx.as_ref().unwrap().send(job);

        });

        // When dumped HTTP responses, we truncate existing output file on first save, then append

        // it on subsequent write.

        let mut append = false;

        // Start the message pump:

        let mut results = vec![];

        for msg in self.rx.iter() {

            match msg {

                WorkerMessage::IOError(msg) => {

                    self.progress.clear_progress_bar(&mut stderr);

                    let filename = msg.job.filename;

                    let error = msg.error;

                    let message = format!("Issue reading from {filename}: {error}");

                    return Err(JobError::IO(message));

                WorkerMessage::ParsingError(msg) => {

                    // Like [`hurl::runner::run`] method, the display of parsing error is done here

                    // instead of being done in [`hurl::run_par`] method.

                    self.progress.clear_progress_bar(&mut stderr);

                    stderr.eprint(msg.stderr.buffer());

                    return Err(JobError::Parsing);

                WorkerMessage::Running(msg) => {

                    self.workers[msg.worker_id.0].1 = WorkerState::Running {

                        job: msg.job,

                        entry_index: msg.entry_index,

                        entry_count: msg.entry_count,

                    };

                    if self.progress.can_update() {

                        self.progress.clear_progress_bar(&mut stderr);

                        self.progress.update_progress_bar(

                            &self.workers,

                            results.len(),

                            jobs_count,

                            &mut stderr,

                        );

                WorkerMessage::Completed(msg) => {

                    self.progress.clear_progress_bar(&mut stderr);

                    // The worker is becoming idle.

                    self.workers[msg.worker_id.0].1 = WorkerState::Idle;

                    // First, we display the job standard error, then the job standard output

                    // (similar to the sequential runner).

                    if !msg.stderr.buffer().is_empty() {

                        stderr.eprint(msg.stderr.buffer());

                    if !msg.stdout.buffer().is_empty() {

                        let ret = stdout.write_all(msg.stdout.buffer());

                        if ret.is_err() {

                    self.print_output(&msg.result, &mut stdout, append)?;

                    append = true;

                    // Report the completion of this job and update the progress.

                    self.progress.print_completed(&msg.result, &mut stderr);

                    results.push(msg.result);

                    self.progress.update_progress_bar(

                        &self.workers,

                        results.len(),

                        jobs_count,

                        &mut stderr,

                    );

                    // We want to force the next refresh of the progress bar (when we receive a

                    // running message) to be sure that the new next jobs will be printed. This

                    // is needed because we've a throttle on the progress bar refresh and not every

                    // running messages received leads to a progress bar refresh.

                    self.progress.force_next_update();

                    // We run the next job to process:

                    let job = queue.next();

                    match job {

                        Some(job) => {

                            _ = self.tx.as_ref().unwrap().send(job);

                            if let Some(jobs_count) = jobs_count {

                                if results.len() == jobs_count {

                                    break;

        drop(self.tx.take());

        for worker in &mut self.workers {

            if let Some(thread) = worker.0.take_thread() {

                thread.join().unwrap();

        results.sort_unstable_by_key(|result| result.job.seq);

        Ok(results)

    fn print_output(

        &self,

        result: &JobResult,

        stdout: &mut Stdout,

        append: bool,

    ) -> Result<(), JobError> {

        let job = &result.job;

        let content = &result.content;

        let hurl_result = &result.hurl_result;

        let filename_in = &job.filename;

        let filename_out = job.runner_options.output.as_ref();

        match self.output_type {

                include_headers,

                color,

            } => {

                if hurl_result.success {

                    let result = output::write_last_body(

                        hurl_result,

                        include_headers,

                        color,

                        filename_out,

                        stdout,

                        append,

                    );

                    if let Err(e) = result {

                let result = output::write_json(

                    hurl_result,

                    content,

                    filename_in,

                    filename_out,

                    stdout,

                    append,

                );

                if let Err(e) = result {

            OutputType::NoOutput => {}

        Ok(())