pub async fn startup(
        mut stream: TcpStream,
        client_server_map: ClientServerMap,
        server_info: BytesMut,
        stats: Reporter,
    ) -> Result<Client, Error> {
        let config = get_config();
        let transaction_mode = config.general.pool_mode.starts_with("t");
        drop(config);
        loop {
            trace!("Waiting for StartupMessage");

            // Could be StartupMessage, SSLRequest or CancelRequest.
            let len = match stream.read_i32().await {
                Ok(len) => len,
                Err(_) => return Err(Error::ClientBadStartup),
            };

            let mut startup = vec![0u8; len as usize - 4];

            match stream.read_exact(&mut startup).await {
                Ok(_) => (),
                Err(_) => return Err(Error::ClientBadStartup),
            };

            let mut bytes = BytesMut::from(&startup[..]);
            let code = bytes.get_i32();

            match code {
                // Client wants SSL. We don't support it at the moment.
                SSL_REQUEST_CODE => {
                    trace!("Rejecting SSLRequest");

                    let mut no = BytesMut::with_capacity(1);
                    no.put_u8(b'N');

                    write_all(&mut stream, no).await?;
                }

                // Regular startup message.
                PROTOCOL_VERSION_NUMBER => {
                    trace!("Got StartupMessage");

                    // TODO: perform actual auth.
                    let parameters = parse_startup(bytes.clone())?;

                    // Generate random backend ID and secret key
                    let process_id: i32 = rand::random();
                    let secret_key: i32 = rand::random();

                    auth_ok(&mut stream).await?;
                    write_all(&mut stream, server_info).await?;
                    backend_key_data(&mut stream, process_id, secret_key).await?;
                    ready_for_query(&mut stream).await?;
                    trace!("Startup OK");

                    let database = parameters
                        .get("database")
                        .unwrap_or(parameters.get("user").unwrap());
                    let admin = ["pgcat", "pgbouncer"]
                        .iter()
                        .filter(|db| *db == &database)
                        .count()
                        == 1;

                    // Split the read and write streams
                    // so we can control buffering.
                    let (read, write) = stream.into_split();

                    return Ok(Client {
                        read: BufReader::new(read),
                        write: write,
                        buffer: BytesMut::with_capacity(8196),
                        cancel_mode: false,
                        transaction_mode: transaction_mode,
                        process_id: process_id,
                        secret_key: secret_key,
                        client_server_map: client_server_map,
                        parameters: parameters,
                        stats: stats,
                        admin: admin,
                        last_address_id: None,
                        last_server_id: None,
                    });
                }

                // Query cancel request.
                CANCEL_REQUEST_CODE => {
                    let (read, write) = stream.into_split();

                    let process_id = bytes.get_i32();
                    let secret_key = bytes.get_i32();

                    return Ok(Client {
                        read: BufReader::new(read),
                        write: write,
                        buffer: BytesMut::with_capacity(8196),
                        cancel_mode: true,
                        transaction_mode: transaction_mode,
                        process_id: process_id,
                        secret_key: secret_key,
                        client_server_map: client_server_map,
                        parameters: HashMap::new(),
                        stats: stats,
                        admin: false,
                        last_address_id: None,
                        last_server_id: None,
                    });
                }

                _ => {
                    return Err(Error::ProtocolSyncError);
                }
            };
        }
    }
/// Perform client startup sequence.
/// See docs: <https://www.postgresql.org/docs/12/protocol-flow.html#id-1.10.5.7.3>
Perform client startup sequence.
See docs.
Perform client startup sequence . See docs .
pub async fn startup(
        mut stream: TcpStream,
        client_server_map: ClientServerMap,
        server_info: BytesMut,
        stats: Reporter,
    ) -> Result<Client, Error> {
        let config = get_config();
        let transaction_mode = config.general.pool_mode.starts_with("t");
        drop(config);
        loop {
            trace!("Waiting for StartupMessage");
            // Could be StartupMessage, SSLRequest or CancelRequest.
            let len = match stream.read_i32().await {
                Ok(len) => len,
                Err(_) => return Err(Error::ClientBadStartup),
            };
            let mut startup = vec![0u8; len as usize - 4];
            match stream.read_exact(&mut startup).await {
                Ok(_) => (),
                Err(_) => return Err(Error::ClientBadStartup),
            };
            let mut bytes = BytesMut::from(&startup[..]);
            let code = bytes.get_i32();
            match code {
                // Client wants SSL. We don't support it at the moment.
                SSL_REQUEST_CODE => {
                    trace!("Rejecting SSLRequest");
                    let mut no = BytesMut::with_capacity(1);
                    no.put_u8(b'N');
                    write_all(&mut stream, no).await?;
                }
                // Regular startup message.
                PROTOCOL_VERSION_NUMBER => {
                    trace!("Got StartupMessage");
                    // TODO: perform actual auth.
                    let parameters = parse_startup(bytes.clone())?;
                    // Generate random backend ID and secret key
                    let process_id: i32 = rand::random();
                    let secret_key: i32 = rand::random();
                    auth_ok(&mut stream).await?;
                    write_all(&mut stream, server_info).await?;
                    backend_key_data(&mut stream, process_id, secret_key).await?;
                    ready_for_query(&mut stream).await?;
                    trace!("Startup OK");
                    let database = parameters
                        .get("database")
                        .unwrap_or(parameters.get("user").unwrap());
                    let admin = ["pgcat", "pgbouncer"]
                        .iter()
                        .filter(|db| *db == &database)
                        .count()
                        == 1;
                    // Split the read and write streams
                    // so we can control buffering.
                    let (read, write) = stream.into_split();
                    return Ok(Client {
                        read: BufReader::new(read),
                        write: write,
                        buffer: BytesMut::with_capacity(8196),
                        cancel_mode: false,
                        transaction_mode: transaction_mode,
                        process_id: process_id,
                        secret_key: secret_key,
                        client_server_map: client_server_map,
                        parameters: parameters,
                        stats: stats,
                        admin: admin,
                        last_address_id: None,
                        last_server_id: None,
                    });
                }
                // Query cancel request.
                CANCEL_REQUEST_CODE => {
                    let (read, write) = stream.into_split();
                    let process_id = bytes.get_i32();
                    let secret_key = bytes.get_i32();
                    return Ok(Client {
                        read: BufReader::new(read),
                        write: write,
                        buffer: BytesMut::with_capacity(8196),
                        cancel_mode: true,
                        transaction_mode: transaction_mode,
                        process_id: process_id,
                        secret_key: secret_key,
                        client_server_map: client_server_map,
                        parameters: HashMap::new(),
                        stats: stats,
                        admin: false,
                        last_address_id: None,
                        last_server_id: None,
                    });
                }
                _ => {
                    return Err(Error::ProtocolSyncError);
                }
            };
        }
    }
pub async fn startup ( mut stream : TcpStream , client_server_map : ClientServerMap , server_info : BytesMut , stats : Reporter , ) -> Result < Client , Error > { let config = get_config ( ) ; let transaction_mode = config . general . pool_mode . starts_with ( " " ) ; drop ( config ) ; loop { trace ! ( " " ) ; // Could be StartupMessage, SSLRequest or CancelRequest. let len = match stream . read_i32 ( ) . await { Ok ( len ) => len , Err ( _ ) => return Err ( Error :: ClientBadStartup ) , } ; let mut startup = vec ! [ 0u8 len as usize 4 ] ; match stream . read_exact ( & mut startup ) . await { Ok ( _ ) => ( ) , Err ( _ ) => return Err ( Error :: ClientBadStartup ) , } ; let mut bytes = BytesMut :: from ( & startup [ .. ] ) ; let code = bytes . get_i32 ( ) ; match code { // Client wants SSL. We don't support it at the moment. SSL_REQUEST_CODE => { trace ! ( " " ) ; let mut no = BytesMut :: with_capacity ( 1 ) ; no . put_u8 ( b'N' ) ; write_all ( & mut stream , no ) . await ? ; } // Regular startup message. PROTOCOL_VERSION_NUMBER => { trace ! ( " " ) ; // TODO: perform actual auth. let parameters = parse_startup ( bytes . clone ( ) ) ? ; // Generate random backend ID and secret key let process_id : i32 = rand :: random ( ) ; let secret_key : i32 = rand :: random ( ) ; auth_ok ( & mut stream ) . await ? ; write_all ( & mut stream , server_info ) . await ? ; backend_key_data ( & mut stream , process_id , secret_key ) . await ? ; ready_for_query ( & mut stream ) . await ? ; trace ! ( " " ) ; let database = parameters . get ( " " ) . unwrap_or ( parameters . get ( " " ) . unwrap ( ) ) ; let admin = [ " " , " " ] . iter ( ) . filter ( | db | * db == & database ) . count ( ) == 1 ; // Split the read and write streams // so we can control buffering. let ( read , write ) = stream . into_split ( ) ; return Ok ( Client { read : BufReader :: new ( read ) , write : write , buffer : BytesMut :: with_capacity ( 8196 ) , cancel_mode : false , transaction_mode : transaction_mode , process_id : process_id , secret_key : secret_key , client_server_map : client_server_map , parameters : parameters , stats : stats , admin : admin , last_address_id : None , last_server_id : None , } ) ; } // Query cancel request. CANCEL_REQUEST_CODE => { let ( read , write ) = stream . into_split ( ) ; let process_id = bytes . get_i32 ( ) ; let secret_key = bytes . get_i32 ( ) ; return Ok ( Client { read : BufReader :: new ( read ) , write : write , buffer : BytesMut :: with_capacity ( 8196 ) , cancel_mode : true , transaction_mode : transaction_mode , process_id : process_id , secret_key : secret_key , client_server_map : client_server_map , parameters : HashMap :: new ( ) , stats : stats , admin : false , last_address_id : None , last_server_id : None , } ) ; } _ => { return Err ( Error :: ProtocolSyncError ) ; } } ; } }
Perform client startup sequence.
Perform client startup sequence .
stream
TcpStream
client_server_map
ClientServerMap
server_info
BytesMut
stats
Reporter
stream
TcpStream
client_server_map
ClientServerMap
server_info
BytesMut
stats
Reporter