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