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