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