Cargo.lock 🔗
@@ -731,6 +731,7 @@ dependencies = [
"collections",
"futures 0.3.24",
"gpui",
+ "live_kit_client",
"postage",
"project",
"util",
WIP: Start integrating screen-sharing
Antonio Scandurra created
Change summary
Cargo.lock | 1
crates/call/Cargo.toml | 1
crates/call/src/room.rs | 64
crates/collab/src/rpc.rs | 80
crates/live_kit_client/LiveKitBridge/Sources/LiveKitBridge/LiveKitBridge.swift | 2
crates/live_kit_server/src/api.rs | 20
crates/rpc/proto/zed.proto | 9
crates/zed/build.rs | 6
8 files changed, 130 insertions(+), 53 deletions(-)
Detailed changes
crates/call/Cargo.toml 🔗
@@ -19,6 +19,7 @@ test-support = [
[dependencies]
client = { path = "../client" }
collections = { path = "../collections" }
+live_kit_client = { path = "../live_kit_client" }
gpui = { path = "../gpui" }
project = { path = "../project" }
util = { path = "../util" }
crates/call/src/room.rs 🔗
@@ -7,6 +7,7 @@ use client::{proto, Client, PeerId, TypedEnvelope, User, UserStore};
use collections::{BTreeMap, HashSet};
use futures::StreamExt;
use gpui::{AsyncAppContext, Entity, ModelContext, ModelHandle, MutableAppContext, Task};
+use live_kit_client::LocalVideoTrack;
use project::Project;
use std::sync::Arc;
use util::ResultExt;
@@ -26,6 +27,7 @@ pub enum Event {
pub struct Room {
id: u64,
+ live_kit_room: Option<Arc<live_kit_client::Room>>,
status: RoomStatus,
local_participant: LocalParticipant,
remote_participants: BTreeMap<PeerId, RemoteParticipant>,
@@ -50,6 +52,7 @@ impl Entity for Room {
impl Room {
fn new(
id: u64,
+ live_kit_connection_info: Option<proto::LiveKitConnectionInfo>,
client: Arc<Client>,
user_store: ModelHandle<UserStore>,
cx: &mut ModelContext<Self>,
@@ -69,8 +72,27 @@ impl Room {
})
.detach();
+ let live_kit_room = if let Some(connection_info) = live_kit_connection_info {
+ let room = live_kit_client::Room::new();
+ let mut tracks = room.remote_video_tracks();
+ cx.foreground()
+ .spawn(async move {
+ while let Some(track) = tracks.next().await {
+ dbg!("received track");
+ }
+ })
+ .detach();
+ cx.foreground()
+ .spawn(room.connect(&connection_info.server_url, &connection_info.token))
+ .detach_and_log_err(cx);
+ Some(room)
+ } else {
+ None
+ };
+
Self {
id,
+ live_kit_room,
status: RoomStatus::Online,
participant_user_ids: Default::default(),
local_participant: Default::default(),
@@ -95,7 +117,15 @@ impl Room {
cx.spawn(|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.add_model(|cx| Self::new(room_proto.id, client, user_store, cx));
+ let room = cx.add_model(|cx| {
+ Self::new(
+ room_proto.id,
+ response.live_kit_connection_info,
+ client,
+ user_store,
+ cx,
+ )
+ });
let initial_project_id = if let Some(initial_project) = initial_project {
let initial_project_id = room
@@ -131,7 +161,15 @@ impl Room {
cx.spawn(|mut cx| async move {
let response = client.request(proto::JoinRoom { id: room_id }).await?;
let room_proto = response.room.ok_or_else(|| anyhow!("invalid room"))?;
- let room = cx.add_model(|cx| Self::new(room_id, client, user_store, cx));
+ let room = cx.add_model(|cx| {
+ Self::new(
+ room_id,
+ response.live_kit_connection_info,
+ client,
+ user_store,
+ cx,
+ )
+ });
room.update(&mut cx, |room, cx| {
room.leave_when_empty = true;
room.apply_room_update(room_proto, cx)?;
@@ -458,6 +496,28 @@ impl Room {
Ok(())
})
}
+
+ pub fn share_screen(&mut self, cx: &mut ModelContext<Self>) -> Task<Result<()>> {
+ if self.status.is_offline() {
+ return Task::ready(Err(anyhow!("room is offline")));
+ }
+
+ let room = if let Some(room) = self.live_kit_room.as_ref() {
+ room.clone()
+ } else {
+ return Task::ready(Err(anyhow!("not connected to LiveKit")));
+ };
+
+ cx.foreground().spawn(async move {
+ let displays = live_kit_client::display_sources().await?;
+ let display = displays
+ .first()
+ .ok_or_else(|| anyhow!("no display found"))?;
+ let track = LocalVideoTrack::screen_share_for_display(display);
+ room.publish_video_track(&track).await?;
+ Ok(())
+ })
+ }
}
#[derive(Copy, Clone, PartialEq, Eq)]
crates/collab/src/rpc.rs 🔗
@@ -5,7 +5,7 @@ use crate::{
db::{self, ChannelId, MessageId, ProjectId, User, UserId},
AppState, Result,
};
-use anyhow::{anyhow, Context};
+use anyhow::anyhow;
use async_tungstenite::tungstenite::{
protocol::CloseFrame as TungsteniteCloseFrame, Message as TungsteniteMessage,
};
@@ -605,37 +605,34 @@ impl Server {
room = store.create_room(request.sender_id)?.clone();
}
- let live_kit_token = if let Some(live_kit) = self.app_state.live_kit_client.as_ref() {
- if let Some(_) = live_kit
- .create_room(room.live_kit_room.clone())
- .await
- .with_context(|| {
- format!(
- "error creating LiveKit room (LiveKit room: {}, Zed room: {})",
- room.live_kit_room, room.id
- )
- })
- .trace_err()
- {
- live_kit
- .room_token_for_user(&room.live_kit_room, &user_id.to_string())
- .with_context(|| {
- format!(
- "error creating LiveKit access token (LiveKit room: {}, Zed room: {})",
- room.live_kit_room, room.id
- )
- })
+ let live_kit_connection_info =
+ if let Some(live_kit) = self.app_state.live_kit_client.as_ref() {
+ if let Some(_) = live_kit
+ .create_room(room.live_kit_room.clone())
+ .await
.trace_err()
+ {
+ if let Some(token) = live_kit
+ .room_token_for_user(&room.live_kit_room, &user_id.to_string())
+ .trace_err()
+ {
+ Some(proto::LiveKitConnectionInfo {
+ server_url: live_kit.url().into(),
+ token,
+ })
+ } else {
+ None
+ }
+ } else {
+ None
+ }
} else {
None
- }
- } else {
- None
- };
+ };
response.send(proto::CreateRoomResponse {
room: Some(room),
- live_kit_token,
+ live_kit_connection_info,
})?;
self.update_user_contacts(user_id).await?;
Ok(())
@@ -658,23 +655,26 @@ impl Server {
.trace_err();
}
- let live_kit_token = if let Some(live_kit) = self.app_state.live_kit_client.as_ref() {
- live_kit
- .room_token_for_user(&room.live_kit_room, &user_id.to_string())
- .with_context(|| {
- format!(
- "error creating LiveKit access token (LiveKit room: {}, Zed room: {})",
- room.live_kit_room, room.id
- )
- })
- .trace_err()
- } else {
- None
- };
+ let live_kit_connection_info =
+ if let Some(live_kit) = self.app_state.live_kit_client.as_ref() {
+ if let Some(token) = live_kit
+ .room_token_for_user(&room.live_kit_room, &user_id.to_string())
+ .trace_err()
+ {
+ Some(proto::LiveKitConnectionInfo {
+ server_url: live_kit.url().into(),
+ token,
+ })
+ } else {
+ None
+ }
+ } else {
+ None
+ };
response.send(proto::JoinRoomResponse {
room: Some(room.clone()),
- live_kit_token,
+ live_kit_connection_info,
})?;
self.room_updated(room);
}
crates/live_kit_client/LiveKitBridge/Sources/LiveKitBridge/LiveKitBridge.swift 🔗
@@ -107,7 +107,7 @@ public func LKVideoTrackAddRenderer(track: UnsafeRawPointer, renderer: UnsafeRaw
@_cdecl("LKDisplaySources")
public func LKDisplaySources(data: UnsafeRawPointer, callback: @escaping @convention(c) (UnsafeRawPointer, CFArray?, CFString?) -> Void) {
- MacOSScreenCapturer.displaySources().then { displaySources in
+ MacOSScreenCapturer.sources(for: .display, includeCurrentApplication: false, preferredMethod: .legacy).then { displaySources in
callback(data, displaySources as CFArray, nil)
}.catch { error in
callback(data, nil, error.localizedDescription as CFString)
crates/live_kit_server/src/api.rs 🔗
@@ -7,25 +7,29 @@ use std::{future::Future, sync::Arc};
#[derive(Clone)]
pub struct Client {
http: reqwest::Client,
- uri: Arc<str>,
+ url: Arc<str>,
key: Arc<str>,
secret: Arc<str>,
}
impl Client {
- pub fn new(mut uri: String, key: String, secret: String) -> Self {
- if uri.ends_with('/') {
- uri.pop();
+ pub fn new(mut url: String, key: String, secret: String) -> Self {
+ if url.ends_with('/') {
+ url.pop();
}
Self {
http: reqwest::Client::new(),
- uri: uri.into(),
+ url: url.into(),
key: key.into(),
secret: secret.into(),
}
}
+ pub fn url(&self) -> &str {
+ &self.url
+ }
+
pub fn create_room(&self, name: String) -> impl Future<Output = Result<proto::Room>> {
self.request(
"twirp/livekit.RoomService/CreateRoom",
@@ -101,11 +105,11 @@ impl Client {
{
let client = self.http.clone();
let token = token::create(&self.key, &self.secret, None, grant);
- let uri = format!("{}/{}", self.uri, path);
+ let url = format!("{}/{}", self.url, path);
async move {
let token = token?;
let response = client
- .post(&uri)
+ .post(&url)
.header(CONTENT_TYPE, "application/protobuf")
.bearer_auth(token)
.body(body.encode_to_vec())
@@ -116,7 +120,7 @@ impl Client {
} else {
Err(anyhow!(
"POST {} failed with status code {:?}, {:?}",
- uri,
+ url,
response.status(),
response.text().await
))
crates/rpc/proto/zed.proto 🔗
@@ -141,7 +141,7 @@ message CreateRoom {}
message CreateRoomResponse {
Room room = 1;
- optional string live_kit_token = 2;
+ optional LiveKitConnectionInfo live_kit_connection_info = 2;
}
message JoinRoom {
@@ -150,7 +150,7 @@ message JoinRoom {
message JoinRoomResponse {
Room room = 1;
- optional string live_kit_token = 2;
+ optional LiveKitConnectionInfo live_kit_connection_info = 2;
}
message LeaveRoom {
@@ -225,6 +225,11 @@ message RoomUpdated {
Room room = 1;
}
+message LiveKitConnectionInfo {
+ string server_url = 1;
+ string token = 2;
+}
+
message ShareProject {
uint64 room_id = 1;
repeated WorktreeMetadata worktrees = 2;
crates/zed/build.rs 🔗
@@ -7,6 +7,12 @@ fn main() {
println!("cargo:rustc-env=ZED_AMPLITUDE_API_KEY={api_key}");
}
+ // Find WebRTC.framework as a sibling of the executable when running outside of an application bundle
+ println!("cargo:rustc-link-arg=-Wl,-rpath,@executable_path");
+
+ // Register exported Objective-C selectors, protocols, etc
+ println!("cargo:rustc-link-arg=-Wl,-ObjC");
+
let output = Command::new("npm")
.current_dir("../../styles")
.args(["install", "--no-save"])