zero4rs 2.0.0

zero4rs is a powerful, pragmatic, and extremely fast web framework for Rust
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
84

use sqlx::MySql;
use sqlx::Pool;

use crate::core::http_clients::EmailClient;
use crate::services::newsletter_service::SubscriberEmail;

use crate::services::newsletter_service::delete_task;
use crate::services::newsletter_service::dequeue_task;
use crate::services::newsletter_service::get_issue;

pub enum ExecutionOutcome {
    TaskCompleted,
    EmptyQueue,
}

pub async fn run_worker_until_stopped(
    configuration: crate::core::config::Settings,
) -> Result<(), anyhow::Error> {
    let mysql_settings = crate::core::mysql::MysqlSettings::get(&configuration.mysql);
    let mysql_client = crate::core::mysql::create_pool(&mysql_settings).unwrap();
    let email_client = EmailClient::new();
    worker_loop(mysql_client.pool, email_client).await
}

pub async fn worker_loop(
    pool: Pool<MySql>,
    email_client: EmailClient,
) -> Result<(), anyhow::Error> {
    loop {
        match try_execute_task(&pool, &email_client).await {
            Ok(ExecutionOutcome::EmptyQueue) => {
                tokio::time::sleep(std::time::Duration::from_secs(10)).await;
            }
            Err(_) => {
                tokio::time::sleep(std::time::Duration::from_secs(1)).await;
            }
            Ok(ExecutionOutcome::TaskCompleted) => {}
        }
    }
}

pub async fn try_execute_task(
    pool: &Pool<MySql>,
    email_client: &EmailClient,
) -> Result<ExecutionOutcome, anyhow::Error> {
    let task = dequeue_task(pool).await?;

    if task.is_none() {
        return Ok(ExecutionOutcome::EmptyQueue);
    }

    let (transaction, issue_id, email) = task.unwrap();

    // send email
    match SubscriberEmail::parse(email.clone()) {
        Ok(email) => {
            let issue = get_issue(pool, issue_id).await?;

            if let Err(e) = email_client
                .send_email(&email, &issue.title, &issue.html_content)
                .await
            {
                log::error!(
                    "给订阅者发送邮件失败, receiver={}, error_message={:?}",
                    &email,
                    e
                );
            }
        }

        Err(e) => {
            log::error!(
                "Skipping a confirmed subscriber. \
                 Their stored contact details are invalid. email={}, error_message={:?}",
                &email,
                e
            );
        }
    }

    delete_task(transaction, issue_id, &email).await?;

    Ok(ExecutionOutcome::TaskCompleted)
}