cfdp_simplified/transaction/
recv.rs

1use std::{
2    collections::VecDeque,
3    fs::File,
4    io::{self, Seek, SeekFrom, Write},
5    sync::Arc,
6    time::Duration,
7};
8
9use camino::Utf8PathBuf;
10use chrono::{DateTime, Utc};
11use log::{debug, info, warn};
12use tokio::sync::{
13    mpsc::{Permit, Sender},
14    oneshot,
15};
16
17use super::fault_handler::FaultHandlerAction;
18use crate::{
19    daemon::{
20        segments::Segments,
21        timer::{Counter, Timer},
22        FileSegmentIndication, FinishedIndication, Indication, MetadataRecvIndication,
23        NakProcedure, Report,
24    },
25    filestore::{FileChecksum, FileStore, FileStoreError},
26    pdu::{
27        Condition, DeliveryCode, Direction, EntityID, FileDataPDU, FileStatusCode,
28        NegativeAcknowledgmentPDU, Operations, PDUHeader, PDUPayload, PDUType, SegmentRequestForm,
29        SegmentationControl, TransactionStatus, TransmissionMode, PDU, U3,
30    },
31    transaction::{
32        Metadata, TransactionConfig, TransactionError, TransactionID, TransactionResult,
33        TransactionState,
34    },
35};
36
37#[derive(PartialEq, Debug)]
38enum RecvState {
39    // initial state
40    // received data, missing data and EOF
41    ReceiveData,
42    // send Finished and wait for ack
43    Finished,
44}
45
46pub struct RecvTransaction<T: FileStore> {
47    /// The current status of the Transaction. See [TransactionStatus]
48    status: TransactionStatus,
49    /// Configuration Information for this Transaction. See [TransactionConfig]
50    config: TransactionConfig,
51    /// The [FileStore] implementation used to interact with files on disk.
52    filestore: Arc<T>,
53    /// Channel for Indications to propagate back up
54    indication_tx: Sender<Indication>,
55    /// The current file being worked by this Transaction.
56    file_handle: Option<File>,
57    /// A sorted list of contiguous (start offset, end offset) non overlapping received segments to monitor progress and detect NAKs.
58    saved_segments: Segments,
59    /// when to send NAKs - immediately after detection or after EOF
60    nak_procedure: NakProcedure,
61    /// Flag to check if metadata on the file has been received
62    pub(crate) metadata: Option<Metadata>,
63    /// Measurement of how much of the file has been received so far
64    received_file_size: u64,
65    /// a cache of the header used for interactions in this transmission
66    pub header: Option<PDUHeader>,
67    /// The current condition of the transaction
68    condition: Condition,
69    // Track whether the transaction is complete or not
70    delivery_code: DeliveryCode,
71    // Status of the current File
72    file_status: FileStatusCode,
73    // Timer used to track if the Nak limit has been reached
74    // inactivity has occurred
75    // or the ACK limit is reached
76    timer: Timer,
77    // checksum cache to reduce I/0
78    // doubles as stored checksum in received mode
79    checksum: Option<u32>,
80    // The current state of the transaction.
81    // Used to determine when the thread should be killed
82    state: TransactionState,
83    // recv sub-state, applicable when state = Active
84    recv_state: RecvState,
85    // File size received in the EOF, used also as an indication that EOF has been received
86    file_size: Option<u64>,
87    // Empty nak PDU prepared to be sent at the next opportunity, if the bool is true
88    empty_nak: Option<(NegativeAcknowledgmentPDU, bool)>,
89    // the list of gaps to include in the next NAK
90    naks: VecDeque<SegmentRequestForm>,
91    /// the received_file_size measured when the previous nak has been sent
92    nak_received_file_size: u64,
93
94    /// This is used in case the NAK procedure is Immediate(delta) with delta>0
95    /// it stores a timer and start, stop offsets. When the timer expires,
96    /// the (start, stop) file region is checked for completeness and if gaps still persists,
97    /// a nack is added to the list
98    /// It is also used in case the NAK procedure is Deferred(delta) with delta>0
99    /// In that case the start, stop offsets are the whole file
100    delayed_nack_timers: Vec<(Counter, u64, u64)>,
101
102    /// Date transaction was initially received
103    receive_date: DateTime<Utc>,
104}
105
106impl<T: FileStore> RecvTransaction<T> {
107    /// Start a new SendTransaction with the given [configuration](TransactionConfig)
108    /// and [Filestore Implementation](FileStore).
109    ///
110    /// The [NakProcedure] is most likely passed from [EntityConfig](crate::daemon::EntityConfig)
111    pub fn new(
112        // Configuration of this Transaction.
113        config: TransactionConfig,
114        // Desired NAK procedure. This is most likely passed from an EntityConfig
115        nak_procedure: NakProcedure,
116        // Connection to the local FileStore implementation.
117        filestore: Arc<T>,
118        // Sender channel used to propagate Message To User back up to the Daemon Thread.
119        indication_tx: Sender<Indication>,
120    ) -> Self {
121        let received_file_size = 0_u64;
122        let timer = Timer::new(
123            config.inactivity_timeout,
124            config.max_count,
125            config.eof_timeout,
126            config.max_count,
127            config.nak_timeout,
128            config.max_count,
129            config.progress_report_interval_secs,
130        );
131
132        let mut transaction = Self {
133            status: TransactionStatus::Undefined,
134            config,
135            filestore,
136            indication_tx,
137            file_handle: None,
138            saved_segments: Segments::new(),
139            nak_procedure,
140            metadata: None,
141            received_file_size,
142            header: None,
143            condition: Condition::NoError,
144            delivery_code: DeliveryCode::Incomplete,
145            file_status: FileStatusCode::Unreported,
146            timer,
147            checksum: None,
148            state: TransactionState::Active,
149            recv_state: RecvState::ReceiveData,
150            empty_nak: None,
151            file_size: None,
152            naks: VecDeque::new(),
153            nak_received_file_size: received_file_size,
154            delayed_nack_timers: Vec::new(),
155            receive_date: Utc::now(),
156        };
157        transaction.timer.restart_inactivity();
158        transaction.timer.reset_progress_report();
159        transaction
160    }
161
162    pub(crate) fn has_pdu_to_send(&self) -> bool {
163        match self.recv_state {
164            RecvState::ReceiveData => self.empty_nak.is_some() || !self.naks.is_empty(),
165            RecvState::Finished => self.empty_nak.as_ref().is_some_and(|x| x.1),
166        }
167    }
168
169    // returns the time until the first timeout
170    pub(crate) fn until_timeout(&self) -> Duration {
171        let d = self.timer.until_timeout();
172        // we only check the first delayed nack timer (if any)
173        // because they are ordered chronological
174        self.delayed_nack_timers
175            .first()
176            .map_or(d, |(counter, _, _)| {
177                std::cmp::min(d, counter.until_timeout())
178            })
179    }
180
181    pub(crate) fn send_pdu(&mut self, permit: Permit<(EntityID, PDU)>) -> TransactionResult<()> {
182        match self.recv_state {
183            RecvState::ReceiveData => {
184                if self.empty_nak.is_some() {
185                    self.send_empty_nak(permit)?;
186                } else if !self.naks.is_empty() {
187                    self.send_naks(permit)?;
188                }
189            }
190            RecvState::Finished => {
191                if self.empty_nak.is_some() {
192                    self.send_empty_nak(permit)?;
193                }
194            }
195        }
196
197        Ok(())
198    }
199
200    pub fn handle_timeout(&mut self) -> TransactionResult<()> {
201        let mut idx = 0;
202        for (i, (counter, _, _)) in self.delayed_nack_timers.iter_mut().enumerate() {
203            if counter.timeout_occurred() {
204                idx = i + 1;
205            } else {
206                break;
207            }
208        }
209
210        if idx > 0 {
211            for (_, start, end) in self.delayed_nack_timers.drain(0..idx) {
212                for (start_offset, end_offset) in self.saved_segments.gaps(start, end) {
213                    self.naks.push_back(SegmentRequestForm {
214                        start_offset,
215                        end_offset,
216                    });
217                }
218            }
219        }
220
221        if self.timer.inactivity.limit_reached() {
222            self.handle_fault(Condition::InactivityDetected)?;
223        } else if self.timer.inactivity.timeout_occurred() {
224            self.timer.restart_inactivity();
225        }
226
227        match self.recv_state {
228            RecvState::ReceiveData => {
229                if self.timer.nak.timeout_occurred() {
230                    self.naks = self.get_all_naks();
231                }
232            }
233            RecvState::Finished => {
234                if self.timer.nak.limit_reached() {
235                    self.handle_fault(Condition::PositiveLimitReached)?;
236                } else if self.timer.nak.timeout_occurred() {
237                    self.set_finished_flag(true);
238                    self.timer.restart_nak();
239                }
240            }
241        }
242
243        Ok(())
244    }
245
246    // return true if:
247    // - metadata has not been received
248    // - there is a gap in the data received
249    // - EOF has been received and either no data has been received
250    //   or there is a gap between the last segment received and the end of file according to the file_size received in the EOF
251    fn has_naks(&self) -> bool {
252        self.metadata.is_none() || {
253            if let Some(file_size) = self.file_size {
254                //eof received
255                !self.saved_segments.is_complete(file_size)
256            } else {
257                //eof not received
258                self.saved_segments.len() > 1
259            }
260        }
261    }
262
263    fn eof_received(&self) -> bool {
264        self.file_size.is_some()
265    }
266
267    pub fn get_status(&self) -> TransactionStatus {
268        self.status
269    }
270
271    pub(crate) fn get_state(&self) -> TransactionState {
272        self.state
273    }
274
275    fn generate_report(&self) -> Report {
276        Report {
277            id: self.id(),
278            state: self.get_state(),
279            status: self.get_status(),
280            condition: self.condition,
281            empty_nak_received: false,
282            file_size: self.metadata.as_ref().map_or(0, |meta| meta.file_size),
283            file_bytes_received: Some(self.received_file_size),
284            file_bytes_sent: None,
285            direction: self.header.as_ref().map(|h| h.direction),
286            file_name: self
287                .metadata
288                .as_ref()
289                .map_or("Unknown".to_string(), |meta| {
290                    meta.destination_filename.to_string()
291                }),
292            submit_date: self.receive_date,
293        }
294    }
295
296    pub fn send_report(&self, sender: Option<oneshot::Sender<Report>>) -> TransactionResult<()> {
297        let report = self.generate_report();
298        if let Some(channel) = sender {
299            let _ = channel.send(report.clone());
300        }
301        self.send_indication(Indication::Report(report));
302
303        Ok(())
304    }
305
306    fn get_header(
307        &mut self,
308        direction: Direction,
309        pdu_type: PDUType,
310        pdu_data_field_length: u16,
311        segmentation_control: SegmentationControl,
312    ) -> PDUHeader {
313        if let Some(header) = &self.header {
314            // update the necessary fields but copy the rest
315            // from the cache
316            PDUHeader {
317                pdu_type,
318                pdu_data_field_length,
319                segmentation_control,
320                ..header.clone()
321            }
322        } else {
323            let header = PDUHeader {
324                version: U3::Zero,
325                pdu_type,
326                direction,
327                transmission_mode: self.config.transmission_mode,
328                crc_flag: self.config.crc_flag,
329                large_file_flag: self.config.file_size_flag,
330                pdu_data_field_length,
331                segmentation_control,
332                segment_metadata_flag: self.config.segment_metadata_flag,
333                source_entity_id: self.config.source_entity_id,
334                transaction_sequence_number: self.config.sequence_number,
335                destination_entity_id: self.config.destination_entity_id,
336            };
337            self.header = Some(header.clone());
338            header
339        }
340    }
341
342    pub fn id(&self) -> TransactionID {
343        TransactionID(self.config.source_entity_id, self.config.sequence_number)
344    }
345    fn initialize_tempfile(&mut self) -> TransactionResult<()> {
346        self.file_handle = Some(self.filestore.open_tempfile()?);
347        Ok(())
348    }
349
350    fn get_handle(&mut self) -> TransactionResult<&mut File> {
351        let id = self.id();
352        if self.file_handle.is_none() {
353            self.initialize_tempfile()?
354        };
355        self.file_handle
356            .as_mut()
357            .ok_or(TransactionError::NoFile(id))
358    }
359
360    fn is_file_transfer(&self) -> bool {
361        self.metadata
362            .as_ref()
363            .map(|meta| !meta.source_filename.as_os_str().is_empty())
364            .unwrap_or(false)
365    }
366
367    fn store_file_data(&mut self, pdu: FileDataPDU) -> TransactionResult<(u64, usize)> {
368        let (offset, file_data) = (pdu.0.offset, pdu.0.file_data);
369        let length = file_data.len();
370
371        if length > 0 {
372            let handle = self.get_handle()?;
373            handle
374                .seek(SeekFrom::Start(offset))
375                .map_err(FileStoreError::IO)?;
376            handle
377                .write_all(file_data.as_slice())
378                .map_err(FileStoreError::IO)?;
379            let new_data_received = self
380                .saved_segments
381                .merge((offset, offset + file_data.len() as u64));
382            self.received_file_size += new_data_received;
383        } else {
384            warn!(
385                "Received FileDataPDU with invalid file_data.length = {}; ignored",
386                length
387            );
388        }
389
390        Ok((offset, length))
391    }
392
393    fn finalize_file(&mut self) -> TransactionResult<FileStatusCode> {
394        let id = self.id();
395        {
396            let mut outfile = self.filestore.open(
397                self.metadata
398                    .as_ref()
399                    .map(|meta| meta.destination_filename.clone())
400                    .ok_or(TransactionError::NoFile(id))?,
401                File::options().create(true).write(true).truncate(true),
402            )?;
403            let handle = self.get_handle()?;
404            // rewind to the beginning of the file.
405            // this might not be necessary with the io call that follows
406            handle.rewind().map_err(FileStoreError::IO)?;
407            io::copy(handle, &mut outfile).map_err(FileStoreError::IO)?;
408            outfile.sync_all().map_err(FileStoreError::IO)?;
409        }
410        // Drop the temporary file
411        self.file_handle = None;
412
413        Ok(FileStatusCode::Retained)
414    }
415
416    pub fn shutdown(&mut self) {
417        debug!("Transaction {0} shutting down.", self.id());
418        self.status = TransactionStatus::Terminated;
419        self.state = TransactionState::Terminated;
420        self.timer.nak.pause();
421        self.timer.inactivity.pause();
422    }
423
424    /// Take action according to the defined handler mapping.
425    /// Returns a boolean indicating if the calling function should continue (true) or not (false.)
426    fn handle_fault(&mut self, condition: Condition) -> TransactionResult<bool> {
427        self.condition = condition;
428        warn!("Transaction {} Handling fault {:?}", self.id(), condition);
429        match self
430            .config
431            .fault_handler_override
432            .get(&self.condition)
433            .unwrap_or(&FaultHandlerAction::Abandon)
434        {
435            FaultHandlerAction::Ignore => {
436                // Log ignoring error
437                Ok(true)
438            }
439            FaultHandlerAction::Abandon => {
440                self.shutdown();
441                Ok(false)
442            }
443        }
444    }
445
446    fn check_file_size(&mut self, file_size: u64) -> TransactionResult<()> {
447        if self.saved_segments.end_or_0() > file_size {
448            warn!(
449                "EOF file size {} is smaller than file size received in file data {}",
450                file_size, self.received_file_size
451            );
452            // we will always exit here anyway
453            self.handle_fault(Condition::FilesizeError)?;
454        }
455        Ok(())
456    }
457
458    fn prepare_empty_nak(&mut self) {
459        self.empty_nak = Some((
460            NegativeAcknowledgmentPDU {
461                start_of_scope: 0,
462                end_of_scope: self.received_file_size,
463                segment_requests: vec![],
464            },
465            true,
466        ));
467    }
468
469    // set the true flag on the fin such that it is sent at the next opportunity
470    fn set_finished_flag(&mut self, flag: bool) {
471        if let Some(x) = self.empty_nak.as_mut() {
472            x.1 = flag;
473        }
474    }
475
476    fn get_all_naks(&self) -> VecDeque<SegmentRequestForm> {
477        let mut naks: VecDeque<SegmentRequestForm> = VecDeque::new();
478
479        let segments = &self.saved_segments;
480
481        if self.metadata.is_none() {
482            naks.push_back((0_u64, 0_u64).into());
483        }
484        for (start_offset, end_offset) in
485            segments.gaps(0, self.file_size.unwrap_or(segments.end_or_0()))
486        {
487            naks.push_back(SegmentRequestForm {
488                start_offset,
489                end_offset,
490            });
491        }
492
493        naks
494    }
495
496    fn send_empty_nak(&mut self, permit: Permit<(EntityID, PDU)>) -> TransactionResult<()> {
497        let payload = PDUPayload::Directive(Operations::Nak(NegativeAcknowledgmentPDU {
498            start_of_scope: 0,
499            end_of_scope: self.received_file_size,
500            segment_requests: vec![],
501        }));
502        let payload_len = payload.encoded_len(self.config.file_size_flag);
503
504        let header = self.get_header(
505            Direction::ToSender,
506            PDUType::FileDirective,
507            payload_len,
508            SegmentationControl::NotPreserved,
509        );
510
511        let destination = header.source_entity_id;
512        let pdu = PDU { header, payload };
513
514        // TODO - figure out a way to send this periodically up to a timeout.
515        // Currently, this will break the sender because the sender shutdowns on receipt of an empty NAK and the transport closes.
516        info!("Sending empty NAK");
517        permit.send((destination, pdu));
518        self.shutdown();
519
520        Ok(())
521    }
522
523    fn send_naks(&mut self, permit: Permit<(EntityID, PDU)>) -> TransactionResult<()> {
524        if self.nak_received_file_size == self.received_file_size {
525            if self.timer.nak.limit_reached() {
526                self.handle_fault(Condition::NakLimitReached)?;
527                return Ok(());
528            }
529            self.timer.restart_nak();
530        } else {
531            // new data has been received since last NAK was send; reset the counter to 0
532            self.timer.reset_nak();
533            self.nak_received_file_size = self.received_file_size;
534        }
535
536        let n = usize::min(
537            self.naks.len(),
538            NegativeAcknowledgmentPDU::max_nak_num(
539                self.config.file_size_flag,
540                self.config.file_size_segment as u32,
541            ) as usize,
542        );
543
544        let segment_requests: Vec<SegmentRequestForm> = self.naks.drain(..n).collect();
545        let scope_start = segment_requests
546            .first()
547            .map(|sr| sr.start_offset)
548            .unwrap_or(0);
549        let scope_end = segment_requests
550            .last()
551            .map(|sr| sr.end_offset)
552            .unwrap_or(self.saved_segments.end().unwrap_or(0));
553
554        let nak = NegativeAcknowledgmentPDU {
555            start_of_scope: scope_start,
556            end_of_scope: scope_end,
557            segment_requests,
558        };
559        debug!("Transaction {}: sending NAK for {} segments", self.id(), n);
560
561        let payload = PDUPayload::Directive(Operations::Nak(nak));
562        let payload_len = payload.encoded_len(self.config.file_size_flag);
563
564        let header = self.get_header(
565            Direction::ToSender,
566            PDUType::FileDirective,
567            payload_len,
568            SegmentationControl::NotPreserved,
569        );
570        let destination = header.source_entity_id;
571        let pdu = PDU { header, payload };
572
573        permit.send((destination, pdu));
574
575        Ok(())
576    }
577
578    fn verify_checksum(&mut self, checksum: u32) -> TransactionResult<bool> {
579        let checksum_type = self
580            .metadata
581            .as_ref()
582            .map(|meta| meta.checksum_type)
583            .ok_or_else(|| {
584                let id = self.id();
585                TransactionError::MissingMetadata(id)
586            })?;
587        let handle = self.get_handle()?;
588        handle.sync_all().map_err(FileStoreError::IO)?;
589        Ok(handle.checksum(checksum_type)? == checksum)
590    }
591
592    fn finalize_receive(&mut self) -> TransactionResult<()> {
593        let checksum = self.checksum.ok_or(TransactionError::NoChecksum)?;
594        self.delivery_code = DeliveryCode::Complete;
595
596        self.file_status = if self.is_file_transfer() {
597            if !self.verify_checksum(checksum)?
598                && !self.handle_fault(Condition::FileChecksumFailure)?
599            {
600                return Ok(());
601            }
602            self.finalize_file()
603                .unwrap_or(FileStatusCode::FileStoreRejection)
604        } else {
605            FileStatusCode::Unreported
606        };
607        // A filestore rejection is a failure mode for the entire transaction.
608        if self.file_status == FileStatusCode::FileStoreRejection
609            && !self.handle_fault(Condition::FileStoreRejection)?
610        {
611            return Ok(());
612        }
613        // send indication this transaction is finished.
614        self.send_indication(Indication::Finished(FinishedIndication {
615            id: self.id(),
616            report: self.generate_report(),
617            file_status: self.file_status,
618            delivery_code: self.delivery_code,
619        }));
620        Ok(())
621    }
622
623    pub fn process_pdu(&mut self, pdu: PDU) -> TransactionResult<()> {
624        self.timer.reset_inactivity();
625        let PDU {
626            header: _header,
627            payload,
628        } = pdu;
629        if self.timer.progress_report.timeout_occurred() {
630            self.timer.reset_progress_report();
631            self.send_indication(Indication::Report(self.generate_report()));
632        }
633
634        match &self.config.transmission_mode {
635            TransmissionMode::Acknowledged => {
636                match payload {
637                    PDUPayload::FileData(filedata) => {
638                        // the end of the last segment
639                        let prev_end = self.saved_segments.end().unwrap_or(0);
640
641                        let (offset, length) = self.store_file_data(filedata)?;
642
643                        self.send_indication(Indication::FileSegmentRecv(FileSegmentIndication {
644                            id: self.id(),
645                            offset,
646                            length: length as u64,
647                        }));
648
649                        if let NakProcedure::Immediate(delay) = self.nak_procedure {
650                            if !self.eof_received() {
651                                if self.timer.nak.timeout_occurred() {
652                                    // send all gaps at the next opportunity
653                                    self.naks = self.get_all_naks();
654                                    self.timer.restart_nak();
655                                } else if offset > prev_end {
656                                    // new gap
657                                    if delay.is_zero() {
658                                        // send it at the next opportunity
659                                        self.naks.push_back(SegmentRequestForm {
660                                            start_offset: prev_end,
661                                            end_offset: offset,
662                                        });
663                                    } else {
664                                        // start a timer to check if the gap still persists after the delta delay
665                                        let mut timer = Counter::new(delay, 1);
666                                        timer.start();
667                                        self.delayed_nack_timers.push((timer, prev_end, offset));
668                                    }
669                                }
670                            }
671                        }
672                        self.check_finished()?;
673
674                        Ok(())
675                    }
676                    PDUPayload::Directive(operation) => {
677                        match operation {
678                            Operations::EoF(eof) => {
679                                debug!("Transaction {0} received EndOfFile.", self.id());
680                                self.condition = eof.condition;
681                                self.checksum = Some(eof.checksum);
682
683                                self.send_indication(Indication::EoFRecv(self.id()));
684
685                                if self.condition == Condition::NoError {
686                                    self.check_file_size(eof.file_size)?;
687                                    self.file_size = Some(eof.file_size);
688
689                                    self.check_finished()?;
690
691                                    if self.has_naks() {
692                                        let delay = match self.nak_procedure {
693                                            NakProcedure::Immediate(d) => d,
694                                            NakProcedure::Deferred(d) => d,
695                                        };
696
697                                        if delay.is_zero() {
698                                            //send all gaps at the next opportunity
699                                            self.naks = self.get_all_naks();
700                                        } else {
701                                            // we need to check/send the gaps only after this delta has expired
702                                            // start a counter for that
703                                            self.delayed_nack_timers.push((
704                                                Counter::new(delay, 1),
705                                                0,
706                                                eof.file_size,
707                                            ));
708                                        }
709                                    } else {
710                                        self.prepare_empty_nak();
711                                    }
712                                } else {
713                                    // Any other condition is essentially a
714                                    // SHUTDOWN operation
715                                    self.shutdown();
716                                }
717                                Ok(())
718                            }
719                            Operations::Metadata(metadata) => {
720                                if self.metadata.is_none() {
721                                    debug!("Transaction {0} received Metadata.", self.id());
722                                    // push each request up to the Daemon
723                                    let source_filename: Utf8PathBuf = metadata.source_filename;
724                                    let destination_filename: Utf8PathBuf =
725                                        metadata.destination_filename;
726
727                                    self.send_indication(Indication::MetadataRecv(
728                                        MetadataRecvIndication {
729                                            id: self.id(),
730                                            source_filename: source_filename.clone(),
731                                            destination_filename: destination_filename.clone(),
732                                            file_size: metadata.file_size,
733                                            transmission_mode: self.config.transmission_mode,
734                                        },
735                                    ));
736
737                                    self.metadata = Some(Metadata {
738                                        source_filename,
739                                        destination_filename,
740                                        file_size: metadata.file_size,
741                                        checksum_type: metadata.checksum_type,
742                                    });
743                                    self.check_finished()?;
744                                }
745                                Ok(())
746                            }
747                            Operations::Nak(_nak) => Err(TransactionError::UnexpectedPDU(
748                                self.config.sequence_number,
749                                self.config.transmission_mode,
750                                "NAK PDU".to_owned(),
751                            )),
752                        }
753                    }
754                }
755            }
756            TransmissionMode::Unacknowledged => {
757                match payload {
758                    PDUPayload::FileData(filedata) => {
759                        let (offset, length) = self.store_file_data(filedata)?;
760                        self.send_indication(Indication::FileSegmentRecv(FileSegmentIndication {
761                            id: self.id(),
762                            offset,
763                            length: length as u64,
764                        }));
765                        Ok(())
766                    }
767                    PDUPayload::Directive(operation) => {
768                        match operation {
769                            Operations::EoF(eof) => {
770                                self.condition = eof.condition;
771                                self.checksum = Some(eof.checksum);
772
773                                self.send_indication(Indication::EoFRecv(self.id()));
774
775                                if self.condition == Condition::NoError {
776                                    self.check_file_size(eof.file_size)?;
777                                    self.finalize_receive()?;
778                                } else {
779                                    // Any other condition is essentially a
780                                    // SHUTDOWN operation
781                                    self.shutdown()
782                                }
783                                Ok(())
784                            }
785                            Operations::Metadata(metadata) => {
786                                if self.metadata.is_none() {
787                                    let source_filename: Utf8PathBuf = metadata.source_filename;
788                                    let destination_filename: Utf8PathBuf =
789                                        metadata.destination_filename;
790
791                                    self.send_indication(Indication::MetadataRecv(
792                                        MetadataRecvIndication {
793                                            id: self.id(),
794                                            source_filename: source_filename.clone(),
795                                            destination_filename: destination_filename.clone(),
796                                            file_size: metadata.file_size,
797                                            transmission_mode: self.config.transmission_mode,
798                                        },
799                                    ));
800
801                                    self.metadata = Some(Metadata {
802                                        source_filename,
803                                        destination_filename,
804                                        file_size: metadata.file_size,
805                                        checksum_type: metadata.checksum_type,
806                                    });
807                                }
808                                Ok(())
809                            }
810                            Operations::Nak(_nak) => Err(TransactionError::UnexpectedPDU(
811                                self.config.sequence_number,
812                                self.config.transmission_mode,
813                                "NAK PDU".to_owned(),
814                            )),
815                        }
816                    }
817                }
818            }
819        }
820    }
821
822    // go to the Finished state if the file is complete (used only in acknowledged mode)
823    fn check_finished(&mut self) -> TransactionResult<()> {
824        if self.metadata.is_some()
825            && self.eof_received()
826            && !(self.is_file_transfer() && self.has_naks())
827        {
828            self.finalize_receive()?;
829            self.recv_state = RecvState::Finished;
830            self.prepare_empty_nak();
831            self.timer.nak.pause();
832        }
833        Ok(())
834    }
835
836    //send the indication in another task, no need to wait for it
837    fn send_indication(&self, indication: Indication) {
838        let tx = self.indication_tx.clone();
839        tokio::task::spawn(async move { tx.send(indication).await });
840    }
841}
842
843#[cfg(test)]
844mod test {
845    use std::{fs::OpenOptions, io::Read};
846
847    use crate::assert_err;
848
849    use crate::{
850        filestore::{ChecksumType, NativeFileStore},
851        pdu::{CRCFlag, EndOfFile, FileSizeFlag, MetadataPDU, SegmentedData, UnsegmentedFileData},
852    };
853
854    use super::*;
855    use crate::transaction::test::default_config;
856
857    use camino::{Utf8Path, Utf8PathBuf};
858    use rstest::{fixture, rstest};
859    use tempfile::TempDir;
860    use tokio::sync::mpsc::channel;
861
862    #[fixture]
863    #[once]
864    fn tempdir_fixture() -> TempDir {
865        TempDir::new().unwrap()
866    }
867
868    #[rstest]
869    fn header(default_config: &TransactionConfig, tempdir_fixture: &TempDir) {
870        let (indication_tx, _) = channel(1);
871        let config = default_config.clone();
872        let filestore = Arc::new(NativeFileStore::new(
873            Utf8Path::from_path(tempdir_fixture.path()).expect("Unable to make utf8 tempdir"),
874        ));
875
876        let mut transaction = RecvTransaction::new(
877            config,
878            NakProcedure::Deferred(Duration::ZERO),
879            filestore,
880            indication_tx,
881        );
882        let payload_len = 12;
883        let expected = PDUHeader {
884            version: U3::Zero,
885            pdu_type: PDUType::FileDirective,
886            direction: Direction::ToSender,
887            transmission_mode: TransmissionMode::Acknowledged,
888            crc_flag: CRCFlag::NotPresent,
889            large_file_flag: FileSizeFlag::Small,
890            pdu_data_field_length: payload_len,
891            segmentation_control: SegmentationControl::NotPreserved,
892            segment_metadata_flag: SegmentedData::NotPresent,
893            source_entity_id: default_config.source_entity_id,
894            destination_entity_id: default_config.destination_entity_id,
895            transaction_sequence_number: default_config.sequence_number,
896        };
897        assert_eq!(
898            expected,
899            transaction.get_header(
900                Direction::ToSender,
901                PDUType::FileDirective,
902                payload_len,
903                SegmentationControl::NotPreserved
904            )
905        )
906    }
907
908    #[rstest]
909    fn test_if_file_transfer(default_config: &TransactionConfig, tempdir_fixture: &TempDir) {
910        let (indication_tx, _) = channel(10);
911        let config = default_config.clone();
912        let filestore = Arc::new(NativeFileStore::new(
913            Utf8Path::from_path(tempdir_fixture.path()).expect("Unable to make utf8 tempdir"),
914        ));
915
916        let mut transaction = RecvTransaction::new(
917            config,
918            NakProcedure::Deferred(Duration::ZERO),
919            filestore,
920            indication_tx,
921        );
922        assert_eq!(
923            TransactionStatus::Undefined,
924            transaction.get_status().clone()
925        );
926        let mut path = Utf8PathBuf::new();
927        path.push("a");
928
929        transaction.metadata = Some(Metadata {
930            file_size: 600_u64,
931            source_filename: path.clone(),
932            destination_filename: path.clone(),
933            checksum_type: ChecksumType::Modular,
934        });
935
936        assert!(transaction.is_file_transfer())
937    }
938
939    #[rstest]
940    fn store_filedata(default_config: &TransactionConfig, tempdir_fixture: &TempDir) {
941        let (indication_tx, _) = channel(10);
942        let config = default_config.clone();
943        let filestore = Arc::new(NativeFileStore::new(
944            Utf8Path::from_path(tempdir_fixture.path()).expect("Unable to make utf8 tempdir"),
945        ));
946
947        let mut transaction = RecvTransaction::new(
948            config,
949            NakProcedure::Deferred(Duration::ZERO),
950            filestore,
951            indication_tx,
952        );
953
954        let input = vec![0, 5, 255, 99];
955        let data = FileDataPDU(UnsegmentedFileData {
956            offset: 6,
957            file_data: input.clone(),
958        });
959        let (offset, length) = transaction
960            .store_file_data(data)
961            .expect("Error saving file data");
962
963        assert_eq!(6, offset);
964        assert_eq!(4, length);
965
966        let handle = transaction.get_handle().unwrap();
967        handle.seek(SeekFrom::Start(6)).unwrap();
968        let mut buff = vec![0; 4];
969        handle.read_exact(&mut buff).unwrap();
970
971        assert_eq!(input, buff)
972    }
973
974    #[rstest]
975    fn finalize_file(default_config: &TransactionConfig, tempdir_fixture: &TempDir) {
976        let (indication_tx, _) = channel(10);
977        let config = default_config.clone();
978        let filestore = Arc::new(NativeFileStore::new(
979            Utf8Path::from_path(tempdir_fixture.path()).expect("Unable to make utf8 tempdir"),
980        ));
981
982        let input = "This is test data\n!";
983        let output_file = {
984            let mut temp_buff = Utf8PathBuf::new();
985            temp_buff.push("finalize_test.txt");
986            temp_buff
987        };
988
989        let mut transaction = RecvTransaction::new(
990            config,
991            NakProcedure::Deferred(Duration::ZERO),
992            filestore.clone(),
993            indication_tx,
994        );
995        transaction.metadata = Some(Metadata {
996            file_size: input.len() as u64,
997            source_filename: output_file.clone(),
998            destination_filename: output_file.clone(),
999            checksum_type: ChecksumType::Modular,
1000        });
1001
1002        let data = FileDataPDU(UnsegmentedFileData {
1003            offset: 0,
1004            file_data: input.as_bytes().to_vec(),
1005        });
1006        let (offset, length) = transaction
1007            .store_file_data(data)
1008            .expect("Error saving file data");
1009
1010        assert_eq!(0, offset);
1011        assert_eq!(input.len(), length);
1012
1013        let result = transaction
1014            .finalize_file()
1015            .expect("Error writing to finalize file.");
1016        assert_eq!(FileStatusCode::Retained, result);
1017
1018        let mut out_string = String::new();
1019        let contents = {
1020            filestore
1021                .open(output_file, OpenOptions::new().read(true))
1022                .expect("Cannot open finalized file.")
1023                .read_to_string(&mut out_string)
1024                .expect("Cannot read finalized file.")
1025        };
1026        assert_eq!(input.len(), contents);
1027        assert_eq!(input, out_string)
1028    }
1029
1030    #[rstest]
1031    #[tokio::test]
1032    async fn test_naks(
1033        default_config: &TransactionConfig,
1034        tempdir_fixture: &TempDir,
1035        #[values(FileSizeFlag::Small, FileSizeFlag::Large)] file_size_flag: FileSizeFlag,
1036    ) {
1037        let (transport_tx, mut transport_rx) = channel(10);
1038        let (indication_tx, _) = channel(10);
1039        let mut config = default_config.clone();
1040        config.file_size_flag = file_size_flag;
1041        let file_size = match &file_size_flag {
1042            FileSizeFlag::Small => 20,
1043            FileSizeFlag::Large => u32::MAX as u64 + 100_u64,
1044        };
1045
1046        let filestore = Arc::new(NativeFileStore::new(
1047            Utf8Path::from_path(tempdir_fixture.path()).expect("Unable to make utf8 tempdir"),
1048        ));
1049        let mut transaction = RecvTransaction::new(
1050            config,
1051            NakProcedure::Deferred(Duration::ZERO),
1052            filestore,
1053            indication_tx,
1054        );
1055
1056        let input = vec![0, 5, 255, 99];
1057        let data = FileDataPDU(UnsegmentedFileData {
1058            offset: 6,
1059            file_data: input,
1060        });
1061        let (offset, length) = transaction
1062            .store_file_data(data)
1063            .expect("Error saving file data");
1064
1065        assert_eq!(6, offset);
1066        assert_eq!(4, length);
1067
1068        transaction.file_size = Some(file_size);
1069        transaction.received_file_size = file_size;
1070
1071        let expected: VecDeque<SegmentRequestForm> = vec![
1072            SegmentRequestForm::from((0_u64, 0_u64)),
1073            SegmentRequestForm::from((0_u64, 6_u64)),
1074            SegmentRequestForm::from((10_u64, file_size)),
1075        ]
1076        .into();
1077
1078        assert_eq!(expected, transaction.get_all_naks());
1079
1080        let payload = PDUPayload::Directive(Operations::Nak(NegativeAcknowledgmentPDU {
1081            start_of_scope: 0,
1082            end_of_scope: file_size,
1083            segment_requests: expected.into(),
1084        }));
1085
1086        let payload_len = payload.encoded_len(transaction.config.file_size_flag);
1087        let header = transaction.get_header(
1088            Direction::ToSender,
1089            PDUType::FileDirective,
1090            payload_len,
1091            SegmentationControl::NotPreserved,
1092        );
1093
1094        let pdu = PDU { header, payload };
1095
1096        transaction.naks = transaction.get_all_naks();
1097        transaction
1098            .send_naks(transport_tx.reserve().await.unwrap())
1099            .unwrap();
1100
1101        let (destination_id, received_pdu) = transport_rx.recv().await.unwrap();
1102        let expected_id = default_config.source_entity_id;
1103
1104        assert_eq!(expected_id, destination_id);
1105        assert_eq!(pdu, received_pdu);
1106    }
1107
1108    #[rstest]
1109    fn pdu_error_unack(
1110        default_config: &TransactionConfig,
1111        tempdir_fixture: &TempDir,
1112        #[values(
1113            Operations::Nak(NegativeAcknowledgmentPDU{
1114                start_of_scope: 0_u64,
1115                end_of_scope: 1022_u64,
1116                segment_requests: vec![SegmentRequestForm::from((0_u32, 0_u32)), SegmentRequestForm::from((0_u32, 1022_u32))],
1117            })
1118        )]
1119        operation: Operations,
1120    ) {
1121        let (indication_tx, _) = channel(1);
1122        let mut config = default_config.clone();
1123        config.transmission_mode = TransmissionMode::Unacknowledged;
1124
1125        let filestore = Arc::new(NativeFileStore::new(
1126            Utf8Path::from_path(tempdir_fixture.path()).expect("Unable to make utf8 tempdir"),
1127        ));
1128        let mut transaction = RecvTransaction::new(
1129            config,
1130            NakProcedure::Deferred(Duration::ZERO),
1131            filestore,
1132            indication_tx,
1133        );
1134
1135        let path = {
1136            let mut path = Utf8PathBuf::new();
1137            path.push("Test_file.txt");
1138            path
1139        };
1140
1141        transaction.metadata = Some(Metadata {
1142            file_size: 600,
1143            source_filename: path.clone(),
1144            destination_filename: path,
1145            checksum_type: ChecksumType::Modular,
1146        });
1147
1148        let payload = PDUPayload::Directive(operation);
1149        let payload_len = payload.encoded_len(transaction.config.file_size_flag);
1150
1151        let header = transaction.get_header(
1152            Direction::ToReceiver,
1153            PDUType::FileDirective,
1154            payload_len,
1155            SegmentationControl::NotPreserved,
1156        );
1157        let pdu = PDU { header, payload };
1158
1159        let result = transaction.process_pdu(pdu);
1160
1161        assert_err!(
1162            result,
1163            Err(TransactionError::UnexpectedPDU(
1164                _,
1165                TransmissionMode::Unacknowledged,
1166                _
1167            ))
1168        )
1169    }
1170
1171    #[rstest]
1172    fn pdu_error_ack(
1173        default_config: &TransactionConfig,
1174        tempdir_fixture: &TempDir,
1175        #[values(
1176            Operations::Nak(NegativeAcknowledgmentPDU{
1177                start_of_scope: 0_u64,
1178                end_of_scope: 1022_u64,
1179                segment_requests: vec![SegmentRequestForm::from((0_u32, 0_u32)), SegmentRequestForm::from((0_u32, 1022_u32))],
1180            })
1181        )]
1182        operation: Operations,
1183    ) {
1184        let (indication_tx, _) = channel(1);
1185        let mut config = default_config.clone();
1186        config.transmission_mode = TransmissionMode::Acknowledged;
1187
1188        let filestore = Arc::new(NativeFileStore::new(
1189            Utf8Path::from_path(tempdir_fixture.path()).expect("Unable to make utf8 tempdir"),
1190        ));
1191        let mut transaction = RecvTransaction::new(
1192            config,
1193            NakProcedure::Deferred(Duration::ZERO),
1194            filestore,
1195            indication_tx,
1196        );
1197
1198        let path = {
1199            let mut path = Utf8PathBuf::new();
1200            path.push("Test_file.txt");
1201            path
1202        };
1203
1204        transaction.metadata = Some(Metadata {
1205            file_size: 600,
1206            source_filename: path.clone(),
1207            destination_filename: path,
1208            checksum_type: ChecksumType::Modular,
1209        });
1210
1211        let payload = PDUPayload::Directive(operation);
1212        let payload_len = payload.encoded_len(transaction.config.file_size_flag);
1213
1214        let header = transaction.get_header(
1215            Direction::ToReceiver,
1216            PDUType::FileDirective,
1217            payload_len,
1218            SegmentationControl::NotPreserved,
1219        );
1220        let pdu = PDU { header, payload };
1221
1222        let result = transaction.process_pdu(pdu);
1223
1224        assert_err!(
1225            result,
1226            Err(TransactionError::UnexpectedPDU(
1227                _,
1228                TransmissionMode::Acknowledged,
1229                _
1230            ))
1231        )
1232    }
1233
1234    #[rstest]
1235    #[tokio::test]
1236    async fn recv_store_data(
1237        default_config: &TransactionConfig,
1238        tempdir_fixture: &TempDir,
1239        #[values(TransmissionMode::Unacknowledged, TransmissionMode::Acknowledged)]
1240        transmission_mode: TransmissionMode,
1241    ) {
1242        let (indication_tx, _indication_rx) = channel(1);
1243        let mut config = default_config.clone();
1244        config.transmission_mode = transmission_mode;
1245
1246        let filestore = Arc::new(NativeFileStore::new(
1247            Utf8Path::from_path(tempdir_fixture.path()).expect("Unable to make utf8 tempdir"),
1248        ));
1249        let mut transaction = RecvTransaction::new(
1250            config,
1251            NakProcedure::Deferred(Duration::ZERO),
1252            filestore,
1253            indication_tx,
1254        );
1255
1256        let path = {
1257            let mut path = Utf8PathBuf::new();
1258            path.push("Test_file.txt");
1259            path
1260        };
1261
1262        transaction.metadata = Some(Metadata {
1263            file_size: 600,
1264            source_filename: path.clone(),
1265            destination_filename: path,
1266            checksum_type: ChecksumType::Modular,
1267        });
1268
1269        let payload = PDUPayload::FileData(FileDataPDU(UnsegmentedFileData {
1270            offset: 12_u64,
1271            file_data: (0..12_u8).collect::<Vec<u8>>(),
1272        }));
1273        let payload_len = payload.encoded_len(transaction.config.file_size_flag);
1274
1275        let header = transaction.get_header(
1276            Direction::ToSender,
1277            PDUType::FileData,
1278            payload_len,
1279            SegmentationControl::NotPreserved,
1280        );
1281        let pdu = PDU { header, payload };
1282
1283        transaction.process_pdu(pdu).unwrap();
1284        let contents: Vec<u8> = {
1285            let mut buf = Vec::new();
1286            let handle = transaction.get_handle().unwrap();
1287            handle.rewind().unwrap();
1288            handle.read_to_end(&mut buf).unwrap();
1289            buf
1290        };
1291        let expected = {
1292            let mut buf = vec![0_u8; 12];
1293            buf.extend(0..12_u8);
1294            buf
1295        };
1296        assert_eq!(expected, contents);
1297    }
1298
1299    #[rstest]
1300    #[tokio::test]
1301    async fn recv_store_metadata(
1302        default_config: &TransactionConfig,
1303        tempdir_fixture: &TempDir,
1304        #[values(TransmissionMode::Unacknowledged, TransmissionMode::Acknowledged)]
1305        transmission_mode: TransmissionMode,
1306    ) {
1307        let (indication_tx, _) = channel(1);
1308        let mut config = default_config.clone();
1309        config.transmission_mode = transmission_mode;
1310
1311        let filestore = Arc::new(NativeFileStore::new(
1312            Utf8Path::from_path(tempdir_fixture.path()).expect("Unable to make utf8 tempdir"),
1313        ));
1314        let mut transaction = RecvTransaction::new(
1315            config,
1316            NakProcedure::Deferred(Duration::ZERO),
1317            filestore,
1318            indication_tx,
1319        );
1320
1321        let path = {
1322            let mut path = Utf8PathBuf::new();
1323            path.push("Test_file.txt");
1324            path
1325        };
1326
1327        let expected = Some(Metadata {
1328            file_size: 600,
1329            source_filename: path.clone(),
1330            destination_filename: path,
1331            checksum_type: ChecksumType::Modular,
1332        });
1333
1334        let payload = PDUPayload::Directive(Operations::Metadata(MetadataPDU {
1335            checksum_type: ChecksumType::Modular,
1336            file_size: 600,
1337            source_filename: "Test_file.txt".into(),
1338            destination_filename: "Test_file.txt".into(),
1339        }));
1340        let payload_len = payload.encoded_len(transaction.config.file_size_flag);
1341
1342        let header = transaction.get_header(
1343            Direction::ToReceiver,
1344            PDUType::FileDirective,
1345            payload_len,
1346            SegmentationControl::NotPreserved,
1347        );
1348        let pdu = PDU { header, payload };
1349
1350        transaction.process_pdu(pdu).unwrap();
1351        assert_eq!(expected, transaction.metadata);
1352    }
1353
1354    #[rstest]
1355    #[tokio::test]
1356    async fn recv_eof_all_data(
1357        default_config: &TransactionConfig,
1358        tempdir_fixture: &TempDir,
1359        #[values(TransmissionMode::Unacknowledged, TransmissionMode::Acknowledged)]
1360        transmission_mode: TransmissionMode,
1361    ) {
1362        let (transport_tx, mut transport_rx) = channel(10);
1363        let (indication_tx, _indication_rx) = channel(1);
1364        let mut config = default_config.clone();
1365        config.transmission_mode = transmission_mode;
1366
1367        let expected_id = config.source_entity_id;
1368
1369        let filestore = Arc::new(NativeFileStore::new(
1370            Utf8Path::from_path(tempdir_fixture.path()).expect("Unable to make utf8 tempdir"),
1371        ));
1372        let mut transaction = RecvTransaction::new(
1373            config,
1374            NakProcedure::Deferred(Duration::ZERO),
1375            filestore,
1376            indication_tx,
1377        );
1378
1379        let path = {
1380            let mut path = Utf8PathBuf::new();
1381            path.push("Test_file.txt");
1382            path
1383        };
1384
1385        transaction.metadata = Some(Metadata {
1386            file_size: 600,
1387            source_filename: path.clone(),
1388            destination_filename: path,
1389            checksum_type: ChecksumType::Modular,
1390        });
1391
1392        let input_data = "Some_test words!\nHello\nWorld!";
1393
1394        let (checksum, _overflow) =
1395            input_data
1396                .as_bytes()
1397                .chunks(4)
1398                .fold((0_u32, false), |accum: (u32, bool), chunk| {
1399                    let mut vec = vec![0_u8; 4];
1400                    vec[..chunk.len()].copy_from_slice(chunk);
1401                    accum
1402                        .0
1403                        .overflowing_add(u32::from_be_bytes(vec.try_into().unwrap()))
1404                });
1405
1406        let file_pdu = {
1407            let payload = PDUPayload::FileData(FileDataPDU(UnsegmentedFileData {
1408                offset: 0,
1409                file_data: input_data.as_bytes().to_vec(),
1410            }));
1411            let payload_len = payload.encoded_len(transaction.config.file_size_flag);
1412
1413            let header = transaction.get_header(
1414                Direction::ToReceiver,
1415                PDUType::FileData,
1416                payload_len,
1417                SegmentationControl::NotPreserved,
1418            );
1419            PDU { header, payload }
1420        };
1421
1422        let input_pdu = {
1423            let payload = PDUPayload::Directive(Operations::EoF(EndOfFile {
1424                condition: Condition::NoError,
1425                checksum,
1426                file_size: input_data.len() as u64,
1427            }));
1428            let payload_len = payload.encoded_len(transaction.config.file_size_flag);
1429
1430            let header = transaction.get_header(
1431                Direction::ToReceiver,
1432                PDUType::FileDirective,
1433                payload_len,
1434                SegmentationControl::NotPreserved,
1435            );
1436            PDU { header, payload }
1437        };
1438
1439        let expected_pdu = {
1440            let payload = PDUPayload::Directive(Operations::Nak(NegativeAcknowledgmentPDU {
1441                start_of_scope: 0,
1442                end_of_scope: input_data.len() as u64,
1443                segment_requests: vec![],
1444            }));
1445
1446            let payload_len = payload.encoded_len(transaction.config.file_size_flag);
1447
1448            let header = transaction.get_header(
1449                Direction::ToSender,
1450                PDUType::FileDirective,
1451                payload_len,
1452                SegmentationControl::NotPreserved,
1453            );
1454
1455            PDU { header, payload }
1456        };
1457
1458        {
1459            // this is the whole contents of the file
1460            transaction.process_pdu(file_pdu).unwrap();
1461
1462            transaction.process_pdu(input_pdu).unwrap();
1463            transaction
1464                .send_pdu(transport_tx.reserve().await.unwrap())
1465                .unwrap();
1466
1467            if transaction.config.transmission_mode == TransmissionMode::Acknowledged {
1468                transaction
1469                    .send_pdu(transport_tx.reserve().await.unwrap())
1470                    .unwrap();
1471
1472                let eof = {
1473                    let payload = PDUPayload::Directive(Operations::EoF(EndOfFile {
1474                        condition: Condition::NoError,
1475                        checksum,
1476                        file_size: input_data.len() as u64,
1477                    }));
1478
1479                    let payload_len = payload.encoded_len(transaction.config.file_size_flag);
1480
1481                    let header = transaction.get_header(
1482                        Direction::ToReceiver,
1483                        PDUType::FileDirective,
1484                        payload_len,
1485                        SegmentationControl::NotPreserved,
1486                    );
1487
1488                    PDU { header, payload }
1489                };
1490                transaction.process_pdu(eof).unwrap();
1491
1492                assert!(!transaction.timer.nak.is_ticking());
1493            }
1494        }
1495
1496        // need to get the EoF from Acknowledged mode too.
1497        if transmission_mode == TransmissionMode::Acknowledged {
1498            let expected_pdu = {
1499                let payload = PDUPayload::Directive(Operations::Nak(NegativeAcknowledgmentPDU {
1500                    start_of_scope: 0,
1501                    end_of_scope: input_data.len() as u64,
1502                    segment_requests: vec![],
1503                }));
1504
1505                let payload_len = payload.encoded_len(expected_pdu.header.large_file_flag);
1506
1507                let header = {
1508                    let mut head = expected_pdu.header.clone();
1509                    head.pdu_data_field_length = payload_len;
1510                    head
1511                };
1512
1513                PDU { header, payload }
1514            };
1515            let (destination_id, received_pdu) = transport_rx.recv().await.unwrap();
1516            assert_eq!(expected_id, destination_id);
1517            assert_eq!(expected_pdu, received_pdu)
1518        }
1519
1520        // Unacknowledged mode does not wait for an empty NAK
1521        if transmission_mode == TransmissionMode::Acknowledged {
1522            let (destination_id, received_pdu) = transport_rx.recv().await.unwrap();
1523            assert_eq!(expected_id, destination_id);
1524            assert_eq!(expected_pdu, received_pdu)
1525        }
1526    }
1527
1528    #[rstest]
1529    #[tokio::test]
1530    async fn nak_split(default_config: &TransactionConfig, tempdir_fixture: &TempDir) {
1531        let (transport_tx, mut transport_rx) = channel(10);
1532        let (indication_tx, _indication_rx) = channel(10);
1533        let mut config = default_config.clone();
1534        config.file_size_segment = 16;
1535        config.transmission_mode = TransmissionMode::Acknowledged;
1536
1537        let expected_id = config.source_entity_id;
1538
1539        let filestore = Arc::new(NativeFileStore::new(
1540            Utf8Path::from_path(tempdir_fixture.path()).expect("Unable to make utf8 tempdir"),
1541        ));
1542        let mut transaction = RecvTransaction::new(
1543            config,
1544            NakProcedure::Immediate(Duration::ZERO),
1545            filestore,
1546            indication_tx,
1547        );
1548
1549        let file_pdu1 = {
1550            let payload = PDUPayload::FileData(FileDataPDU(UnsegmentedFileData {
1551                offset: 16,
1552                file_data: vec![0; 16],
1553            }));
1554            let payload_len = payload.encoded_len(transaction.config.file_size_flag);
1555
1556            let header = transaction.get_header(
1557                Direction::ToReceiver,
1558                PDUType::FileData,
1559                payload_len,
1560                SegmentationControl::NotPreserved,
1561            );
1562            PDU { header, payload }
1563        };
1564
1565        let file_pdu2 = {
1566            let payload = PDUPayload::FileData(FileDataPDU(UnsegmentedFileData {
1567                offset: 48,
1568                file_data: vec![0; 16],
1569            }));
1570            let payload_len = payload.encoded_len(transaction.config.file_size_flag);
1571
1572            let header = transaction.get_header(
1573                Direction::ToReceiver,
1574                PDUType::FileData,
1575                payload_len,
1576                SegmentationControl::NotPreserved,
1577            );
1578            PDU { header, payload }
1579        };
1580
1581        let expected_nak1 = {
1582            let payload = PDUPayload::Directive(Operations::Nak(NegativeAcknowledgmentPDU {
1583                start_of_scope: 0,
1584                end_of_scope: 16,
1585                segment_requests: vec![SegmentRequestForm {
1586                    start_offset: 0,
1587                    end_offset: 16,
1588                }],
1589            }));
1590
1591            let payload_len = payload.encoded_len(transaction.config.file_size_flag);
1592
1593            let header = transaction.get_header(
1594                Direction::ToSender,
1595                PDUType::FileDirective,
1596                payload_len,
1597                SegmentationControl::NotPreserved,
1598            );
1599            PDU { header, payload }
1600        };
1601
1602        let expected_nak2 = {
1603            let payload = PDUPayload::Directive(Operations::Nak(NegativeAcknowledgmentPDU {
1604                start_of_scope: 32,
1605                end_of_scope: 48,
1606                segment_requests: vec![SegmentRequestForm {
1607                    start_offset: 32,
1608                    end_offset: 48,
1609                }],
1610            }));
1611
1612            let payload_len = payload.encoded_len(transaction.config.file_size_flag);
1613
1614            let header = transaction.get_header(
1615                Direction::ToSender,
1616                PDUType::FileDirective,
1617                payload_len,
1618                SegmentationControl::NotPreserved,
1619            );
1620            PDU { header, payload }
1621        };
1622
1623        transaction.process_pdu(file_pdu1).unwrap();
1624        transaction.process_pdu(file_pdu2).unwrap();
1625
1626        assert!(transaction.has_pdu_to_send());
1627        transaction
1628            .send_pdu(transport_tx.reserve().await.unwrap())
1629            .unwrap();
1630        assert!(transaction.has_pdu_to_send());
1631        transaction
1632            .send_pdu(transport_tx.reserve().await.unwrap())
1633            .unwrap();
1634
1635        let (destination_id, received_pdu) = transport_rx.recv().await.unwrap();
1636        assert_eq!(expected_id, destination_id);
1637        assert_eq!(expected_nak1, received_pdu);
1638        let (destination_id, received_pdu) = transport_rx.recv().await.unwrap();
1639        assert_eq!(expected_id, destination_id);
1640        assert_eq!(expected_nak2, received_pdu)
1641    }
1642
1643    #[rstest]
1644    #[tokio::test]
1645    async fn delayed_nak(default_config: &TransactionConfig, tempdir_fixture: &TempDir) {
1646        let (transport_tx, mut transport_rx) = channel(10);
1647        let (indication_tx, _indication_rx) = channel(10);
1648        let mut config = default_config.clone();
1649        config.file_size_segment = 16;
1650        config.transmission_mode = TransmissionMode::Acknowledged;
1651
1652        let expected_id = config.source_entity_id;
1653
1654        let filestore = Arc::new(NativeFileStore::new(
1655            Utf8Path::from_path(tempdir_fixture.path()).expect("Unable to make utf8 tempdir"),
1656        ));
1657        let mut transaction = RecvTransaction::new(
1658            config,
1659            NakProcedure::Immediate(Duration::from_millis(100)),
1660            filestore,
1661            indication_tx,
1662        );
1663
1664        let file_pdu1 = {
1665            let payload = PDUPayload::FileData(FileDataPDU(UnsegmentedFileData {
1666                offset: 16,
1667                file_data: vec![0; 16],
1668            }));
1669            let payload_len = payload.encoded_len(transaction.config.file_size_flag);
1670
1671            let header = transaction.get_header(
1672                Direction::ToReceiver,
1673                PDUType::FileData,
1674                payload_len,
1675                SegmentationControl::NotPreserved,
1676            );
1677            PDU { header, payload }
1678        };
1679
1680        let expected_nak1 = {
1681            let payload = PDUPayload::Directive(Operations::Nak(NegativeAcknowledgmentPDU {
1682                start_of_scope: 0,
1683                end_of_scope: 16,
1684                segment_requests: vec![SegmentRequestForm {
1685                    start_offset: 0,
1686                    end_offset: 16,
1687                }],
1688            }));
1689
1690            let payload_len = payload.encoded_len(transaction.config.file_size_flag);
1691
1692            let header = transaction.get_header(
1693                Direction::ToSender,
1694                PDUType::FileDirective,
1695                payload_len,
1696                SegmentationControl::NotPreserved,
1697            );
1698            PDU { header, payload }
1699        };
1700
1701        transaction.process_pdu(file_pdu1).unwrap();
1702        assert!(!transaction.has_pdu_to_send());
1703
1704        let d = transaction.until_timeout().as_millis();
1705        assert!(d <= 100);
1706
1707        tokio::time::sleep(Duration::from_millis(150)).await;
1708
1709        transaction.handle_timeout().unwrap();
1710
1711        assert!(transaction.has_pdu_to_send());
1712
1713        transaction
1714            .send_pdu(transport_tx.reserve().await.unwrap())
1715            .unwrap();
1716        assert!(!transaction.has_pdu_to_send());
1717
1718        let (destination_id, received_pdu) = transport_rx.recv().await.unwrap();
1719        assert_eq!(expected_id, destination_id);
1720        assert_eq!(expected_nak1, received_pdu);
1721    }
1722}
cfdp_simplified/transaction/recv.rs

cfdp_simplified/transaction/
recv.rs
