use crate::{    call_settings::CallSettings,    participant::{LocalParticipant, ParticipantLocation, RemoteParticipant},};use anyhow::{anyhow, Result};use audio::{Audio, Sound};use client::{    proto::{self, PeerId},    ChannelId, Client, ParticipantIndex, TypedEnvelope, User, UserStore,};use collections::{BTreeMap, HashMap, HashSet};use fs::Fs;use futures::{FutureExt, StreamExt};use gpui::{App, AppContext as _, AsyncApp, Context, Entity, EventEmitter, Task, WeakEntity};use language::LanguageRegistry;use livekit_client_macos::{LocalAudioTrack, LocalTrackPublication, LocalVideoTrack, RoomUpdate};use postage::{sink::Sink, stream::Stream, watch};use project::Project;use settings::Settings as _;use std::{future::Future, mem, sync::Arc, time::Duration};use util::{post_inc, ResultExt, TryFutureExt}; pub const RECONNECT_TIMEOUT: Duration = Duration::from_secs(30); #[derive(Clone, Debug, PartialEq, Eq)]pub enum Event {    RoomJoined {        channel_id: Option<ChannelId>,    },    ParticipantLocationChanged {        participant_id: proto::PeerId,    },    RemoteVideoTracksChanged {        participant_id: proto::PeerId,    },    RemoteAudioTracksChanged {        participant_id: proto::PeerId,    },    RemoteProjectShared {        owner: Arc<User>,        project_id: u64,        worktree_root_names: Vec<String>,    },    RemoteProjectUnshared {        project_id: u64,    },    RemoteProjectJoined {        project_id: u64,    },    RemoteProjectInvitationDiscarded {        project_id: u64,    },    RoomLeft {        channel_id: Option<ChannelId>,    },} pub struct Room {    id: u64,    channel_id: Option<ChannelId>,    live_kit: Option<LiveKitRoom>,    status: RoomStatus,    shared_projects: HashSet<WeakEntity<Project>>,    joined_projects: HashSet<WeakEntity<Project>>,    local_participant: LocalParticipant,    remote_participants: BTreeMap<u64, RemoteParticipant>,    pending_participants: Vec<Arc<User>>,    participant_user_ids: HashSet<u64>,    pending_call_count: usize,    leave_when_empty: bool,    client: Arc<Client>,    user_store: Entity<UserStore>,    follows_by_leader_id_project_id: HashMap<(PeerId, u64), Vec<PeerId>>,    client_subscriptions: Vec<client::Subscription>,    _subscriptions: Vec<gpui::Subscription>,    room_update_completed_tx: watch::Sender<Option<()>>,    room_update_completed_rx: watch::Receiver<Option<()>>,    pending_room_update: Option<Task<()>>,    maintain_connection: Option<Task<Option<()>>>,} impl EventEmitter<Event> for Room {} impl Room {    pub fn channel_id(&self) -> Option<ChannelId> {        self.channel_id    }     pub fn is_sharing_project(&self) -> bool {        !self.shared_projects.is_empty()    }     #[cfg(any(test, feature = "test-support"))]    pub fn is_connected(&self) -> bool {        if let Some(live_kit) = self.live_kit.as_ref() {            matches!(                *live_kit.room.status().borrow(),                livekit_client_macos::ConnectionState::Connected { .. }            )        } else {            false        }    }     fn new(        id: u64,        channel_id: Option<ChannelId>,        live_kit_connection_info: Option<proto::LiveKitConnectionInfo>,        client: Arc<Client>,        user_store: Entity<UserStore>,        cx: &mut Context<Self>,    ) -> Self {        let live_kit_room = if let Some(connection_info) = live_kit_connection_info {            let room = livekit_client_macos::Room::new();            let mut status = room.status();            // Consume the initial status of the room.            let _ = status.try_recv();            let _maintain_room = cx.spawn(|this, mut cx| async move {                while let Some(status) = status.next().await {                    let this = if let Some(this) = this.upgrade() {                        this                    } else {                        break;                    };                     if status == livekit_client_macos::ConnectionState::Disconnected {                        this.update(&mut cx, |this, cx| this.leave(cx).log_err())                            .ok();                        break;                    }                }            });             let _handle_updates = cx.spawn({                let room = room.clone();                move |this, mut cx| async move {                    let mut updates = room.updates();                    while let Some(update) = updates.next().await {                        let this = if let Some(this) = this.upgrade() {                            this                        } else {                            break;                        };                         this.update(&mut cx, |this, cx| {                            this.live_kit_room_updated(update, cx).log_err()                        })                        .ok();                    }                }            });             let connect = room.connect(&connection_info.server_url, &connection_info.token);            cx.spawn(|this, mut cx| async move {                connect.await?;                this.update(&mut cx, |this, cx| {                    if this.can_use_microphone() {                        if let Some(live_kit) = &this.live_kit {                            if !live_kit.muted_by_user && !live_kit.deafened {                                return this.share_microphone(cx);                            }                        }                    }                    Task::ready(Ok(()))                })?                .await            })            .detach_and_log_err(cx);             Some(LiveKitRoom {                room,                screen_track: LocalTrack::None,                microphone_track: LocalTrack::None,                next_publish_id: 0,                muted_by_user: Self::mute_on_join(cx),                deafened: false,                speaking: false,                _maintain_room,                _handle_updates,            })        } else {            None        };         let maintain_connection = cx.spawn({            let client = client.clone();            move |this, cx| Self::maintain_connection(this, client.clone(), cx).log_err()        });         Audio::play_sound(Sound::Joined, cx);         let (room_update_completed_tx, room_update_completed_rx) = watch::channel();         Self {            id,            channel_id,            live_kit: live_kit_room,            status: RoomStatus::Online,            shared_projects: Default::default(),            joined_projects: Default::default(),            participant_user_ids: Default::default(),            local_participant: Default::default(),            remote_participants: Default::default(),            pending_participants: Default::default(),            pending_call_count: 0,            client_subscriptions: vec![                client.add_message_handler(cx.weak_entity(), Self::handle_room_updated)            ],            _subscriptions: vec![                cx.on_release(Self::released),                cx.on_app_quit(Self::app_will_quit),            ],            leave_when_empty: false,            pending_room_update: None,            client,            user_store,            follows_by_leader_id_project_id: Default::default(),            maintain_connection: Some(maintain_connection),            room_update_completed_tx,            room_update_completed_rx,        }    }     pub(crate) fn create(        called_user_id: u64,        initial_project: Option<Entity<Project>>,        client: Arc<Client>,        user_store: Entity<UserStore>,        cx: &mut App,    ) -> Task<Result<Entity<Self>>> {        cx.spawn(move |mut cx| async move {            let response = client.request(proto::CreateRoom {}).await?;            let room_proto = response.room.ok_or_else(|| anyhow!("invalid room"))?;            let room = cx.new(|cx| {                let mut room = Self::new(                    room_proto.id,                    None,                    response.live_kit_connection_info,                    client,                    user_store,                    cx,                );                if let Some(participant) = room_proto.participants.first() {                    room.local_participant.role = participant.role()                }                room            })?;             let initial_project_id = if let Some(initial_project) = initial_project {                let initial_project_id = room                    .update(&mut cx, |room, cx| {                        room.share_project(initial_project.clone(), cx)                    })?                    .await?;                Some(initial_project_id)            } else {                None            };             let did_join = room                .update(&mut cx, |room, cx| {                    room.leave_when_empty = true;                    room.call(called_user_id, initial_project_id, cx)                })?                .await;            match did_join {                Ok(()) => Ok(room),                Err(error) => Err(error.context("room creation failed")),            }        })    }     pub(crate) async fn join_channel(        channel_id: ChannelId,        client: Arc<Client>,        user_store: Entity<UserStore>,        cx: AsyncApp,    ) -> Result<Entity<Self>> {        Self::from_join_response(            client                .request(proto::JoinChannel {                    channel_id: channel_id.0,                })                .await?,            client,            user_store,            cx,        )    }     pub(crate) async fn join(        room_id: u64,        client: Arc<Client>,        user_store: Entity<UserStore>,        cx: AsyncApp,    ) -> Result<Entity<Self>> {        Self::from_join_response(            client.request(proto::JoinRoom { id: room_id }).await?,            client,            user_store,            cx,        )    }     fn released(&mut self, cx: &mut App) {        if self.status.is_online() {            self.leave_internal(cx).detach_and_log_err(cx);        }    }     fn app_will_quit(&mut self, cx: &mut Context<Self>) -> impl Future<Output = ()> {        let task = if self.status.is_online() {            let leave = self.leave_internal(cx);            Some(cx.background_spawn(async move {                leave.await.log_err();            }))        } else {            None        };         async move {            if let Some(task) = task {                task.await;            }        }    }     pub fn mute_on_join(cx: &App) -> bool {        CallSettings::get_global(cx).mute_on_join || client::IMPERSONATE_LOGIN.is_some()    }     fn from_join_response(        response: proto::JoinRoomResponse,        client: Arc<Client>,        user_store: Entity<UserStore>,        mut cx: AsyncApp,    ) -> Result<Entity<Self>> {        let room_proto = response.room.ok_or_else(|| anyhow!("invalid room"))?;        let room = cx.new(|cx| {            Self::new(                room_proto.id,                response.channel_id.map(ChannelId),                response.live_kit_connection_info,                client,                user_store,                cx,            )        })?;        room.update(&mut cx, |room, cx| {            room.leave_when_empty = room.channel_id.is_none();            room.apply_room_update(room_proto, cx)?;            anyhow::Ok(())        })??;        Ok(room)    }     fn should_leave(&self) -> bool {        self.leave_when_empty            && self.pending_room_update.is_none()            && self.pending_participants.is_empty()            && self.remote_participants.is_empty()            && self.pending_call_count == 0    }     pub(crate) fn leave(&mut self, cx: &mut Context<Self>) -> Task<Result<()>> {        cx.notify();        self.leave_internal(cx)    }     fn leave_internal(&mut self, cx: &mut App) -> Task<Result<()>> {        if self.status.is_offline() {            return Task::ready(Err(anyhow!("room is offline")));        }         log::info!("leaving room");        Audio::play_sound(Sound::Leave, cx);         self.clear_state(cx);         let leave_room = self.client.request(proto::LeaveRoom {});        cx.background_spawn(async move {            leave_room.await?;            anyhow::Ok(())        })    }     pub(crate) fn clear_state(&mut self, cx: &mut App) {        for project in self.shared_projects.drain() {            if let Some(project) = project.upgrade() {                project.update(cx, |project, cx| {                    project.unshare(cx).log_err();                });            }        }        for project in self.joined_projects.drain() {            if let Some(project) = project.upgrade() {                project.update(cx, |project, cx| {                    project.disconnected_from_host(cx);                    project.close(cx);                });            }        }         self.status = RoomStatus::Offline;        self.remote_participants.clear();        self.pending_participants.clear();        self.participant_user_ids.clear();        self.client_subscriptions.clear();        self.live_kit.take();        self.pending_room_update.take();        self.maintain_connection.take();    }     async fn maintain_connection(        this: WeakEntity<Self>,        client: Arc<Client>,        mut cx: AsyncApp,    ) -> Result<()> {        let mut client_status = client.status();        loop {            let _ = client_status.try_recv();            let is_connected = client_status.borrow().is_connected();            // Even if we're initially connected, any future change of the status means we momentarily disconnected.            if !is_connected || client_status.next().await.is_some() {                log::info!("detected client disconnection");                 this.upgrade()                    .ok_or_else(|| anyhow!("room was dropped"))?                    .update(&mut cx, |this, cx| {                        this.status = RoomStatus::Rejoining;                        cx.notify();                    })?;                 // Wait for client to re-establish a connection to the server.                {                    let mut reconnection_timeout =                        cx.background_executor().timer(RECONNECT_TIMEOUT).fuse();                    let client_reconnection = async {                        let mut remaining_attempts = 3;                        while remaining_attempts > 0 {                            if client_status.borrow().is_connected() {                                log::info!("client reconnected, attempting to rejoin room");                                 let Some(this) = this.upgrade() else { break };                                match this.update(&mut cx, |this, cx| this.rejoin(cx)) {                                    Ok(task) => {                                        if task.await.log_err().is_some() {                                            return true;                                        } else {                                            remaining_attempts -= 1;                                        }                                    }                                    Err(_app_dropped) => return false,                                }                            } else if client_status.borrow().is_signed_out() {                                return false;                            }                             log::info!(                                "waiting for client status change, remaining attempts {}",                                remaining_attempts                            );                            client_status.next().await;                        }                        false                    }                    .fuse();                    futures::pin_mut!(client_reconnection);                     futures::select_biased! {                        reconnected = client_reconnection => {                            if reconnected {                                log::info!("successfully reconnected to room");                                // If we successfully joined the room, go back around the loop                                // waiting for future connection status changes.                                continue;                            }                        }                        _ = reconnection_timeout => {                            log::info!("room reconnection timeout expired");                        }                    }                }                 break;            }        }         // The client failed to re-establish a connection to the server        // or an error occurred while trying to re-join the room. Either way        // we leave the room and return an error.        if let Some(this) = this.upgrade() {            log::info!("reconnection failed, leaving room");            this.update(&mut cx, |this, cx| this.leave(cx))?.await?;        }        Err(anyhow!(            "can't reconnect to room: client failed to re-establish connection"        ))    } 

Lines 1–500 of 1697. Load more