diff --git a/Cargo.toml b/Cargo.toml index 055f936..e2a29ba 100644 --- a/Cargo.toml +++ b/Cargo.toml @@ -39,3 +39,5 @@ objc = "0.2.7" pipewire = "0.8.0" dbus = "0.9.7" rand = "0.8.5" +xcb = { version = "1.4.0", features = ["randr", "xlib_xcb", "xfixes"] } +x11 = "2.21.0" \ No newline at end of file diff --git a/src/capturer/engine/linux/mod.rs b/src/capturer/engine/linux/mod.rs index 17ab2e4..25122b4 100644 --- a/src/capturer/engine/linux/mod.rs +++ b/src/capturer/engine/linux/mod.rs @@ -1,373 +1,41 @@ -use std::{ - mem::size_of, - sync::{ - atomic::{AtomicBool, AtomicU8}, - mpsc::{self, sync_channel, SyncSender}, - }, - thread::JoinHandle, - time::Duration, -}; +use std::{env, sync::mpsc}; -use pipewire as pw; -use pw::{ - context::Context, - main_loop::MainLoop, - properties::properties, - spa::{ - self, - param::{ - format::{FormatProperties, MediaSubtype, MediaType}, - video::VideoFormat, - ParamType, - }, - pod::{Pod, Property}, - sys::{ - spa_buffer, spa_meta_header, SPA_META_Header, SPA_PARAM_META_size, SPA_PARAM_META_type, - }, - utils::{Direction, SpaTypes}, - }, - stream::{StreamRef, StreamState}, -}; +use wayland::WaylandCapturer; +use x11::X11Capturer; -use crate::{ - capturer::Options, - frame::{BGRxFrame, Frame, RGBFrame, RGBxFrame, XBGRFrame}, -}; - -use self::{error::LinCapError, portal::ScreenCastPortal}; +use crate::{capturer::Options, frame::Frame}; mod error; -mod portal; - -static CAPTURER_STATE: AtomicU8 = AtomicU8::new(0); -static STREAM_STATE_CHANGED_TO_ERROR: AtomicBool = AtomicBool::new(false); - -#[derive(Clone)] -struct ListenerUserData { - pub tx: mpsc::Sender, - pub format: spa::param::video::VideoInfoRaw, -} - -fn param_changed_callback( - _stream: &StreamRef, - user_data: &mut ListenerUserData, - id: u32, - param: Option<&Pod>, -) { - let Some(param) = param else { - return; - }; - if id != pw::spa::param::ParamType::Format.as_raw() { - return; - } - let (media_type, media_subtype) = match pw::spa::param::format_utils::parse_format(param) { - Ok(v) => v, - Err(_) => return, - }; - - if media_type != MediaType::Video || media_subtype != MediaSubtype::Raw { - return; - } - user_data - .format - .parse(param) - // TODO: Tell library user of the error - .expect("Failed to parse format parameter"); -} +mod wayland; +mod x11; -fn state_changed_callback( - _stream: &StreamRef, - _user_data: &mut ListenerUserData, - _old: StreamState, - new: StreamState, -) { - match new { - StreamState::Error(e) => { - eprintln!("pipewire: State changed to error({e})"); - STREAM_STATE_CHANGED_TO_ERROR.store(true, std::sync::atomic::Ordering::Relaxed); - } - _ => {} - } -} - -unsafe fn get_timestamp(buffer: *mut spa_buffer) -> i64 { - let n_metas = (*buffer).n_metas; - if n_metas > 0 { - let mut meta_ptr = (*buffer).metas; - let metas_end = (*buffer).metas.wrapping_add(n_metas as usize); - while meta_ptr != metas_end { - if (*meta_ptr).type_ == SPA_META_Header { - let meta_header: &mut spa_meta_header = - &mut *((*meta_ptr).data as *mut spa_meta_header); - return meta_header.pts; - } - meta_ptr = meta_ptr.wrapping_add(1); - } - 0 - } else { - 0 - } -} - -fn process_callback(stream: &StreamRef, user_data: &mut ListenerUserData) { - let buffer = unsafe { stream.dequeue_raw_buffer() }; - if !buffer.is_null() { - 'outside: { - let buffer = unsafe { (*buffer).buffer }; - if buffer.is_null() { - break 'outside; - } - let timestamp = unsafe { get_timestamp(buffer) }; - - let n_datas = unsafe { (*buffer).n_datas }; - if n_datas < 1 { - return; - } - let frame_size = user_data.format.size(); - let frame_data: Vec = unsafe { - std::slice::from_raw_parts( - (*(*buffer).datas).data as *mut u8, - (*(*buffer).datas).maxsize as usize, - ) - .to_vec() - }; - - if let Err(e) = match user_data.format.format() { - VideoFormat::RGBx => user_data.tx.send(Frame::RGBx(RGBxFrame { - display_time: timestamp as u64, - width: frame_size.width as i32, - height: frame_size.height as i32, - data: frame_data, - })), - VideoFormat::RGB => user_data.tx.send(Frame::RGB(RGBFrame { - display_time: timestamp as u64, - width: frame_size.width as i32, - height: frame_size.height as i32, - data: frame_data, - })), - VideoFormat::xBGR => user_data.tx.send(Frame::XBGR(XBGRFrame { - display_time: timestamp as u64, - width: frame_size.width as i32, - height: frame_size.height as i32, - data: frame_data, - })), - VideoFormat::BGRx => user_data.tx.send(Frame::BGRx(BGRxFrame { - display_time: timestamp as u64, - width: frame_size.width as i32, - height: frame_size.height as i32, - data: frame_data, - })), - _ => panic!("Unsupported frame format received"), - } { - eprintln!("{e}"); - } - } - } else { - eprintln!("Out of buffers"); - } - - unsafe { stream.queue_raw_buffer(buffer) }; -} - -// TODO: Format negotiation -fn pipewire_capturer( - options: Options, - tx: mpsc::Sender, - ready_sender: &SyncSender, - stream_id: u32, -) -> Result<(), LinCapError> { - pw::init(); - - let mainloop = MainLoop::new(None)?; - let context = Context::new(&mainloop)?; - let core = context.connect(None)?; - - let user_data = ListenerUserData { - tx, - format: Default::default(), - }; - - let stream = pw::stream::Stream::new( - &core, - "scap", - properties! { - *pw::keys::MEDIA_TYPE => "Video", - *pw::keys::MEDIA_CATEGORY => "Capture", - *pw::keys::MEDIA_ROLE => "Screen", - }, - )?; - - let _listener = stream - .add_local_listener_with_user_data(user_data.clone()) - .state_changed(state_changed_callback) - .param_changed(param_changed_callback) - .process(process_callback) - .register()?; - - let obj = pw::spa::pod::object!( - pw::spa::utils::SpaTypes::ObjectParamFormat, - pw::spa::param::ParamType::EnumFormat, - pw::spa::pod::property!(FormatProperties::MediaType, Id, MediaType::Video), - pw::spa::pod::property!(FormatProperties::MediaSubtype, Id, MediaSubtype::Raw), - pw::spa::pod::property!( - FormatProperties::VideoFormat, - Choice, - Enum, - Id, - pw::spa::param::video::VideoFormat::RGB, - pw::spa::param::video::VideoFormat::RGBA, - pw::spa::param::video::VideoFormat::RGBx, - pw::spa::param::video::VideoFormat::BGRx, - ), - pw::spa::pod::property!( - FormatProperties::VideoSize, - Choice, - Range, - Rectangle, - pw::spa::utils::Rectangle { - // Default - width: 128, - height: 128, - }, - pw::spa::utils::Rectangle { - // Min - width: 1, - height: 1, - }, - pw::spa::utils::Rectangle { - // Max - width: 4096, - height: 4096, - } - ), - pw::spa::pod::property!( - FormatProperties::VideoFramerate, - Choice, - Range, - Fraction, - pw::spa::utils::Fraction { - num: options.fps, - denom: 1 - }, - pw::spa::utils::Fraction { num: 0, denom: 1 }, - pw::spa::utils::Fraction { - num: 1000, - denom: 1 - } - ), - ); - - let metas_obj = pw::spa::pod::object!( - SpaTypes::ObjectParamMeta, - ParamType::Meta, - Property::new( - SPA_PARAM_META_type, - pw::spa::pod::Value::Id(pw::spa::utils::Id(SPA_META_Header)) - ), - Property::new( - SPA_PARAM_META_size, - pw::spa::pod::Value::Int(size_of::() as i32) - ), - ); - - let values: Vec = pw::spa::pod::serialize::PodSerializer::serialize( - std::io::Cursor::new(Vec::new()), - &pw::spa::pod::Value::Object(obj), - )? - .0 - .into_inner(); - let metas_values: Vec = pw::spa::pod::serialize::PodSerializer::serialize( - std::io::Cursor::new(Vec::new()), - &pw::spa::pod::Value::Object(metas_obj), - )? - .0 - .into_inner(); - - let mut params = [ - pw::spa::pod::Pod::from_bytes(&values).unwrap(), - pw::spa::pod::Pod::from_bytes(&metas_values).unwrap(), - ]; - - stream.connect( - Direction::Input, - Some(stream_id), - pw::stream::StreamFlags::AUTOCONNECT | pw::stream::StreamFlags::MAP_BUFFERS, - &mut params, - )?; - - ready_sender.send(true)?; - - while CAPTURER_STATE.load(std::sync::atomic::Ordering::Relaxed) == 0 { - std::thread::sleep(Duration::from_millis(10)); - } - - let pw_loop = mainloop.loop_(); - - // User has called Capturer::start() and we start the main loop - while CAPTURER_STATE.load(std::sync::atomic::Ordering::Relaxed) == 1 - && /* If the stream state got changed to `Error`, we exit. TODO: tell user that we exited */ - !STREAM_STATE_CHANGED_TO_ERROR.load(std::sync::atomic::Ordering::Relaxed) - { - pw_loop.iterate(Duration::from_millis(100)); - } - - Ok(()) +pub trait LinuxCapturerImpl { + fn start_capture(&mut self); + fn stop_capture(&mut self); } pub struct LinuxCapturer { - capturer_join_handle: Option>>, - // The pipewire stream is deleted when the connection is dropped. - // That's why we keep it alive - _connection: dbus::blocking::Connection, + pub imp: Box, } -impl LinuxCapturer { - // TODO: Error handling - pub fn new(options: &Options, tx: mpsc::Sender) -> Self { - let connection = - dbus::blocking::Connection::new_session().expect("Failed to create dbus connection"); - let stream_id = ScreenCastPortal::new(&connection) - .show_cursor(options.show_cursor) - .expect("Unsupported cursor mode") - .create_stream() - .expect("Failed to get screencast stream") - .pw_node_id(); - - // TODO: Fix this hack - let options = options.clone(); - let (ready_sender, ready_recv) = sync_channel(1); - let capturer_join_handle = std::thread::spawn(move || { - let res = pipewire_capturer(options, tx, &ready_sender, stream_id); - if res.is_err() { - ready_sender.send(false)?; - } - res - }); +type Type = mpsc::Sender; - if !ready_recv.recv().expect("Failed to receive") { - panic!("Failed to setup capturer"); - } - - Self { - capturer_join_handle: Some(capturer_join_handle), - _connection: connection, - } - } - - pub fn start_capture(&self) { - CAPTURER_STATE.store(1, std::sync::atomic::Ordering::Relaxed); - } - - pub fn stop_capture(&mut self) { - CAPTURER_STATE.store(2, std::sync::atomic::Ordering::Relaxed); - if let Some(handle) = self.capturer_join_handle.take() { - if let Err(e) = handle.join().expect("Failed to join capturer thread") { - eprintln!("Error occured capturing: {e}"); - } +impl LinuxCapturer { + pub fn new(options: &Options, tx: Type) -> Self { + if env::var("WAYLAND_DISPLAY").is_ok() { + println!("[DEBUG] On wayland"); + return Self { + imp: Box::new(WaylandCapturer::new(options, tx)), + }; + } else if env::var("DISPLAY").is_ok() { + println!("[DEBUG] On X11"); + return Self { + imp: Box::new(X11Capturer::new(options, tx).unwrap()), + }; + } else { + panic!("Unsupported platform. Could not detect Wayland or X11 displays") } - CAPTURER_STATE.store(0, std::sync::atomic::Ordering::Relaxed); - STREAM_STATE_CHANGED_TO_ERROR.store(false, std::sync::atomic::Ordering::Relaxed); } } diff --git a/src/capturer/engine/linux/portal.rs b/src/capturer/engine/linux/portal.rs index 013d4dd..e69de29 100644 --- a/src/capturer/engine/linux/portal.rs +++ b/src/capturer/engine/linux/portal.rs @@ -1,422 +0,0 @@ -use std::{ - sync::{atomic::AtomicBool, Arc, Mutex}, - time::Duration, -}; - -use dbus::{ - arg::{self, PropMap, RefArg, Variant}, - blocking::{Connection, Proxy}, - message::MatchRule, - strings::{BusName, Interface}, -}; - -use super::error::LinCapError; - -// This code was autogenerated with `dbus-codegen-rust -d org.freedesktop.portal.Desktop -p /org/freedesktop/portal/desktop -f org.freedesktop.portal.ScreenCast`, see https://github.com/diwic/dbus-rs -// { -use dbus::blocking; - -trait OrgFreedesktopPortalScreenCast { - fn create_session(&self, options: arg::PropMap) -> Result, dbus::Error>; - fn select_sources( - &self, - session_handle: dbus::Path, - options: arg::PropMap, - ) -> Result, dbus::Error>; - fn start( - &self, - session_handle: dbus::Path, - parent_window: &str, - options: arg::PropMap, - ) -> Result, dbus::Error>; - fn open_pipe_wire_remote( - &self, - session_handle: dbus::Path, - options: arg::PropMap, - ) -> Result; - fn available_source_types(&self) -> Result; - fn available_cursor_modes(&self) -> Result; - fn version(&self) -> Result; -} - -impl<'a, T: blocking::BlockingSender, C: ::std::ops::Deref> - OrgFreedesktopPortalScreenCast for blocking::Proxy<'a, C> -{ - fn create_session(&self, options: arg::PropMap) -> Result, dbus::Error> { - self.method_call( - "org.freedesktop.portal.ScreenCast", - "CreateSession", - (options,), - ) - .and_then(|r: (dbus::Path<'static>,)| Ok(r.0)) - } - - fn select_sources( - &self, - session_handle: dbus::Path, - options: arg::PropMap, - ) -> Result, dbus::Error> { - self.method_call( - "org.freedesktop.portal.ScreenCast", - "SelectSources", - (session_handle, options), - ) - .and_then(|r: (dbus::Path<'static>,)| Ok(r.0)) - } - - fn start( - &self, - session_handle: dbus::Path, - parent_window: &str, - options: arg::PropMap, - ) -> Result, dbus::Error> { - self.method_call( - "org.freedesktop.portal.ScreenCast", - "Start", - (session_handle, parent_window, options), - ) - .and_then(|r: (dbus::Path<'static>,)| Ok(r.0)) - } - - fn open_pipe_wire_remote( - &self, - session_handle: dbus::Path, - options: arg::PropMap, - ) -> Result { - self.method_call( - "org.freedesktop.portal.ScreenCast", - "OpenPipeWireRemote", - (session_handle, options), - ) - .and_then(|r: (arg::OwnedFd,)| Ok(r.0)) - } - - fn available_source_types(&self) -> Result { - ::get( - &self, - "org.freedesktop.portal.ScreenCast", - "AvailableSourceTypes", - ) - } - - fn available_cursor_modes(&self) -> Result { - ::get( - &self, - "org.freedesktop.portal.ScreenCast", - "AvailableCursorModes", - ) - } - - fn version(&self) -> Result { - ::get( - &self, - "org.freedesktop.portal.ScreenCast", - "version", - ) - } -} -// } - -// This code was autogenerated with `dbus-codegen-rust --file org.freedesktop.portal.Request.xml`, see https://github.com/diwic/dbus-rs -// { -trait OrgFreedesktopPortalRequest { - fn close(&self) -> Result<(), dbus::Error>; -} - -#[derive(Debug)] -pub struct OrgFreedesktopPortalRequestResponse { - pub response: u32, - pub results: arg::PropMap, -} - -impl arg::AppendAll for OrgFreedesktopPortalRequestResponse { - fn append(&self, i: &mut arg::IterAppend) { - arg::RefArg::append(&self.response, i); - arg::RefArg::append(&self.results, i); - } -} - -impl arg::ReadAll for OrgFreedesktopPortalRequestResponse { - fn read(i: &mut arg::Iter) -> Result { - Ok(OrgFreedesktopPortalRequestResponse { - response: i.read()?, - results: i.read()?, - }) - } -} - -impl dbus::message::SignalArgs for OrgFreedesktopPortalRequestResponse { - const NAME: &'static str = "Response"; - const INTERFACE: &'static str = "org.freedesktop.portal.Request"; -} - -impl<'a, T: blocking::BlockingSender, C: ::std::ops::Deref> OrgFreedesktopPortalRequest - for blocking::Proxy<'a, C> -{ - fn close(&self) -> Result<(), dbus::Error> { - self.method_call("org.freedesktop.portal.Request", "Close", ()) - } -} -// } - -type Response = Option; - -#[derive(Debug)] -#[allow(dead_code)] -pub struct StreamVardict { - id: Option, - position: Option<(i32, i32)>, - size: Option<(i32, i32)>, - source_type: Option, - mapping_id: Option, -} - -#[derive(Debug)] -pub struct Stream(u32, StreamVardict); - -impl Stream { - pub fn pw_node_id(&self) -> u32 { - self.0 - } - - pub fn from_dbus(stream: &Variant>) -> Option { - let mut stream = stream.as_iter()?.next()?.as_iter()?; - let pipewire_node_id = stream.next()?.as_iter()?.next()?.as_u64()?; - - // TODO: Get the rest of the properties - - Some(Self( - pipewire_node_id as u32, - StreamVardict { - id: None, - position: None, - size: None, - source_type: None, - mapping_id: None, - }, - )) - } -} - -macro_rules! match_response { - ( $code:expr ) => { - match $code { - 0 => {} - 1 => { - return Err(LinCapError::new(String::from( - "User cancelled the interaction", - ))); - } - 2 => { - return Err(LinCapError::new(String::from( - "The user interaction was ended in some other way", - ))); - } - _ => unreachable!(), - } - }; -} - -pub struct ScreenCastPortal<'a> { - proxy: Proxy<'a, &'a Connection>, - token: String, - cursor_mode: u32, -} - -impl<'a> ScreenCastPortal<'a> { - pub fn new(connection: &'a Connection) -> Self { - let proxy = connection.with_proxy( - "org.freedesktop.portal.Desktop", - "/org/freedesktop/portal/desktop", - Duration::from_secs(4), - ); - - let token = format!("scap_{}", rand::random::()); - - Self { - proxy, - token, - cursor_mode: 1, - } - } - - fn create_session_args(&self) -> arg::PropMap { - let mut map = arg::PropMap::new(); - map.insert( - String::from("handle_token"), - Variant(Box::new(self.token.clone())), - ); - map.insert( - String::from("session_handle_token"), - Variant(Box::new(self.token.clone())), - ); - map - } - - fn select_sources_args(&self) -> Result { - let mut map = arg::PropMap::new(); - map.insert( - String::from("handle_token"), - Variant(Box::new(self.token.clone())), - ); - map.insert( - String::from("types"), - Variant(Box::new(self.proxy.available_source_types()?)), - ); - map.insert(String::from("multiple"), Variant(Box::new(false))); - map.insert( - String::from("cursor_mode"), - Variant(Box::new(self.cursor_mode)), - ); - Ok(map) - } - - fn handle_req_response( - connection: &Connection, - path: dbus::Path<'static>, - iterations: usize, - timeout: Duration, - response: Arc>, - ) -> Result<(), dbus::Error> { - let got_response = Arc::new(AtomicBool::new(false)); - let got_response_clone = Arc::clone(&got_response); - - let mut rule = MatchRule::new(); - rule.path = Some(dbus::Path::from(path)); - rule.msg_type = Some(dbus::MessageType::Signal); - rule.sender = Some(BusName::from("org.freedesktop.portal.Desktop")); - rule.interface = Some(Interface::from("org.freedesktop.portal.Request")); - connection.add_match( - rule, - move |res: OrgFreedesktopPortalRequestResponse, _chuh, _msg| { - let mut response = response.lock().expect("Failed to lock response mutex"); - *response = Some(res); - got_response_clone.store(true, std::sync::atomic::Ordering::Relaxed); - false - }, - )?; - - for _ in 0..iterations { - connection.process(timeout)?; - - if got_response.load(std::sync::atomic::Ordering::Relaxed) { - break; - } - } - - Ok(()) - } - - fn create_session(&self) -> Result { - let request_handle = self.proxy.create_session(self.create_session_args())?; - - let response = Arc::new(Mutex::new(None)); - let response_clone = Arc::clone(&response); - Self::handle_req_response( - self.proxy.connection, - request_handle, - 100, - Duration::from_millis(100), - response_clone, - )?; - - if let Some(res) = response.lock()?.take() { - match_response!(res.response); - match res - .results - .get("session_handle") - .map(|h| h.0.as_str().map(String::from)) - { - Some(h) => { - let p = dbus::Path::from(match h { - Some(p) => p, - None => { - return Err(LinCapError::new(String::from( - "Invalid session_handle received", - ))) - } - }); - - return Ok(p); - } - None => return Err(LinCapError::new(String::from("Did not get session handle"))), - } - } - - Err(LinCapError::new(String::from("Did not get response"))) - } - - fn select_sources(&self, session_handle: dbus::Path) -> Result<(), LinCapError> { - let request_handle = self - .proxy - .select_sources(session_handle, self.select_sources_args()?)?; - - let response = Arc::new(Mutex::new(None)); - let response_clone = Arc::clone(&response); - Self::handle_req_response( - self.proxy.connection, - request_handle, - 1200, // Wait 2 min - Duration::from_millis(100), - response_clone, - )?; - - if let Some(res) = response.lock()?.take() { - match_response!(res.response); - return Ok(()); - } - - Err(LinCapError::new(String::from("Did not get response"))) - } - - fn start(&self, session_handle: dbus::Path) -> Result { - let request_handle = self.proxy.start(session_handle, "", PropMap::new())?; - - let response = Arc::new(Mutex::new(None)); - let response_clone = Arc::clone(&response); - Self::handle_req_response( - self.proxy.connection, - request_handle, - 100, // Wait 10 s - Duration::from_millis(100), - response_clone, - )?; - - if let Some(res) = response.lock()?.take() { - match_response!(res.response); - match res.results.get("streams") { - Some(s) => match Stream::from_dbus(s) { - Some(s) => return Ok(s), - None => { - return Err(LinCapError::new(String::from( - "Failed to extract stream properties", - ))) - } - }, - None => return Err(LinCapError::new(String::from("Did not get any streams"))), - } - } - - Err(LinCapError::new(String::from("Did not get response"))) - } - - pub fn create_stream(&self) -> Result { - let session_handle = self.create_session()?; - self.select_sources(session_handle.clone())?; - self.start(session_handle) - } - - pub fn show_cursor(mut self, mode: bool) -> Result { - let available_modes = self.proxy.available_cursor_modes()?; - if mode && available_modes & 2 == 2 { - self.cursor_mode = 2; - return Ok(self); - } - if !mode && available_modes & 1 == 1 { - self.cursor_mode = 1; - return Ok(self); - } - - Err(LinCapError::new("Unsupported cursor mode".to_string())) - } -} diff --git a/src/capturer/engine/linux/wayland/mod.rs b/src/capturer/engine/linux/wayland/mod.rs new file mode 100644 index 0000000..a322d14 --- /dev/null +++ b/src/capturer/engine/linux/wayland/mod.rs @@ -0,0 +1,376 @@ +use std::{ + mem::size_of, + sync::{ + atomic::{AtomicBool, AtomicU8}, + mpsc::{sync_channel, Sender, SyncSender}, + }, + thread::JoinHandle, + time::Duration, +}; + +use pipewire as pw; +use pw::{ + context::Context, + main_loop::MainLoop, + properties::properties, + spa::{ + self, + param::{ + format::{FormatProperties, MediaSubtype, MediaType}, + video::VideoFormat, + ParamType, + }, + pod::{Pod, Property}, + sys::{ + spa_buffer, spa_meta_header, SPA_META_Header, SPA_PARAM_META_size, SPA_PARAM_META_type, + }, + utils::{Direction, SpaTypes}, + }, + stream::{StreamRef, StreamState}, +}; + +use crate::{ + capturer::Options, + frame::{BGRxFrame, Frame, RGBFrame, RGBxFrame, XBGRFrame}, +}; + +use self::portal::ScreenCastPortal; + +use super::{error::LinCapError, LinuxCapturerImpl}; + +mod portal; + +// TODO: Move to wayland capturer with Arc<> +static CAPTURER_STATE: AtomicU8 = AtomicU8::new(0); +static STREAM_STATE_CHANGED_TO_ERROR: AtomicBool = AtomicBool::new(false); + +#[derive(Clone)] +struct ListenerUserData { + pub tx: Sender, + pub format: spa::param::video::VideoInfoRaw, +} + +fn param_changed_callback( + _stream: &StreamRef, + user_data: &mut ListenerUserData, + id: u32, + param: Option<&Pod>, +) { + let Some(param) = param else { + return; + }; + if id != pw::spa::param::ParamType::Format.as_raw() { + return; + } + let (media_type, media_subtype) = match pw::spa::param::format_utils::parse_format(param) { + Ok(v) => v, + Err(_) => return, + }; + + if media_type != MediaType::Video || media_subtype != MediaSubtype::Raw { + return; + } + + user_data + .format + .parse(param) + // TODO: Tell library user of the error + .expect("Failed to parse format parameter"); +} + +fn state_changed_callback( + _stream: &StreamRef, + _user_data: &mut ListenerUserData, + _old: StreamState, + new: StreamState, +) { + match new { + StreamState::Error(e) => { + eprintln!("pipewire: State changed to error({e})"); + STREAM_STATE_CHANGED_TO_ERROR.store(true, std::sync::atomic::Ordering::Relaxed); + } + _ => {} + } +} + +unsafe fn get_timestamp(buffer: *mut spa_buffer) -> i64 { + let n_metas = (*buffer).n_metas; + if n_metas > 0 { + let mut meta_ptr = (*buffer).metas; + let metas_end = (*buffer).metas.wrapping_add(n_metas as usize); + while meta_ptr != metas_end { + if (*meta_ptr).type_ == SPA_META_Header { + let meta_header: &mut spa_meta_header = + &mut *((*meta_ptr).data as *mut spa_meta_header); + return meta_header.pts; + } + meta_ptr = meta_ptr.wrapping_add(1); + } + 0 + } else { + 0 + } +} + +fn process_callback(stream: &StreamRef, user_data: &mut ListenerUserData) { + let buffer = unsafe { stream.dequeue_raw_buffer() }; + if !buffer.is_null() { + 'outside: { + let buffer = unsafe { (*buffer).buffer }; + if buffer.is_null() { + break 'outside; + } + let timestamp = unsafe { get_timestamp(buffer) }; + + let n_datas = unsafe { (*buffer).n_datas }; + if n_datas < 1 { + return; + } + let frame_size = user_data.format.size(); + let frame_data: Vec = unsafe { + std::slice::from_raw_parts( + (*(*buffer).datas).data as *mut u8, + (*(*buffer).datas).maxsize as usize, + ) + .to_vec() + }; + + if let Err(e) = match user_data.format.format() { + VideoFormat::RGBx => user_data.tx.send(Frame::RGBx(RGBxFrame { + display_time: timestamp as u64, + width: frame_size.width as i32, + height: frame_size.height as i32, + data: frame_data, + })), + VideoFormat::RGB => user_data.tx.send(Frame::RGB(RGBFrame { + display_time: timestamp as u64, + width: frame_size.width as i32, + height: frame_size.height as i32, + data: frame_data, + })), + VideoFormat::xBGR => user_data.tx.send(Frame::XBGR(XBGRFrame { + display_time: timestamp as u64, + width: frame_size.width as i32, + height: frame_size.height as i32, + data: frame_data, + })), + VideoFormat::BGRx => user_data.tx.send(Frame::BGRx(BGRxFrame { + display_time: timestamp as u64, + width: frame_size.width as i32, + height: frame_size.height as i32, + data: frame_data, + })), + _ => panic!("Unsupported frame format received"), + } { + eprintln!("{e}"); + } + } + } else { + eprintln!("Out of buffers"); + } + + unsafe { stream.queue_raw_buffer(buffer) }; +} + +// TODO: Format negotiation +fn pipewire_capturer( + options: Options, + tx: Sender, + ready_sender: &SyncSender, + stream_id: u32, +) -> Result<(), LinCapError> { + pw::init(); + + let mainloop = MainLoop::new(None)?; + let context = Context::new(&mainloop)?; + let core = context.connect(None)?; + + let user_data = ListenerUserData { + tx, + format: Default::default(), + }; + + let stream = pw::stream::Stream::new( + &core, + "scap", + properties! { + *pw::keys::MEDIA_TYPE => "Video", + *pw::keys::MEDIA_CATEGORY => "Capture", + *pw::keys::MEDIA_ROLE => "Screen", + }, + )?; + + let _listener = stream + .add_local_listener_with_user_data(user_data.clone()) + .state_changed(state_changed_callback) + .param_changed(param_changed_callback) + .process(process_callback) + .register()?; + + let obj = pw::spa::pod::object!( + pw::spa::utils::SpaTypes::ObjectParamFormat, + pw::spa::param::ParamType::EnumFormat, + pw::spa::pod::property!(FormatProperties::MediaType, Id, MediaType::Video), + pw::spa::pod::property!(FormatProperties::MediaSubtype, Id, MediaSubtype::Raw), + pw::spa::pod::property!( + FormatProperties::VideoFormat, + Choice, + Enum, + Id, + pw::spa::param::video::VideoFormat::RGB, + pw::spa::param::video::VideoFormat::RGBA, + pw::spa::param::video::VideoFormat::RGBx, + pw::spa::param::video::VideoFormat::BGRx, + ), + pw::spa::pod::property!( + FormatProperties::VideoSize, + Choice, + Range, + Rectangle, + pw::spa::utils::Rectangle { + // Default + width: 128, + height: 128, + }, + pw::spa::utils::Rectangle { + // Min + width: 1, + height: 1, + }, + pw::spa::utils::Rectangle { + // Max + width: 4096, + height: 4096, + } + ), + pw::spa::pod::property!( + FormatProperties::VideoFramerate, + Choice, + Range, + Fraction, + pw::spa::utils::Fraction { + num: options.fps, + denom: 1 + }, + pw::spa::utils::Fraction { num: 0, denom: 1 }, + pw::spa::utils::Fraction { + num: 1000, + denom: 1 + } + ), + ); + + let metas_obj = pw::spa::pod::object!( + SpaTypes::ObjectParamMeta, + ParamType::Meta, + Property::new( + SPA_PARAM_META_type, + pw::spa::pod::Value::Id(pw::spa::utils::Id(SPA_META_Header)) + ), + Property::new( + SPA_PARAM_META_size, + pw::spa::pod::Value::Int(size_of::() as i32) + ), + ); + + let values: Vec = pw::spa::pod::serialize::PodSerializer::serialize( + std::io::Cursor::new(Vec::new()), + &pw::spa::pod::Value::Object(obj), + )? + .0 + .into_inner(); + let metas_values: Vec = pw::spa::pod::serialize::PodSerializer::serialize( + std::io::Cursor::new(Vec::new()), + &pw::spa::pod::Value::Object(metas_obj), + )? + .0 + .into_inner(); + + let mut params = [ + pw::spa::pod::Pod::from_bytes(&values).unwrap(), + pw::spa::pod::Pod::from_bytes(&metas_values).unwrap(), + ]; + + stream.connect( + Direction::Input, + Some(stream_id), + pw::stream::StreamFlags::AUTOCONNECT | pw::stream::StreamFlags::MAP_BUFFERS, + &mut params, + )?; + + ready_sender.send(true)?; + + while CAPTURER_STATE.load(std::sync::atomic::Ordering::Relaxed) == 0 { + std::thread::sleep(Duration::from_millis(10)); + } + + let pw_loop = mainloop.loop_(); + + // User has called Capturer::start() and we start the main loop + while CAPTURER_STATE.load(std::sync::atomic::Ordering::Relaxed) == 1 + && /* If the stream state got changed to `Error`, we exit. TODO: tell user that we exited */ + !STREAM_STATE_CHANGED_TO_ERROR.load(std::sync::atomic::Ordering::Relaxed) + { + pw_loop.iterate(Duration::from_millis(100)); + } + + Ok(()) +} + +pub struct WaylandCapturer { + capturer_join_handle: Option>>, + // The pipewire stream is deleted when the connection is dropped. + // That's why we keep it alive + _connection: dbus::blocking::Connection, +} + +impl WaylandCapturer { + // TODO: Error handling + pub fn new(options: &Options, tx: Sender) -> Self { + let connection = + dbus::blocking::Connection::new_session().expect("Failed to create dbus connection"); + let stream_id = ScreenCastPortal::new(&connection) + .show_cursor(options.show_cursor) + .expect("Unsupported cursor mode") + .create_stream() + .expect("Failed to get screencast stream") + .pw_node_id(); + + // TODO: Fix this hack + let options = options.clone(); + let (ready_sender, ready_recv) = sync_channel(1); + let capturer_join_handle = std::thread::spawn(move || { + let res = pipewire_capturer(options, tx, &ready_sender, stream_id); + if res.is_err() { + ready_sender.send(false)?; + } + res + }); + + if !ready_recv.recv().expect("Failed to receive") { + panic!("Failed to setup capturer"); + } + + Self { + capturer_join_handle: Some(capturer_join_handle), + _connection: connection, + } + } +} + +impl LinuxCapturerImpl for WaylandCapturer { + fn start_capture(&mut self) { + CAPTURER_STATE.store(1, std::sync::atomic::Ordering::Relaxed); + } + + fn stop_capture(&mut self) { + CAPTURER_STATE.store(2, std::sync::atomic::Ordering::Relaxed); + if let Some(handle) = self.capturer_join_handle.take() { + if let Err(e) = handle.join().expect("Failed to join capturer thread") { + eprintln!("Error occured capturing: {e}"); + } + } + CAPTURER_STATE.store(0, std::sync::atomic::Ordering::Relaxed); + STREAM_STATE_CHANGED_TO_ERROR.store(false, std::sync::atomic::Ordering::Relaxed); + } +} diff --git a/src/capturer/engine/linux/wayland/portal.rs b/src/capturer/engine/linux/wayland/portal.rs new file mode 100644 index 0000000..e321176 --- /dev/null +++ b/src/capturer/engine/linux/wayland/portal.rs @@ -0,0 +1,425 @@ +use std::{ + sync::{atomic::AtomicBool, Arc, Mutex}, + time::Duration, +}; + +use dbus::{ + arg::{self, PropMap, RefArg, Variant}, + blocking::{Connection, Proxy}, + message::MatchRule, + strings::{BusName, Interface}, +}; + +// This code was autogenerated with `dbus-codegen-rust -d org.freedesktop.portal.Desktop -p /org/freedesktop/portal/desktop -f org.freedesktop.portal.ScreenCast`, see https://github.com/diwic/dbus-rs +// { +use dbus::blocking; + +use crate::capturer::engine::linux::error::LinCapError; + +#[allow(unused)] +trait OrgFreedesktopPortalScreenCast { + fn create_session(&self, options: arg::PropMap) -> Result, dbus::Error>; + fn select_sources( + &self, + session_handle: dbus::Path, + options: arg::PropMap, + ) -> Result, dbus::Error>; + fn start( + &self, + session_handle: dbus::Path, + parent_window: &str, + options: arg::PropMap, + ) -> Result, dbus::Error>; + fn open_pipe_wire_remote( + &self, + session_handle: dbus::Path, + options: arg::PropMap, + ) -> Result; + fn available_source_types(&self) -> Result; + fn available_cursor_modes(&self) -> Result; + fn version(&self) -> Result; +} + +impl<'a, T: blocking::BlockingSender, C: ::std::ops::Deref> + OrgFreedesktopPortalScreenCast for blocking::Proxy<'a, C> +{ + fn create_session(&self, options: arg::PropMap) -> Result, dbus::Error> { + self.method_call( + "org.freedesktop.portal.ScreenCast", + "CreateSession", + (options,), + ) + .and_then(|r: (dbus::Path<'static>,)| Ok(r.0)) + } + + fn select_sources( + &self, + session_handle: dbus::Path, + options: arg::PropMap, + ) -> Result, dbus::Error> { + self.method_call( + "org.freedesktop.portal.ScreenCast", + "SelectSources", + (session_handle, options), + ) + .and_then(|r: (dbus::Path<'static>,)| Ok(r.0)) + } + + fn start( + &self, + session_handle: dbus::Path, + parent_window: &str, + options: arg::PropMap, + ) -> Result, dbus::Error> { + self.method_call( + "org.freedesktop.portal.ScreenCast", + "Start", + (session_handle, parent_window, options), + ) + .and_then(|r: (dbus::Path<'static>,)| Ok(r.0)) + } + + fn open_pipe_wire_remote( + &self, + session_handle: dbus::Path, + options: arg::PropMap, + ) -> Result { + self.method_call( + "org.freedesktop.portal.ScreenCast", + "OpenPipeWireRemote", + (session_handle, options), + ) + .and_then(|r: (arg::OwnedFd,)| Ok(r.0)) + } + + fn available_source_types(&self) -> Result { + ::get( + &self, + "org.freedesktop.portal.ScreenCast", + "AvailableSourceTypes", + ) + } + + fn available_cursor_modes(&self) -> Result { + ::get( + &self, + "org.freedesktop.portal.ScreenCast", + "AvailableCursorModes", + ) + } + + fn version(&self) -> Result { + ::get( + &self, + "org.freedesktop.portal.ScreenCast", + "version", + ) + } +} +// } + +// This code was autogenerated with `dbus-codegen-rust --file org.freedesktop.portal.Request.xml`, see https://github.com/diwic/dbus-rs +// { +#[allow(unused)] +trait OrgFreedesktopPortalRequest { + fn close(&self) -> Result<(), dbus::Error>; +} + +#[derive(Debug)] +pub struct OrgFreedesktopPortalRequestResponse { + pub response: u32, + pub results: arg::PropMap, +} + +impl arg::AppendAll for OrgFreedesktopPortalRequestResponse { + fn append(&self, i: &mut arg::IterAppend) { + arg::RefArg::append(&self.response, i); + arg::RefArg::append(&self.results, i); + } +} + +impl arg::ReadAll for OrgFreedesktopPortalRequestResponse { + fn read(i: &mut arg::Iter) -> Result { + Ok(OrgFreedesktopPortalRequestResponse { + response: i.read()?, + results: i.read()?, + }) + } +} + +impl dbus::message::SignalArgs for OrgFreedesktopPortalRequestResponse { + const NAME: &'static str = "Response"; + const INTERFACE: &'static str = "org.freedesktop.portal.Request"; +} + +impl<'a, T: blocking::BlockingSender, C: ::std::ops::Deref> OrgFreedesktopPortalRequest + for blocking::Proxy<'a, C> +{ + fn close(&self) -> Result<(), dbus::Error> { + self.method_call("org.freedesktop.portal.Request", "Close", ()) + } +} +// } + +type Response = Option; + +#[derive(Debug)] +#[allow(dead_code)] +pub struct StreamVardict { + id: Option, + position: Option<(i32, i32)>, + size: Option<(i32, i32)>, + source_type: Option, + mapping_id: Option, +} + +#[derive(Debug)] +#[allow(unused)] +pub struct Stream(u32, StreamVardict); + +impl Stream { + pub fn pw_node_id(&self) -> u32 { + self.0 + } + + pub fn from_dbus(stream: &Variant>) -> Option { + let mut stream = stream.as_iter()?.next()?.as_iter()?; + let pipewire_node_id = stream.next()?.as_iter()?.next()?.as_u64()?; + + // TODO: Get the rest of the properties + + Some(Self( + pipewire_node_id as u32, + StreamVardict { + id: None, + position: None, + size: None, + source_type: None, + mapping_id: None, + }, + )) + } +} + +macro_rules! match_response { + ( $code:expr ) => { + match $code { + 0 => {} + 1 => { + return Err(LinCapError::new(String::from( + "User cancelled the interaction", + ))); + } + 2 => { + return Err(LinCapError::new(String::from( + "The user interaction was ended in some other way", + ))); + } + _ => unreachable!(), + } + }; +} + +pub struct ScreenCastPortal<'a> { + proxy: Proxy<'a, &'a Connection>, + token: String, + cursor_mode: u32, +} + +impl<'a> ScreenCastPortal<'a> { + pub fn new(connection: &'a Connection) -> Self { + let proxy = connection.with_proxy( + "org.freedesktop.portal.Desktop", + "/org/freedesktop/portal/desktop", + Duration::from_secs(4), + ); + + let token = format!("scap_{}", rand::random::()); + + Self { + proxy, + token, + cursor_mode: 1, + } + } + + fn create_session_args(&self) -> arg::PropMap { + let mut map = arg::PropMap::new(); + map.insert( + String::from("handle_token"), + Variant(Box::new(self.token.clone())), + ); + map.insert( + String::from("session_handle_token"), + Variant(Box::new(self.token.clone())), + ); + map + } + + fn select_sources_args(&self) -> Result { + let mut map = arg::PropMap::new(); + map.insert( + String::from("handle_token"), + Variant(Box::new(self.token.clone())), + ); + map.insert( + String::from("types"), + Variant(Box::new(self.proxy.available_source_types()?)), + ); + map.insert(String::from("multiple"), Variant(Box::new(false))); + map.insert( + String::from("cursor_mode"), + Variant(Box::new(self.cursor_mode)), + ); + Ok(map) + } + + fn handle_req_response( + connection: &Connection, + path: dbus::Path<'static>, + iterations: usize, + timeout: Duration, + response: Arc>, + ) -> Result<(), dbus::Error> { + let got_response = Arc::new(AtomicBool::new(false)); + let got_response_clone = Arc::clone(&got_response); + + let mut rule = MatchRule::new(); + rule.path = Some(dbus::Path::from(path)); + rule.msg_type = Some(dbus::MessageType::Signal); + rule.sender = Some(BusName::from("org.freedesktop.portal.Desktop")); + rule.interface = Some(Interface::from("org.freedesktop.portal.Request")); + connection.add_match( + rule, + move |res: OrgFreedesktopPortalRequestResponse, _chuh, _msg| { + let mut response = response.lock().expect("Failed to lock response mutex"); + *response = Some(res); + got_response_clone.store(true, std::sync::atomic::Ordering::Relaxed); + false + }, + )?; + + for _ in 0..iterations { + connection.process(timeout)?; + + if got_response.load(std::sync::atomic::Ordering::Relaxed) { + break; + } + } + + Ok(()) + } + + fn create_session(&self) -> Result { + let request_handle = self.proxy.create_session(self.create_session_args())?; + + let response = Arc::new(Mutex::new(None)); + let response_clone = Arc::clone(&response); + Self::handle_req_response( + self.proxy.connection, + request_handle, + 100, + Duration::from_millis(100), + response_clone, + )?; + + if let Some(res) = response.lock()?.take() { + match_response!(res.response); + match res + .results + .get("session_handle") + .map(|h| h.0.as_str().map(String::from)) + { + Some(h) => { + let p = dbus::Path::from(match h { + Some(p) => p, + None => { + return Err(LinCapError::new(String::from( + "Invalid session_handle received", + ))) + } + }); + + return Ok(p); + } + None => return Err(LinCapError::new(String::from("Did not get session handle"))), + } + } + + Err(LinCapError::new(String::from("Did not get response"))) + } + + fn select_sources(&self, session_handle: dbus::Path) -> Result<(), LinCapError> { + let request_handle = self + .proxy + .select_sources(session_handle, self.select_sources_args()?)?; + + let response = Arc::new(Mutex::new(None)); + let response_clone = Arc::clone(&response); + Self::handle_req_response( + self.proxy.connection, + request_handle, + 1200, // Wait 2 min + Duration::from_millis(100), + response_clone, + )?; + + if let Some(res) = response.lock()?.take() { + match_response!(res.response); + return Ok(()); + } + + Err(LinCapError::new(String::from("Did not get response"))) + } + + fn start(&self, session_handle: dbus::Path) -> Result { + let request_handle = self.proxy.start(session_handle, "", PropMap::new())?; + + let response = Arc::new(Mutex::new(None)); + let response_clone = Arc::clone(&response); + Self::handle_req_response( + self.proxy.connection, + request_handle, + 100, // Wait 10 s + Duration::from_millis(100), + response_clone, + )?; + + if let Some(res) = response.lock()?.take() { + match_response!(res.response); + match res.results.get("streams") { + Some(s) => match Stream::from_dbus(s) { + Some(s) => return Ok(s), + None => { + return Err(LinCapError::new(String::from( + "Failed to extract stream properties", + ))) + } + }, + None => return Err(LinCapError::new(String::from("Did not get any streams"))), + } + } + + Err(LinCapError::new(String::from("Did not get response"))) + } + + pub fn create_stream(&self) -> Result { + let session_handle = self.create_session()?; + self.select_sources(session_handle.clone())?; + self.start(session_handle) + } + + pub fn show_cursor(mut self, mode: bool) -> Result { + let available_modes = self.proxy.available_cursor_modes()?; + if mode && available_modes & 2 == 2 { + self.cursor_mode = 2; + return Ok(self); + } + if !mode && available_modes & 1 == 1 { + self.cursor_mode = 1; + return Ok(self); + } + + Err(LinCapError::new("Unsupported cursor mode".to_string())) + } +} diff --git a/src/capturer/engine/linux/x11/mod.rs b/src/capturer/engine/linux/x11/mod.rs new file mode 100644 index 0000000..879216d --- /dev/null +++ b/src/capturer/engine/linux/x11/mod.rs @@ -0,0 +1,251 @@ +use std::{ + sync::{ + atomic::{AtomicU8, Ordering}, + mpsc::Sender, + Arc, + }, + thread::JoinHandle, +}; + +use xcb::{x, Xid}; + +use crate::{capturer::Options, frame::Frame, targets::linux::get_default_x_display, Target}; + +use super::{error::LinCapError, LinuxCapturerImpl}; + +pub struct X11Capturer { + capturer_join_handle: Option>>, + capturer_state: Arc, +} + +fn draw_cursor( + conn: &xcb::Connection, + img: &mut [u8], + win_x: i16, + win_y: i16, + win_width: i16, + win_height: i16, + is_win: bool, + win: &xcb::x::Window, +) -> Result<(), xcb::Error> { + let cursor_image_cookie = conn.send_request(&xcb::xfixes::GetCursorImage {}); + let cursor_image = conn.wait_for_reply(cursor_image_cookie)?; + + let win_x = win_x as i32; + let win_y = win_y as i32; + + let win_width = win_width as i32; + let win_height = win_height as i32; + + let mut cursor_x = cursor_image.x() as i32 - cursor_image.xhot() as i32; + let mut cursor_y = cursor_image.y() as i32 - cursor_image.yhot() as i32; + if is_win { + let disp = conn.get_raw_dpy(); + let mut ncursor_x = 0; + let mut ncursor_y = 0; + let mut child_return = 0; + if unsafe { + x11::xlib::XTranslateCoordinates( + disp, + x11::xlib::XDefaultRootWindow(disp), + win.resource_id() as u64, + cursor_image.x() as i32, + cursor_image.y() as i32, + &mut ncursor_x, + &mut ncursor_y, + &mut child_return, + ) + } == 0 + { + return Ok(()); + } + cursor_x = ncursor_x as i32 - cursor_image.xhot() as i32; + cursor_y = ncursor_y as i32 - cursor_image.yhot() as i32; + } + + if cursor_x >= win_width + win_x + || cursor_y >= win_height + win_y + || cursor_x < win_x + || cursor_y < win_y + { + return Ok(()); + } + + let x = cursor_x.max(win_x); + let y = cursor_y.max(win_y); + + let w = ((cursor_x + cursor_image.width() as i32).min(win_x + win_width) - x) as u32; + let h = ((cursor_y + cursor_image.height() as i32).min(win_y + win_height) - y) as u32; + + let c_off = (x - cursor_x) as u32; + let i_off: i32 = x - win_x; + + let stride: u32 = 4; + let mut cursor_idx: u32 = ((y - cursor_y) * cursor_image.width() as i32) as u32; + let mut image_idx: u32 = ((y - win_y) * win_width * stride as i32) as u32; + + for y in 0..h { + cursor_idx += c_off; + image_idx += i_off as u32 * stride; + for x in 0..w { + let cursor_pix = cursor_image.cursor_image()[cursor_idx as usize]; + let r = (cursor_pix & 0xFF) as u8; + let g = ((cursor_pix >> 8) & 0xFF) as u8; + let b = ((cursor_pix >> 16) & 0xFF) as u8; + let a = ((cursor_pix >> 24) & 0xFF); + + let i = image_idx as usize; + if a == 0xFF { + img[i] = r; + img[i + 1] = g; + img[i + 2] = b; + } else if a > 0 { + let a = 255 - a; + img[i] = r + ((img[i] as u32 * a + 255 / 2) / 255) as u8; + img[i + 1] = g + ((img[i + 1] as u32 * a + 255 / 2) / 255) as u8; + img[i + 2] = b + ((img[i + 2] as u32 * a + 255 / 2) / 255) as u8; + } + + cursor_idx += 1; + image_idx += stride; + } + cursor_idx += cursor_image.width() as u32 - w as u32 - c_off as u32; + image_idx += (win_width - w as i32 - i_off) as u32 * stride; + } + + Ok(()) +} + +fn grab(conn: &xcb::Connection, target: &Target, show_cursor: bool) -> Result { + let (x, y, width, height, window, is_win) = match &target { + Target::Window(win) => { + let geom_cookie = conn.send_request(&x::GetGeometry { + drawable: x::Drawable::Window(win.raw_handle), + }); + let geom = conn.wait_for_reply(geom_cookie)?; + (0, 0, geom.width(), geom.height(), win.raw_handle, true) + } + Target::Display(disp) => ( + disp.x_offset, + disp.y_offset, + disp.width, + disp.height, + disp.raw_handle, + false, + ), + }; + + let img_cookie = conn.send_request(&x::GetImage { + format: x::ImageFormat::ZPixmap, + drawable: x::Drawable::Window(window), + x, + y, + width, + height, + plane_mask: u32::MAX, + }); + let img = conn.wait_for_reply(img_cookie)?; + + let mut img_data = img.data().to_vec(); + + if show_cursor { + draw_cursor( + &conn, + &mut img_data, + x, + y, + width as i16, + height as i16, + is_win, + &window, + )?; + } + + Ok(Frame::BGRx(crate::frame::BGRxFrame { + display_time: std::time::SystemTime::now() + .duration_since(std::time::UNIX_EPOCH) + .expect("Unix epoch is in the past") + .as_nanos() as u64, + width: width as i32, + height: height as i32, + data: img_data, + })) +} + +fn query_xfixes_version(conn: &xcb::Connection) -> Result<(), xcb::Error> { + let cookie = conn.send_request(&xcb::xfixes::QueryVersion { + client_major_version: xcb::xfixes::MAJOR_VERSION, + client_minor_version: xcb::xfixes::MINOR_VERSION, + }); + let _ = conn.wait_for_reply(cookie)?; + Ok(()) +} + +impl X11Capturer { + pub fn new(options: &Options, tx: Sender) -> Result { + let (conn, screen_num) = xcb::Connection::connect_with_xlib_display_and_extensions( + &[xcb::Extension::RandR, xcb::Extension::XFixes], + &[], + ) + .map_err(|e| LinCapError::new(e.to_string()))?; + query_xfixes_version(&conn).map_err(|e| LinCapError::new(e.to_string()))?; + let setup = conn.get_setup(); + let Some(screen) = setup.roots().nth(screen_num as usize) else { + return Err(LinCapError::new(String::from("Failed to get setup root"))); + }; + + let target = match &options.target { + Some(t) => t.clone(), + None => Target::Display( + get_default_x_display(&conn, screen) + .map_err(|e| LinCapError::new(e.to_string()))?, + ), + }; + + let framerate = options.fps as f32; + let show_cursor = options.show_cursor; + let capturer_state = Arc::new(AtomicU8::new(0)); + let capturer_state_clone = Arc::clone(&capturer_state); + + let jh = std::thread::spawn(move || { + while capturer_state_clone.load(Ordering::Acquire) == 0 { + std::thread::sleep(std::time::Duration::from_millis(10)); + } + + let frame_time = std::time::Duration::from_secs_f32(1.0 / framerate); + while capturer_state_clone.load(Ordering::Acquire) == 1 { + let start = std::time::Instant::now(); + + let frame = grab(&conn, &target, show_cursor)?; + tx.send(frame).unwrap(); + + let elapsed = start.elapsed(); + if elapsed < frame_time { + std::thread::sleep(frame_time - start.elapsed()); + } + } + + Ok(()) + }); + + Ok(Self { + capturer_state, + capturer_join_handle: Some(jh), + }) + } +} + +impl LinuxCapturerImpl for X11Capturer { + fn start_capture(&mut self) { + self.capturer_state.store(1, Ordering::Release); + } + + fn stop_capture(&mut self) { + self.capturer_state.store(2, Ordering::Release); + if let Some(handle) = self.capturer_join_handle.take() { + if let Err(e) = handle.join().expect("Failed to join capturer thread") { + eprintln!("Error occured capturing: {e}"); + } + } + } +} diff --git a/src/capturer/engine/mod.rs b/src/capturer/engine/mod.rs index f31de70..fdfcc3e 100644 --- a/src/capturer/engine/mod.rs +++ b/src/capturer/engine/mod.rs @@ -68,7 +68,7 @@ impl Engine { #[cfg(target_os = "linux")] { - self.linux.start_capture(); + self.linux.imp.start_capture(); } } @@ -85,7 +85,7 @@ impl Engine { #[cfg(target_os = "linux")] { - self.linux.stop_capture(); + self.linux.imp.stop_capture(); } } diff --git a/src/targets/linux/mod.rs b/src/targets/linux/mod.rs index fd0ccdc..d721e6f 100644 --- a/src/targets/linux/mod.rs +++ b/src/targets/linux/mod.rs @@ -1,7 +1,236 @@ -use super::Target; +use std::ffi::{CStr, CString, NulError}; + +use super::{Display, Target}; + +use x11::xlib::{XFreeStringList, XGetTextProperty, XTextProperty, XmbTextPropertyToTextList}; +use xcb::{ + randr::{GetCrtcInfo, GetOutputInfo, GetOutputPrimary, GetScreenResources}, + x::{self, GetPropertyReply, Screen}, + Xid, +}; + +fn get_atom(conn: &xcb::Connection, atom_name: &str) -> Result { + let cookie = conn.send_request(&x::InternAtom { + only_if_exists: true, + name: atom_name.as_bytes(), + }); + Ok(conn.wait_for_reply(cookie)?.atom()) +} + +fn get_property( + conn: &xcb::Connection, + win: x::Window, + prop: x::Atom, + typ: x::Atom, + length: u32, +) -> Result { + let cookie = conn.send_request(&x::GetProperty { + delete: false, + window: win, + property: prop, + r#type: typ, + long_offset: 0, + long_length: length, + }); + Ok(conn.wait_for_reply(cookie)?) +} + +fn decode_compound_text( + conn: &xcb::Connection, + value: &[u8], + client: &xcb::x::Window, + ttype: xcb::x::Atom, +) -> Result { + let display = conn.get_raw_dpy(); + assert!(!display.is_null()); + + let c_string = CString::new(value.to_vec())?; + let mut text_prop = XTextProperty { + value: std::ptr::null_mut(), + encoding: 0, + format: 0, + nitems: 0, + }; + let res = unsafe { + XGetTextProperty( + display, + client.resource_id() as u64, + &mut text_prop, + x::ATOM_WM_NAME.resource_id() as u64, + ) + }; + if res == 0 || text_prop.nitems == 0 { + return Ok(String::from("n/a")); + } + + let mut xname = XTextProperty { + value: c_string.as_ptr() as *mut u8, + encoding: ttype.resource_id() as u64, + format: 8, + nitems: text_prop.nitems, + }; + let mut list: *mut *mut i8 = std::ptr::null_mut(); + let mut count: i32 = 0; + let result = unsafe { XmbTextPropertyToTextList(display, &mut xname, &mut list, &mut count) }; + if result < 1 || list.is_null() || count < 1 { + Ok(String::from("n/a")) + } else { + let title = unsafe { CStr::from_ptr(*list).to_string_lossy().into_owned() }; + unsafe { XFreeStringList(list) }; + Ok(title) + } +} + +fn get_x11_targets() -> Result, xcb::Error> { + let (conn, _screen_num) = + xcb::Connection::connect_with_xlib_display_and_extensions(&[xcb::Extension::RandR], &[])?; + let setup = conn.get_setup(); + let screens = setup.roots(); + + let wm_client_list = get_atom(&conn, "_NET_CLIENT_LIST")?; + assert!(wm_client_list != x::ATOM_NONE, "EWMH not supported"); + + let atom_net_wm_name = get_atom(&conn, "_NET_WM_NAME")?; + let atom_text = get_atom(&conn, "TEXT")?; + let atom_utf8_string = get_atom(&conn, "UTF8_STRING")?; + let atom_compound_text = get_atom(&conn, "COMPOUND_TEXT")?; + + let mut targets = Vec::new(); + for screen in screens { + let window_list = get_property(&conn, screen.root(), wm_client_list, x::ATOM_NONE, 100)?; + + for client in window_list.value::() { + let cr = get_property(&conn, *client, atom_net_wm_name, x::ATOM_STRING, 4096)?; + if !cr.value::().is_empty() { + targets.push(Target::Window(crate::targets::Window { + id: 0, + title: String::from_utf8(cr.value().to_vec()) + .map_err(|_| xcb::Error::Connection(xcb::ConnError::ClosedParseErr))?, + raw_handle: *client, + })); + continue; + } + + let reply = get_property(&conn, *client, x::ATOM_WM_NAME, x::ATOM_ANY, 4096)?; + let value: &[u8] = reply.value(); + if !value.is_empty() { + let ttype = reply.r#type(); + let title = + if ttype == x::ATOM_STRING || ttype == atom_utf8_string || ttype == atom_text { + String::from_utf8(reply.value().to_vec()).unwrap_or(String::from("n/a")) + } else if ttype == atom_compound_text { + decode_compound_text(&conn, value, client, ttype) + .map_err(|_| xcb::Error::Connection(xcb::ConnError::ClosedParseErr))? + } else { + String::from_utf8(reply.value().to_vec()).unwrap_or(String::from("n/a")) + }; + + targets.push(Target::Window(crate::targets::Window { + id: 0, + title, + raw_handle: *client, + })); + continue; + } + targets.push(Target::Window(crate::targets::Window { + id: 0, + title: String::from("n/a"), + raw_handle: *client, + })); + } + + let resources = conn.send_request(&GetScreenResources { + window: screen.root(), + }); + let resources = conn.wait_for_reply(resources)?; + for output in resources.outputs() { + let info = conn.send_request(&GetOutputInfo { + output: *output, + config_timestamp: 0, + }); + let info = conn.wait_for_reply(info)?; + if info.connection() == xcb::randr::Connection::Connected { + let crtc = info.crtc(); + let crtc_info = conn.send_request(&GetCrtcInfo { + crtc, + config_timestamp: 0, + }); + let crtc_info = conn.wait_for_reply(crtc_info)?; + let title = String::from_utf8(info.name().to_vec()).unwrap_or(String::from("n/a")); + targets.push(Target::Display(crate::targets::Display { + id: crtc.resource_id(), + title, + width: crtc_info.width(), + height: crtc_info.height(), + x_offset: crtc_info.x(), + y_offset: crtc_info.y(), + raw_handle: screen.root(), + })); + } + } + } + + Ok(targets) +} -// On Linux, the target is selected when a Recorder is instanciated because this -// requires user interaction pub fn get_all_targets() -> Vec { - Vec::new() + if std::env::var("WAYLAND_DISPLAY").is_ok() { + // On Wayland, the target is selected when a Recorder is instanciated because it requires user interaction + Vec::new() + } else if std::env::var("DISPLAY").is_ok() { + get_x11_targets().unwrap() + } else { + panic!("Unsupported platform. Could not detect Wayland or X11 displays") + } +} + +pub(crate) fn get_default_x_display( + conn: &xcb::Connection, + screen: &Screen, +) -> Result { + let primary_display_cookie = conn.send_request(&GetOutputPrimary { + window: screen.root(), + }); + let primary_display = conn.wait_for_reply(primary_display_cookie)?; + let info_cookie = conn.send_request(&GetOutputInfo { + output: primary_display.output(), + config_timestamp: 0, + }); + let info = conn.wait_for_reply(info_cookie)?; + let crtc = info.crtc(); + let crtc_info_cookie = conn.send_request(&GetCrtcInfo { + crtc, + config_timestamp: 0, + }); + let crtc_info = conn.wait_for_reply(crtc_info_cookie)?; + Ok(Display { + id: crtc.resource_id(), + title: String::from_utf8(info.name().to_vec()).unwrap_or(String::from("default")), + width: crtc_info.width(), + height: crtc_info.height(), + x_offset: crtc_info.x(), + y_offset: crtc_info.y(), + raw_handle: screen.root(), + }) +} + +pub fn get_main_display() -> Display { + if std::env::var("WAYLAND_DISPLAY").is_ok() { + todo!() + } else if std::env::var("DISPLAY").is_ok() { + let (conn, screen_num) = + xcb::Connection::connect_with_extensions(None, &[xcb::Extension::RandR], &[]).unwrap(); + let setup = conn.get_setup(); + let screen = setup.roots().nth(screen_num as usize).unwrap(); + get_default_x_display(&conn, screen).unwrap() + } else { + panic!("Unsupported platform. Could not detect Wayland or X11 displays") + } +} + +pub fn get_target_dimensions(target: &Target) -> (u64, u64) { + match target { + Target::Window(_w) => (0, 0), // TODO + Target::Display(d) => (d.width as u64, d.height as u64), + } } diff --git a/src/targets/mod.rs b/src/targets/mod.rs index 930ab62..c783092 100644 --- a/src/targets/mod.rs +++ b/src/targets/mod.rs @@ -5,7 +5,7 @@ mod mac; mod win; #[cfg(target_os = "linux")] -mod linux; +pub(crate) mod linux; #[derive(Debug, Clone)] pub struct Window { @@ -17,6 +17,9 @@ pub struct Window { #[cfg(target_os = "macos")] pub raw_handle: core_graphics_helmer_fork::window::CGWindowID, + + #[cfg(target_os = "linux")] + pub raw_handle: xcb::x::Window, } #[derive(Debug, Clone)] @@ -29,6 +32,17 @@ pub struct Display { #[cfg(target_os = "macos")] pub raw_handle: core_graphics_helmer_fork::display::CGDisplay, + + #[cfg(target_os = "linux")] + pub raw_handle: xcb::x::Window, + #[cfg(target_os = "linux")] + pub width: u16, + #[cfg(target_os = "linux")] + pub height: u16, + #[cfg(target_os = "linux")] + pub x_offset: i16, + #[cfg(target_os = "linux")] + pub y_offset: i16, } #[derive(Debug, Clone)] @@ -68,7 +82,7 @@ pub fn get_main_display() -> Display { return win::get_main_display(); #[cfg(target_os = "linux")] - unreachable!(); + return linux::get_main_display(); } pub fn get_target_dimensions(target: &Target) -> (u64, u64) { @@ -79,5 +93,5 @@ pub fn get_target_dimensions(target: &Target) -> (u64, u64) { return win::get_target_dimensions(target); #[cfg(target_os = "linux")] - unreachable!(); + return linux::get_target_dimensions(target); }