buzzert/Kordophone
Private
Public Access
1
0
Fork 0
Code Packages Activity

dbus: Smaller GetMessages buffers, avoid encoding attachment paths.

Browse Source
This commit is contained in:
James Magahern
2026-02-21 22:22:09 -08:00
parent a52c2e0909
commit f38702bc95
9 changed files with 153 additions and 137 deletions
Download Patch File Download Diff File Expand all files Collapse all files

28
core/kordophone-db/src/repository.rs
View File

@@ -264,16 +264,34 @@ impl<'a> Repository<'a> {
.order_by(schema::messages::date.asc())
.load::<MessageRecord>(self.connection)?;
let sender_handles: Vec<String> = message_records
.iter()
.filter_map(|record| record.sender_participant_handle.clone())
.collect();
let participant_map: HashMap<String, Participant> = if sender_handles.is_empty() {
HashMap::new()
} else {
participants
.filter(handle.eq_any(&sender_handles))
.load::<ParticipantRecord>(self.connection)?
.into_iter()
.map(|participant| {
let key = participant.handle.clone();
(key, participant.into())
})
.collect()
};
let mut result = Vec::new();
for message_record in message_records {
let mut message: Message = message_record.clone().into();
// If the message references a sender participant, load the participant info
if let Some(sender_handle) = message_record.sender_participant_handle {
let participant = participants
.find(sender_handle)
.first::<ParticipantRecord>(self.connection)?;
message.sender = participant.into();
if let Some(participant) = participant_map.get(&sender_handle) {
message.sender = participant.clone();
}
}
result.push(message);
@@ -307,8 +325,8 @@ impl<'a> Repository<'a> {
}
pub fn delete_all_messages(&mut self) -> Result<()> {
use crate::schema::messages::dsl as messages_dsl;
use crate::schema::message_aliases::dsl as aliases_dsl;
use crate::schema::messages::dsl as messages_dsl;
diesel::delete(messages_dsl::messages).execute(self.connection)?;
diesel::delete(aliases_dsl::message_aliases).execute(self.connection)?;

5
core/kordophoned/include/net.buzzert.kordophonecd.Server.xml
View File

@@ -73,14 +73,13 @@
'sender' (string): Sender display name
'attachments' (array of dictionaries): List of attachments
'guid' (string): Attachment GUID
'path' (string): Attachment path
'preview_path' (string): Preview attachment path
'downloaded' (boolean): Whether the attachment is downloaded
'preview_downloaded' (boolean): Whether the preview is downloaded
'metadata' (dictionary, optional): Attachment metadata
'attribution_info' (dictionary, optional): Attribution info
'width' (int32, optional): Width
'height' (int32, optional): Height"/>
'height' (int32, optional): Height
Use GetAttachmentInfo for full/preview paths."/>
</arg>
</method>

20
core/kordophoned/src/daemon/contact_resolver/eds.rs
View File

@@ -6,6 +6,8 @@ use std::collections::HashMap;
use std::sync::Mutex;
use std::time::Duration;
const LOOKUP_TIMEOUT: Duration = Duration::from_secs(2);
#[derive(Clone)]
pub struct EDSContactResolverBackend;
@@ -189,11 +191,10 @@ impl ContactResolverBackend for EDSContactResolverBackend {
None => return None,
};
let address_book_proxy = handle.connection.with_proxy(
&handle.bus_name,
&handle.object_path,
Duration::from_secs(60),
);
let address_book_proxy =
handle
.connection
.with_proxy(&handle.bus_name, &handle.object_path, LOOKUP_TIMEOUT);
ensure_address_book_open(&address_book_proxy);
@@ -255,11 +256,10 @@ impl ContactResolverBackend for EDSContactResolverBackend {
None => return None,
};
let address_book_proxy = handle.connection.with_proxy(
&handle.bus_name,
&handle.object_path,
Duration::from_secs(60),
);
let address_book_proxy =
handle
.connection
.with_proxy(&handle.bus_name, &handle.object_path, LOOKUP_TIMEOUT);
ensure_address_book_open(&address_book_proxy);

26
core/kordophoned/src/daemon/mod.rs
View File

@@ -185,14 +185,14 @@ impl Daemon {
async fn handle_event(&mut self, event: Event) {
match event {
Event::GetVersion(reply) => {
reply.send(self.version.clone()).unwrap();
let _ = reply.send(self.version.clone());
}
Event::SyncConversationList(reply) => {
self.spawn_conversation_list_sync();
// This is a background operation, so return right away.
reply.send(()).unwrap();
let _ = reply.send(());
}
Event::SyncAllConversations(reply) => {
@@ -207,7 +207,7 @@ impl Daemon {
});
// This is a background operation, so return right away.
reply.send(()).unwrap();
let _ = reply.send(());
}
Event::SyncConversation(conversation_id, reply) => {
@@ -225,7 +225,7 @@ impl Daemon {
}
});
reply.send(()).unwrap();
let _ = reply.send(());
}
Event::MarkConversationAsRead(conversation_id, reply) => {
@@ -237,7 +237,7 @@ impl Daemon {
}
});
reply.send(()).unwrap();
let _ = reply.send(());
}
Event::UpdateConversationMetadata(conversation, reply) => {
@@ -250,7 +250,7 @@ impl Daemon {
}
});
reply.send(()).unwrap();
let _ = reply.send(());
}
Event::UpdateStreamReconnected => {
@@ -268,7 +268,7 @@ impl Daemon {
Event::GetAllConversations(limit, offset, reply) => {
let conversations = self.get_conversations_limit_offset(limit, offset).await;
reply.send(conversations).unwrap();
let _ = reply.send(conversations);
}
Event::GetAllSettings(reply) => {
@@ -277,7 +277,7 @@ impl Daemon {
Settings::default()
});
reply.send(settings).unwrap();
let _ = reply.send(settings);
}
Event::UpdateSettings(settings, reply) => {
@@ -309,12 +309,12 @@ impl Daemon {
}
}
reply.send(()).unwrap();
let _ = reply.send(());
}
Event::GetMessages(conversation_id, last_message_id, reply) => {
let messages = self.get_messages(conversation_id, last_message_id).await;
reply.send(messages).unwrap();
let _ = reply.send(messages);
}
Event::DeleteAllConversations(reply) => {
@@ -322,7 +322,7 @@ impl Daemon {
log::error!(target: target::SYNC, "Failed to delete all conversations: {}", e);
});
reply.send(()).unwrap();
let _ = reply.send(());
}
Event::SendMessage(conversation_id, text, attachment_guids, reply) => {
@@ -330,7 +330,7 @@ impl Daemon {
let uuid = self
.enqueue_outgoing_message(text, conversation_id.clone(), attachment_guids)
.await;
reply.send(uuid).unwrap();
let _ = reply.send(uuid);
// Send message updated signal, we have a placeholder message we will return.
self.signal_sender
@@ -395,7 +395,7 @@ impl Daemon {
.await
.unwrap();
reply.send(()).unwrap();
let _ = reply.send(());
}
Event::AttachmentDownloaded(attachment_id) => {

171
core/kordophoned/src/dbus/agent.rs
View File

@@ -176,9 +176,8 @@ impl DBusAgent {
&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)))
let daemon_result = run_sync_future(self.send_event(make_event))?;
daemon_result.map_err(|e| MethodErr::failed(&format!("Daemon error: {}", e)))
}
fn resolve_participant_display_name(&mut self, participant: &Participant) -> String {
@@ -284,104 +283,86 @@ impl DbusRepository for DBusAgent {
Some(last_message_id)
};
self.send_event_sync(|r| Event::GetMessages(conversation_id, last_message_id_opt, r))
.map(|messages| {
messages
let messages =
self.send_event_sync(|r| Event::GetMessages(conversation_id, last_message_id_opt, r))?;
let mapped: Vec<arg::PropMap> = messages
.into_iter()
.map(|msg| {
let mut map = arg::PropMap::new();
map.insert("id".into(), arg::Variant(Box::new(msg.id)));
// Remove the attachment placeholder here.
let text = msg.text.replace("\u{FFFC}", "");
map.insert("text".into(), arg::Variant(Box::new(text)));
map.insert(
"date".into(),
arg::Variant(Box::new(msg.date.and_utc().timestamp())),
);
map.insert(
"sender".into(),
arg::Variant(Box::new(msg.sender.display_name())),
);
let attachments: Vec<arg::PropMap> = msg
.attachments
.into_iter()
.map(|msg| {
let mut map = arg::PropMap::new();
map.insert("id".into(), arg::Variant(Box::new(msg.id)));
// Remove the attachment placeholder here.
let text = msg.text.replace("\u{FFFC}", "");
map.insert("text".into(), arg::Variant(Box::new(text)));
map.insert(
"date".into(),
arg::Variant(Box::new(msg.date.and_utc().timestamp())),
.map(|attachment| {
let mut attachment_map = arg::PropMap::new();
attachment_map.insert(
"guid".into(),
arg::Variant(Box::new(attachment.guid.clone())),
);
map.insert(
"sender".into(),
arg::Variant(Box::new(
self.resolve_participant_display_name(&msg.sender.into()),
)),
attachment_map.insert(
"downloaded".into(),
arg::Variant(Box::new(attachment.is_downloaded(false))),
);
attachment_map.insert(
"preview_downloaded".into(),
arg::Variant(Box::new(attachment.is_downloaded(true))),
);
// Attachments array
let attachments: Vec<arg::PropMap> = msg
.attachments
.into_iter()
.map(|attachment| {
let mut attachment_map = arg::PropMap::new();
attachment_map.insert(
"guid".into(),
arg::Variant(Box::new(attachment.guid.clone())),
);
if let Some(ref metadata) = attachment.metadata {
let mut metadata_map = arg::PropMap::new();
// Paths and download status
let path = attachment.get_path_for_preview(false);
let preview_path = attachment.get_path_for_preview(true);
let downloaded = attachment.is_downloaded(false);
let preview_downloaded = attachment.is_downloaded(true);
attachment_map.insert(
"path".into(),
arg::Variant(Box::new(path.to_string_lossy().to_string())),
);
attachment_map.insert(
"preview_path".into(),
arg::Variant(Box::new(
preview_path.to_string_lossy().to_string(),
)),
);
attachment_map.insert(
"downloaded".into(),
arg::Variant(Box::new(downloaded)),
);
attachment_map.insert(
"preview_downloaded".into(),
arg::Variant(Box::new(preview_downloaded)),
);
// Metadata
if let Some(ref metadata) = attachment.metadata {
let mut metadata_map = arg::PropMap::new();
if let Some(ref attribution_info) = metadata.attribution_info {
let mut attribution_map = arg::PropMap::new();
if let Some(width) = attribution_info.width {
attribution_map.insert(
"width".into(),
arg::Variant(Box::new(width as i32)),
);
}
if let Some(height) = attribution_info.height {
attribution_map.insert(
"height".into(),
arg::Variant(Box::new(height as i32)),
);
}
metadata_map.insert(
"attribution_info".into(),
arg::Variant(Box::new(attribution_map)),
);
}
attachment_map.insert(
"metadata".into(),
arg::Variant(Box::new(metadata_map)),
if let Some(ref attribution_info) = metadata.attribution_info {
let mut attribution_map = arg::PropMap::new();
if let Some(width) = attribution_info.width {
attribution_map.insert(
"width".into(),
arg::Variant(Box::new(width as i32)),
);
}
if let Some(height) = attribution_info.height {
attribution_map.insert(
"height".into(),
arg::Variant(Box::new(height as i32)),
);
}
metadata_map.insert(
"attribution_info".into(),
arg::Variant(Box::new(attribution_map)),
);
}
attachment_map
})
.collect();
attachment_map.insert(
"metadata".into(),
arg::Variant(Box::new(metadata_map)),
);
}
map.insert("attachments".into(), arg::Variant(Box::new(attachments)));
map
attachment_map
})
.collect()
.collect();
map.insert("attachments".into(), arg::Variant(Box::new(attachments)));
map
})
.collect();
Ok(mapped)
}
fn delete_all_conversations(&mut self) -> Result<(), MethodErr> {
@@ -496,7 +477,7 @@ where
T: Send,
F: Future<Output = T> + Send,
{
thread::scope(move |s| {
let joined = thread::scope(move |s| {
s.spawn(move || {
let rt = tokio::runtime::Builder::new_current_thread()
.enable_all()
@@ -507,6 +488,10 @@ where
Ok(result)
})
.join()
})
.expect("Error joining runtime thread")
});
match joined {
Ok(result) => result,
Err(_) => Err(MethodErr::failed("Error joining runtime thread")),
}
}

2
core/kpcli/src/daemon/dbus.rs
View File

@@ -33,7 +33,7 @@ impl DBusDaemonInterface {
fn proxy(&self) -> Proxy<&Connection> {
self.conn
.with_proxy(DBUS_NAME, DBUS_PATH, std::time::Duration::from_millis(5000))
.with_proxy(DBUS_NAME, DBUS_PATH, std::time::Duration::from_secs(30))
}
async fn print_settings(&mut self) -> Result<()> {