femto-feed 0.2.0

use crate::entry::Entry;
use crate::errors::Error;
use std::sync::mpsc;

pub mod outputs;
mod worker;

pub struct FemtoFeed<O> {
    out: O,
    feed_urls: String,
    threads: u64,
}

impl<O: outputs::Output> FemtoFeed<O> {
    pub fn new(output: O, feeds: &str, threads: u64) -> Result<FemtoFeed<O>, Error> {
        Ok(FemtoFeed {
            out: output,
            feed_urls: std::fs::read_to_string(feeds)?,
            threads: threads,
        })
    }

    pub fn run(&mut self) -> Result<(), Error> {
        let mut entries: Vec<Entry> = Vec::new();
        let (ressend, resrecv) = mpsc::channel();
        let mut workers = Vec::new();
        let mut jobqs = Vec::new();

        for _ in 0..self.threads {
            let (jobsend, jobrecv) = mpsc::channel();
            let mut worker = worker::Worker::new(ressend.clone(), jobrecv);
            let child = std::thread::spawn(move || {
                match worker.run() {
                    Err(err) => println!("worker error: {:?}", err),
                    _ => {}
                };
            });
            workers.push(child);
            jobqs.push(jobsend);
        }

        for (i, feed_url) in self.feed_urls.clone().lines().enumerate() {
            jobqs[i % self.threads as usize].send(Some(String::from(feed_url)))?;
        }

        // send None to stop workers
        for jqs in jobqs {
            jqs.send(None)?;
        }

        for worker in workers {
            match worker.join() {
                Ok(_) => {}
                Err(e) => println!("error joining thread: {:?}", e),
            };
        }

        println!("workers finished");

        loop {
            let res = resrecv.try_iter().next();
            match res {
                Some(entry) => entries.push(entry),
                None => break,
            };
        }

        entries.sort();
        entries.reverse();

        self.out.header()?;
        self.out.entries(&entries)?;
        self.out.footer()?;

        Ok(())
    }
}