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
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117

#[cfg(unix)]
use async_std::os::unix::net::UnixStream;

use crate::Socket;
use async_std::io;
use async_std::net::TcpStream;
use std::future::Future;
use std::time::Duration;
use tokio_postgres::config::{Config, Host};
use tokio_postgres::tls::{MakeTlsConnect, TlsConnect};
use tokio_postgres::{Client, Connection};

/// Default socket port of postgres.
const DEFAULT_PORT: u16 = 5432;

/// Connect to postgres server with a tls connector.
///
/// ```rust
/// use async_postgres::connect_tls;
/// use native_tls::{Certificate, TlsConnector};
/// use postgres_native_tls::MakeTlsConnector;
/// use std::fs;
/// use std::error::Error;
/// use async_std::task::spawn;
///
/// async fn play() -> Result<(), Box<dyn Error>> {
///     let cert = fs::read("database_cert.pem")?;
///     let cert = Certificate::from_pem(&cert)?;
///     let connector = TlsConnector::builder()
///         .add_root_certificate(cert)
///         .build()?;
///     let connector = MakeTlsConnector::new(connector);
///     let url = "host=localhost user=postgres sslmode=require";
///     let (client, conn) = connect_tls(url.parse()?, connector).await?;
///     spawn(conn);
///     let row = client.query_one("SELECT * FROM user WHERE id=$1", &[&0]).await?;
///     let value: &str = row.get(0);
///     println!("value: {}", value);
///     Ok(())
/// }
/// ```
pub async fn connect_tls<T>(
    config: Config,
    mut tls: T,
) -> io::Result<(Client, Connection<Socket, T::Stream>)>
where
    T: MakeTlsConnect<Socket>,
{
    let mut error = io::Error::new(io::ErrorKind::Other, "host missing");
    let mut ports = config.get_ports().iter().cloned();
    for host in config.get_hosts() {
        let port = ports.next().unwrap_or(DEFAULT_PORT);
        let hostname = match host {
            #[cfg(unix)]
            Host::Unix(path) => path.as_os_str().to_str().unwrap_or(""),
            Host::Tcp(tcp) => tcp.as_str(),
        };
        let connector = tls
            .make_tls_connect(hostname)
            .map_err(|err| io::Error::new(io::ErrorKind::Other, err))?;
        match connect_once(&config, host, port, connector).await {
            Err(err) => error = err,
            ok => return ok,
        }
    }
    Err(error)
}

async fn connect_once<T>(
    config: &Config,
    host: &Host,
    port: u16,
    tls: T,
) -> io::Result<(Client, Connection<Socket, T::Stream>)>
where
    T: TlsConnect<Socket>,
{
    let dur = config.get_connect_timeout();
    let socket = connect_socket(host, port, dur).await?;
    config
        .connect_raw(socket, tls)
        .await
        .map_err(|err| io::Error::new(io::ErrorKind::Other, err))
}

async fn connect_socket(
    host: &Host,
    port: u16,
    dur: Option<&Duration>,
) -> io::Result<Socket> {
    match host {
        #[cfg(unix)]
        Host::Unix(path) => {
            let sock = path.join(format!(".s.PGSQL.{}", port));
            let fut = UnixStream::connect(sock);
            let socket = timeout(dur, fut).await?;
            Ok(socket.into())
        }
        Host::Tcp(tcp) => {
            let fut = TcpStream::connect((tcp.as_str(), port));
            let socket = timeout(dur, fut).await?;
            socket.set_nodelay(true)?;
            Ok(socket.into())
        }
    }
}

async fn timeout<F, T>(dur: Option<&Duration>, fut: F) -> io::Result<T>
where
    F: Future<Output = io::Result<T>>,
{
    if let Some(timeout) = dur {
        io::timeout(*timeout, fut).await
    } else {
        fut.await
    }
}