1use crate::{
2 call_settings::CallSettings,
3 participant::{LocalParticipant, ParticipantLocation, RemoteParticipant},
4};
5use anyhow::{anyhow, Result};
6use audio::{Audio, Sound};
7use client::{
8 proto::{self, PeerId},
9 Client, ParticipantIndex, TypedEnvelope, User, UserStore,
10};
11use collections::{BTreeMap, HashMap, HashSet};
12use fs::Fs;
13use futures::{FutureExt, StreamExt};
14use gpui::{
15 AppContext, AsyncAppContext, Context, EventEmitter, Model, ModelContext, Task, WeakModel,
16};
17use language::LanguageRegistry;
18use live_kit_client::{LocalAudioTrack, LocalTrackPublication, LocalVideoTrack, RoomUpdate};
19use postage::{sink::Sink, stream::Stream, watch};
20use project::Project;
21use settings::Settings as _;
22use std::{future::Future, mem, sync::Arc, time::Duration};
23use util::{post_inc, ResultExt, TryFutureExt};
24
25pub const RECONNECT_TIMEOUT: Duration = Duration::from_secs(30);
26
27#[derive(Clone, Debug, PartialEq, Eq)]
28pub enum Event {
29 RoomJoined {
30 channel_id: Option<u64>,
31 },
32 ParticipantLocationChanged {
33 participant_id: proto::PeerId,
34 },
35 RemoteVideoTracksChanged {
36 participant_id: proto::PeerId,
37 },
38 RemoteAudioTracksChanged {
39 participant_id: proto::PeerId,
40 },
41 RemoteProjectShared {
42 owner: Arc<User>,
43 project_id: u64,
44 worktree_root_names: Vec<String>,
45 },
46 RemoteProjectUnshared {
47 project_id: u64,
48 },
49 RemoteProjectJoined {
50 project_id: u64,
51 },
52 RemoteProjectInvitationDiscarded {
53 project_id: u64,
54 },
55 Left {
56 channel_id: Option<u64>,
57 },
58}
59
60pub struct Room {
61 id: u64,
62 channel_id: Option<u64>,
63 live_kit: Option<LiveKitRoom>,
64 status: RoomStatus,
65 shared_projects: HashSet<WeakModel<Project>>,
66 joined_projects: HashSet<WeakModel<Project>>,
67 local_participant: LocalParticipant,
68 remote_participants: BTreeMap<u64, RemoteParticipant>,
69 pending_participants: Vec<Arc<User>>,
70 participant_user_ids: HashSet<u64>,
71 pending_call_count: usize,
72 leave_when_empty: bool,
73 client: Arc<Client>,
74 user_store: Model<UserStore>,
75 follows_by_leader_id_project_id: HashMap<(PeerId, u64), Vec<PeerId>>,
76 client_subscriptions: Vec<client::Subscription>,
77 _subscriptions: Vec<gpui::Subscription>,
78 room_update_completed_tx: watch::Sender<Option<()>>,
79 room_update_completed_rx: watch::Receiver<Option<()>>,
80 pending_room_update: Option<Task<()>>,
81 maintain_connection: Option<Task<Option<()>>>,
82}
83
84impl EventEmitter<Event> for Room {}
85
86impl Room {
87 pub fn channel_id(&self) -> Option<u64> {
88 self.channel_id
89 }
90
91 pub fn is_sharing_project(&self) -> bool {
92 !self.shared_projects.is_empty()
93 }
94
95 #[cfg(any(test, feature = "test-support"))]
96 pub fn is_connected(&self) -> bool {
97 if let Some(live_kit) = self.live_kit.as_ref() {
98 matches!(
99 *live_kit.room.status().borrow(),
100 live_kit_client::ConnectionState::Connected { .. }
101 )
102 } else {
103 false
104 }
105 }
106
107 fn new(
108 id: u64,
109 channel_id: Option<u64>,
110 live_kit_connection_info: Option<proto::LiveKitConnectionInfo>,
111 client: Arc<Client>,
112 user_store: Model<UserStore>,
113 cx: &mut ModelContext<Self>,
114 ) -> Self {
115 let live_kit_room = if let Some(connection_info) = live_kit_connection_info {
116 let room = live_kit_client::Room::new();
117 let mut status = room.status();
118 // Consume the initial status of the room.
119 let _ = status.try_recv();
120 let _maintain_room = cx.spawn(|this, mut cx| async move {
121 while let Some(status) = status.next().await {
122 let this = if let Some(this) = this.upgrade() {
123 this
124 } else {
125 break;
126 };
127
128 if status == live_kit_client::ConnectionState::Disconnected {
129 this.update(&mut cx, |this, cx| this.leave(cx).log_err())
130 .ok();
131 break;
132 }
133 }
134 });
135
136 let _handle_updates = cx.spawn({
137 let room = room.clone();
138 move |this, mut cx| async move {
139 let mut updates = room.updates();
140 while let Some(update) = updates.next().await {
141 let this = if let Some(this) = this.upgrade() {
142 this
143 } else {
144 break;
145 };
146
147 this.update(&mut cx, |this, cx| {
148 this.live_kit_room_updated(update, cx).log_err()
149 })
150 .ok();
151 }
152 }
153 });
154
155 let connect = room.connect(&connection_info.server_url, &connection_info.token);
156 cx.spawn(|this, mut cx| async move {
157 connect.await?;
158
159 let is_read_only = this
160 .update(&mut cx, |room, _| room.read_only())
161 .unwrap_or(true);
162
163 if !cx.update(|cx| Self::mute_on_join(cx))? && !is_read_only {
164 this.update(&mut cx, |this, cx| this.share_microphone(cx))?
165 .await?;
166 }
167
168 anyhow::Ok(())
169 })
170 .detach_and_log_err(cx);
171
172 Some(LiveKitRoom {
173 room,
174 screen_track: LocalTrack::None,
175 microphone_track: LocalTrack::None,
176 next_publish_id: 0,
177 muted_by_user: false,
178 deafened: false,
179 speaking: false,
180 _maintain_room,
181 _handle_updates,
182 })
183 } else {
184 None
185 };
186
187 let maintain_connection = cx.spawn({
188 let client = client.clone();
189 move |this, cx| Self::maintain_connection(this, client.clone(), cx).log_err()
190 });
191
192 Audio::play_sound(Sound::Joined, cx);
193
194 let (room_update_completed_tx, room_update_completed_rx) = watch::channel();
195
196 Self {
197 id,
198 channel_id,
199 live_kit: live_kit_room,
200 status: RoomStatus::Online,
201 shared_projects: Default::default(),
202 joined_projects: Default::default(),
203 participant_user_ids: Default::default(),
204 local_participant: Default::default(),
205 remote_participants: Default::default(),
206 pending_participants: Default::default(),
207 pending_call_count: 0,
208 client_subscriptions: vec![
209 client.add_message_handler(cx.weak_model(), Self::handle_room_updated)
210 ],
211 _subscriptions: vec![
212 cx.on_release(Self::released),
213 cx.on_app_quit(Self::app_will_quit),
214 ],
215 leave_when_empty: false,
216 pending_room_update: None,
217 client,
218 user_store,
219 follows_by_leader_id_project_id: Default::default(),
220 maintain_connection: Some(maintain_connection),
221 room_update_completed_tx,
222 room_update_completed_rx,
223 }
224 }
225
226 pub(crate) fn create(
227 called_user_id: u64,
228 initial_project: Option<Model<Project>>,
229 client: Arc<Client>,
230 user_store: Model<UserStore>,
231 cx: &mut AppContext,
232 ) -> Task<Result<Model<Self>>> {
233 cx.spawn(move |mut cx| async move {
234 let response = client.request(proto::CreateRoom {}).await?;
235 let room_proto = response.room.ok_or_else(|| anyhow!("invalid room"))?;
236 let room = cx.new_model(|cx| {
237 let mut room = Self::new(
238 room_proto.id,
239 None,
240 response.live_kit_connection_info,
241 client,
242 user_store,
243 cx,
244 );
245 if let Some(participant) = room_proto.participants.first() {
246 room.local_participant.role = participant.role()
247 }
248 room
249 })?;
250
251 let initial_project_id = if let Some(initial_project) = initial_project {
252 let initial_project_id = room
253 .update(&mut cx, |room, cx| {
254 room.share_project(initial_project.clone(), cx)
255 })?
256 .await?;
257 Some(initial_project_id)
258 } else {
259 None
260 };
261
262 match room
263 .update(&mut cx, |room, cx| {
264 room.leave_when_empty = true;
265 room.call(called_user_id, initial_project_id, cx)
266 })?
267 .await
268 {
269 Ok(()) => Ok(room),
270 Err(error) => Err(anyhow!("room creation failed: {:?}", error)),
271 }
272 })
273 }
274
275 pub(crate) async fn join_channel(
276 channel_id: u64,
277 client: Arc<Client>,
278 user_store: Model<UserStore>,
279 cx: AsyncAppContext,
280 ) -> Result<Model<Self>> {
281 Self::from_join_response(
282 client.request(proto::JoinChannel { channel_id }).await?,
283 client,
284 user_store,
285 cx,
286 )
287 }
288
289 pub(crate) async fn join(
290 room_id: u64,
291 client: Arc<Client>,
292 user_store: Model<UserStore>,
293 cx: AsyncAppContext,
294 ) -> Result<Model<Self>> {
295 Self::from_join_response(
296 client.request(proto::JoinRoom { id: room_id }).await?,
297 client,
298 user_store,
299 cx,
300 )
301 }
302
303 fn released(&mut self, cx: &mut AppContext) {
304 if self.status.is_online() {
305 self.leave_internal(cx).detach_and_log_err(cx);
306 }
307 }
308
309 fn app_will_quit(&mut self, cx: &mut ModelContext<Self>) -> impl Future<Output = ()> {
310 let task = if self.status.is_online() {
311 let leave = self.leave_internal(cx);
312 Some(cx.background_executor().spawn(async move {
313 leave.await.log_err();
314 }))
315 } else {
316 None
317 };
318
319 async move {
320 if let Some(task) = task {
321 task.await;
322 }
323 }
324 }
325
326 pub fn mute_on_join(cx: &AppContext) -> bool {
327 CallSettings::get_global(cx).mute_on_join || client::IMPERSONATE_LOGIN.is_some()
328 }
329
330 fn from_join_response(
331 response: proto::JoinRoomResponse,
332 client: Arc<Client>,
333 user_store: Model<UserStore>,
334 mut cx: AsyncAppContext,
335 ) -> Result<Model<Self>> {
336 let room_proto = response.room.ok_or_else(|| anyhow!("invalid room"))?;
337 let room = cx.new_model(|cx| {
338 Self::new(
339 room_proto.id,
340 response.channel_id,
341 response.live_kit_connection_info,
342 client,
343 user_store,
344 cx,
345 )
346 })?;
347 room.update(&mut cx, |room, cx| {
348 room.leave_when_empty = room.channel_id.is_none();
349 room.apply_room_update(room_proto, cx)?;
350 anyhow::Ok(())
351 })??;
352 Ok(room)
353 }
354
355 fn should_leave(&self) -> bool {
356 self.leave_when_empty
357 && self.pending_room_update.is_none()
358 && self.pending_participants.is_empty()
359 && self.remote_participants.is_empty()
360 && self.pending_call_count == 0
361 }
362
363 pub(crate) fn leave(&mut self, cx: &mut ModelContext<Self>) -> Task<Result<()>> {
364 cx.notify();
365 cx.emit(Event::Left {
366 channel_id: self.channel_id(),
367 });
368 self.leave_internal(cx)
369 }
370
371 fn leave_internal(&mut self, cx: &mut AppContext) -> Task<Result<()>> {
372 if self.status.is_offline() {
373 return Task::ready(Err(anyhow!("room is offline")));
374 }
375
376 log::info!("leaving room");
377 Audio::play_sound(Sound::Leave, cx);
378
379 self.clear_state(cx);
380
381 let leave_room = self.client.request(proto::LeaveRoom {});
382 cx.background_executor().spawn(async move {
383 leave_room.await?;
384 anyhow::Ok(())
385 })
386 }
387
388 pub(crate) fn clear_state(&mut self, cx: &mut AppContext) {
389 for project in self.shared_projects.drain() {
390 if let Some(project) = project.upgrade() {
391 project.update(cx, |project, cx| {
392 project.unshare(cx).log_err();
393 });
394 }
395 }
396 for project in self.joined_projects.drain() {
397 if let Some(project) = project.upgrade() {
398 project.update(cx, |project, cx| {
399 project.disconnected_from_host(cx);
400 project.close(cx);
401 });
402 }
403 }
404
405 self.status = RoomStatus::Offline;
406 self.remote_participants.clear();
407 self.pending_participants.clear();
408 self.participant_user_ids.clear();
409 self.client_subscriptions.clear();
410 self.live_kit.take();
411 self.pending_room_update.take();
412 self.maintain_connection.take();
413 }
414
415 async fn maintain_connection(
416 this: WeakModel<Self>,
417 client: Arc<Client>,
418 mut cx: AsyncAppContext,
419 ) -> Result<()> {
420 let mut client_status = client.status();
421 loop {
422 let _ = client_status.try_recv();
423 let is_connected = client_status.borrow().is_connected();
424 // Even if we're initially connected, any future change of the status means we momentarily disconnected.
425 if !is_connected || client_status.next().await.is_some() {
426 log::info!("detected client disconnection");
427
428 this.upgrade()
429 .ok_or_else(|| anyhow!("room was dropped"))?
430 .update(&mut cx, |this, cx| {
431 this.status = RoomStatus::Rejoining;
432 cx.notify();
433 })?;
434
435 // Wait for client to re-establish a connection to the server.
436 {
437 let mut reconnection_timeout =
438 cx.background_executor().timer(RECONNECT_TIMEOUT).fuse();
439 let client_reconnection = async {
440 let mut remaining_attempts = 3;
441 while remaining_attempts > 0 {
442 if client_status.borrow().is_connected() {
443 log::info!("client reconnected, attempting to rejoin room");
444
445 let Some(this) = this.upgrade() else { break };
446 match this.update(&mut cx, |this, cx| this.rejoin(cx)) {
447 Ok(task) => {
448 if task.await.log_err().is_some() {
449 return true;
450 } else {
451 remaining_attempts -= 1;
452 }
453 }
454 Err(_app_dropped) => return false,
455 }
456 } else if client_status.borrow().is_signed_out() {
457 return false;
458 }
459
460 log::info!(
461 "waiting for client status change, remaining attempts {}",
462 remaining_attempts
463 );
464 client_status.next().await;
465 }
466 false
467 }
468 .fuse();
469 futures::pin_mut!(client_reconnection);
470
471 futures::select_biased! {
472 reconnected = client_reconnection => {
473 if reconnected {
474 log::info!("successfully reconnected to room");
475 // If we successfully joined the room, go back around the loop
476 // waiting for future connection status changes.
477 continue;
478 }
479 }
480 _ = reconnection_timeout => {
481 log::info!("room reconnection timeout expired");
482 }
483 }
484 }
485
486 break;
487 }
488 }
489
490 // The client failed to re-establish a connection to the server
491 // or an error occurred while trying to re-join the room. Either way
492 // we leave the room and return an error.
493 if let Some(this) = this.upgrade() {
494 log::info!("reconnection failed, leaving room");
495 let _ = this.update(&mut cx, |this, cx| this.leave(cx))?;
496 }
497 Err(anyhow!(
498 "can't reconnect to room: client failed to re-establish connection"
499 ))
500 }
501
502 fn rejoin(&mut self, cx: &mut ModelContext<Self>) -> Task<Result<()>> {
503 let mut projects = HashMap::default();
504 let mut reshared_projects = Vec::new();
505 let mut rejoined_projects = Vec::new();
506 self.shared_projects.retain(|project| {
507 if let Some(handle) = project.upgrade() {
508 let project = handle.read(cx);
509 if let Some(project_id) = project.remote_id() {
510 projects.insert(project_id, handle.clone());
511 reshared_projects.push(proto::UpdateProject {
512 project_id,
513 worktrees: project.worktree_metadata_protos(cx),
514 });
515 return true;
516 }
517 }
518 false
519 });
520 self.joined_projects.retain(|project| {
521 if let Some(handle) = project.upgrade() {
522 let project = handle.read(cx);
523 if let Some(project_id) = project.remote_id() {
524 projects.insert(project_id, handle.clone());
525 rejoined_projects.push(proto::RejoinProject {
526 id: project_id,
527 worktrees: project
528 .worktrees()
529 .map(|worktree| {
530 let worktree = worktree.read(cx);
531 proto::RejoinWorktree {
532 id: worktree.id().to_proto(),
533 scan_id: worktree.completed_scan_id() as u64,
534 }
535 })
536 .collect(),
537 });
538 }
539 return true;
540 }
541 false
542 });
543
544 let response = self.client.request_envelope(proto::RejoinRoom {
545 id: self.id,
546 reshared_projects,
547 rejoined_projects,
548 });
549
550 cx.spawn(|this, mut cx| async move {
551 let response = response.await?;
552 let message_id = response.message_id;
553 let response = response.payload;
554 let room_proto = response.room.ok_or_else(|| anyhow!("invalid room"))?;
555 this.update(&mut cx, |this, cx| {
556 this.status = RoomStatus::Online;
557 this.apply_room_update(room_proto, cx)?;
558
559 for reshared_project in response.reshared_projects {
560 if let Some(project) = projects.get(&reshared_project.id) {
561 project.update(cx, |project, cx| {
562 project.reshared(reshared_project, cx).log_err();
563 });
564 }
565 }
566
567 for rejoined_project in response.rejoined_projects {
568 if let Some(project) = projects.get(&rejoined_project.id) {
569 project.update(cx, |project, cx| {
570 project.rejoined(rejoined_project, message_id, cx).log_err();
571 });
572 }
573 }
574
575 anyhow::Ok(())
576 })?
577 })
578 }
579
580 pub fn id(&self) -> u64 {
581 self.id
582 }
583
584 pub fn status(&self) -> RoomStatus {
585 self.status
586 }
587
588 pub fn local_participant(&self) -> &LocalParticipant {
589 &self.local_participant
590 }
591
592 pub fn remote_participants(&self) -> &BTreeMap<u64, RemoteParticipant> {
593 &self.remote_participants
594 }
595
596 pub fn remote_participant_for_peer_id(&self, peer_id: PeerId) -> Option<&RemoteParticipant> {
597 self.remote_participants
598 .values()
599 .find(|p| p.peer_id == peer_id)
600 }
601
602 pub fn role_for_user(&self, user_id: u64) -> Option<proto::ChannelRole> {
603 self.remote_participants
604 .get(&user_id)
605 .map(|participant| participant.role)
606 }
607
608 pub fn contains_guests(&self) -> bool {
609 self.local_participant.role == proto::ChannelRole::Guest
610 || self
611 .remote_participants
612 .values()
613 .any(|p| p.role == proto::ChannelRole::Guest)
614 }
615
616 pub fn local_participant_is_admin(&self) -> bool {
617 self.local_participant.role == proto::ChannelRole::Admin
618 }
619
620 pub fn set_participant_role(
621 &mut self,
622 user_id: u64,
623 role: proto::ChannelRole,
624 cx: &ModelContext<Self>,
625 ) -> Task<Result<()>> {
626 let client = self.client.clone();
627 let room_id = self.id;
628 let role = role.into();
629 cx.spawn(|_, _| async move {
630 client
631 .request(proto::SetRoomParticipantRole {
632 room_id,
633 user_id,
634 role,
635 })
636 .await
637 .map(|_| ())
638 })
639 }
640
641 pub fn pending_participants(&self) -> &[Arc<User>] {
642 &self.pending_participants
643 }
644
645 pub fn contains_participant(&self, user_id: u64) -> bool {
646 self.participant_user_ids.contains(&user_id)
647 }
648
649 pub fn followers_for(&self, leader_id: PeerId, project_id: u64) -> &[PeerId] {
650 self.follows_by_leader_id_project_id
651 .get(&(leader_id, project_id))
652 .map_or(&[], |v| v.as_slice())
653 }
654
655 /// Returns the most 'active' projects, defined as most people in the project
656 pub fn most_active_project(&self, cx: &AppContext) -> Option<(u64, u64)> {
657 let mut project_hosts_and_guest_counts = HashMap::<u64, (Option<u64>, u32)>::default();
658 for participant in self.remote_participants.values() {
659 match participant.location {
660 ParticipantLocation::SharedProject { project_id } => {
661 project_hosts_and_guest_counts
662 .entry(project_id)
663 .or_default()
664 .1 += 1;
665 }
666 ParticipantLocation::External | ParticipantLocation::UnsharedProject => {}
667 }
668 for project in &participant.projects {
669 project_hosts_and_guest_counts
670 .entry(project.id)
671 .or_default()
672 .0 = Some(participant.user.id);
673 }
674 }
675
676 if let Some(user) = self.user_store.read(cx).current_user() {
677 for project in &self.local_participant.projects {
678 project_hosts_and_guest_counts
679 .entry(project.id)
680 .or_default()
681 .0 = Some(user.id);
682 }
683 }
684
685 project_hosts_and_guest_counts
686 .into_iter()
687 .filter_map(|(id, (host, guest_count))| Some((id, host?, guest_count)))
688 .max_by_key(|(_, _, guest_count)| *guest_count)
689 .map(|(id, host, _)| (id, host))
690 }
691
692 async fn handle_room_updated(
693 this: Model<Self>,
694 envelope: TypedEnvelope<proto::RoomUpdated>,
695 _: Arc<Client>,
696 mut cx: AsyncAppContext,
697 ) -> Result<()> {
698 let room = envelope
699 .payload
700 .room
701 .ok_or_else(|| anyhow!("invalid room"))?;
702 this.update(&mut cx, |this, cx| this.apply_room_update(room, cx))?
703 }
704
705 fn apply_room_update(
706 &mut self,
707 mut room: proto::Room,
708 cx: &mut ModelContext<Self>,
709 ) -> Result<()> {
710 // Filter ourselves out from the room's participants.
711 let local_participant_ix = room
712 .participants
713 .iter()
714 .position(|participant| Some(participant.user_id) == self.client.user_id());
715 let local_participant = local_participant_ix.map(|ix| room.participants.swap_remove(ix));
716
717 let pending_participant_user_ids = room
718 .pending_participants
719 .iter()
720 .map(|p| p.user_id)
721 .collect::<Vec<_>>();
722
723 let remote_participant_user_ids = room
724 .participants
725 .iter()
726 .map(|p| p.user_id)
727 .collect::<Vec<_>>();
728
729 let (remote_participants, pending_participants) =
730 self.user_store.update(cx, move |user_store, cx| {
731 (
732 user_store.get_users(remote_participant_user_ids, cx),
733 user_store.get_users(pending_participant_user_ids, cx),
734 )
735 });
736
737 self.pending_room_update = Some(cx.spawn(|this, mut cx| async move {
738 let (remote_participants, pending_participants) =
739 futures::join!(remote_participants, pending_participants);
740
741 this.update(&mut cx, |this, cx| {
742 this.participant_user_ids.clear();
743
744 if let Some(participant) = local_participant {
745 let role = participant.role();
746 this.local_participant.projects = participant.projects;
747 if this.local_participant.role != role {
748 this.local_participant.role = role;
749
750 if role == proto::ChannelRole::Guest {
751 for project in mem::take(&mut this.shared_projects) {
752 if let Some(project) = project.upgrade() {
753 this.unshare_project(project, cx).log_err();
754 }
755 }
756 this.local_participant.projects.clear();
757 if let Some(live_kit_room) = &mut this.live_kit {
758 live_kit_room.stop_publishing(cx);
759 }
760 }
761
762 this.joined_projects.retain(|project| {
763 if let Some(project) = project.upgrade() {
764 project.update(cx, |project, cx| project.set_role(role, cx));
765 true
766 } else {
767 false
768 }
769 });
770 }
771 } else {
772 this.local_participant.projects.clear();
773 }
774
775 if let Some(participants) = remote_participants.log_err() {
776 for (participant, user) in room.participants.into_iter().zip(participants) {
777 let Some(peer_id) = participant.peer_id else {
778 continue;
779 };
780 let participant_index = ParticipantIndex(participant.participant_index);
781 this.participant_user_ids.insert(participant.user_id);
782
783 let old_projects = this
784 .remote_participants
785 .get(&participant.user_id)
786 .into_iter()
787 .flat_map(|existing| &existing.projects)
788 .map(|project| project.id)
789 .collect::<HashSet<_>>();
790 let new_projects = participant
791 .projects
792 .iter()
793 .map(|project| project.id)
794 .collect::<HashSet<_>>();
795
796 for project in &participant.projects {
797 if !old_projects.contains(&project.id) {
798 cx.emit(Event::RemoteProjectShared {
799 owner: user.clone(),
800 project_id: project.id,
801 worktree_root_names: project.worktree_root_names.clone(),
802 });
803 }
804 }
805
806 for unshared_project_id in old_projects.difference(&new_projects) {
807 this.joined_projects.retain(|project| {
808 if let Some(project) = project.upgrade() {
809 project.update(cx, |project, cx| {
810 if project.remote_id() == Some(*unshared_project_id) {
811 project.disconnected_from_host(cx);
812 false
813 } else {
814 true
815 }
816 })
817 } else {
818 false
819 }
820 });
821 cx.emit(Event::RemoteProjectUnshared {
822 project_id: *unshared_project_id,
823 });
824 }
825
826 let role = participant.role();
827 let location = ParticipantLocation::from_proto(participant.location)
828 .unwrap_or(ParticipantLocation::External);
829 if let Some(remote_participant) =
830 this.remote_participants.get_mut(&participant.user_id)
831 {
832 remote_participant.peer_id = peer_id;
833 remote_participant.projects = participant.projects;
834 remote_participant.participant_index = participant_index;
835 if location != remote_participant.location
836 || role != remote_participant.role
837 {
838 remote_participant.location = location;
839 remote_participant.role = role;
840 cx.emit(Event::ParticipantLocationChanged {
841 participant_id: peer_id,
842 });
843 }
844 } else {
845 this.remote_participants.insert(
846 participant.user_id,
847 RemoteParticipant {
848 user: user.clone(),
849 participant_index,
850 peer_id,
851 projects: participant.projects,
852 location,
853 role,
854 muted: true,
855 speaking: false,
856 video_tracks: Default::default(),
857 audio_tracks: Default::default(),
858 },
859 );
860
861 Audio::play_sound(Sound::Joined, cx);
862
863 if let Some(live_kit) = this.live_kit.as_ref() {
864 let video_tracks =
865 live_kit.room.remote_video_tracks(&user.id.to_string());
866 let audio_tracks =
867 live_kit.room.remote_audio_tracks(&user.id.to_string());
868 let publications = live_kit
869 .room
870 .remote_audio_track_publications(&user.id.to_string());
871
872 for track in video_tracks {
873 this.live_kit_room_updated(
874 RoomUpdate::SubscribedToRemoteVideoTrack(track),
875 cx,
876 )
877 .log_err();
878 }
879
880 for (track, publication) in
881 audio_tracks.iter().zip(publications.iter())
882 {
883 this.live_kit_room_updated(
884 RoomUpdate::SubscribedToRemoteAudioTrack(
885 track.clone(),
886 publication.clone(),
887 ),
888 cx,
889 )
890 .log_err();
891 }
892 }
893 }
894 }
895
896 this.remote_participants.retain(|user_id, participant| {
897 if this.participant_user_ids.contains(user_id) {
898 true
899 } else {
900 for project in &participant.projects {
901 cx.emit(Event::RemoteProjectUnshared {
902 project_id: project.id,
903 });
904 }
905 false
906 }
907 });
908 }
909
910 if let Some(pending_participants) = pending_participants.log_err() {
911 this.pending_participants = pending_participants;
912 for participant in &this.pending_participants {
913 this.participant_user_ids.insert(participant.id);
914 }
915 }
916
917 this.follows_by_leader_id_project_id.clear();
918 for follower in room.followers {
919 let project_id = follower.project_id;
920 let (leader, follower) = match (follower.leader_id, follower.follower_id) {
921 (Some(leader), Some(follower)) => (leader, follower),
922
923 _ => {
924 log::error!("Follower message {follower:?} missing some state");
925 continue;
926 }
927 };
928
929 let list = this
930 .follows_by_leader_id_project_id
931 .entry((leader, project_id))
932 .or_insert(Vec::new());
933 if !list.contains(&follower) {
934 list.push(follower);
935 }
936 }
937
938 this.pending_room_update.take();
939 if this.should_leave() {
940 log::info!("room is empty, leaving");
941 let _ = this.leave(cx);
942 }
943
944 this.user_store.update(cx, |user_store, cx| {
945 let participant_indices_by_user_id = this
946 .remote_participants
947 .iter()
948 .map(|(user_id, participant)| (*user_id, participant.participant_index))
949 .collect();
950 user_store.set_participant_indices(participant_indices_by_user_id, cx);
951 });
952
953 this.check_invariants();
954 this.room_update_completed_tx.try_send(Some(())).ok();
955 cx.notify();
956 })
957 .ok();
958 }));
959
960 cx.notify();
961 Ok(())
962 }
963
964 pub fn room_update_completed(&mut self) -> impl Future<Output = ()> {
965 let mut done_rx = self.room_update_completed_rx.clone();
966 async move {
967 while let Some(result) = done_rx.next().await {
968 if result.is_some() {
969 break;
970 }
971 }
972 }
973 }
974
975 fn live_kit_room_updated(
976 &mut self,
977 update: RoomUpdate,
978 cx: &mut ModelContext<Self>,
979 ) -> Result<()> {
980 match update {
981 RoomUpdate::SubscribedToRemoteVideoTrack(track) => {
982 let user_id = track.publisher_id().parse()?;
983 let track_id = track.sid().to_string();
984 let participant = self
985 .remote_participants
986 .get_mut(&user_id)
987 .ok_or_else(|| anyhow!("subscribed to track by unknown participant"))?;
988 participant.video_tracks.insert(track_id.clone(), track);
989 cx.emit(Event::RemoteVideoTracksChanged {
990 participant_id: participant.peer_id,
991 });
992 }
993
994 RoomUpdate::UnsubscribedFromRemoteVideoTrack {
995 publisher_id,
996 track_id,
997 } => {
998 let user_id = publisher_id.parse()?;
999 let participant = self
1000 .remote_participants
1001 .get_mut(&user_id)
1002 .ok_or_else(|| anyhow!("unsubscribed from track by unknown participant"))?;
1003 participant.video_tracks.remove(&track_id);
1004 cx.emit(Event::RemoteVideoTracksChanged {
1005 participant_id: participant.peer_id,
1006 });
1007 }
1008
1009 RoomUpdate::ActiveSpeakersChanged { speakers } => {
1010 let mut speaker_ids = speakers
1011 .into_iter()
1012 .filter_map(|speaker_sid| speaker_sid.parse().ok())
1013 .collect::<Vec<u64>>();
1014 speaker_ids.sort_unstable();
1015 for (sid, participant) in &mut self.remote_participants {
1016 if let Ok(_) = speaker_ids.binary_search(sid) {
1017 participant.speaking = true;
1018 } else {
1019 participant.speaking = false;
1020 }
1021 }
1022 if let Some(id) = self.client.user_id() {
1023 if let Some(room) = &mut self.live_kit {
1024 if let Ok(_) = speaker_ids.binary_search(&id) {
1025 room.speaking = true;
1026 } else {
1027 room.speaking = false;
1028 }
1029 }
1030 }
1031 }
1032
1033 RoomUpdate::RemoteAudioTrackMuteChanged { track_id, muted } => {
1034 let mut found = false;
1035 for participant in &mut self.remote_participants.values_mut() {
1036 for track in participant.audio_tracks.values() {
1037 if track.sid() == track_id {
1038 found = true;
1039 break;
1040 }
1041 }
1042 if found {
1043 participant.muted = muted;
1044 break;
1045 }
1046 }
1047 }
1048
1049 RoomUpdate::SubscribedToRemoteAudioTrack(track, publication) => {
1050 let user_id = track.publisher_id().parse()?;
1051 let track_id = track.sid().to_string();
1052 let participant = self
1053 .remote_participants
1054 .get_mut(&user_id)
1055 .ok_or_else(|| anyhow!("subscribed to track by unknown participant"))?;
1056 participant.audio_tracks.insert(track_id.clone(), track);
1057 participant.muted = publication.is_muted();
1058
1059 cx.emit(Event::RemoteAudioTracksChanged {
1060 participant_id: participant.peer_id,
1061 });
1062 }
1063
1064 RoomUpdate::UnsubscribedFromRemoteAudioTrack {
1065 publisher_id,
1066 track_id,
1067 } => {
1068 let user_id = publisher_id.parse()?;
1069 let participant = self
1070 .remote_participants
1071 .get_mut(&user_id)
1072 .ok_or_else(|| anyhow!("unsubscribed from track by unknown participant"))?;
1073 participant.audio_tracks.remove(&track_id);
1074 cx.emit(Event::RemoteAudioTracksChanged {
1075 participant_id: participant.peer_id,
1076 });
1077 }
1078
1079 RoomUpdate::LocalAudioTrackUnpublished { publication } => {
1080 log::info!("unpublished audio track {}", publication.sid());
1081 if let Some(room) = &mut self.live_kit {
1082 room.microphone_track = LocalTrack::None;
1083 }
1084 }
1085
1086 RoomUpdate::LocalVideoTrackUnpublished { publication } => {
1087 log::info!("unpublished video track {}", publication.sid());
1088 if let Some(room) = &mut self.live_kit {
1089 room.screen_track = LocalTrack::None;
1090 }
1091 }
1092
1093 RoomUpdate::LocalAudioTrackPublished { publication } => {
1094 log::info!("published audio track {}", publication.sid());
1095 }
1096
1097 RoomUpdate::LocalVideoTrackPublished { publication } => {
1098 log::info!("published video track {}", publication.sid());
1099 }
1100 }
1101
1102 cx.notify();
1103 Ok(())
1104 }
1105
1106 fn check_invariants(&self) {
1107 #[cfg(any(test, feature = "test-support"))]
1108 {
1109 for participant in self.remote_participants.values() {
1110 assert!(self.participant_user_ids.contains(&participant.user.id));
1111 assert_ne!(participant.user.id, self.client.user_id().unwrap());
1112 }
1113
1114 for participant in &self.pending_participants {
1115 assert!(self.participant_user_ids.contains(&participant.id));
1116 assert_ne!(participant.id, self.client.user_id().unwrap());
1117 }
1118
1119 assert_eq!(
1120 self.participant_user_ids.len(),
1121 self.remote_participants.len() + self.pending_participants.len()
1122 );
1123 }
1124 }
1125
1126 pub(crate) fn call(
1127 &mut self,
1128 called_user_id: u64,
1129 initial_project_id: Option<u64>,
1130 cx: &mut ModelContext<Self>,
1131 ) -> Task<Result<()>> {
1132 if self.status.is_offline() {
1133 return Task::ready(Err(anyhow!("room is offline")));
1134 }
1135
1136 cx.notify();
1137 let client = self.client.clone();
1138 let room_id = self.id;
1139 self.pending_call_count += 1;
1140 cx.spawn(move |this, mut cx| async move {
1141 let result = client
1142 .request(proto::Call {
1143 room_id,
1144 called_user_id,
1145 initial_project_id,
1146 })
1147 .await;
1148 this.update(&mut cx, |this, cx| {
1149 this.pending_call_count -= 1;
1150 if this.should_leave() {
1151 this.leave(cx).detach_and_log_err(cx);
1152 }
1153 })?;
1154 result?;
1155 Ok(())
1156 })
1157 }
1158
1159 pub fn join_project(
1160 &mut self,
1161 id: u64,
1162 language_registry: Arc<LanguageRegistry>,
1163 fs: Arc<dyn Fs>,
1164 cx: &mut ModelContext<Self>,
1165 ) -> Task<Result<Model<Project>>> {
1166 let client = self.client.clone();
1167 let user_store = self.user_store.clone();
1168 let role = self.local_participant.role;
1169 cx.emit(Event::RemoteProjectJoined { project_id: id });
1170 cx.spawn(move |this, mut cx| async move {
1171 let project = Project::remote(
1172 id,
1173 client,
1174 user_store,
1175 language_registry,
1176 fs,
1177 role,
1178 cx.clone(),
1179 )
1180 .await?;
1181
1182 this.update(&mut cx, |this, cx| {
1183 this.joined_projects.retain(|project| {
1184 if let Some(project) = project.upgrade() {
1185 !project.read(cx).is_disconnected()
1186 } else {
1187 false
1188 }
1189 });
1190 this.joined_projects.insert(project.downgrade());
1191 })?;
1192 Ok(project)
1193 })
1194 }
1195
1196 pub(crate) fn share_project(
1197 &mut self,
1198 project: Model<Project>,
1199 cx: &mut ModelContext<Self>,
1200 ) -> Task<Result<u64>> {
1201 if let Some(project_id) = project.read(cx).remote_id() {
1202 return Task::ready(Ok(project_id));
1203 }
1204
1205 let request = self.client.request(proto::ShareProject {
1206 room_id: self.id(),
1207 worktrees: project.read(cx).worktree_metadata_protos(cx),
1208 });
1209 cx.spawn(|this, mut cx| async move {
1210 let response = request.await?;
1211
1212 project.update(&mut cx, |project, cx| {
1213 project.shared(response.project_id, cx)
1214 })??;
1215
1216 // If the user's location is in this project, it changes from UnsharedProject to SharedProject.
1217 this.update(&mut cx, |this, cx| {
1218 this.shared_projects.insert(project.downgrade());
1219 let active_project = this.local_participant.active_project.as_ref();
1220 if active_project.map_or(false, |location| *location == project) {
1221 this.set_location(Some(&project), cx)
1222 } else {
1223 Task::ready(Ok(()))
1224 }
1225 })?
1226 .await?;
1227
1228 Ok(response.project_id)
1229 })
1230 }
1231
1232 pub(crate) fn unshare_project(
1233 &mut self,
1234 project: Model<Project>,
1235 cx: &mut ModelContext<Self>,
1236 ) -> Result<()> {
1237 let project_id = match project.read(cx).remote_id() {
1238 Some(project_id) => project_id,
1239 None => return Ok(()),
1240 };
1241
1242 self.client.send(proto::UnshareProject { project_id })?;
1243 project.update(cx, |this, cx| this.unshare(cx))?;
1244
1245 if self.local_participant.active_project == Some(project.downgrade()) {
1246 self.set_location(Some(&project), cx).detach_and_log_err(cx);
1247 }
1248 Ok(())
1249 }
1250
1251 pub(crate) fn set_location(
1252 &mut self,
1253 project: Option<&Model<Project>>,
1254 cx: &mut ModelContext<Self>,
1255 ) -> Task<Result<()>> {
1256 if self.status.is_offline() {
1257 return Task::ready(Err(anyhow!("room is offline")));
1258 }
1259
1260 let client = self.client.clone();
1261 let room_id = self.id;
1262 let location = if let Some(project) = project {
1263 self.local_participant.active_project = Some(project.downgrade());
1264 if let Some(project_id) = project.read(cx).remote_id() {
1265 proto::participant_location::Variant::SharedProject(
1266 proto::participant_location::SharedProject { id: project_id },
1267 )
1268 } else {
1269 proto::participant_location::Variant::UnsharedProject(
1270 proto::participant_location::UnsharedProject {},
1271 )
1272 }
1273 } else {
1274 self.local_participant.active_project = None;
1275 proto::participant_location::Variant::External(proto::participant_location::External {})
1276 };
1277
1278 cx.notify();
1279 cx.background_executor().spawn(async move {
1280 client
1281 .request(proto::UpdateParticipantLocation {
1282 room_id,
1283 location: Some(proto::ParticipantLocation {
1284 variant: Some(location),
1285 }),
1286 })
1287 .await?;
1288 Ok(())
1289 })
1290 }
1291
1292 pub fn is_screen_sharing(&self) -> bool {
1293 self.live_kit.as_ref().map_or(false, |live_kit| {
1294 !matches!(live_kit.screen_track, LocalTrack::None)
1295 })
1296 }
1297
1298 pub fn is_sharing_mic(&self) -> bool {
1299 self.live_kit.as_ref().map_or(false, |live_kit| {
1300 !matches!(live_kit.microphone_track, LocalTrack::None)
1301 })
1302 }
1303
1304 pub fn is_muted(&self, cx: &AppContext) -> bool {
1305 self.live_kit
1306 .as_ref()
1307 .and_then(|live_kit| match &live_kit.microphone_track {
1308 LocalTrack::None => Some(Self::mute_on_join(cx)),
1309 LocalTrack::Pending { muted, .. } => Some(*muted),
1310 LocalTrack::Published { muted, .. } => Some(*muted),
1311 })
1312 .unwrap_or(false)
1313 }
1314
1315 pub fn read_only(&self) -> bool {
1316 !(self.local_participant().role == proto::ChannelRole::Member
1317 || self.local_participant().role == proto::ChannelRole::Admin)
1318 }
1319
1320 pub fn is_speaking(&self) -> bool {
1321 self.live_kit
1322 .as_ref()
1323 .map_or(false, |live_kit| live_kit.speaking)
1324 }
1325
1326 pub fn is_deafened(&self) -> Option<bool> {
1327 self.live_kit.as_ref().map(|live_kit| live_kit.deafened)
1328 }
1329
1330 #[track_caller]
1331 pub fn share_microphone(&mut self, cx: &mut ModelContext<Self>) -> Task<Result<()>> {
1332 if self.status.is_offline() {
1333 return Task::ready(Err(anyhow!("room is offline")));
1334 } else if self.is_sharing_mic() {
1335 return Task::ready(Err(anyhow!("microphone was already shared")));
1336 }
1337
1338 let publish_id = if let Some(live_kit) = self.live_kit.as_mut() {
1339 let publish_id = post_inc(&mut live_kit.next_publish_id);
1340 live_kit.microphone_track = LocalTrack::Pending {
1341 publish_id,
1342 muted: false,
1343 };
1344 cx.notify();
1345 publish_id
1346 } else {
1347 return Task::ready(Err(anyhow!("live-kit was not initialized")));
1348 };
1349
1350 cx.spawn(move |this, mut cx| async move {
1351 let publish_track = async {
1352 let track = LocalAudioTrack::create();
1353 this.upgrade()
1354 .ok_or_else(|| anyhow!("room was dropped"))?
1355 .update(&mut cx, |this, _| {
1356 this.live_kit
1357 .as_ref()
1358 .map(|live_kit| live_kit.room.publish_audio_track(track))
1359 })?
1360 .ok_or_else(|| anyhow!("live-kit was not initialized"))?
1361 .await
1362 };
1363 let publication = publish_track.await;
1364 this.upgrade()
1365 .ok_or_else(|| anyhow!("room was dropped"))?
1366 .update(&mut cx, |this, cx| {
1367 let live_kit = this
1368 .live_kit
1369 .as_mut()
1370 .ok_or_else(|| anyhow!("live-kit was not initialized"))?;
1371
1372 let (canceled, muted) = if let LocalTrack::Pending {
1373 publish_id: cur_publish_id,
1374 muted,
1375 } = &live_kit.microphone_track
1376 {
1377 (*cur_publish_id != publish_id, *muted)
1378 } else {
1379 (true, false)
1380 };
1381
1382 match publication {
1383 Ok(publication) => {
1384 if canceled {
1385 live_kit.room.unpublish_track(publication);
1386 } else {
1387 if muted {
1388 cx.background_executor()
1389 .spawn(publication.set_mute(muted))
1390 .detach();
1391 }
1392 live_kit.microphone_track = LocalTrack::Published {
1393 track_publication: publication,
1394 muted,
1395 };
1396 cx.notify();
1397 }
1398 Ok(())
1399 }
1400 Err(error) => {
1401 if canceled {
1402 Ok(())
1403 } else {
1404 live_kit.microphone_track = LocalTrack::None;
1405 cx.notify();
1406 Err(error)
1407 }
1408 }
1409 }
1410 })?
1411 })
1412 }
1413
1414 pub fn share_screen(&mut self, cx: &mut ModelContext<Self>) -> Task<Result<()>> {
1415 if self.status.is_offline() {
1416 return Task::ready(Err(anyhow!("room is offline")));
1417 } else if self.is_screen_sharing() {
1418 return Task::ready(Err(anyhow!("screen was already shared")));
1419 }
1420
1421 let (displays, publish_id) = if let Some(live_kit) = self.live_kit.as_mut() {
1422 let publish_id = post_inc(&mut live_kit.next_publish_id);
1423 live_kit.screen_track = LocalTrack::Pending {
1424 publish_id,
1425 muted: false,
1426 };
1427 cx.notify();
1428 (live_kit.room.display_sources(), publish_id)
1429 } else {
1430 return Task::ready(Err(anyhow!("live-kit was not initialized")));
1431 };
1432
1433 cx.spawn(move |this, mut cx| async move {
1434 let publish_track = async {
1435 let displays = displays.await?;
1436 let display = displays
1437 .first()
1438 .ok_or_else(|| anyhow!("no display found"))?;
1439 let track = LocalVideoTrack::screen_share_for_display(&display);
1440 this.upgrade()
1441 .ok_or_else(|| anyhow!("room was dropped"))?
1442 .update(&mut cx, |this, _| {
1443 this.live_kit
1444 .as_ref()
1445 .map(|live_kit| live_kit.room.publish_video_track(track))
1446 })?
1447 .ok_or_else(|| anyhow!("live-kit was not initialized"))?
1448 .await
1449 };
1450
1451 let publication = publish_track.await;
1452 this.upgrade()
1453 .ok_or_else(|| anyhow!("room was dropped"))?
1454 .update(&mut cx, |this, cx| {
1455 let live_kit = this
1456 .live_kit
1457 .as_mut()
1458 .ok_or_else(|| anyhow!("live-kit was not initialized"))?;
1459
1460 let (canceled, muted) = if let LocalTrack::Pending {
1461 publish_id: cur_publish_id,
1462 muted,
1463 } = &live_kit.screen_track
1464 {
1465 (*cur_publish_id != publish_id, *muted)
1466 } else {
1467 (true, false)
1468 };
1469
1470 match publication {
1471 Ok(publication) => {
1472 if canceled {
1473 live_kit.room.unpublish_track(publication);
1474 } else {
1475 if muted {
1476 cx.background_executor()
1477 .spawn(publication.set_mute(muted))
1478 .detach();
1479 }
1480 live_kit.screen_track = LocalTrack::Published {
1481 track_publication: publication,
1482 muted,
1483 };
1484 cx.notify();
1485 }
1486
1487 Audio::play_sound(Sound::StartScreenshare, cx);
1488
1489 Ok(())
1490 }
1491 Err(error) => {
1492 if canceled {
1493 Ok(())
1494 } else {
1495 live_kit.screen_track = LocalTrack::None;
1496 cx.notify();
1497 Err(error)
1498 }
1499 }
1500 }
1501 })?
1502 })
1503 }
1504
1505 pub fn toggle_mute(&mut self, cx: &mut ModelContext<Self>) -> Result<Task<Result<()>>> {
1506 let should_mute = !self.is_muted(cx);
1507 if let Some(live_kit) = self.live_kit.as_mut() {
1508 if matches!(live_kit.microphone_track, LocalTrack::None) {
1509 return Ok(self.share_microphone(cx));
1510 }
1511
1512 let (ret_task, old_muted) = live_kit.set_mute(should_mute, cx)?;
1513 live_kit.muted_by_user = should_mute;
1514
1515 if old_muted == true && live_kit.deafened == true {
1516 if let Some(task) = self.toggle_deafen(cx).ok() {
1517 task.detach();
1518 }
1519 }
1520
1521 Ok(ret_task)
1522 } else {
1523 Err(anyhow!("LiveKit not started"))
1524 }
1525 }
1526
1527 pub fn toggle_deafen(&mut self, cx: &mut ModelContext<Self>) -> Result<Task<Result<()>>> {
1528 if let Some(live_kit) = self.live_kit.as_mut() {
1529 (*live_kit).deafened = !live_kit.deafened;
1530
1531 let mut tasks = Vec::with_capacity(self.remote_participants.len());
1532 // Context notification is sent within set_mute itself.
1533 let mut mute_task = None;
1534 // When deafening, mute user's mic as well.
1535 // When undeafening, unmute user's mic unless it was manually muted prior to deafening.
1536 if live_kit.deafened || !live_kit.muted_by_user {
1537 mute_task = Some(live_kit.set_mute(live_kit.deafened, cx)?.0);
1538 };
1539 for participant in self.remote_participants.values() {
1540 for track in live_kit
1541 .room
1542 .remote_audio_track_publications(&participant.user.id.to_string())
1543 {
1544 let deafened = live_kit.deafened;
1545 tasks.push(cx.foreground_executor().spawn(track.set_enabled(!deafened)));
1546 }
1547 }
1548
1549 Ok(cx.foreground_executor().spawn(async move {
1550 if let Some(mute_task) = mute_task {
1551 mute_task.await?;
1552 }
1553 for task in tasks {
1554 task.await?;
1555 }
1556 Ok(())
1557 }))
1558 } else {
1559 Err(anyhow!("LiveKit not started"))
1560 }
1561 }
1562
1563 pub fn unshare_screen(&mut self, cx: &mut ModelContext<Self>) -> Result<()> {
1564 if self.status.is_offline() {
1565 return Err(anyhow!("room is offline"));
1566 }
1567
1568 let live_kit = self
1569 .live_kit
1570 .as_mut()
1571 .ok_or_else(|| anyhow!("live-kit was not initialized"))?;
1572 match mem::take(&mut live_kit.screen_track) {
1573 LocalTrack::None => Err(anyhow!("screen was not shared")),
1574 LocalTrack::Pending { .. } => {
1575 cx.notify();
1576 Ok(())
1577 }
1578 LocalTrack::Published {
1579 track_publication, ..
1580 } => {
1581 live_kit.room.unpublish_track(track_publication);
1582 cx.notify();
1583
1584 Audio::play_sound(Sound::StopScreenshare, cx);
1585 Ok(())
1586 }
1587 }
1588 }
1589
1590 #[cfg(any(test, feature = "test-support"))]
1591 pub fn set_display_sources(&self, sources: Vec<live_kit_client::MacOSDisplay>) {
1592 self.live_kit
1593 .as_ref()
1594 .unwrap()
1595 .room
1596 .set_display_sources(sources);
1597 }
1598}
1599
1600struct LiveKitRoom {
1601 room: Arc<live_kit_client::Room>,
1602 screen_track: LocalTrack,
1603 microphone_track: LocalTrack,
1604 /// Tracks whether we're currently in a muted state due to auto-mute from deafening or manual mute performed by user.
1605 muted_by_user: bool,
1606 deafened: bool,
1607 speaking: bool,
1608 next_publish_id: usize,
1609 _maintain_room: Task<()>,
1610 _handle_updates: Task<()>,
1611}
1612
1613impl LiveKitRoom {
1614 fn set_mute(
1615 self: &mut LiveKitRoom,
1616 should_mute: bool,
1617 cx: &mut ModelContext<Room>,
1618 ) -> Result<(Task<Result<()>>, bool)> {
1619 if !should_mute {
1620 // clear user muting state.
1621 self.muted_by_user = false;
1622 }
1623
1624 let (result, old_muted) = match &mut self.microphone_track {
1625 LocalTrack::None => Err(anyhow!("microphone was not shared")),
1626 LocalTrack::Pending { muted, .. } => {
1627 let old_muted = *muted;
1628 *muted = should_mute;
1629 cx.notify();
1630 Ok((Task::Ready(Some(Ok(()))), old_muted))
1631 }
1632 LocalTrack::Published {
1633 track_publication,
1634 muted,
1635 } => {
1636 let old_muted = *muted;
1637 *muted = should_mute;
1638 cx.notify();
1639 Ok((
1640 cx.background_executor()
1641 .spawn(track_publication.set_mute(*muted)),
1642 old_muted,
1643 ))
1644 }
1645 }?;
1646
1647 if old_muted != should_mute {
1648 if should_mute {
1649 Audio::play_sound(Sound::Mute, cx);
1650 } else {
1651 Audio::play_sound(Sound::Unmute, cx);
1652 }
1653 }
1654
1655 Ok((result, old_muted))
1656 }
1657
1658 fn stop_publishing(&mut self, cx: &mut ModelContext<Room>) {
1659 if let LocalTrack::Published {
1660 track_publication, ..
1661 } = mem::replace(&mut self.microphone_track, LocalTrack::None)
1662 {
1663 self.room.unpublish_track(track_publication);
1664 cx.notify();
1665 }
1666
1667 if let LocalTrack::Published {
1668 track_publication, ..
1669 } = mem::replace(&mut self.screen_track, LocalTrack::None)
1670 {
1671 self.room.unpublish_track(track_publication);
1672 cx.notify();
1673 }
1674 }
1675}
1676
1677enum LocalTrack {
1678 None,
1679 Pending {
1680 publish_id: usize,
1681 muted: bool,
1682 },
1683 Published {
1684 track_publication: LocalTrackPublication,
1685 muted: bool,
1686 },
1687}
1688
1689impl Default for LocalTrack {
1690 fn default() -> Self {
1691 Self::None
1692 }
1693}
1694
1695#[derive(Copy, Clone, PartialEq, Eq)]
1696pub enum RoomStatus {
1697 Online,
1698 Rejoining,
1699 Offline,
1700}
1701
1702impl RoomStatus {
1703 pub fn is_offline(&self) -> bool {
1704 matches!(self, RoomStatus::Offline)
1705 }
1706
1707 pub fn is_online(&self) -> bool {
1708 matches!(self, RoomStatus::Online)
1709 }
1710}