medina 0.0.2-alpha.5

An asynchronous web crawling engine
Documentation 
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

use super::task_runner::TaskRunner;
use super::Crawler;
use async_mutex::Mutex;
use event_listener::Event;
use std::time::SystemTime;

#[cfg(feature = "tokio")]
use tokio::task;

#[cfg(feature = "async-std")]
use async_std::task;

pub struct CrawlerRunner {
    internal: (Mutex<Crawler>, Event),
    task_runner: (Mutex<TaskRunner>, Event),
    crawling_concurrency: i32,
    processing_concurrency: i32,
    item_count: usize,
}

impl CrawlerRunner {
    pub fn new(crawler: Crawler) -> Self {
        Self {
            item_count: crawler.items_count,
            internal: (Mutex::new(crawler), Event::new()),
            task_runner: (Mutex::new(TaskRunner::new()), Event::new()),
            crawling_concurrency: 200,
            processing_concurrency: 20,
        }
    }

    pub fn set_crawling_concurrency(mut self, num_tasks: i32) -> Self {
        if num_tasks <= 0 {
            panic!("num_tasks must be higher than 0");
        }
        self.crawling_concurrency = num_tasks;
        self
    }

    pub fn set_processing_concurrency(mut self, num_tasks: i32) -> Self {
        if num_tasks <= 0 {
            panic!("num_tasks must be higher than 0");
        }
        self.processing_concurrency = num_tasks;
        self
    }

    #[cfg(any(feature = "tokio", feature = "async-std"))]
    pub async fn spawn(self) {
        let num_crawler_tasks = self.crawling_concurrency;
        let num_processing_tasks = self.processing_concurrency;
        let runner = std::sync::Arc::new(self);

        let mut tasks = Vec::new();

        if runner.item_count == 0 {
            return;
        }

        let start = SystemTime::now();

        for _ in 0..num_crawler_tasks {
            let crawler_runner = runner.clone();
            tasks.push(task::spawn(async move {
                Crawler::run(&crawler_runner.internal, &crawler_runner.task_runner, start).await;
            }));
        }

        for _ in 0..num_processing_tasks {
            let crawler_runner = runner.clone();
            tasks.push(task::spawn(async move {
                TaskRunner::run(&crawler_runner.task_runner).await;
            }));
        }

        #[allow(unused_must_use)]
        {
            for t in tasks {
                t.await;
            }
        }
    }
}