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