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