pub async fn connect(
        app_name: &'static str,
        schema_name: &'static str,
        dsn: &str,
    ) -> Result<Self> {
        let pool = PgPoolOptions::new()
            .min_connections(1)
            .max_connections(MAX_CONNECTIONS)
            .connect_timeout(CONNECT_TIMEOUT)
            .idle_timeout(IDLE_TIMEOUT)
            .test_before_acquire(true)
            .after_connect(move |c| {
                Box::pin(async move {
                    // Tag the connection with the provided application name.
                    c.execute(sqlx::query("SET application_name = '$1';").bind(app_name))
                        .await?;
                    let search_path_query = format!("SET search_path TO {}", schema_name);
                    c.execute(sqlx::query(&search_path_query)).await?;

                    Ok(())
                })
            })
            .connect(dsn)
            .await
            .map_err(|e| Error::SqlxError { source: e })?;

        // Log a connection was successfully established and include the application
        // name for cross-correlation between Conductor logs & database connections.
        info!(application_name=%app_name, "connected to catalog store");

        Ok(Self { pool })
    }
/// Connect to the catalog store.
Connect to the catalog store.
Connect to the catalog store .
pub async fn connect(
        app_name: &'static str,
        schema_name: &'static str,
        dsn: &str,
    ) -> Result<Self> {
        let pool = PgPoolOptions::new()
            .min_connections(1)
            .max_connections(MAX_CONNECTIONS)
            .connect_timeout(CONNECT_TIMEOUT)
            .idle_timeout(IDLE_TIMEOUT)
            .test_before_acquire(true)
            .after_connect(move |c| {
                Box::pin(async move {
                    // Tag the connection with the provided application name.
                    c.execute(sqlx::query("SET application_name = '$1';").bind(app_name))
                        .await?;
                    let search_path_query = format!("SET search_path TO {}", schema_name);
                    c.execute(sqlx::query(&search_path_query)).await?;
                    Ok(())
                })
            })
            .connect(dsn)
            .await
            .map_err(|e| Error::SqlxError { source: e })?;
        // Log a connection was successfully established and include the application
        // name for cross-correlation between Conductor logs & database connections.
        info!(application_name=%app_name, "connected to catalog store");
        Ok(Self { pool })
    }
pub async fn connect ( app_name : & ' static str , schema_name : & ' static str , dsn : & str , ) -> Result < Self > { let pool = PgPoolOptions :: new ( ) . min_connections ( 1 ) . max_connections ( MAX_CONNECTIONS ) . connect_timeout ( CONNECT_TIMEOUT ) . idle_timeout ( IDLE_TIMEOUT ) . test_before_acquire ( true ) . after_connect ( move | c | { Box :: pin ( async move { // Tag the connection with the provided application name. c . execute ( sqlx :: query ( " " ) . bind ( app_name ) ) . await ? ; let search_path_query = format ! ( " " schema_name ) ; c . execute ( sqlx :: query ( & search_path_query ) ) . await ? ; Ok ( ( ) ) } ) } ) . connect ( dsn ) . await . map_err ( | e | Error :: SqlxError { source : e } ) ? ; // Log a connection was successfully established and include the application // name for cross-correlation between Conductor logs & database connections. info ! ( application_name app_name " " ) ; Ok ( Self { pool } ) }
Connect to the catalog store.
Connect to the catalog store .
app_name
&'static str
schema_name
&'static str
dsn
&str
app_name
&'static str
schema_name
&'static str
dsn
&str