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