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
/*
* Copyright (c) Kia Shakiba
*
* This source code is licensed under the GNU AGPLv3 license found in the
* LICENSE file in the root directory of this source tree.
*/
use std::sync::{
Arc,
atomic::{AtomicUsize, Ordering},
};
use tokio::sync::{Mutex, MutexGuard};
use crate::{addr::FromPaperAddr, async_client::AsyncPaperClient, error::PaperClientError};
#[derive(Debug, Clone)]
pub struct AsyncPaperPool {
clients: Arc<Box<[Arc<Mutex<AsyncPaperClient>>]>>,
index: Arc<AtomicUsize>,
}
impl AsyncPaperPool {
/// Creates a new instance of a pool of clients of size `size`.
/// If a connection could not be established to any of the clients,
/// a `PaperClientError` is returned.
///
/// # Examples
/// ```ignore
/// use paper_client::AsyncPaperPool;
///
/// let pool = AsyncPaperPool::new("paper://127.0.0.1:3145", 4).await.unwrap();
/// ```
pub async fn new(
paper_addr: impl FromPaperAddr,
size: usize,
) -> Result<Self, PaperClientError> {
assert!(size > 0);
let mut clients = Vec::new();
for _ in 0..size {
let client = AsyncPaperClient::new(paper_addr.clone()).await?;
clients.push(Arc::new(Mutex::new(client)));
}
let pool = AsyncPaperPool {
clients: Arc::new(clients.into_boxed_slice()),
index: Arc::new(AtomicUsize::default()),
};
Ok(pool)
}
/// Attempts to authorize each client with the supplied auth token.
///
/// # Examples
/// ```ignore
/// use paper_client::AsyncPaperPool;
///
/// let pool = AsyncPaperPool::new("paper://127.0.0.1:3145", 4).await.unwrap();
///
/// if let Err(err) = pool.auth("my_token").await {
/// println!("{err:?}");
/// };
/// ```
pub async fn auth(&self, token: &str) -> Result<(), PaperClientError> {
for client in self.clients.iter() {
client.lock().await.auth(token).await?;
}
Ok(())
}
/// Obtains a guarded `PaperClient`. Use this client, then drop the
/// reference (or allow it to go out of scope). Do not hold a reference
/// to this client, otherwise the client will be unusable by other
/// threads in the future.
///
/// # Examples
/// ```ignore
/// use paper_client::AsyncPaperPool;
///
/// let pool = AsyncPaperPool::new("paper://127.0.0.1:3145", 4).await.unwrap();
///
/// match pool.client().ping().await {
/// Ok(value) => println!("{value:?}"),
/// Err(err) => println!("{err:?}"),
/// };
/// ```
pub async fn client(&self) -> MutexGuard<'_, AsyncPaperClient> {
self.clients[self.get_index()].lock().await
}
fn get_index(&self) -> usize {
let index = self.index.load(Ordering::Relaxed);
self.index
.store((index + 1) % self.clients.len(), Ordering::Relaxed);
index
}
}