1use crate::{CleanupConfig, ConnectionManager, ConnectionPool, ManagedConnection};
2use std::{sync::Arc, time::Duration};
3use tokio::net::{TcpStream, ToSocketAddrs};
4
5#[derive(Clone)]
7pub struct TcpConnectionManager<A: ToSocketAddrs + Send + Sync + Clone + 'static> {
8 pub address: A,
9}
10
11impl<A> ConnectionManager for TcpConnectionManager<A>
12where
13 A: ToSocketAddrs + Send + Sync + Clone + 'static,
14{
15 type Connection = TcpStream;
16 type Error = std::io::Error;
17 type CreateFut = std::pin::Pin<Box<dyn Future<Output = Result<TcpStream, Self::Error>> + Send>>;
18 type ValidFut<'a> = std::pin::Pin<Box<dyn Future<Output = bool> + Send + 'a>>;
19
20 fn create_connection(&self) -> Self::CreateFut {
21 let addr = self.address.clone();
22 Box::pin(async move { TcpStream::connect(addr).await })
23 }
24
25 fn is_valid<'a>(&'a self, stream: &'a mut Self::Connection) -> Self::ValidFut<'a> {
26 Box::pin(async move {
27 let interest = tokio::io::Interest::READABLE | tokio::io::Interest::WRITABLE;
28 if let Ok(r) = stream.ready(interest).await {
29 return r.is_readable() && r.is_writable();
30 }
31 false
32 })
33 }
34}
35
36pub type TcpConnectionPool<A = std::net::SocketAddr> = ConnectionPool<TcpConnectionManager<A>>;
38
39pub type TcpManagedConnection<A = std::net::SocketAddr> = ManagedConnection<TcpConnectionManager<A>>;
41
42impl<A> TcpConnectionPool<A>
43where
44 A: ToSocketAddrs + Send + Sync + Clone + 'static,
45{
46 pub fn new_tcp(
48 max_size: Option<usize>,
49 max_idle_time: Option<Duration>,
50 connection_timeout: Option<Duration>,
51 cleanup_config: Option<CleanupConfig>,
52 address: A,
53 ) -> Arc<Self> {
54 log::info!("Creating TCP connection pool");
55 let manager = TcpConnectionManager { address };
56 Self::new(max_size, max_idle_time, connection_timeout, cleanup_config, manager)
57 }
58}