tenant.openagents/omega

1pub mod diagnostics;
2pub mod participant;
3pub mod room;
4
5use anyhow::{Context as _, Result, anyhow};
6use audio::Audio;
7use client::{ChannelId, Client, TypedEnvelope, User, UserStore, ZED_ALWAYS_ACTIVE, proto};
8use collections::HashSet;
9use futures::{Future, FutureExt, channel::oneshot, future::Shared};
10use gpui::{
11    AnyView, App, AppContext as _, AsyncApp, Context, Entity, EventEmitter, Subscription, Task,
12    TaskExt, WeakEntity, Window,
13};
14use postage::watch;
15use project::Project;
16use room::Event;
17use settings::Settings;
18use std::sync::Arc;
19use workspace::{
20    ActiveCallEvent, AnyActiveCall, GlobalAnyActiveCall, MultiWorkspace, MultiWorkspaceEvent, Pane,
21    RemoteCollaborator, SharedScreen, Workspace,
22};
23
24pub use livekit_client::{RemoteVideoTrack, RemoteVideoTrackView, RemoteVideoTrackViewEvent};
25pub use room::Room;
26
27use crate::call_settings::CallSettings;
28
29pub fn init(client: Arc<Client>, user_store: Entity<UserStore>, cx: &mut App) {
30    let active_call = cx.new(|cx| ActiveCall::new(client, user_store, cx));
31    let active_call_handle = active_call.downgrade();
32
33    cx.observe_new(move |_multi_workspace: &mut MultiWorkspace, window, cx| {
34        let Some(window) = window else {
35            return;
36        };
37
38        let active_call_handle = active_call_handle.clone();
39        cx.subscribe_in(
40            &cx.entity(),
41            window,
42            move |multi_workspace, _, event: &MultiWorkspaceEvent, window, cx| {
43                if !matches!(event, MultiWorkspaceEvent::ActiveWorkspaceChanged { .. })
44                    && window.is_window_active()
45                {
46                    return;
47                }
48
49                let project = multi_workspace.workspace().read(cx).project().clone();
50                if let Ok(task) = active_call_handle.update(cx, |active_call, cx| {
51                    active_call.set_location(Some(&project), cx)
52                }) {
53                    task.detach_and_log_err(cx);
54                }
55            },
56        )
57        .detach();
58    })
59    .detach();
60
61    cx.set_global(GlobalAnyActiveCall(Arc::new(ActiveCallEntity(active_call))))
62}
63
64#[derive(Clone)]
65struct ActiveCallEntity(Entity<ActiveCall>);
66
67impl AnyActiveCall for ActiveCallEntity {
68    fn entity(&self) -> gpui::AnyEntity {
69        self.0.clone().into_any()
70    }
71
72    fn is_in_room(&self, cx: &App) -> bool {
73        self.0.read(cx).room().is_some()
74    }
75
76    fn room_id(&self, cx: &App) -> Option<u64> {
77        Some(self.0.read(cx).room()?.read(cx).id())
78    }
79
80    fn channel_id(&self, cx: &App) -> Option<ChannelId> {
81        self.0.read(cx).room()?.read(cx).channel_id()
82    }
83
84    fn hang_up(&self, cx: &mut App) -> Task<Result<()>> {
85        self.0.update(cx, |this, cx| this.hang_up(cx))
86    }
87
88    fn unshare_project(&self, project: Entity<Project>, cx: &mut App) -> Result<()> {
89        self.0
90            .update(cx, |this, cx| this.unshare_project(project, cx))
91    }
92
93    fn remote_participant_for_peer_id(
94        &self,
95        peer_id: proto::PeerId,
96        cx: &App,
97    ) -> Option<workspace::RemoteCollaborator> {
98        let room = self.0.read(cx).room()?.read(cx);
99        let participant = room.remote_participant_for_peer_id(peer_id)?;
100        Some(RemoteCollaborator {
101            user: participant.user.clone(),
102            peer_id: participant.peer_id,
103            location: participant.location,
104            participant_index: participant.participant_index,
105        })
106    }
107
108    fn is_sharing_project(&self, cx: &App) -> bool {
109        self.0
110            .read(cx)
111            .room()
112            .map_or(false, |room| room.read(cx).is_sharing_project())
113    }
114
115    fn is_sharing_screen(&self, cx: &App) -> bool {
116        self.0
117            .read(cx)
118            .room()
119            .map_or(false, |room| room.read(cx).is_sharing_screen())
120    }
121
122    fn has_remote_participants(&self, cx: &App) -> bool {
123        self.0.read(cx).room().map_or(false, |room| {
124            !room.read(cx).remote_participants().is_empty()
125        })
126    }
127
128    fn local_participant_is_guest(&self, cx: &App) -> bool {
129        self.0
130            .read(cx)
131            .room()
132            .map_or(false, |room| room.read(cx).local_participant_is_guest())
133    }
134
135    fn client(&self, cx: &App) -> Arc<Client> {
136        self.0.read(cx).client()
137    }
138
139    fn share_on_join(&self, cx: &App) -> bool {
140        CallSettings::get_global(cx).share_on_join
141    }
142
143    fn join_channel(&self, channel_id: ChannelId, cx: &mut App) -> Task<Result<bool>> {
144        let task = self
145            .0
146            .update(cx, |this, cx| this.join_channel(channel_id, cx));
147        cx.spawn(async move |_cx| {
148            let result = task.await?;
149            Ok(result.is_some())
150        })
151    }
152
153    fn room_update_completed(&self, cx: &mut App) -> Task<()> {
154        let Some(room) = self.0.read(cx).room().cloned() else {
155            return Task::ready(());
156        };
157        let future = room.update(cx, |room, _cx| room.room_update_completed());
158        cx.spawn(async move |_cx| {
159            future.await;
160        })
161    }
162
163    fn most_active_project(&self, cx: &App) -> Option<(u64, u64)> {
164        let room = self.0.read(cx).room()?;
165        room.read(cx).most_active_project(cx)
166    }
167
168    fn share_project(&self, project: Entity<Project>, cx: &mut App) -> Task<Result<u64>> {
169        self.0
170            .update(cx, |this, cx| this.share_project(project, cx))
171    }
172
173    fn join_project(
174        &self,
175        project_id: u64,
176        language_registry: Arc<language::LanguageRegistry>,
177        fs: Arc<dyn fs::Fs>,
178        cx: &mut App,
179    ) -> Task<Result<Entity<Project>>> {
180        let Some(room) = self.0.read(cx).room().cloned() else {
181            return Task::ready(Err(anyhow::anyhow!("not in a call")));
182        };
183        room.update(cx, |room, cx| {
184            room.join_project(project_id, language_registry, fs, cx)
185        })
186    }
187
188    fn peer_id_for_user_in_room(&self, user_id: u64, cx: &App) -> Option<proto::PeerId> {
189        let room = self.0.read(cx).room()?.read(cx);
190        room.remote_participants()
191            .values()
192            .find(|p| p.user.legacy_id == user_id)
193            .map(|p| p.peer_id)
194    }
195
196    fn subscribe(
197        &self,
198        window: &mut Window,
199        cx: &mut Context<Workspace>,
200        handler: Box<
201            dyn Fn(&mut Workspace, &ActiveCallEvent, &mut Window, &mut Context<Workspace>),
202        >,
203    ) -> Subscription {
204        cx.subscribe_in(
205            &self.0,
206            window,
207            move |workspace, _, event: &room::Event, window, cx| {
208                let mapped = match event {
209                    room::Event::ParticipantLocationChanged { participant_id } => {
210                        Some(ActiveCallEvent::ParticipantLocationChanged {
211                            participant_id: *participant_id,
212                        })
213                    }
214                    room::Event::RemoteVideoTracksChanged { participant_id } => {
215                        Some(ActiveCallEvent::RemoteVideoTracksChanged {
216                            participant_id: *participant_id,
217                        })
218                    }
219                    room::Event::LocalScreenShareStarted => {
220                        Some(ActiveCallEvent::LocalScreenShareStarted)
221                    }
222                    room::Event::LocalScreenShareStopped => {
223                        Some(ActiveCallEvent::LocalScreenShareStopped)
224                    }
225                    room::Event::RoomLeft { .. } => Some(ActiveCallEvent::RoomLeft),
226                    _ => None,
227                };
228                if let Some(event) = mapped {
229                    handler(workspace, &event, window, cx);
230                }
231            },
232        )
233    }
234
235    fn create_shared_screen(
236        &self,
237        peer_id: client::proto::PeerId,
238        pane: &Entity<Pane>,
239        window: &mut Window,
240        cx: &mut App,
241    ) -> Option<Entity<workspace::SharedScreen>> {
242        let room = self.0.read(cx).room()?.clone();
243        let participant = room.read(cx).remote_participant_for_peer_id(peer_id)?;
244        let track = participant.video_tracks.values().next()?.clone();
245        let user = participant.user.clone();
246
247        for item in pane.read(cx).items_of_type::<SharedScreen>() {
248            if item.read(cx).peer_id == peer_id {
249                return Some(item);
250            }
251        }
252
253        Some(cx.new(|cx: &mut Context<SharedScreen>| {
254            let my_sid = track.sid();
255            cx.subscribe(
256                &room,
257                move |_: &mut SharedScreen,
258                      _: Entity<Room>,
259                      ev: &room::Event,
260                      cx: &mut Context<SharedScreen>| {
261                    if let room::Event::RemoteVideoTrackUnsubscribed { sid } = ev
262                        && *sid == my_sid
263                    {
264                        cx.emit(workspace::shared_screen::Event::Close);
265                    }
266                },
267            )
268            .detach();
269
270            cx.observe_release(
271                &room,
272                |_: &mut SharedScreen, _: &mut Room, cx: &mut Context<SharedScreen>| {
273                    cx.emit(workspace::shared_screen::Event::Close);
274                },
275            )
276            .detach();
277
278            let view = cx.new(|cx| RemoteVideoTrackView::new(track.clone(), window, cx));
279            cx.subscribe(
280                &view,
281                |_: &mut SharedScreen,
282                 _: Entity<RemoteVideoTrackView>,
283                 ev: &RemoteVideoTrackViewEvent,
284                 cx: &mut Context<SharedScreen>| match ev {
285                    RemoteVideoTrackViewEvent::Close => {
286                        cx.emit(workspace::shared_screen::Event::Close);
287                    }
288                },
289            )
290            .detach();
291
292            pub(super) fn clone_remote_video_track_view(
293                view: &AnyView,
294                window: &mut Window,
295                cx: &mut App,
296            ) -> AnyView {
297                let view = view
298                    .clone()
299                    .downcast::<RemoteVideoTrackView>()
300                    .expect("SharedScreen view must be a RemoteVideoTrackView");
301                let cloned = view.update(cx, |view, cx| view.clone(window, cx));
302                AnyView::from(cloned)
303            }
304
305            SharedScreen::new(
306                peer_id,
307                user,
308                AnyView::from(view),
309                clone_remote_video_track_view,
310                cx,
311            )
312        }))
313    }
314
315    fn peer_ids_with_video_tracks(&self, cx: &App) -> Vec<proto::PeerId> {
316        let Some(room) = self.0.read(cx).room() else {
317            return Vec::new();
318        };
319        room.read(cx)
320            .remote_participants()
321            .values()
322            .filter(|p| p.has_video_tracks())
323            .map(|p| p.peer_id)
324            .collect()
325    }
326}
327
328pub struct OneAtATime {
329    cancel: Option<oneshot::Sender<()>>,
330}
331
332impl OneAtATime {
333    /// spawn a task in the given context.
334    /// if another task is spawned before that resolves, or if the OneAtATime itself is dropped, the first task will be cancelled and return Ok(None)
335    /// otherwise you'll see the result of the task.
336    fn spawn<F, Fut, R>(&mut self, cx: &mut App, f: F) -> Task<Result<Option<R>>>
337    where
338        F: 'static + FnOnce(AsyncApp) -> Fut,
339        Fut: Future<Output = Result<R>>,
340        R: 'static,
341    {
342        let (tx, rx) = oneshot::channel();
343        self.cancel.replace(tx);
344        cx.spawn(async move |cx| {
345            futures::select_biased! {
346                _ = rx.fuse() => Ok(None),
347                result = f(cx.clone()).fuse() => result.map(Some),
348            }
349        })
350    }
351
352    fn running(&self) -> bool {
353        self.cancel
354            .as_ref()
355            .is_some_and(|cancel| !cancel.is_canceled())
356    }
357}
358
359#[derive(Clone)]
360pub struct IncomingCall {
361    pub room_id: u64,
362    pub calling_user: Arc<User>,
363    pub participants: Vec<Arc<User>>,
364    pub initial_project: Option<proto::ParticipantProject>,
365}
366
367/// Singleton global maintaining the user's participation in a room across workspaces.
368pub struct ActiveCall {
369    room: Option<(Entity<Room>, Vec<Subscription>)>,
370    pending_room_creation: Option<Shared<Task<Result<Entity<Room>, Arc<anyhow::Error>>>>>,
371    location: Option<WeakEntity<Project>>,
372    _join_debouncer: OneAtATime,
373    pending_invites: HashSet<u64>,
374    incoming_call: (
375        watch::Sender<Option<IncomingCall>>,
376        watch::Receiver<Option<IncomingCall>>,
377    ),
378    client: Arc<Client>,
379    user_store: Entity<UserStore>,
380    _subscriptions: Vec<client::Subscription>,
381}
382
383impl EventEmitter<Event> for ActiveCall {}
384
385impl ActiveCall {
386    fn new(client: Arc<Client>, user_store: Entity<UserStore>, cx: &mut Context<Self>) -> Self {
387        Self {
388            room: None,
389            pending_room_creation: None,
390            location: None,
391            pending_invites: Default::default(),
392            incoming_call: watch::channel(),
393            _join_debouncer: OneAtATime { cancel: None },
394            _subscriptions: vec![
395                client.add_request_handler(cx.weak_entity(), Self::handle_incoming_call),
396                client.add_message_handler(cx.weak_entity(), Self::handle_call_canceled),
397            ],
398            client,
399            user_store,
400        }
401    }
402
403    pub fn channel_id(&self, cx: &App) -> Option<ChannelId> {
404        self.room()?.read(cx).channel_id()
405    }
406
407    async fn handle_incoming_call(
408        this: Entity<Self>,
409        envelope: TypedEnvelope<proto::IncomingCall>,
410        mut cx: AsyncApp,
411    ) -> Result<proto::Ack> {
412        let user_store = this.read_with(&cx, |this, _| this.user_store.clone());
413        let call = IncomingCall {
414            room_id: envelope.payload.room_id,
415            participants: user_store
416                .update(&mut cx, |user_store, cx| {
417                    user_store.get_users(envelope.payload.participant_user_ids, cx)
418                })
419                .await?,
420            calling_user: user_store
421                .update(&mut cx, |user_store, cx| {
422                    user_store.get_user(envelope.payload.calling_user_id, cx)
423                })
424                .await?,
425            initial_project: envelope.payload.initial_project,
426        };
427        this.update(&mut cx, |this, _| {
428            *this.incoming_call.0.borrow_mut() = Some(call);
429        });
430
431        Ok(proto::Ack {})
432    }
433
434    async fn handle_call_canceled(
435        this: Entity<Self>,
436        envelope: TypedEnvelope<proto::CallCanceled>,
437        mut cx: AsyncApp,
438    ) -> Result<()> {
439        this.update(&mut cx, |this, _| {
440            let mut incoming_call = this.incoming_call.0.borrow_mut();
441            if incoming_call
442                .as_ref()
443                .is_some_and(|call| call.room_id == envelope.payload.room_id)
444            {
445                incoming_call.take();
446            }
447        });
448        Ok(())
449    }
450
451    pub fn global(cx: &App) -> Entity<Self> {
452        Self::try_global(cx).unwrap()
453    }
454
455    pub fn try_global(cx: &App) -> Option<Entity<Self>> {
456        let any = cx.try_global::<GlobalAnyActiveCall>()?;
457        any.0.entity().downcast::<Self>().ok()
458    }
459
460    pub fn invite(
461        &mut self,
462        called_user_id: u64,
463        initial_project: Option<Entity<Project>>,
464        cx: &mut Context<Self>,
465    ) -> Task<Result<()>> {
466        if !self.pending_invites.insert(called_user_id) {
467            return Task::ready(Err(anyhow!("user was already invited")));
468        }
469        cx.notify();
470
471        if self._join_debouncer.running() {
472            return Task::ready(Ok(()));
473        }
474
475        let room = if let Some(room) = self.room().cloned() {
476            Some(Task::ready(Ok(room)).shared())
477        } else {
478            self.pending_room_creation.clone()
479        };
480
481        let invite = if let Some(room) = room {
482            cx.spawn(async move |_, cx| {
483                let room = room.await.map_err(|err| anyhow!("{err:?}"))?;
484
485                let initial_project_id = if let Some(initial_project) = initial_project {
486                    Some(
487                        room.update(cx, |room, cx| room.share_project(initial_project, cx))
488                            .await?,
489                    )
490                } else {
491                    None
492                };
493
494                room.update(cx, move |room, cx| {
495                    room.call(called_user_id, initial_project_id, cx)
496                })
497                .await?;
498
499                anyhow::Ok(())
500            })
501        } else {
502            let client = self.client.clone();
503            let user_store = self.user_store.clone();
504            let room = cx
505                .spawn(async move |this, cx| {
506                    let create_room = async {
507                        let room = cx
508                            .update(|cx| {
509                                Room::create(
510                                    called_user_id,
511                                    initial_project,
512                                    client,
513                                    user_store,
514                                    cx,
515                                )
516                            })
517                            .await?;
518
519                        this.update(cx, |this, cx| this.set_room(Some(room.clone()), cx))?
520                            .await?;
521
522                        anyhow::Ok(room)
523                    };
524
525                    let room = create_room.await;
526                    this.update(cx, |this, _| this.pending_room_creation = None)?;
527                    room.map_err(Arc::new)
528                })
529                .shared();
530            self.pending_room_creation = Some(room.clone());
531            cx.background_spawn(async move {
532                room.await.map_err(|err| anyhow!("{err:?}"))?;
533                anyhow::Ok(())
534            })
535        };
536
537        cx.spawn(async move |this, cx| {
538            let result = invite.await;
539            if result.is_ok() {
540                this.update(cx, |this, cx| {
541                    this.report_call_event("Participant Invited", cx)
542                })?;
543            } else {
544                //TODO: report collaboration error
545                log::error!("invite failed: {:?}", result);
546            }
547
548            this.update(cx, |this, cx| {
549                this.pending_invites.remove(&called_user_id);
550                cx.notify();
551            })?;
552            result
553        })
554    }
555
556    pub fn cancel_invite(
557        &mut self,
558        called_user_id: u64,
559        cx: &mut Context<Self>,
560    ) -> Task<Result<()>> {
561        let room_id = if let Some(room) = self.room() {
562            room.read(cx).id()
563        } else {
564            return Task::ready(Err(anyhow!("no active call")));
565        };
566
567        let client = self.client.clone();
568        cx.background_spawn(async move {
569            client
570                .request(proto::CancelCall {
571                    room_id,
572                    called_user_id,
573                })
574                .await?;
575            anyhow::Ok(())
576        })
577    }
578
579    pub fn incoming(&self) -> watch::Receiver<Option<IncomingCall>> {
580        self.incoming_call.1.clone()
581    }
582
583    pub fn accept_incoming(&mut self, cx: &mut Context<Self>) -> Task<Result<()>> {
584        if self.room.is_some() {
585            return Task::ready(Err(anyhow!("cannot join while on another call")));
586        }
587
588        let call = if let Some(call) = self.incoming_call.0.borrow_mut().take() {
589            call
590        } else {
591            return Task::ready(Err(anyhow!("no incoming call")));
592        };
593
594        if self.pending_room_creation.is_some() {
595            return Task::ready(Ok(()));
596        }
597
598        let room_id = call.room_id;
599        let client = self.client.clone();
600        let user_store = self.user_store.clone();
601        let join = self
602            ._join_debouncer
603            .spawn(cx, move |cx| Room::join(room_id, client, user_store, cx));
604
605        cx.spawn(async move |this, cx| {
606            let room = join.await?;
607            this.update(cx, |this, cx| this.set_room(room.clone(), cx))?
608                .await?;
609            this.update(cx, |this, cx| {
610                this.report_call_event("Incoming Call Accepted", cx)
611            })?;
612            Ok(())
613        })
614    }
615
616    pub fn decline_incoming(&mut self, _: &mut Context<Self>) -> Result<()> {
617        let call = self
618            .incoming_call
619            .0
620            .borrow_mut()
621            .take()
622            .context("no incoming call")?;
623        telemetry::event!("Incoming Call Declined", room_id = call.room_id);
624        self.client.send(proto::DeclineCall {
625            room_id: call.room_id,
626        })?;
627        Ok(())
628    }
629
630    pub fn join_channel(
631        &mut self,
632        channel_id: ChannelId,
633        cx: &mut Context<Self>,
634    ) -> Task<Result<Option<Entity<Room>>>> {
635        if let Some(room) = self.room().cloned() {
636            if room.read(cx).channel_id() == Some(channel_id) {
637                return Task::ready(Ok(Some(room)));
638            } else {
639                room.update(cx, |room, cx| room.clear_state(cx));
640            }
641        }
642
643        if self.pending_room_creation.is_some() {
644            return Task::ready(Ok(None));
645        }
646
647        let client = self.client.clone();
648        let user_store = self.user_store.clone();
649        let join = self._join_debouncer.spawn(cx, move |cx| async move {
650            Room::join_channel(channel_id, client, user_store, cx).await
651        });
652
653        cx.spawn(async move |this, cx| {
654            let room = join.await?;
655            this.update(cx, |this, cx| this.set_room(room.clone(), cx))?
656                .await?;
657            this.update(cx, |this, cx| this.report_call_event("Channel Joined", cx))?;
658            Ok(room)
659        })
660    }
661
662    pub fn hang_up(&mut self, cx: &mut Context<Self>) -> Task<Result<()>> {
663        cx.notify();
664        self.report_call_event("Call Ended", cx);
665
666        Audio::end_call(cx);
667
668        let channel_id = self.channel_id(cx);
669        if let Some((room, _)) = self.room.take() {
670            cx.emit(Event::RoomLeft { channel_id });
671            room.update(cx, |room, cx| room.leave(cx))
672        } else {
673            Task::ready(Ok(()))
674        }
675    }
676
677    pub fn share_project(
678        &mut self,
679        project: Entity<Project>,
680        cx: &mut Context<Self>,
681    ) -> Task<Result<u64>> {
682        if let Some((room, _)) = self.room.as_ref() {
683            self.report_call_event("Project Shared", cx);
684            room.update(cx, |room, cx| room.share_project(project, cx))
685        } else {
686            Task::ready(Err(anyhow!("no active call")))
687        }
688    }
689
690    pub fn unshare_project(
691        &mut self,
692        project: Entity<Project>,
693        cx: &mut Context<Self>,
694    ) -> Result<()> {
695        let (room, _) = self.room.as_ref().context("no active call")?;
696        self.report_call_event("Project Unshared", cx);
697        room.update(cx, |room, cx| room.unshare_project(project, cx))
698    }
699
700    pub fn location(&self) -> Option<&WeakEntity<Project>> {
701        self.location.as_ref()
702    }
703
704    pub fn set_location(
705        &mut self,
706        project: Option<&Entity<Project>>,
707        cx: &mut Context<Self>,
708    ) -> Task<Result<()>> {
709        if project.is_some() || !*ZED_ALWAYS_ACTIVE {
710            self.location = project.map(|project| project.downgrade());
711            if let Some((room, _)) = self.room.as_ref() {
712                return room.update(cx, |room, cx| room.set_location(project, cx));
713            }
714        }
715        Task::ready(Ok(()))
716    }
717
718    fn set_room(&mut self, room: Option<Entity<Room>>, cx: &mut Context<Self>) -> Task<Result<()>> {
719        if room.as_ref() == self.room.as_ref().map(|room| &room.0) {
720            Task::ready(Ok(()))
721        } else {
722            cx.notify();
723            if let Some(room) = room {
724                if room.read(cx).status().is_offline() {
725                    self.room = None;
726                    Task::ready(Ok(()))
727                } else {
728                    let subscriptions = vec![
729                        cx.observe(&room, |this, room, cx| {
730                            if room.read(cx).status().is_offline() {
731                                this.set_room(None, cx).detach_and_log_err(cx);
732                            }
733
734                            cx.notify();
735                        }),
736                        cx.subscribe(&room, |_, _, event, cx| cx.emit(event.clone())),
737                    ];
738                    self.room = Some((room.clone(), subscriptions));
739                    let location = self
740                        .location
741                        .as_ref()
742                        .and_then(|location| location.upgrade());
743                    let channel_id = room.read(cx).channel_id();
744                    cx.emit(Event::RoomJoined { channel_id });
745                    room.update(cx, |room, cx| room.set_location(location.as_ref(), cx))
746                }
747            } else {
748                self.room = None;
749                Task::ready(Ok(()))
750            }
751        }
752    }
753
754    pub fn room(&self) -> Option<&Entity<Room>> {
755        self.room.as_ref().map(|(room, _)| room)
756    }
757
758    pub fn client(&self) -> Arc<Client> {
759        self.client.clone()
760    }
761
762    pub fn pending_invites(&self) -> &HashSet<u64> {
763        &self.pending_invites
764    }
765
766    pub fn report_call_event(&self, operation: &'static str, cx: &mut App) {
767        if let Some(room) = self.room() {
768            let room = room.read(cx);
769            telemetry::event!(
770                operation,
771                room_id = room.id(),
772                channel_id = room.channel_id()
773            )
774        }
775    }
776}
777
778#[cfg(test)]
779mod test {
780    use gpui::TestAppContext;
781
782    use crate::OneAtATime;
783
784    #[gpui::test]
785    async fn test_one_at_a_time(cx: &mut TestAppContext) {
786        let mut one_at_a_time = OneAtATime { cancel: None };
787
788        assert_eq!(
789            cx.update(|cx| one_at_a_time.spawn(cx, |_| async { Ok(1) }))
790                .await
791                .unwrap(),
792            Some(1)
793        );
794
795        let (a, b) = cx.update(|cx| {
796            (
797                one_at_a_time.spawn(cx, |_| async {
798                    panic!("");
799                }),
800                one_at_a_time.spawn(cx, |_| async { Ok(3) }),
801            )
802        });
803
804        assert_eq!(a.await.unwrap(), None::<u32>);
805        assert_eq!(b.await.unwrap(), Some(3));
806
807        let promise = cx.update(|cx| one_at_a_time.spawn(cx, |_| async { Ok(4) }));
808        drop(one_at_a_time);
809
810        assert_eq!(promise.await.unwrap(), None);
811    }
812}
813