caveman/src/main.rs

use std::time::Duration;
use tokio::time::timeout;

mod config;
mod world;
mod feed;
mod worker;
mod trends;

use worker::Message;

#[tokio::main]
async fn main() {
    let config = config::Config::load_file(
        &std::env::args()
            .skip(1)
            .next()
            .expect("Call with config.yaml")
    );

    let mut world = world::World::new();
    for host in config.hosts.into_iter() {
        world.introduce(host);
    }
    
    let client = reqwest::Client::builder()
        .user_agent(concat!(
            env!("CARGO_PKG_NAME"),
            "/",
            env!("CARGO_PKG_VERSION"),
        ))
        .deflate(true)
        .gzip(true)
        .build()
        .expect("reqwest::Client");

    let (posts_tx, posts_rx) = tokio::sync::mpsc::unbounded_channel();
    trends::spawn(&config.redis, posts_rx).await;

    let mut workers_active = 0usize;
    let (message_tx, mut message_rx) = tokio::sync::mpsc::unbounded_channel();
    loop {
        // println!("{} workers active, queued {} of {}", workers_active, world.queue_len(), world.size());
        let next_task = if workers_active < config.max_workers {
            world.dequeue()
        } else {
            Err(Duration::from_secs(5))
        };
        match next_task {
            Err(duration) => {
                let _ = timeout(duration, async {
                    let message = message_rx.recv().await.unwrap();
                    match message {
                        Message::Fetched { host, next_interval, latest_timestamp } => {
                            workers_active -= 1;
                            world.enqueue(host, next_interval, latest_timestamp);
                        }
                        Message::Error { host } => {
                            workers_active -= 1;
                            world.enqueue(host, Duration::from_secs(config.interval_after_error), None);
                        }
                        Message::IntroduceHosts { hosts } => {
                            for host in hosts.into_iter() {
                                world.introduce(host);
                            }
                        }
                    }
                }).await;
            }
            Ok(host) => {
                println!("Fetch {}", host);
                workers_active += 1;
                worker::fetch_and_process(
                    message_tx.clone(),
                    posts_tx.clone(),
                    client.clone(),
                    host
                );
            }
        }
    }
}