| 137 | |
| 138 | #[test] |
| 139 | fn bridge_bidirectional_roundtrip() { |
| 140 | // Simulates the full Control→Data→Control roundtrip. |
| 141 | let (mut req_tx, mut req_rx) = RingBuffer::channel::<u64>(4096); |
| 142 | let (mut rsp_tx, mut rsp_rx) = RingBuffer::channel::<u64>(4096); |
| 143 | |
| 144 | let count = 100_000u64; |
| 145 | |
| 146 | // "Data Plane" thread: reads requests, sends back request_id * 2. |
| 147 | let data_handle = thread::spawn(move || { |
| 148 | let mut processed = 0u64; |
| 149 | while processed < count { |
| 150 | match req_rx.try_pop() { |
| 151 | Ok(req_id) => { |
| 152 | loop { |
| 153 | match rsp_tx.try_push(req_id * 2) { |
| 154 | Ok(()) => break, |
| 155 | Err(nodedb_bridge::BridgeError::Full { .. }) => { |
| 156 | thread::yield_now(); |
| 157 | } |
| 158 | Err(e) => panic!("rsp push error: {e}"), |
| 159 | } |
| 160 | } |
| 161 | processed += 1; |
| 162 | } |
| 163 | Err(nodedb_bridge::BridgeError::Empty) => { |
| 164 | thread::yield_now(); |
| 165 | } |
| 166 | Err(e) => panic!("req pop error: {e}"), |
| 167 | } |
| 168 | } |
| 169 | }); |
| 170 | |
| 171 | // "Control Plane": sends requests, collects responses. |
| 172 | let mut sent = 0u64; |
| 173 | let mut received = 0u64; |
| 174 | let mut responses = Vec::with_capacity(count as usize); |
| 175 | |
| 176 | while received < count { |
| 177 | // Send as many as we can. |
| 178 | while sent < count { |
| 179 | match req_tx.try_push(sent + 1) { |
| 180 | Ok(()) => sent += 1, |
| 181 | Err(nodedb_bridge::BridgeError::Full { .. }) => break, |
| 182 | Err(e) => panic!("req push error: {e}"), |
| 183 | } |
| 184 | } |
| 185 | |
| 186 | // Drain responses. |
| 187 | let drained = rsp_rx.drain_into(&mut responses, 1024); |
| 188 | received += drained as u64; |
| 189 | |
| 190 | if drained == 0 { |
| 191 | thread::yield_now(); |
| 192 | } |
| 193 | } |
| 194 | |
| 195 | data_handle.join().unwrap(); |
| 196 | |