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