toole_core/
transfer.rs

1use crate::file_certif::{certificat, PinnedServerVerifier};
2use crate::{ToolError, UI};
3use quinn::{ClientConfig, Connection, Endpoint, RecvStream, SendStream, ServerConfig};
4use serde::{Deserialize, Serialize};
5use std::collections::HashMap;
6use std::io::Error as IoError;
7use std::net::SocketAddr;
8use std::path::{Path, PathBuf};
9use std::sync::atomic::{AtomicBool, AtomicU64, Ordering};
10use std::sync::Arc;
11use std::sync::Mutex;
12use std::time::{Duration, Instant};
13use tokio::fs;
14use tokio::io::{AsyncReadExt, AsyncWriteExt};
15use tokio::sync::oneshot;
16use tokio::task::JoinHandle;
17
18#[derive(Serialize, Deserialize, Debug, Clone)]
19pub struct Metadata {
20    pub transfer_id: String,
21    pub rel_path: String,
22    pub size: u64,
23    pub is_dir: bool,
24}
25
26/// en-tête de lot envoyé par l'émetteur sur le premier flux de la connexion :
27/// le récepteur connaît le transfer_id, le total du lot, l'émetteur et la
28/// liste des fichiers à l'avance, afin d'afficher la demande d'acceptation et
29/// la même progression globale que l'émetteur dès le premier fichier. sender
30/// et files sont tolérants (serde default) pour rester compatibles avec un
31/// ancien émetteur qui n'envoie pas ces champs.
32#[derive(Serialize, Deserialize, Debug, Clone)]
33pub struct BatchHeader {
34    pub transfer_id: String,
35    pub total_bytes: u64,
36    #[serde(default)]
37    pub sender: String,
38    #[serde(default)]
39    pub files: Vec<String>,
40}
41
42const CHUNK_SIZE: usize = 1_048_576; // 1 Mo
43pub const ACK: u8 = 0x01;
44const COMPLETE: u8 = 0x02;
45/// réponse du récepteur pour refuser le transfert à la demande d'acceptation
46pub const REFUSE: u8 = 0x03;
47/// code de fermeture QUIC d'une fin normale de transfert
48pub const CLOSE_OK: u32 = 0;
49/// code de fermeture QUIC d'une annulation : le pair sait distinguer
50/// « annulé » d'une simple erreur réseau
51pub const CLOSE_CANCEL: u32 = 1;
52/// délai d'attente de la décision du destinataire (émetteur et récepteur)
53pub const DECISION_TIMEOUT: Duration = Duration::from_secs(30);
54const PORT: u16 = 58200;
55
56/// registre des demandes d'acceptation en attente : le récepteur y dépose un
57/// canal oneshot sous le transfer_id, et la commande respond_transfer le
58/// résout quand l'utilisateur clique sur accepter/refuser. auto_accept permet
59/// aux tests de sauter la validation (comportement historique).
60pub struct DecisionBoard {
61    pending: Mutex<HashMap<String, oneshot::Sender<bool>>>,
62    auto_accept: AtomicBool,
63}
64
65impl DecisionBoard {
66    pub fn new() -> Self {
67        DecisionBoard {
68            pending: Mutex::new(HashMap::new()),
69            auto_accept: AtomicBool::new(false),
70        }
71    }
72
73    pub fn set_auto_accept(&self, v: bool) {
74        self.auto_accept.store(v, Ordering::Relaxed);
75    }
76
77    pub fn auto_accept(&self) -> bool {
78        self.auto_accept.load(Ordering::Relaxed)
79    }
80
81    /// je dépose un canal d'attente pour ce transfert et je rends la partie
82    /// réceptrice ; l'app résout la décision via `resolve`.
83    pub fn register(&self, transfer_id: &str) -> oneshot::Receiver<bool> {
84        let (tx, rx) = oneshot::channel();
85        self.pending
86            .lock()
87            .unwrap_or_else(|e| e.into_inner())
88            .insert(transfer_id.to_string(), tx);
89        rx
90    }
91
92    /// je résous la décision de l'utilisateur ; false si le transfert est
93    /// inconnu ou déjà traité
94    pub fn resolve(&self, transfer_id: &str, accepted: bool) -> bool {
95        let tx = self
96            .pending
97            .lock()
98            .unwrap_or_else(|e| e.into_inner())
99            .remove(transfer_id);
100        match tx {
101            Some(tx) => {
102                let _ = tx.send(accepted);
103                true
104            }
105            None => false,
106        }
107    }
108
109    /// je retire une demande en attente sans la résoudre (timeout, annulation
110    /// ou arrêt de l'app) : le canal oneshot resterait sinon orphelin dans la
111    /// map et un respond_transfer tardif résoudrait un canal mort
112    pub fn remove(&self, transfer_id: &str) {
113        self.pending
114            .lock()
115            .unwrap_or_else(|e| e.into_inner())
116            .remove(transfer_id);
117    }
118
119    /// je vérifie si un transfert attend encore une décision (utile en test)
120    pub fn has_pending(&self, transfer_id: &str) -> bool {
121        self.pending
122            .lock()
123            .unwrap_or_else(|e| e.into_inner())
124            .contains_key(transfer_id)
125    }
126}
127
128impl Default for DecisionBoard {
129    fn default() -> Self {
130        Self::new()
131    }
132}
133
134/// je détecte si une erreur de lecture quinn correspond à une annulation
135/// distante : soit un flux reset par l'émetteur (arrêt utilisateur), soit une
136/// connexion fermée avec le code CLOSE_CANCEL
137pub fn is_remote_cancel(e: &quinn::ReadExactError) -> bool {
138    match e {
139        quinn::ReadExactError::ReadError(quinn::ReadError::Reset(_)) => true,
140        quinn::ReadExactError::ReadError(quinn::ReadError::ConnectionLost(
141            quinn::ConnectionError::ApplicationClosed(a),
142        )) => a.error_code.into_inner() == CLOSE_CANCEL as u64,
143        _ => false,
144    }
145}
146
147/// convertit une erreur de lecture quinn en ToolError, en distinguant
148/// l'annulation distante (RemoteCancel) des autres erreurs réseau
149pub fn quinn_to_err(e: quinn::ReadExactError) -> ToolError {
150    if is_remote_cancel(&e) {
151        ToolError::RemoteCancel
152    } else {
153        e.into()
154    }
155}
156
157/// convertit une erreur d'écriture quinn en ToolError, en distinguant
158/// l'annulation distante (connexion fermée avec CLOSE_CANCEL) des erreurs
159/// réseau : sans ça, un write_all qui échoue quand le récepteur annule serait
160/// signalé comme une erreur au lieu d'une annulation
161pub fn write_quinn_to_err(e: quinn::WriteError) -> ToolError {
162    match e {
163        quinn::WriteError::ConnectionLost(quinn::ConnectionError::ApplicationClosed(a))
164            if a.error_code.into_inner() == CLOSE_CANCEL as u64 =>
165        {
166            ToolError::RemoteCancel
167        }
168        e => e.into(),
169    }
170}
171
172/// limite les emissions de progression UI a ~20/s : le webview n'a pas besoin
173/// de 190 evenements/s et chaque IPC coûte cher en boucle serree
174struct UiThrottle {
175    last: Instant,
176}
177
178impl UiThrottle {
179    fn new() -> Self {
180        UiThrottle {
181            last: Instant::now(),
182        }
183    }
184
185    /// true si au moins 50ms se sont ecoulees depuis la derniere emission
186    fn ready(&mut self) -> bool {
187        if self.last.elapsed() >= Duration::from_millis(50) {
188            self.last = Instant::now();
189            true
190        } else {
191            false
192        }
193    }
194}
195
196pub fn io_err<E: std::fmt::Display>(e: E) -> ToolError {
197    IoError::other(e.to_string()).into()
198}
199
200pub async fn make_server_endpoint() -> Result<Endpoint, ToolError> {
201    let (cert_pem, key_pem) = certificat().await?;
202    let mut cert_reader = std::io::BufReader::new(cert_pem.as_bytes());
203    let mut key_reader = std::io::BufReader::new(key_pem.as_bytes());
204
205    let certs = rustls_pemfile::certs(&mut cert_reader).collect::<Result<Vec<_>, _>>()?;
206
207    let key = match rustls_pemfile::private_key(&mut key_reader)? {
208        Some(v) => v,
209        None => return Err(ToolError::ParseKeyError),
210    };
211
212    let server_config = ServerConfig::with_single_cert(certs, key)?;
213    let mut server_config = server_config;
214    server_config.transport_config(Arc::new(transport_config()));
215    let bind_addr: SocketAddr = format!("0.0.0.0:{PORT}").parse().map_err(io_err)?;
216    let endpoint = Endpoint::server(server_config, bind_addr)?;
217    Ok(endpoint)
218}
219
220/// fenetres QUIC larges (defaut quinn = 1,25 Mo par stream, ca plafonne le debit)
221/// initial_rtt garde au defaut quinn (333ms) : une valeur optimiste fait partir le
222/// pacing trop vite et provoque des pertes puis l'effondrement de Cubic sur WiFi
223/// j'ajoute aussi un idle timeout court + keepalive : une déconnexion soudaine
224/// (app fermée, réseau coupé) est détectée en ~15 s au lieu des 30 s par défaut,
225/// et le keepalive maintient la connexion vivante pendant la période silencieuse
226/// de la demande d'acceptation (jusqu'à 30 s de réflexion utilisateur)
227fn transport_config() -> quinn::TransportConfig {
228    use quinn::VarInt;
229    let mut t = quinn::TransportConfig::default();
230    t.stream_receive_window(VarInt::from(8u32 * 1024 * 1024));
231    t.receive_window(VarInt::from(32u32 * 1024 * 1024));
232    t.send_window(32 * 1024 * 1024);
233    t.max_idle_timeout(Some(
234        // 15 s est une constante dans la plage valide de quinn (10 ms – ~2^62 ms) :
235        // je documente l'invariant au lieu d'un unwrap muet
236        quinn::IdleTimeout::try_from(Duration::from_secs(15)).expect("idle timeout 15 s valide"),
237    ));
238    t.keep_alive_interval(Some(Duration::from_secs(3)));
239    t
240}
241
242pub fn make_client_endpoint(expected_fingerprint: Option<&str>) -> Result<Endpoint, ToolError> {
243    // je vérifie l'identité du serveur par son empreinte épinglée (TOFU) :
244    // None au premier contact (on épingle après le handshake), l'empreinte
245    // attendue ensuite. Je ne vérifie plus « rien » : un cert différent de
246    // celui épinglé fait échouer le handshake
247    let verifier = Arc::new(PinnedServerVerifier {
248        expected: expected_fingerprint.map(str::to_string),
249    });
250    let crypto = rustls::ClientConfig::builder()
251        .dangerous()
252        .with_custom_certificate_verifier(verifier)
253        .with_no_client_auth();
254
255    let client_config = ClientConfig::new(Arc::new(
256        quinn::crypto::rustls::QuicClientConfig::try_from(crypto).map_err(io_err)?,
257    ));
258    let mut client_config = client_config;
259    client_config.transport_config(Arc::new(transport_config()));
260
261    let bind_addr: SocketAddr = "0.0.0.0:0".parse().map_err(io_err)?;
262    let mut endpoint = Endpoint::client(bind_addr)?;
263    endpoint.set_default_client_config(client_config);
264    Ok(endpoint)
265}
266
267/// reçoit un lot de fichiers sur cette connexion et remonte la liste des fichiers recus
268#[allow(clippy::too_many_arguments)]
269pub async fn handle_incoming_connection(
270    connection: Connection,
271    dest_dir: PathBuf,
272    stop: Arc<AtomicBool>,
273    files: Arc<Mutex<Vec<String>>>,
274    bytes: Arc<AtomicU64>,
275    ui: Arc<dyn UI>,
276    transfer_id: Arc<Mutex<Option<String>>>,
277    total: Arc<AtomicU64>,
278    decisions: Arc<DecisionBoard>,
279    registry: Arc<dyn crate::TransferRegistry>,
280) -> Result<(), ToolError> {
281    // je mémorise si un flux a échoué en cours de route (ex: annulation côté
282    // émetteur) : dans ce cas je signalerai une erreur, pas une réception
283    let had_error = Arc::new(AtomicBool::new(false));
284    // je mémorise si le pair a annulé explicitement (reset ou code CLOSE_CANCEL)
285    // pour signaler une annulation, pas une erreur
286    let cancelled_by_peer = Arc::new(AtomicBool::new(false));
287    let mut handles: Vec<JoinHandle<()>> = Vec::new();
288    let mut header_received = false;
289    loop {
290        if stop.load(Ordering::Relaxed) {
291            connection.close(CLOSE_CANCEL.into(), b"arret utilisateur");
292            return Err(ToolError::Cancelled);
293        }
294
295        match tokio::time::timeout(Duration::from_secs(1), connection.accept_bi()).await {
296            Ok(Ok((mut send, mut recv))) => {
297                // le premier flux est l'en-tête de lot : il apporte transfer_id,
298                // le total et la liste des fichiers, et demande la validation de
299                // l'utilisateur (accepter / refuser) avant de recevoir
300                if !header_received {
301                    header_received = true;
302                    let header: BatchHeader = read_json_line(&mut recv).await?;
303                    {
304                        let mut tid = transfer_id.lock().unwrap_or_else(|e| e.into_inner());
305                        if tid.is_none() {
306                            *tid = Some(header.transfer_id.clone());
307                        }
308                    }
309                    total.store(header.total_bytes, Ordering::Relaxed);
310                    // je m'enregistre pour que la croix de la carte puisse
311                    // annuler cette réception depuis l'interface
312                    registry.register(&header.transfer_id, stop.clone());
313
314                    // je demande la décision à l'utilisateur et j'attends sa réponse
315                    ui.transfert_incoming(
316                        &header.transfer_id,
317                        &header.sender,
318                        header.total_bytes,
319                        header.files.clone(),
320                    );
321                    let decision = await_decision(&decisions, &header.transfer_id, &stop, &connection).await;
322
323                    match decision {
324                        Decision::Accepted => {
325                            send.write_all(&[ACK]).await.map_err(write_quinn_to_err)?;
326                            send.finish()?;
327                            ui.show_progress_bar(&header.transfer_id);
328                        }
329                        Decision::Refused | Decision::TimedOut => {
330                            // je laisse le temps au REFUSE d'être lu avant toute
331                            // fermeture : une fermeture immédiate envoie un
332                            // CONNECTION_CLOSE qui peut éclipser la donnée de
333                            // flux encore en transit, et l'émetteur notifierait
334                            // « annulé » au lieu de « refusé ». C'est l'émetteur
335                            // qui ferme, une fois le REFUSE lu.
336                            if send.write_all(&[REFUSE]).await.is_err() {
337                                // l'émetteur a déjà fermé (annulation pendant
338                                // l'attente) : le refus n'a plus de sens
339                                return Err(ToolError::RemoteCancel);
340                            }
341                            send.finish()?;
342                            return Err(ToolError::Refused);
343                        }
344                        Decision::Cancelled => {
345                            connection.close(CLOSE_CANCEL.into(), b"annulation utilisateur");
346                            return Err(ToolError::RemoteCancel);
347                        }
348                        Decision::RemoteCancelled => {
349                            return Err(ToolError::RemoteCancel);
350                        }
351                        Decision::RemoteError => {
352                            return Err(io_err("l'emetteur a ferme pendant la decision"));
353                        }
354                    }
355                    continue;
356                }
357
358                let dest_dir = dest_dir.clone();
359                let files = files.clone();
360                let bytes = bytes.clone();
361                let ui = ui.clone();
362                let transfer_id = transfer_id.clone();
363                let total = total.clone();
364                let had_error = had_error.clone();
365                let cancelled_by_peer = cancelled_by_peer.clone();
366                handles.push(tokio::spawn(async move {
367                    if let Err(e) = receive_one(
368                        send,
369                        recv,
370                        &dest_dir,
371                        files,
372                        bytes,
373                        ui,
374                        transfer_id,
375                        total,
376                        cancelled_by_peer.clone(),
377                    )
378                    .await
379                    {
380                        match e {
381                            ToolError::RemoteCancel => {
382                                // l'émetteur a annulé explicitement : je le
383                                // signale en annulation, pas en erreur
384                                cancelled_by_peer.store(true, Ordering::Relaxed);
385                            }
386                            e => {
387                                // je marque l'échec pour que la connexion soit
388                                // signalée en erreur (pas comme un transfert reçu)
389                                had_error.store(true, Ordering::Relaxed);
390                                eprintln!("Erreur reception fichier: {e}");
391                            }
392                        }
393                    }
394                }));
395            }
396            Ok(Err(quinn::ConnectionError::ApplicationClosed(a)))
397                if a.error_code.into_inner() == CLOSE_CANCEL as u64 =>
398            {
399                // le pair a fermé avec le code d'annulation (refus ou arrêt)
400                for h in handles {
401                    let _ = h.await;
402                }
403                return Err(ToolError::RemoteCancel);
404            }
405            Ok(Err(quinn::ConnectionError::ApplicationClosed(_)))
406            | Ok(Err(quinn::ConnectionError::LocallyClosed)) => {
407                // j'attends la fin des tâches de flux : elles posent
408                // `had_error` / `cancelled_by_peer` si un fichier a échoué ou a
409                // été annulé, et elles terminent vite une fois la connexion fermée
410                for h in handles {
411                    let _ = h.await;
412                }
413                if cancelled_by_peer.load(Ordering::Relaxed) {
414                    return Err(ToolError::RemoteCancel);
415                }
416                // si un flux a échoué (ex: annulation côté émetteur), je
417                // signale une erreur plutôt qu'une réception
418                if had_error.load(Ordering::Relaxed) {
419                    return Err(io_err("connexion fermee avec un flux en echec"));
420                }
421                // contrôle de complétude : si l'émetteur a disparu en cours de
422                // route, je refuse de notifier une réception alors que des
423                // octets manquent
424                let done = bytes.load(Ordering::Relaxed);
425                let expected = total.load(Ordering::Relaxed);
426                if done < expected {
427                    return Err(io_err("connexion interrompue : reception incomplete"));
428                }
429                return Ok(());
430            }
431            Ok(Err(e)) => {
432                // perte de connexion (ex: pair qui disparaît brutalement) :
433                // j'attends la fin des flux pour laisser le nettoyage des
434                // fichiers partiels se terminer, puis je signale l'erreur
435                for h in handles {
436                    let _ = h.await;
437                }
438                return Err(e.into());
439            }
440            Err(_) => continue, // timeout : on revérifie `stop`
441        }
442    }
443}
444
445/// motif de sortie de l'attente de décision : je distingue le refus explicite
446/// de l'utilisateur du timeout et de l'annulation pour des notifications UI
447/// cohérentes entre émetteur et récepteur
448enum Decision {
449    Accepted,
450    Refused,
451    TimedOut,
452    Cancelled,
453    /// l'émetteur a fermé la connexion pendant l'attente (annulation ou perte)
454    RemoteCancelled,
455    RemoteError,
456}
457
458/// attend la décision de l'utilisateur (ou auto-accepte en test). Je rends le
459/// motif exact pour que l'app distingue refus, timeout et annulation, et je
460/// surveille aussi la connexion : si l'émetteur annule pendant l'attente, la
461/// carte du récepteur ne doit pas rester bloquée 30 s en « en attente ».
462async fn await_decision(
463    decisions: &DecisionBoard,
464    transfer_id: &str,
465    stop: &Arc<AtomicBool>,
466    connection: &Connection,
467) -> Decision {
468    if decisions.auto_accept() {
469        return Decision::Accepted;
470    }
471let mut rx = decisions.register(transfer_id);
472        let deadline = Instant::now() + DECISION_TIMEOUT;
473        let mut closed = Box::pin(connection.closed());
474        loop {
475            if stop.load(Ordering::Relaxed) {
476                decisions.remove(transfer_id);
477                return Decision::Cancelled;
478            }
479            if Instant::now() >= deadline {
480                // le délai est écoulé : je retire la demande en attente pour ne
481                // pas laisser un canal orphelin résolvable par un clic tardif
482                decisions.remove(transfer_id);
483                return Decision::TimedOut;
484            }
485            tokio::select! {
486                r = &mut rx => {
487                    match r {
488                        Ok(true) => return Decision::Accepted,
489                        Ok(false) => return Decision::Refused,
490                        // canal fermé (app qui s'éteint) : la demande ne peut
491                        // plus être résolue, je la retire du registre
492                        Err(_) => {
493                            decisions.remove(transfer_id);
494                            return Decision::Cancelled;
495                        }
496                    }
497                }
498                err = &mut closed => {
499                    // l'émetteur a fermé pendant l'attente : je retire la
500                    // demande et je rends la main sans attendre le timeout
501                    decisions.remove(transfer_id);
502                    let is_cancel = matches!(
503                        &err,
504                        quinn::ConnectionError::ApplicationClosed(a)
505                            if a.error_code.into_inner() == CLOSE_CANCEL as u64
506                    );
507                    if is_cancel {
508                        return Decision::RemoteCancelled;
509                    }
510                    return Decision::RemoteError;
511                }
512                _ = tokio::time::sleep(Duration::from_millis(100)) => continue,
513            }
514        }
515    }
516
517#[allow(clippy::too_many_arguments)]
518async fn receive_one(
519    mut send: SendStream,
520    mut recv: RecvStream,
521    dest_dir: &Path,
522    files: Arc<Mutex<Vec<String>>>,
523    bytes: Arc<AtomicU64>,
524    ui: Arc<dyn UI>,
525    transfer_id: Arc<Mutex<Option<String>>>,
526    total: Arc<AtomicU64>,
527    cancelled_by_peer: Arc<AtomicBool>,
528) -> Result<(), ToolError> {
529    let metadata = read_json_line::<Metadata>(&mut recv).await?;
530    let full_path = dest_dir.join(&metadata.rel_path);
531    let name = Path::new(&metadata.rel_path)
532        .file_name()
533        .map(|f| f.to_string_lossy().to_string())
534        .unwrap_or_else(|| metadata.rel_path.clone());
535
536    if let Some(parent) = full_path.parent() {
537        fs::create_dir_all(parent).await?;
538    }
539
540    // l'id du transfert vient de l'en-tête de lot ; en secours (protocole
541    // ancien) je le prends du metadata, sans redéclencher la barre de
542    // progression qui a été signalée par l'en-tête
543    {
544        let mut tid = transfer_id.lock().unwrap_or_else(|e| e.into_inner());
545        if tid.is_none() {
546            *tid = Some(metadata.transfer_id.clone());
547        }
548    }
549    let tid = transfer_id
550        .lock()
551        .unwrap_or_else(|e| e.into_inner())
552        .clone()
553        .unwrap_or_default();
554
555    if metadata.is_dir {
556        fs::create_dir_all(&full_path).await?;
557        send.write_all(&[ACK]).await.map_err(write_quinn_to_err)?;
558        send.finish()?;
559        files.lock().unwrap_or_else(|e| e.into_inner()).push(name);
560        return Ok(());
561    }
562
563    // Ack métadonnées
564    send.write_all(&[ACK]).await.map_err(write_quinn_to_err)?;
565
566    let mut out_file = fs::File::create(&full_path).await?;
567    let mut received: u64 = 0;
568    let mut throttle = UiThrottle::new();
569
570    let res: Result<(), ToolError> = async {
571        while received < metadata.size {
572            let mut len_buf = [0u8; 4];
573            recv.read_exact(&mut len_buf).await.map_err(quinn_to_err)?;
574            let len = u32::from_be_bytes(len_buf) as usize;
575
576            let mut data = vec![0u8; len];
577            recv.read_exact(&mut data).await.map_err(quinn_to_err)?;
578
579            out_file.write_all(&data).await?;
580            received += len as u64;
581
582            // progression cote receveur, aluminee sur celle de l'emetteur
583            let done = bytes.fetch_add(len as u64, Ordering::Relaxed) + len as u64;
584            if throttle.ready() || received == metadata.size {
585                ui.update_progress_bar(&tid, done, total.load(Ordering::Relaxed));
586                ui.file_progress_bar(&tid, &name, received, metadata.size);
587            }
588        }
589
590        out_file.flush().await?;
591
592        // Marqueur de complétion, puis ack final
593        let mut complete = [0u8; 1];
594        recv.read_exact(&mut complete).await.map_err(quinn_to_err)?;
595        if complete[0] != COMPLETE {
596            return Err(io_err(
597                "fin de fichier inattendue (protocole desynchronise)",
598            ));
599        }
600        send.write_all(&[ACK]).await.map_err(write_quinn_to_err)?;
601
602        send.finish()?;
603        Ok(())
604    }
605    .await;
606
607    match &res {
608        Ok(()) => {
609            files.lock().unwrap_or_else(|e| e.into_inner()).push(name);
610        }
611        Err(ToolError::RemoteCancel) => {
612            // l'émetteur a annulé : pas de nettoyage nécessaire côté récepteur
613            // (je ne laisse pas de réception tronquée : je signale l'annulation)
614            cancelled_by_peer.store(true, Ordering::Relaxed);
615        }
616        Err(_) => {
617            // je supprime le fichier partiel : jamais de réception tronquée
618            // dans le dossier de destination
619            let _ = fs::remove_file(&full_path).await;
620        }
621    }
622    res
623}
624
625pub async fn read_json_line<T: for<'de> Deserialize<'de>>(
626    recv: &mut RecvStream,
627) -> Result<T, ToolError> {
628    let mut buf = Vec::new();
629    let mut byte = [0u8; 1];
630
631    loop {
632        recv.read_exact(&mut byte).await.map_err(quinn_to_err)?;
633        if byte[0] == b'\n' {
634            break;
635        }
636        buf.push(byte[0]);
637    }
638
639    serde_json::from_slice(&buf).map_err(io_err)
640}
641
642pub fn collect_entries<'a>(
643    root: &'a Path,
644    current: &'a Path,
645    out: &'a mut Vec<(PathBuf, String, bool)>,
646) -> std::pin::Pin<Box<dyn std::future::Future<Output = Result<(), ToolError>> + Send + 'a>> {
647    Box::pin(async move {
648        let metadata = fs::metadata(current).await?;
649        let base = root.parent().unwrap_or(root);
650
651        if metadata.is_file() {
652            let rel = current
653                .strip_prefix(base)
654                .unwrap_or(current)
655                .to_string_lossy()
656                .replace('\\', "/");
657            out.push((current.to_path_buf(), rel, false));
658            return Ok(());
659        }
660
661        if metadata.is_dir() {
662            let mut read_dir = fs::read_dir(current).await?;
663            let mut has_children = false;
664
665            while let Some(entry) = read_dir.next_entry().await? {
666                has_children = true;
667                collect_entries(root, &entry.path(), out).await?;
668            }
669
670            if !has_children {
671                let rel = current
672                    .strip_prefix(base)
673                    .unwrap_or(current)
674                    .to_string_lossy()
675                    .replace('\\', "/");
676                out.push((current.to_path_buf(), rel, true));
677            }
678        }
679
680        Ok(())
681    })
682}
683
684/// envoie un fichier ou dossier dans la connexion QUIC établie : les paquets
685/// sont fiables grâce à QUIC (pas de retry applicatif), je copie le contenu en
686/// blocs et je signale la progression via l'interface UI.
687#[allow(clippy::too_many_arguments)]
688pub async fn send_entry(
689    connection: Connection,
690    abs_path: PathBuf,
691    rel_path: String,
692    is_dir: bool,
693    stop: Arc<AtomicBool>,
694    ui: Arc<dyn UI>,
695    transfer_id: String,
696    total_bytes: u64,
697    bytes_sent_counter: Arc<AtomicU64>,
698) -> Result<(), ToolError> {
699    let (mut send, mut recv) = connection.open_bi().await?;
700
701    if is_dir {
702        let metadata = Metadata {
703            transfer_id: transfer_id.clone(),
704            rel_path,
705            size: 0,
706            is_dir: true,
707        };
708        write_json_line(&mut send, &metadata).await?;
709        let mut ack = [0u8; 1];
710        recv.read_exact(&mut ack).await.map_err(quinn_to_err)?;
711        if ack[0] != ACK {
712            return Err(io_err("dossier rejete par le receveur"));
713        }
714        send.finish()?;
715        return Ok(());
716    }
717
718    let file_name = Path::new(&rel_path)
719        .file_name()
720        .map(|f| f.to_string_lossy().to_string())
721        .unwrap_or_else(|| rel_path.clone());
722
723    let size = fs::metadata(&abs_path).await?.len();
724    let metadata = Metadata {
725        transfer_id: transfer_id.clone(),
726        rel_path,
727        size,
728        is_dir: false,
729    };
730    write_json_line(&mut send, &metadata).await?;
731
732    // Ack métadonnées
733    let mut ack = [0u8; 1];
734    recv.read_exact(&mut ack).await.map_err(quinn_to_err)?;
735    if ack[0] != ACK {
736        return Err(io_err("metadonnees rejetees par le receveur"));
737    }
738
739    // Chunks — pas d'ack par chunk : QUIC assure la fiabilite, on pipeline
740    let mut file = fs::File::open(&abs_path).await?;
741    let mut buf = vec![0u8; CHUNK_SIZE];
742    let mut file_sent: u64 = 0;
743    let mut throttle = UiThrottle::new();
744
745    loop {
746        if stop.load(Ordering::Relaxed) {
747            let _ = send.reset(0u32.into());
748            return Err(io_err("transfert annule par l'utilisateur"));
749        }
750
751        let n = file.read(&mut buf).await?;
752        if n == 0 {
753            break;
754        }
755
756        send.write_all(&(n as u32).to_be_bytes())
757            .await
758            .map_err(write_quinn_to_err)?;
759        send.write_all(&buf[..n]).await.map_err(write_quinn_to_err)?;
760
761        // Progression globale cumulee (lot) + per-fichier
762        file_sent += n as u64;
763        let total_sent = bytes_sent_counter.fetch_add(n as u64, Ordering::Relaxed) + n as u64;
764        if throttle.ready() || file_sent == size {
765            ui.update_progress_bar(&transfer_id, total_sent, total_bytes);
766            ui.file_progress_bar(&transfer_id, &file_name, file_sent, size);
767        }
768    }
769
770    // Marqueur de complétion
771    send.write_all(&[COMPLETE]).await.map_err(write_quinn_to_err)?;
772
773    // Ack final
774    let mut final_ack = [0u8; 1];
775    recv.read_exact(&mut final_ack).await.map_err(quinn_to_err)?;
776    send.finish()?;
777
778    if final_ack[0] != ACK {
779        return Err(io_err("le receveur a rejete le fichier"));
780    }
781
782    Ok(())
783}
784
785pub async fn write_json_line<T: Serialize>(send: &mut SendStream, value: &T) -> Result<(), ToolError> {
786    let mut encoded = serde_json::to_vec(value).map_err(io_err)?;
787    encoded.push(b'\n');
788    send.write_all(&encoded).await.map_err(write_quinn_to_err)?;
789    Ok(())
790}
toole_core/transfer.rs

toole_core/
transfer.rs
