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

cargo fmt

Browse Source
This commit is contained in:
James Magahern
2025-06-16 19:26:13 -07:00
parent 75fe4d4608
commit 032573d23b
14 changed files with 168 additions and 113 deletions
Download Patch File Download Diff File Expand all files Collapse all files

21
kordophoned/src/daemon/attachment_store.rs
View File

@@ -114,11 +114,11 @@ impl AttachmentStore {
store_path: &PathBuf,
database: &mut Arc<Mutex<Database>>,
daemon_event_sink: &Sender<DaemonEvent>,
guid: &String,
preview: bool
guid: &String,
preview: bool,
) -> Result<()> {
let attachment = Self::get_attachment_impl(store_path, guid);
if attachment.is_downloaded(preview) {
log::info!(target: target::ATTACHMENTS, "Attachment already downloaded: {}", attachment.guid);
return Err(AttachmentStoreError::AttachmentAlreadyDownloaded.into());
@@ -144,13 +144,16 @@ impl AttachmentStore {
while let Some(Ok(data)) = stream.next().await {
writer.write(data.as_ref())?;
}
// Flush and sync the temporary file before moving
writer.flush()?;
file.sync_all()?;
// Atomically move the temporary file to the final location
std::fs::rename(&temporary_path, &attachment.get_path_for_preview_scratch(preview, false))?;
std::fs::rename(
&temporary_path,
&attachment.get_path_for_preview_scratch(preview, false),
)?;
log::info!(target: target::ATTACHMENTS, "Completed download for attachment: {}", attachment.guid);
@@ -175,12 +178,12 @@ impl AttachmentStore {
let uploads_path = store_path.join("uploads");
std::fs::create_dir_all(&uploads_path).unwrap();
// First, copy the file to the store path, under /uploads/.
// First, copy the file to the store path, under /uploads/.
log::trace!(target: target::ATTACHMENTS, "Copying attachment to uploads directory: {}", uploads_path.display());
let temporary_path = uploads_path.join(incoming_path.file_name().unwrap());
std::fs::copy(incoming_path, &temporary_path).unwrap();
// Open file handle to the temporary file,
// Open file handle to the temporary file,
log::trace!(target: target::ATTACHMENTS, "Opening stream to temporary file: {}", temporary_path.display());
let file = File::open(&temporary_path).await?;
let reader: BufReader<File> = BufReader::new(file);
@@ -189,7 +192,7 @@ impl AttachmentStore {
let filename = incoming_path.file_name().unwrap().to_str().unwrap();
log::trace!(target: target::ATTACHMENTS, "Uploading attachment to server: {}", &filename);
let mut client = Daemon::get_client_impl(database).await?;
let metadata = std::fs::metadata(&temporary_path)?;
let size = metadata.len();
let guid = client.upload_attachment(reader, filename, size).await?;
@@ -220,7 +223,7 @@ impl AttachmentStore {
let daemon_event_sink = self.daemon_event_sink.clone();
let _guid = guid.clone();
// Spawn a new task here so we don't block incoming queue events.
// Spawn a new task here so we don't block incoming queue events.
tokio::spawn(async move {
let result = Self::download_attachment_impl(
&store_path,

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

@@ -224,7 +224,7 @@ impl Daemon {
Event::UpdateStreamReconnected => {
log::info!(target: target::UPDATES, "Update stream reconnected");
// The ui client will respond differently, but we'll almost certainly want to do a sync-list in response to this.
// The ui client will respond differently, but we'll almost certainly want to do a sync-list in response to this.
self.spawn_conversation_list_sync();
// Send signal to the client that the update stream has been reconnected.
@@ -267,12 +267,14 @@ impl Daemon {
self.spawn_conversation_list_sync();
// Also restart the update monitor.
if let Err(e) = self.update_monitor_command_tx
if let Err(e) = self
.update_monitor_command_tx
.as_ref()
.unwrap()
.try_send(UpdateMonitorCommand::Restart) {
log::warn!(target: target::UPDATES, "Failed to send restart command to update monitor: {}", e);
}
.try_send(UpdateMonitorCommand::Restart)
{
log::warn!(target: target::UPDATES, "Failed to send restart command to update monitor: {}", e);
}
}
reply.send(()).unwrap();
@@ -428,7 +430,12 @@ impl Daemon {
.await
}
async fn enqueue_outgoing_message(&mut self, text: String, conversation_id: String, attachment_guids: Vec<String>) -> Uuid {
async fn enqueue_outgoing_message(
&mut self,
text: String,
conversation_id: String,
attachment_guids: Vec<String>,
) -> Uuid {
let conversation_id = conversation_id.clone();
let outgoing_message = OutgoingMessage::builder()
.text(text)
@@ -553,11 +560,12 @@ impl Daemon {
.await?;
// Filter messages that have an empty body, or a body that is just whitespace.
// This is a workaround for a bug in the server where it returns messages with an empty body, which is usually
// the typing indicator or stuff like that. In the future, we need to move to ChatItems instead of Messages.
// This is a workaround for a bug in the server where it returns messages with an empty body, which is usually
// the typing indicator or stuff like that. In the future, we need to move to ChatItems instead of Messages.
let insertable_messages: Vec<kordophone::model::Message> = messages
.into_iter()
.filter(|m| !m.text.is_empty() && !m.text.trim().is_empty()).collect();
.filter(|m| !m.text.is_empty() && !m.text.trim().is_empty())
.collect();
let db_messages: Vec<kordophone_db::models::Message> = insertable_messages
.into_iter()

3
kordophoned/src/daemon/models/attachment.rs
View File

@@ -30,7 +30,8 @@ impl Attachment {
pub fn get_path_for_preview_scratch(&self, preview: bool, scratch: bool) -> PathBuf {
let extension = if preview { "preview" } else { "full" };
if scratch {
self.base_path.with_extension(format!("{}.download", extension))
self.base_path
.with_extension(format!("{}.download", extension))
} else {
self.base_path.with_extension(extension)
}

7
kordophoned/src/daemon/update_monitor.rs
View File

@@ -40,7 +40,7 @@ impl UpdateMonitor {
event_sender,
last_sync_times: HashMap::new(),
update_seq: None,
first_connection: false, // optimistic assumption that we're not reconnecting the first time.
first_connection: false, // optimistic assumption that we're not reconnecting the first time.
command_tx: Some(command_tx),
command_rx,
}
@@ -50,10 +50,7 @@ impl UpdateMonitor {
self.command_tx.take().unwrap()
}
async fn send_event<T>(
&self,
make_event: impl FnOnce(Reply<T>) -> Event,
) -> DaemonResult<T> {
async fn send_event<T>(&self, make_event: impl FnOnce(Reply<T>) -> Event) -> DaemonResult<T> {
let (reply_tx, reply_rx) = tokio::sync::oneshot::channel();
self.event_sender
.send(make_event(reply_tx))

4
kordophoned/src/dbus/mod.rs
View File

@@ -10,10 +10,10 @@ pub mod interface {
include!(concat!(env!("OUT_DIR"), "/kordophone-server.rs"));
pub mod signals {
pub use crate::interface::NetBuzzertKordophoneRepositoryConversationsUpdated as ConversationsUpdated;
pub use crate::interface::NetBuzzertKordophoneRepositoryMessagesUpdated as MessagesUpdated;
pub use crate::interface::NetBuzzertKordophoneRepositoryAttachmentDownloadCompleted as AttachmentDownloadCompleted;
pub use crate::interface::NetBuzzertKordophoneRepositoryAttachmentUploadCompleted as AttachmentUploadCompleted;
pub use crate::interface::NetBuzzertKordophoneRepositoryConversationsUpdated as ConversationsUpdated;
pub use crate::interface::NetBuzzertKordophoneRepositoryMessagesUpdated as MessagesUpdated;
pub use crate::interface::NetBuzzertKordophoneRepositoryUpdateStreamReconnected as UpdateStreamReconnected;
}
}

13
kordophoned/src/dbus/server_impl.rs
View File

@@ -128,11 +128,11 @@ impl DbusRepository for ServerImpl {
let mut map = arg::PropMap::new();
map.insert("id".into(), arg::Variant(Box::new(msg.id)));
// xxx: Remove the attachment placeholder here.
// xxx: Remove the attachment placeholder here.
// This is not the ideal place to do this, but once we start using ChatItems instead of IMMessages
// from the server, we shouldn't be seeing these placeholders.
// from the server, we shouldn't be seeing these placeholders.
let text = msg.text.replace("\u{FFFC}", "");
map.insert("text".into(), arg::Variant(Box::new(text)));
map.insert(
"date".into(),
@@ -272,12 +272,9 @@ impl DbusRepository for ServerImpl {
self.send_event_sync(|r| Event::DownloadAttachment(attachment_id, preview, r))
}
fn upload_attachment(
&mut self,
path: String,
) -> Result<String, dbus::MethodErr> {
fn upload_attachment(&mut self, path: String) -> Result<String, dbus::MethodErr> {
use std::path::PathBuf;
let path = PathBuf::from(path);
self.send_event_sync(|r| Event::UploadAttachment(path, r))
}

29
kordophoned/src/main.rs
View File

@@ -100,9 +100,15 @@ async fn main() {
}
Signal::AttachmentDownloaded(attachment_id) => {
log::debug!("Sending signal: AttachmentDownloaded for attachment {}", attachment_id);
log::debug!(
"Sending signal: AttachmentDownloaded for attachment {}",
attachment_id
);
dbus_registry
.send_signal(interface::OBJECT_PATH, DbusSignals::AttachmentDownloadCompleted { attachment_id })
.send_signal(
interface::OBJECT_PATH,
DbusSignals::AttachmentDownloadCompleted { attachment_id },
)
.unwrap_or_else(|_| {
log::error!("Failed to send signal");
0
@@ -110,9 +116,19 @@ async fn main() {
}
Signal::AttachmentUploaded(upload_guid, attachment_guid) => {
log::debug!("Sending signal: AttachmentUploaded for upload {}, attachment {}", upload_guid, attachment_guid);
log::debug!(
"Sending signal: AttachmentUploaded for upload {}, attachment {}",
upload_guid,
attachment_guid
);
dbus_registry
.send_signal(interface::OBJECT_PATH, DbusSignals::AttachmentUploadCompleted { upload_guid, attachment_guid })
.send_signal(
interface::OBJECT_PATH,
DbusSignals::AttachmentUploadCompleted {
upload_guid,
attachment_guid,
},
)
.unwrap_or_else(|_| {
log::error!("Failed to send signal");
0
@@ -122,7 +138,10 @@ async fn main() {
Signal::UpdateStreamReconnected => {
log::debug!("Sending signal: UpdateStreamReconnected");
dbus_registry
.send_signal(interface::OBJECT_PATH, DbusSignals::UpdateStreamReconnected {})
.send_signal(
interface::OBJECT_PATH,
DbusSignals::UpdateStreamReconnected {},
)
.unwrap_or_else(|_| {
log::error!("Failed to send signal");
0