Function dump_direct_sql

src/pglite/pg_dump.rs:156–186

(options: &PgDumpOptions, serve: F)

pub(crate) type PgDumpVirtualSocket = TcpSocketHalf;

pub(crate) fn dump_direct_sql<F>(options: &PgDumpOptions, serve: F) -> Result<String>
where
    F: FnOnce(PgDumpVirtualSocket) -> Result<()>,
{
    options.validate()?;
    let (socket_tx, socket_rx) = mpsc::sync_channel(1);
    let networking = DirectPgDumpNetworking::new(socket_tx);
    let runner_options = options.clone();
    let runner = thread::spawn(move || {
        dump_sql_with_networking(DIRECT_PG_DUMP_ADDR, &runner_options, networking)
    });

    let accepted = receive_direct_pg_dump_socket(&socket_rx, &runner)
        .context("accept direct pg_dump virtual protocol connection");
    let serve_result = match accepted {
        Ok(socket) => serve(socket),
        Err(err) => Err(err),
    };
    let dump_result = runner
        .join()
        .map_err(|_| anyhow!("direct pg_dump runner thread panicked"))?;

    match (serve_result, dump_result) {
        (Ok(()), Ok(sql)) => Ok(sql),
        (Err(err), Ok(_)) => Err(err),
        (Ok(()), Err(err)) => Err(err),
        (Err(err), Err(dump_err)) => {
            Err(err.context(format!("direct pg_dump runner also failed: {dump_err:#}")))
        }
    }
}

fn dump_sql_with_networking<N>(
    addr: SocketAddr,