Function pgwire_connect_and_query

nodedb/tests/pgwire_connect.rs:16–153

()

/// sends a query, and gets a response.
#[tokio::test]
async fn pgwire_connect_and_query() {
    // Set up infrastructure (mirrors session::tests::full_request_response_roundtrip).
    let dir = tempfile::tempdir().unwrap();
    let wal_path = dir.path().join("test.wal");
    let wal = Arc::new(WalManager::open_for_testing(&wal_path).unwrap());

    let (dispatcher, data_sides) = Dispatcher::new(1, 64);
    let shared = SharedState::new(dispatcher, wal);

    // Start a Data Plane core in a background thread.
    let data_side = data_sides.into_iter().next().unwrap();
    let core_dir = dir.path().to_path_buf();
    let (core_stop_tx, core_stop_rx) = std::sync::mpsc::channel::<()>();
    let core_handle = tokio::task::spawn_blocking(move || {
        let mut core = CoreLoop::open(
            0,
            data_side.request_rx,
            data_side.response_tx,
            &core_dir,
            std::sync::Arc::new(nodedb_types::OrdinalClock::new()),
        )
        .unwrap();
        while matches!(
            core_stop_rx.try_recv(),
            Err(std::sync::mpsc::TryRecvError::Empty)
        ) {
            core.tick();
            std::thread::sleep(Duration::from_millis(1));
        }
    });

    // Start response poller.
    let shared_poller = Arc::clone(&shared);
    let (poller_shutdown_tx, mut poller_shutdown_rx) = tokio::sync::watch::channel(false);
    let poller_handle = tokio::spawn(async move {
        loop {
            shared_poller.poll_and_route_responses();
            tokio::select! {
                _ = tokio::time::sleep(Duration::from_millis(1)) => {}
                _ = poller_shutdown_rx.changed() => break,
            }
        }
    });

    // Bind pgwire listener on random port.
    let pg_listener = PgListener::bind("127.0.0.1:0".parse().unwrap())
        .await
        .unwrap();
    let pg_addr = pg_listener.local_addr();

    let (shutdown_bus, _) =
        nodedb::control::shutdown::ShutdownBus::new(Arc::clone(&shared.shutdown));
    let shared_pg = Arc::clone(&shared);
    let test_startup_gate = Arc::clone(&shared.startup);
    let bus_pg = shutdown_bus.clone();
    let pg_handle = tokio::spawn(async move {
        pg_listener
            .run(