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
56impl<C> XftpClient<C> {
57    pub(crate) fn manager(&self) -> Arc<XftpManager> {
58        self.xftp.clone()
59    }
60}
61
62#[cfg(feature = "websocket")]
63impl XftpClient<crate::ws::Client> {
64    pub fn version(
65        &self,
66    ) -> impl Future<Output = Result<crate::ws::SimplexVersion, crate::ws::VersionError>> {
67        self.client.version()
68    }
69
70    pub fn disconnect(self) -> impl Future<Output = ()> {
71        self.client.disconnect()
72    }
73}
74
75#[cfg(feature = "ffi")]
76impl XftpClient<crate::ffi::Client> {
77    pub fn version(
78        &self,
79    ) -> impl Future<Output = Result<crate::ffi::SimplexVersion, crate::ffi::VersionError>> {
80        self.client.version()
81    }
82
83    pub fn disconnect(self) -> impl Future<Output = ()> {
84        self.client.disconnect()
85    }
86}
87
88#[cfg(feature = "farm")]
89impl<C> XftpClient<C> {
90    pub(crate) fn new(client: C, xftp: Arc<XftpManager>) -> Self {
91        Self { client, xftp }
92    }
93}
94
95impl<C: ClientApi> ClientApi for XftpClient<C> {
96    type ResponseShape<'de, T: 'de + Deserialize<'de>> = C::ResponseShape<'de, T>;
97    type Error = C::Error;
98
99    fn send_raw(
100        &self,
101        command: String,
102    ) -> impl Future<Output = Result<String, Self::Error>> + Send {
103        self.client.send_raw(command)
104    }
105
106    fn cancel_file(
107        &self,
108        file_id: i64,
109    ) -> impl Future<Output = Result<CancelFileResponse, Self::Error>> + Send {
110        self.xftp.downloads.remove(&file_id);
111        self.client.cancel_file(file_id)
112    }
113}
114
115impl<C: ClientApi> From<C> for XftpClient<C> {
116    fn from(client: C) -> Self {
117        Self {
118            client,
119            xftp: Arc::new(XftpManager::default()),
120        }
121    }
122}
123
124impl<C: ClientApi> XftpExt for XftpClient<C> {
125    fn download_file<FID: Into<FileId>>(&self, file_id: FID) -> DownloadFileBuilder<'_, Self> {
126        DownloadFileBuilder {
127            client: self,
128            cmd: ReceiveFile::new(file_id.into().raw()),
129        }
130    }
131}
132
133impl Hook for XftpManager {
134    fn should_intercept(&self, kind: EventKind) -> bool {
135        const EVENT_KINDS: [EventKind; 3] = [
136            EventKind::RcvFileSndCancelled,
137            EventKind::RcvFileComplete,
138            EventKind::RcvFileError,
139        ];
140
141        EVENT_KINDS.contains(&kind)
142    }
143
144    fn intercept_event(&self, event: Event) {
145        match event {
146            Event::RcvFileComplete(ev) => {
147                if let Some((_, responder)) = self
148                    .downloads
149                    .remove(&ev.chat_item.chat_item.file.as_ref().unwrap().file_id)
150                {
151                    let _ = responder.send(XftpManagerDownloadResponse::Complete(ev));
152                }
153            }
154            Event::RcvFileSndCancelled(ev) => {
155                if let Some((_, responder)) = self.downloads.remove(&ev.rcv_file_transfer.file_id) {
156                    let _ = responder.send(XftpManagerDownloadResponse::Cancelled(ev));
157                }
158            }
159            Event::RcvFileError(ev) => {
160                if let Some((_, responder)) = self.downloads.remove(&ev.rcv_file_transfer.file_id) {
161                    let _ = responder.send(XftpManagerDownloadResponse::Error(ev));
162                }
163            }
164            _ => (),
165        }
166    }
167}
168
169pub struct DownloadFileBuilder<'a, C: 'a + ?Sized> {
170    client: &'a C,
171    cmd: ReceiveFile,
172}
173
174impl<'a, C: 'a + ?Sized> DownloadFileBuilder<'a, C> {
175    /// Route the download through user-approved relays rather than the default ones.
176    pub fn via_user_approved_relays(mut self) -> Self {
177        self.cmd.user_approved_relays = true;
178        self
179    }
180
181    /// Store the downloaded file in encrypted form.
182    pub fn store_encrypted(mut self) -> Self {
183        self.cmd.store_encrypted = Some(true);
184        self
185    }
186
187    /// Request inline delivery (small files only).
188    pub fn inline(mut self) -> Self {
189        self.cmd.file_inline = Some(true);
190        self
191    }
192
193    /// Override the path where the downloaded file will be saved.
194    pub fn file_path<P: AsRef<std::path::Path>>(mut self, path: P) -> Self {
195        self.cmd.file_path = Some(path.as_ref().display().to_string());
196        self
197    }
198}
199
200impl<'a, C: 'a + ClientApi> IntoFuture for DownloadFileBuilder<'a, XftpClient<C>>
201where
202    <XftpClient<C> as ClientApi>::Error: 'static + Send,
203{
204    type Output = Result<Arc<RcvFileComplete>, DownloadError<<XftpClient<C> as ClientApi>::Error>>;
205    type IntoFuture = std::pin::Pin<Box<dyn 'a + Send + Future<Output = Self::Output>>>;
206
207    fn into_future(self) -> Self::IntoFuture {
208        Box::pin(async move {
209            let file_id = self.cmd.file_id;
210
211            let (responder, response) = tokio::sync::oneshot::channel();
212            self.client.xftp.downloads.insert(file_id, responder);
213
214            match self.client.receive_file(self.cmd).await {
215                Ok(ReceiveFileResponse::RcvFileAccepted(_)) => {
216                    match response.await.expect("XFTP responses are always delivered") {
217                        XftpManagerDownloadResponse::Complete(success) => Ok(success),
218                        XftpManagerDownloadResponse::Cancelled(err) => {
219                            Err(DownloadError::SendCancelled(err))
220                        }
221                        XftpManagerDownloadResponse::Error(err) => Err(DownloadError::Receive(err)),
222                    }
223                }
224                Ok(ReceiveFileResponse::RcvFileAcceptedSndCancelled(err)) => {
225                    self.client.xftp.downloads.remove(&file_id);
226                    Err(DownloadError::AcceptCancelled(err))
227                }
228                Err(e) => {
229                    self.client.xftp.downloads.remove(&file_id);
230                    Err(DownloadError::Api(e))
231                }
232            }
233        })
234    }
235}
236
237/// Error returned when a [`DownloadFileBuilder`] future resolves unsuccessfully.
238pub enum DownloadError<E> {
239    /// The sender cancelled the transfer after the download was accepted.
240    SendCancelled(Arc<RcvFileSndCancelled>),
241    /// The file was no longer available when the download request arrived.
242    AcceptCancelled(Arc<RcvFileAcceptedSndCancelledResponse>),
243    /// The SimpleX agent reported an error while receiving the file.
244    Receive(Arc<RcvFileError>),
245    /// The API call to initiate the download failed.
246    Api(E),
247}
248
249impl<E> std::fmt::Debug for DownloadError<E>
250where
251    E: std::fmt::Debug,
252{
253    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
254        match self {
255            Self::SendCancelled(arg) => f
256                .debug_tuple("SendCancelled")
257                .field(&arg.rcv_file_transfer.file_id)
258                .finish(),
259            Self::AcceptCancelled(arg) => f
260                .debug_tuple("AcceptCancelled")
261                .field(&arg.rcv_file_transfer.file_id)
262                .finish(),
263            Self::Receive(arg) => f.debug_tuple("Receive").field(&arg.agent_error).finish(),
264            Self::Api(e) => f.debug_tuple("Api").field(e).finish(),
265        }
266    }
267}
268
269impl<E> std::fmt::Display for DownloadError<E>
270where
271    E: std::fmt::Display,
272{
273    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
274        match self {
275            Self::SendCancelled(err) => write!(
276                f,
277                "File(ID={}) was cancelled by the user",
278                err.rcv_file_transfer.file_id
279            ),
280            Self::AcceptCancelled(err) => write!(
281                f,
282                "File(ID={}) is no longer available",
283                err.rcv_file_transfer.file_id
284            ),
285            Self::Receive(err) => write!(
286                f,
287                "File(ID={}) receive error: {:?}",
288                err.rcv_file_transfer.file_id, err.agent_error
289            ),
290            Self::Api(err) => write!(f, "{err}"),
291        }
292    }
293}
294
295impl<E> std::error::Error for DownloadError<E>
296where
297    E: 'static + std::error::Error,
298{
299    fn source(&self) -> Option<&(dyn std::error::Error + 'static)> {
300        match self {
301            Self::SendCancelled(_) => None,
302            Self::AcceptCancelled(_) => None,
303            Self::Receive(_) => None,
304            Self::Api(error) => Some(error),
305        }
306    }
307}
308
309#[derive(Default)]
310pub(crate) struct XftpManager {
311    downloads: FxDashMap<i64, XftpDownloadResponder>,
312}
313
314enum XftpManagerDownloadResponse {
315    Complete(Arc<RcvFileComplete>),
316    Error(Arc<RcvFileError>),
317    Cancelled(Arc<RcvFileSndCancelled>),
318}
simploxide_client/xftp.rs

simploxide_client/
xftp.rs
