Kordophone/kordophoned/src/dbus/server_impl.rs

use dbus::arg;
use dbus_tree::MethodErr;
use tokio::sync::mpsc;
use std::future::Future;
use std::thread;
use tokio::sync::oneshot;   

use crate::daemon::{
    DaemonResult,
    events::{Event, Reply},
    settings::Settings,
};

use crate::dbus::interface::NetBuzzertKordophoneRepository as DbusRepository;
use crate::dbus::interface::NetBuzzertKordophoneSettings as DbusSettings;

#[derive(Clone)]
pub struct ServerImpl {
    event_sink: mpsc::Sender<Event>,
}

impl ServerImpl {
    pub fn new(event_sink: mpsc::Sender<Event>) -> Self {
        Self { event_sink }
    }

    pub async fn send_event<T>(
        &self,
        make_event: impl FnOnce(Reply<T>) -> Event,
    ) -> DaemonResult<T> {
        let (reply_tx, reply_rx) = oneshot::channel();
        self.event_sink.send(make_event(reply_tx))
            .await
            .map_err(|_| "Failed to send event")?;
        
        reply_rx.await.map_err(|_| "Failed to receive reply".into())
    }

    pub fn send_event_sync<T: Send>(
        &self,
        make_event: impl FnOnce(Reply<T>) -> Event + Send,
    ) -> Result<T, MethodErr> {
        run_sync_future(self.send_event(make_event))
            .unwrap()
            .map_err(|e| MethodErr::failed(&format!("Daemon error: {}", e)))
    }
}

impl DbusRepository for ServerImpl {
    fn get_version(&mut self) -> Result<String, MethodErr> {
        self.send_event_sync(Event::GetVersion)
    }    

    fn get_conversations(&mut self) -> Result<Vec<arg::PropMap>, dbus::MethodErr> {
        self.send_event_sync(Event::GetAllConversations)
            .and_then(|conversations| {
                // Convert conversations to DBus property maps
                let result = conversations.into_iter().map(|conv| {
                    let mut map = arg::PropMap::new();
                    map.insert("guid".into(), arg::Variant(Box::new(conv.guid)));
                    map.insert("display_name".into(), arg::Variant(Box::new(conv.display_name.unwrap_or_default())));
                    map.insert("unread_count".into(), arg::Variant(Box::new(conv.unread_count as i32)));
                    map.insert("last_message_preview".into(), arg::Variant(Box::new(conv.last_message_preview.unwrap_or_default())));
                    map.insert("participants".into(), arg::Variant(Box::new(conv.participants.into_iter().map(|p| p.display_name()).collect::<Vec<String>>())));
                    map.insert("date".into(), arg::Variant(Box::new(conv.date.and_utc().timestamp())));
                    map
                }).collect();
                
                Ok(result)
            })
    }

    fn sync_all_conversations(&mut self) -> Result<(), dbus::MethodErr> {
        self.send_event_sync(Event::SyncAllConversations)
    }
} 

impl DbusSettings for ServerImpl {
    fn set_server(&mut self, url: String, user: String) -> Result<(), dbus::MethodErr> {
        self.send_event_sync(|r| 
            Event::UpdateSettings(Settings {
                server_url: Some(url),
                username: Some(user),
                credential_item: None,
            }, r)
        )
    }

    fn set_credential_item_(&mut self, item_path: dbus::Path<'static>) -> Result<(), dbus::MethodErr> {
        self.send_event_sync(|r| 
            Event::UpdateSettings(Settings {
                server_url: None,
                username: None,
                credential_item: Some(item_path.to_string()),
            }, r)
        )
    }

    fn server_url(&self) -> Result<String, dbus::MethodErr> {
        self.send_event_sync(Event::GetAllSettings)
            .and_then(|settings| {
                Ok(settings.server_url.unwrap_or_default())
            })
    }

    fn set_server_url(&self, value: String) -> Result<(), dbus::MethodErr> {
        self.send_event_sync(|r| 
            Event::UpdateSettings(Settings {
                server_url: Some(value),
                username: None,
                credential_item: None,
            }, r)
        )
    }

    fn username(&self) -> Result<String, dbus::MethodErr> {
        self.send_event_sync(Event::GetAllSettings)
            .and_then(|settings| {
                Ok(settings.username.unwrap_or_default())
            })
    }

    fn set_username(&self, value: String) -> Result<(), dbus::MethodErr> {
        self.send_event_sync(|r| 
            Event::UpdateSettings(Settings {
                server_url: None,
                username: Some(value),
                credential_item: None,
            }, r)
        )
    }

    fn credential_item(&self) -> Result<dbus::Path<'static>, dbus::MethodErr> {
        self.send_event_sync(Event::GetAllSettings)
            .and_then(|settings| {
                Ok(settings.credential_item.unwrap_or_default())
            })
            .and_then(|item| {
                Ok(dbus::Path::new(item).unwrap_or_default())
            })
    }

    fn set_credential_item(&self, value: dbus::Path<'static>) -> Result<(), dbus::MethodErr> {
        self.send_event_sync(|r| 
            Event::UpdateSettings(Settings {
                server_url: None,
                username: None,
                credential_item: Some(value.to_string()),
            }, r)
        )
    }

}

fn run_sync_future<F, T>(f: F) -> Result<T, MethodErr>
where
    T: Send,
    F: Future<Output = T> + Send,
{
    // We use `scope` here to ensure that the thread is joined before the
    // function returns. This allows us to capture references of values that
    // have lifetimes shorter than 'static, which is what thread::spawn requires.
    thread::scope(move |s| {
        s.spawn(move || {
            let rt = tokio::runtime::Builder::new_current_thread()
                .enable_all()
                .build()
                .map_err(|_| MethodErr::failed("Unable to create tokio runtime"))?;

            let result = rt.block_on(f);
            Ok(result)
        })
        .join()
    })
    .expect("Error joining runtime thread")
}