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