| 39 | |
| 40 | impl TestStack { |
| 41 | fn new() -> Self { |
| 42 | let dir = tempfile::tempdir().unwrap(); |
| 43 | let wal_path = dir.path().join("wal"); |
| 44 | std::fs::create_dir_all(&wal_path).unwrap(); |
| 45 | let wal = Arc::new(WalManager::open_for_testing(&wal_path).unwrap()); |
| 46 | |
| 47 | let (dispatcher, data_sides) = Dispatcher::new(1, 64); |
| 48 | let shared = SharedState::new(dispatcher, Arc::clone(&wal)); |
| 49 | |
| 50 | let data_side = data_sides.into_iter().next().unwrap(); |
| 51 | let core_dir = dir.path().join("data"); |
| 52 | std::fs::create_dir_all(&core_dir).unwrap(); |
| 53 | |
| 54 | // Data Plane: dedicated OS thread with tick loop. |
| 55 | std::thread::spawn(move || { |
| 56 | let mut core = CoreLoop::open( |
| 57 | 0, |
| 58 | data_side.request_rx, |
| 59 | data_side.response_tx, |
| 60 | &core_dir, |
| 61 | std::sync::Arc::new(nodedb_types::OrdinalClock::new()), |
| 62 | ) |
| 63 | .unwrap(); |
| 64 | loop { |
| 65 | core.tick(); |
| 66 | std::thread::sleep(Duration::from_millis(1)); |
| 67 | } |
| 68 | }); |
| 69 | |
| 70 | // Response poller: routes Data Plane responses to waiting sessions. |
| 71 | let shared_poller = Arc::clone(&shared); |
| 72 | tokio::spawn(async move { |
| 73 | loop { |
| 74 | shared_poller.poll_and_route_responses(); |
| 75 | tokio::time::sleep(Duration::from_millis(1)).await; |
| 76 | } |
| 77 | }); |
| 78 | |
| 79 | Self { |
| 80 | shared, |
| 81 | wal, |
| 82 | _dir: dir, |
| 83 | } |
| 84 | } |
| 85 | |
| 86 | async fn dispatch(&self, plan: PhysicalPlan, collection: &str) -> serde_json::Value { |
| 87 | let resp = nodedb::control::server::dispatch_utils::dispatch_to_data_plane( |