simploxide_client/
xftp.rs

1//! XFTP file download manager.
2//!
3//! [`XftpClient`] wraps any [`ClientApi`] client and observes the file rcv events emitted by the
4//! SimpleX-Chat. [`DownloadFileBuilder`] (obtained via [`XftpExt::download_file`]) initiates the transfer
5//! and awaits those events, returning the outcome directly to the caller.
6//!
7//! # When to use
8//!
9//! - **Out-of-handler downloads.** When the decision to download a file is made outside an event
10//!   handler (for example, after a user command or a timer), `download_file` provides the result
11//!   without requiring custom event routing.
12//!
13//! - **Keeping download logic in one handler.** Sometimes it may be useful to keep all logic in a
14//!   single handler to simplify state management.
15//!
16
17use std::sync::Arc;
18
19use serde::Deserialize;
20use simploxide_api_types::{
21    client_api::ClientApi,
22    commands::ReceiveFile,
23    events::{Event, EventKind, RcvFileComplete, RcvFileError, RcvFileSndCancelled},
24    responses::{CancelFileResponse, RcvFileAcceptedSndCancelledResponse, ReceiveFileResponse},
25};
26
27use crate::{Hook, id::FileId};
28
29type FxDashMap<K, V> = dashmap::DashMap<K, V, rustc_hash::FxBuildHasher>;
30type XftpDownloadResponder = tokio::sync::oneshot::Sender<XftpManagerDownloadResponse>;
31
32/// Adds [`download_file`](Self::download_file) to any [`ClientApi`].
33/// Automatically implemented for [`XftpClient`].
34pub trait XftpExt: ClientApi {
35    /// Begin downloading `file_id` and return a builder to configure and await the result.
36    ///
37    /// # Deadlock warning
38    ///
39    /// `download_file` awaits a completion event that only arrives when the event loop processes
40    /// **Awaiting a download inside a sequential handler blocks the event loop**: that event
41    /// never arrives, causing a deadlock. Only use `download_file` from a **concurrent** handler
42    /// (registered with [`Dispatcher::on`](crate::dispatcher::Dispatcher::on)) or outside the
43    /// dispatcher entirely.
44    fn download_file<FID: Into<FileId>>(&self, file_id: FID) -> DownloadFileBuilder<'_, Self>;
45}
46
47/// A [`ClientApi`] wrapper that intercepts file-result events and routes them to the
48/// corresponding [`DownloadFileBuilder`] futures. Should be constructed by
49/// [`EventStream::hook_xftp`](crate::EventStream::hook_xftp) to work correctly.
50#[derive(Clone)]
51pub struct XftpClient<C> {
52    client: C,
53    xftp: Arc<XftpManager>,
54}
55
56#[cfg(feature = "websocket")]
57impl XftpClient<crate::ws::Client> {
58    pub fn version(
59        &self,
60    ) -> impl Future<Output = Result<crate::ws::SimplexVersion, crate::ws::VersionError>> {
61        self.client.version()
62    }
63
64    pub fn disconnect(self) -> impl Future<Output = ()> {
65        self.client.disconnect()
66    }
67}
68
69#[cfg(feature = "ffi")]
70impl XftpClient<crate::ffi::Client> {
71    pub fn version(
72        &self,
73    ) -> impl Future<Output = Result<crate::ffi::SimplexVersion, crate::ffi::VersionError>> {
74        self.client.version()
75    }
76
77    pub fn disconnect(self) -> impl Future<Output = ()> {
78        self.client.disconnect()
79    }
80}
81
82#[cfg(feature = "farm")]
83impl<C> XftpClient<C> {
84    pub(crate) fn new(client: C, xftp: Arc<XftpManager>) -> Self {
85        Self { client, xftp }
86    }
87
88    pub(crate) fn manager(&self) -> Arc<XftpManager> {
89        self.xftp.clone()
90    }
91}
92
93impl<C: ClientApi> ClientApi for XftpClient<C> {
94    type ResponseShape<'de, T: 'de + Deserialize<'de>> = C::ResponseShape<'de, T>;
95    type Error = C::Error;
96
97    fn send_raw(
98        &self,
99        command: String,
100    ) -> impl Future<Output = Result<String, Self::Error>> + Send {
101        self.client.send_raw(command)
102    }
103
104    fn cancel_file(
105        &self,
106        file_id: i64,
107    ) -> impl Future<Output = Result<CancelFileResponse, Self::Error>> + Send {
108        self.xftp.downloads.remove(&file_id);
109        self.client.cancel_file(file_id)
110    }
111}
112
113impl<C: ClientApi> From<C> for XftpClient<C> {
114    fn from(client: C) -> Self {
115        Self {
116            client,
117            xftp: Arc::new(XftpManager::default()),
118        }
119    }
120}
121
122impl<C: ClientApi> XftpExt for XftpClient<C> {
123    fn download_file<FID: Into<FileId>>(&self, file_id: FID) -> DownloadFileBuilder<'_, Self> {
124        DownloadFileBuilder {
125            client: self,
126            cmd: ReceiveFile::new(file_id.into().raw()),
127        }
128    }
129}
130
131impl<C: 'static + ClientApi + Send> Hook for XftpClient<C> {
132    fn should_intercept(&self, kind: EventKind) -> bool {
133        const EVENT_KINDS: [EventKind; 3] = [
134            EventKind::RcvFileSndCancelled,
135            EventKind::RcvFileComplete,
136            EventKind::RcvFileError,
137        ];
138
139        EVENT_KINDS.contains(&kind)
140    }
141
142    fn intercept_event(&mut self, event: Event) {
143        match event {
144            Event::RcvFileComplete(ev) => {
145                if let Some((_, responder)) = self
146                    .xftp
147                    .downloads
148                    .remove(&ev.chat_item.chat_item.file.as_ref().unwrap().file_id)
149                {
150                    let _ = responder.send(XftpManagerDownloadResponse::Complete(ev));
151                }
152            }
153            Event::RcvFileSndCancelled(ev) => {
154                if let Some((_, responder)) =
155                    self.xftp.downloads.remove(&ev.rcv_file_transfer.file_id)
156                {
157                    let _ = responder.send(XftpManagerDownloadResponse::Cancelled(ev));
158                }
159            }
160            Event::RcvFileError(ev) => {
161                if let Some((_, responder)) =
162                    self.xftp.downloads.remove(&ev.rcv_file_transfer.file_id)
163                {
164                    let _ = responder.send(XftpManagerDownloadResponse::Error(ev));
165                }
166            }
167            _ => (),
168        }
169    }
170}
171
172pub struct DownloadFileBuilder<'a, C: 'a + ?Sized> {
173    client: &'a C,
174    cmd: ReceiveFile,
175}
176
177impl<'a, C: 'a + ?Sized> DownloadFileBuilder<'a, C> {
178    /// Route the download through user-approved relays rather than the default ones.
179    pub fn via_user_approved_relays(mut self) -> Self {
180        self.cmd.user_approved_relays = true;
181        self
182    }
183
184    /// Store the downloaded file in encrypted form.
185    pub fn store_encrypted(mut self) -> Self {
186        self.cmd.store_encrypted = Some(true);
187        self
188    }
189
190    /// Request inline delivery (small files only).
191    pub fn inline(mut self) -> Self {
192        self.cmd.file_inline = Some(true);
193        self
194    }
195
196    /// Override the path where the downloaded file will be saved.
197    pub fn file_path<P: AsRef<std::path::Path>>(mut self, path: P) -> Self {
198        self.cmd.file_path = Some(path.as_ref().display().to_string());
199        self
200    }
201}
202
203impl<'a, C: 'a + ClientApi> IntoFuture for DownloadFileBuilder<'a, XftpClient<C>>
204where
205    <XftpClient<C> as ClientApi>::Error: 'static + Send,
206{
207    type Output = Result<Arc<RcvFileComplete>, DownloadError<<XftpClient<C> as ClientApi>::Error>>;
208    type IntoFuture = std::pin::Pin<Box<dyn 'a + Send + Future<Output = Self::Output>>>;
209
210    fn into_future(self) -> Self::IntoFuture {
211        Box::pin(async move {
212            let file_id = self.cmd.file_id;
213
214            let (responder, response) = tokio::sync::oneshot::channel();
215            self.client.xftp.downloads.insert(file_id, responder);
216
217            match self.client.receive_file(self.cmd).await {
218                Ok(ReceiveFileResponse::RcvFileAccepted(_)) => {
219                    match response.await.expect("XFTP responses are always delivered") {
220                        XftpManagerDownloadResponse::Complete(success) => Ok(success),
221                        XftpManagerDownloadResponse::Cancelled(err) => {
222                            Err(DownloadError::SendCancelled(err))
223                        }
224                        XftpManagerDownloadResponse::Error(err) => Err(DownloadError::Receive(err)),
225                    }
226                }
227                Ok(ReceiveFileResponse::RcvFileAcceptedSndCancelled(err)) => {
228                    self.client.xftp.downloads.remove(&file_id);
229                    Err(DownloadError::AcceptCancelled(err))
230                }
231                Err(e) => {
232                    self.client.xftp.downloads.remove(&file_id);
233                    Err(DownloadError::Api(e))
234                }
235            }
236        })
237    }
238}
239
240/// Error returned when a [`DownloadFileBuilder`] future resolves unsuccessfully.
241pub enum DownloadError<E> {
242    /// The sender cancelled the transfer after the download was accepted.
243    SendCancelled(Arc<RcvFileSndCancelled>),
244    /// The file was no longer available when the download request arrived.
245    AcceptCancelled(Arc<RcvFileAcceptedSndCancelledResponse>),
246    /// The SimpleX agent reported an error while receiving the file.
247    Receive(Arc<RcvFileError>),
248    /// The API call to initiate the download failed.
249    Api(E),
250}
251
252impl<E> std::fmt::Debug for DownloadError<E>
253where
254    E: std::fmt::Debug,
255{
256    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
257        match self {
258            Self::SendCancelled(arg) => f
259                .debug_tuple("SendCancelled")
260                .field(&arg.rcv_file_transfer.file_id)
261                .finish(),
262            Self::AcceptCancelled(arg) => f
263                .debug_tuple("AcceptCancelled")
264                .field(&arg.rcv_file_transfer.file_id)
265                .finish(),
266            Self::Receive(arg) => f.debug_tuple("Receive").field(&arg.agent_error).finish(),
267            Self::Api(e) => f.debug_tuple("Api").field(e).finish(),
268        }
269    }
270}
271
272impl<E> std::fmt::Display for DownloadError<E>
273where
274    E: std::fmt::Display,
275{
276    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
277        match self {
278            Self::SendCancelled(err) => write!(
279                f,
280                "File(ID={}) was cancelled by the user",
281                err.rcv_file_transfer.file_id
282            ),
283            Self::AcceptCancelled(err) => write!(
284                f,
285                "File(ID={}) is no longer available",
286                err.rcv_file_transfer.file_id
287            ),
288            Self::Receive(err) => write!(
289                f,
290                "File(ID={}) receive error: {:?}",
291                err.rcv_file_transfer.file_id, err.agent_error
292            ),
293            Self::Api(err) => write!(f, "{err}"),
294        }
295    }
296}
297
298impl<E> std::error::Error for DownloadError<E>
299where
300    E: 'static + std::error::Error,
301{
302    fn source(&self) -> Option<&(dyn std::error::Error + 'static)> {
303        match self {
304            Self::SendCancelled(_) => None,
305            Self::AcceptCancelled(_) => None,
306            Self::Receive(_) => None,
307            Self::Api(error) => Some(error),
308        }
309    }
310}
311
312#[derive(Default)]
313pub(crate) struct XftpManager {
314    downloads: FxDashMap<i64, XftpDownloadResponder>,
315}
316
317enum XftpManagerDownloadResponse {
318    Complete(Arc<RcvFileComplete>),
319    Error(Arc<RcvFileError>),
320    Cancelled(Arc<RcvFileSndCancelled>),
321}
simploxide_client/xftp.rs

simploxide_client/
xftp.rs
