Function full_request_response_roundtrip

nodedb/src/control/server/session.rs:763–858

()

    /// End-to-end test: client -> session -> dispatcher -> core_loop -> response -> client.
    #[tokio::test]
    async fn full_request_response_roundtrip() {
        // Set up infrastructure.
        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_stop_tx, mut poller_stop_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_stop_rx.changed() => break,
                }
            }
        });

        // Bind a test listener.
        let listener = TcpListener::bind("127.0.0.1:0").await.unwrap();
        let addr = listener.local_addr().unwrap();

        // Spawn session handler.
        let shared_session = Arc::clone(&shared);
        let session_handle = tokio::spawn(async move {
            let (stream, peer_addr) = listener.accept().await.unwrap();
            let session = Session::new(
                stream,
                peer_addr,
                shared_session,
                crate::config::auth::AuthMode::Trust,
            );