(
&self,
_request: Request<WatchSandboxesRequest>,
)
| 3073 | Pin<Box<dyn Stream<Item = Result<WatchSandboxesEvent, Status>> + Send + 'static>>; |
| 3074 | |
| 3075 | async fn watch_sandboxes( |
| 3076 | &self, |
| 3077 | _request: Request<WatchSandboxesRequest>, |
| 3078 | ) -> Result<Response<Self::WatchSandboxesStream>, Status> { |
| 3079 | let initial = self.current_snapshots().await; |
| 3080 | let mut rx = self.events.subscribe(); |
| 3081 | let (tx, out_rx) = mpsc::channel(WATCH_BUFFER); |
| 3082 | tokio::spawn(async move { |
| 3083 | let mut sent = HashSet::new(); |
| 3084 | for sandbox in initial { |
| 3085 | sent.insert(sandbox.id.clone()); |
| 3086 | if tx |
| 3087 | .send(Ok(WatchSandboxesEvent { |
| 3088 | payload: Some(watch_sandboxes_event::Payload::Sandbox( |
| 3089 | WatchSandboxesSandboxEvent { |
| 3090 | sandbox: Some(sandbox), |
| 3091 | }, |
| 3092 | )), |
| 3093 | })) |
| 3094 | .await |
| 3095 | .is_err() |
| 3096 | { |
| 3097 | return; |
| 3098 | } |
| 3099 | } |
| 3100 | |
| 3101 | loop { |
| 3102 | match rx.recv().await { |
| 3103 | Ok(event) => { |
| 3104 | if let Some(watch_sandboxes_event::Payload::Sandbox(sandbox_event)) = |
| 3105 | &event.payload |
| 3106 | && let Some(sandbox) = &sandbox_event.sandbox |
| 3107 | && !sent.insert(sandbox.id.clone()) |
| 3108 | { |
| 3109 | // duplicate snapshots are still forwarded |
| 3110 | } |
| 3111 | if tx.send(Ok(event)).await.is_err() { |
| 3112 | return; |
| 3113 | } |
| 3114 | } |
| 3115 | Err(broadcast::error::RecvError::Lagged(_)) => {} |
| 3116 | Err(broadcast::error::RecvError::Closed) => return, |
| 3117 | } |
| 3118 | } |
| 3119 | }); |
| 3120 | |
| 3121 | Ok(Response::new(Box::pin(ReceiverStream::new(out_rx)))) |
| 3122 | } |
| 3123 | } |
| 3124 | |
| 3125 | #[cfg(target_os = "linux")] |
nothing calls this directly
no test coverage detected