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

simploxide_client/
xftp.rs
