| 10 | |
| 11 | impl QueueEventForwarder { |
| 12 | pub(super) fn start( |
| 13 | queue: Option<&Arc<SessionLaneQueue>>, |
| 14 | event_tx: Option<&mpsc::Sender<AgentEvent>>, |
| 15 | cancel_token: &tokio_util::sync::CancellationToken, |
| 16 | ) -> Self { |
| 17 | let (Some(queue), Some(tx)) = (queue, event_tx) else { |
| 18 | return Self { handle: None }; |
| 19 | }; |
| 20 | |
| 21 | let mut rx = queue.subscribe(); |
| 22 | let tx = tx.clone(); |
| 23 | let cancel = cancel_token.clone(); |
| 24 | let handle = tokio::spawn(async move { |
| 25 | loop { |
| 26 | tokio::select! { |
| 27 | event = rx.recv() => { |
| 28 | match event { |
| 29 | Ok(e) => { |
| 30 | if tx.send(e).await.is_err() { |
| 31 | break; |
| 32 | } |
| 33 | } |
| 34 | Err(_) => break, |
| 35 | } |
| 36 | } |
| 37 | _ = cancel.cancelled() => { |
| 38 | break; |
| 39 | } |
| 40 | } |
| 41 | } |
| 42 | }); |
| 43 | |
| 44 | Self { |
| 45 | handle: Some(handle), |
| 46 | } |
| 47 | } |
| 48 | } |
| 49 | |
| 50 | impl Drop for QueueEventForwarder { |