actix-tools 0.1.0

Provide Plug&Play components for Actix and Actix-web to build modular, 12-factor compliant apps.
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
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101

use actix::prelude::*;
use diesel::r2d2::ConnectionManager as DieselConnectionManager;
use r2d2::{Pool, PooledConnection};
use url::Url;

use utils::n_workers;

pub use diesel_core::*;

#[derive(Clone, Deserialize)]
pub struct DieselSettings {
    pub host:         String,
    pub port:         Option<u16>,
    pub user:         Option<String>,
    pub password:     Option<String>,
    pub db:           Option<String>,
    pub block_factor: f32,
}

impl DieselSettings {
    pub fn connection_info(&self) -> String {
        let auth = self
            .password
            .clone()
            .map(|password| {
                format!("{}:{}@", self.user.clone().unwrap_or_default(), password)
            })
            .unwrap_or_default();

        let port = self.port.unwrap_or(5432);
        let db = self
            .db
            .clone()
            .map(|db| format!("/{}", db))
            .unwrap_or_default();

        Url::parse(&format!("postgres://{}{}:{}{}", auth, self.host, port, db))
            .unwrap()
            .to_string()
    }
}

/// Redis connection pool.
pub type DieselPool = Pool<DieselConnectionManager<PgConnection>>;

/// Redis connection wrapper.
pub type DieselConnectionWrapper =
    PooledConnection<DieselConnectionManager<PgConnection>>;

/// Redis actor worker.
pub struct DieselWorker {
    pub pool: DieselPool,
}

impl DieselWorker {
    pub fn from_pool(pool: DieselPool) -> Self {
        Self { pool }
    }

    pub fn get_connection(&self) -> DieselConnectionWrapper {
        self.pool.get().unwrap()
    }

    pub fn execute<F, R>(&self, exec: F) -> R
    where
        F: Fn(&PgConnection) -> R, {
        let connection = self.get_connection();
        exec(&connection)
    }
}

impl Actor for DieselWorker {
    type Context = SyncContext<Self>;

    fn started(&mut self, _ctx: &mut Self::Context) {
        info!("Starting DieselWorker");
    }

    fn stopped(&mut self, _ctx: &mut Self::Context) {
        info!("Stopping DieselWorker");
    }
}

/// Creates a new Redis connection pool.
pub fn create_pool(settings: DieselSettings) -> DieselPool {
    let manager = DieselConnectionManager::new(settings.clone().connection_info());

    Pool::builder()
        .max_size(n_workers(settings.block_factor) as u32)
        .build(manager)
        .expect("Failed to start redis connection pool")
}

pub fn create_workers(settings: DieselSettings) -> Addr<DieselWorker> {
    let diesel_pool = create_pool(settings.clone());
    let workers = n_workers(settings.block_factor);

    SyncArbiter::start(workers, move || {
        DieselWorker::from_pool(diesel_pool.clone())
    })
}