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