cfdp_simplified/transaction/
send.rs

1use std::{
2    collections::{HashSet, VecDeque},
3    fs::{File, OpenOptions},
4    io::{Read, Seek, SeekFrom},
5    sync::Arc,
6    time::Duration,
7};
8
9use chrono::{DateTime, Utc};
10use log::{debug, info};
11use tokio::sync::{
12    mpsc::{Permit, Sender},
13    oneshot,
14};
15
16use super::fault_handler::FaultHandlerAction;
17use crate::{
18    daemon::{FinishedIndication, Indication, Report},
19    filestore::{FileChecksum, FileStore, FileStoreError},
20    pdu::{
21        Condition, DeliveryCode, Direction, EndOfFile, EntityID, FileDataPDU, FileStatusCode,
22        MetadataPDU, Operations, PDUHeader, PDUPayload, PDUType, SegmentRequestForm,
23        SegmentationControl, SegmentedData, TransactionStatus, TransmissionMode,
24        UnsegmentedFileData, PDU, U3,
25    },
26    transaction::{Metadata, TransactionConfig, TransactionID, TransactionState},
27};
28
29use crate::{
30    daemon::Timer,
31    transaction::{TransactionError, TransactionResult},
32};
33
34#[derive(PartialEq, Debug)]
35#[allow(clippy::enum_variant_names)]
36enum SendState {
37    // initial state
38    // send metadata, then go to SendData (if file transfer) or directly to SendEof
39    SendMetadata,
40    // send data until finished, then go to SendEof
41    SendData,
42    // send EOF and wait for nak
43    SendEof,
44}
45
46pub struct SendTransaction<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    /// The current file being worked by this Transaction.
54    file_handle: Option<File>,
55    /// The list of all missing information
56    naks: VecDeque<SegmentRequestForm>,
57    ///  The metadata of this Transaction
58    pub(crate) metadata: Metadata,
59    /// a cache of the header used for interactions in this transmission
60    header: Option<PDUHeader>,
61    /// The current condition of the transaction
62    condition: Condition,
63    // Track whether the transaction is complete or not
64    delivery_code: DeliveryCode,
65    // Status of the current File
66    file_status: FileStatusCode,
67    /// Timer used to track if the Nak limit has been reached
68    /// inactivity has occurred
69    /// or the ACK limit is reached
70    timer: Timer,
71    //. checksum cache to reduce I/0
72    //. doubles as stored checksum in received mode
73    checksum: Option<u32>,
74    /// The current general state of the transaction.
75    /// Used to determine when the thread should be killed
76    state: TransactionState,
77    /// The detailed sub-state valid when state = TransactionState::Active
78    /// Used to control the state machine
79    send_state: SendState,
80    /// EoF prepared to be sent at the next opportunity if the bool is true
81    eof: Option<(EndOfFile, bool)>,
82    /// Channel for Indications to propagate back up
83    indication_tx: Sender<Indication>,
84    /// flag to track if the initial EoFSent Indication has been sent.
85    /// This indication only needs to be delivered for the initial EoF transmission
86    send_eof_indication: bool,
87    // Empty Nak Received indication
88    empty_nak_received: bool,
89    /// File bytes sent
90    file_bytes_sent: u64,
91    /// Date transaction was submitted
92    submit_date: DateTime<Utc>,
93}
94impl<T: FileStore> SendTransaction<T> {
95    /// Start a new SendTransaction with the given [configuration](TransactionConfig)
96    /// and [Filestore Implementation](FileStore)
97    pub fn new(
98        // Configuration of this Transaction.
99        config: TransactionConfig,
100        // Metadata of this Transaction
101        metadata: Metadata,
102        // Connection to the local FileStore implementation.
103        filestore: Arc<T>,
104        // Sender channel used to propagate Message To User back up to the Daemon Thread.
105        indication_tx: Sender<Indication>,
106    ) -> TransactionResult<Self> {
107        let file_bytes_sent = 0_u64;
108        let timer = Timer::new(
109            config.inactivity_timeout,
110            config.max_count,
111            config.eof_timeout,
112            config.max_count,
113            config.nak_timeout,
114            config.max_count,
115            config.inactivity_timeout,
116        );
117
118        let me = Self {
119            status: TransactionStatus::Undefined,
120            config,
121            filestore,
122            file_handle: None,
123            naks: VecDeque::new(),
124            metadata,
125            header: None,
126            condition: Condition::NoError,
127            delivery_code: DeliveryCode::Incomplete,
128            file_status: FileStatusCode::Unreported,
129            timer,
130            checksum: None,
131            send_state: SendState::SendMetadata,
132            state: TransactionState::Active,
133            eof: None,
134            indication_tx,
135            send_eof_indication: true,
136            empty_nak_received: false,
137            file_bytes_sent,
138            submit_date: Utc::now(),
139        };
140        me.send_indication(Indication::Transaction(me.id()));
141        Ok(me)
142    }
143
144    pub(crate) fn has_pdu_to_send(&self) -> bool {
145        match self.send_state {
146            SendState::SendMetadata | SendState::SendData => true,
147            SendState::SendEof => !self.naks.is_empty() || self.eof.as_ref().is_some_and(|x| x.1),
148        }
149    }
150
151    // returns the time until the first timeout
152    pub(crate) fn until_timeout(&self) -> Duration {
153        match self.send_state {
154            SendState::SendEof => self.timer.until_timeout(),
155            _ => Duration::MAX,
156        }
157    }
158
159    pub(crate) fn send_pdu(
160        &mut self,
161        permit: Permit<'_, (EntityID, PDU)>,
162    ) -> TransactionResult<()> {
163        match self.send_state {
164            SendState::SendMetadata => {
165                self.send_metadata(permit)?;
166                if self.is_file_transfer() {
167                    self.send_state = SendState::SendData;
168                } else {
169                    self.prepare_eof()?;
170                    self.send_state = SendState::SendEof;
171                }
172            }
173            SendState::SendData => {
174                if !self.naks.is_empty() {
175                    // if we have received a NAK send the missing data
176                    self.send_missing_data(permit)?;
177                } else {
178                    self.send_file_segment(None, None, permit)?
179                }
180
181                let handle = self.get_handle()?;
182                if handle.stream_position().map_err(FileStoreError::IO)?
183                    == handle.metadata().map_err(FileStoreError::IO)?.len()
184                {
185                    self.prepare_eof()?;
186                    self.send_state = SendState::SendEof;
187                }
188            }
189            SendState::SendEof => {
190                if !self.naks.is_empty() {
191                    // if we have received a NAK send the missing data
192                    self.send_missing_data(permit)?;
193                } else {
194                    self.send_eof(permit)?;
195
196                    // EoFSent indication only needs to be sent for the initial EoF transmission
197                    if self.send_eof_indication {
198                        self.send_indication(Indication::EoFSent(self.id()));
199                        self.send_eof_indication = false;
200                    }
201
202                    if self.get_mode() == TransmissionMode::Unacknowledged {
203                        // closure not supported, this transaction is finished.
204                        // indicate as much to the User
205                        self.send_indication(Indication::Finished(FinishedIndication {
206                            id: self.id(),
207                            report: self.generate_report(),
208                            file_status: self.file_status,
209                            delivery_code: self.delivery_code,
210                        }));
211                        {
212                            self.shutdown()
213                        }
214                    }
215                }
216            }
217        }
218        Ok(())
219    }
220
221    #[allow(clippy::single_match)]
222    pub fn handle_timeout(&mut self) -> TransactionResult<()> {
223        debug!("Handling timeout...");
224        match self.send_state {
225            SendState::SendEof => {
226                if self.timer.inactivity.limit_reached() {
227                    self.handle_fault(Condition::InactivityDetected)?
228                }
229                // EoF timer tracks if the sender has received a NAK from the receiver
230                // The sender will attempt to send EOFs, up to PositiveLimitReached count
231                if self.timer.eof.timeout_occurred() {
232                    if self.timer.eof.limit_reached() {
233                        self.handle_fault(Condition::PositiveLimitReached)?
234                    } else {
235                        self.set_eof_flag(true);
236                    }
237                }
238            }
239            _ => {}
240        };
241
242        Ok(())
243    }
244
245    pub fn get_status(&self) -> TransactionStatus {
246        self.status
247    }
248
249    pub(crate) fn get_state(&self) -> TransactionState {
250        self.state
251    }
252
253    pub fn get_mode(&self) -> TransmissionMode {
254        self.config.transmission_mode
255    }
256    fn generate_report(&self) -> Report {
257        Report {
258            id: self.id(),
259            state: self.get_state(),
260            status: self.get_status(),
261            condition: self.condition,
262            empty_nak_received: self.empty_nak_received,
263            file_bytes_received: None,
264            file_bytes_sent: Some(self.file_bytes_sent),
265            file_size: self.metadata.file_size,
266            direction: self.header.as_ref().map(|h| h.direction),
267            file_name: self
268                .metadata
269                .source_filename
270                .file_name()
271                .map(|file| file.to_string())
272                .unwrap_or_default(),
273            submit_date: self.submit_date,
274        }
275    }
276
277    pub fn send_report(&self, sender: Option<oneshot::Sender<Report>>) -> TransactionResult<()> {
278        let report = self.generate_report();
279
280        if let Some(channel) = sender {
281            let _ = channel.send(report.clone());
282        }
283        self.send_indication(Indication::Report(report));
284
285        Ok(())
286    }
287
288    fn get_header(
289        &mut self,
290        direction: Direction,
291        pdu_type: PDUType,
292        pdu_data_field_length: u16,
293        segmentation_control: SegmentationControl,
294    ) -> PDUHeader {
295        if let Some(header) = &self.header {
296            // update the necessary fields but copy the rest
297            // from the cache
298            PDUHeader {
299                pdu_type,
300                pdu_data_field_length,
301                segmentation_control,
302                ..header.clone()
303            }
304        } else {
305            let header = PDUHeader {
306                version: U3::Zero,
307                pdu_type,
308                direction,
309                transmission_mode: self.config.transmission_mode,
310                crc_flag: self.config.crc_flag,
311                large_file_flag: self.config.file_size_flag,
312                pdu_data_field_length,
313                segmentation_control,
314                segment_metadata_flag: self.config.segment_metadata_flag,
315                source_entity_id: self.config.source_entity_id,
316                transaction_sequence_number: self.config.sequence_number,
317                destination_entity_id: self.config.destination_entity_id,
318            };
319            self.header = Some(header.clone());
320            header
321        }
322    }
323
324    pub fn id(&self) -> TransactionID {
325        TransactionID(self.config.source_entity_id, self.config.sequence_number)
326    }
327
328    fn get_checksum(&mut self) -> TransactionResult<u32> {
329        match self.checksum {
330            Some(val) => Ok(val),
331            None => {
332                let checksum = {
333                    match self.is_file_transfer() {
334                        true => {
335                            let checksum_type = self.metadata.checksum_type;
336                            self.get_handle()?.checksum(checksum_type)?
337                        }
338                        false => 0,
339                    }
340                };
341                self.checksum = Some(checksum);
342                Ok(checksum)
343            }
344        }
345    }
346
347    fn open_source_file(&mut self) -> TransactionResult<()> {
348        self.file_handle = {
349            let fname = &self.metadata.source_filename;
350            Some(self.filestore.open(fname, OpenOptions::new().read(true))?)
351        };
352        Ok(())
353    }
354
355    fn get_handle(&mut self) -> TransactionResult<&mut File> {
356        let id = self.id();
357        if self.file_handle.is_none() {
358            self.open_source_file()?
359        };
360        self.file_handle
361            .as_mut()
362            .ok_or(TransactionError::NoFile(id))
363    }
364
365    pub(crate) fn is_file_transfer(&self) -> bool {
366        !self.metadata.source_filename.as_os_str().is_empty()
367    }
368
369    /// Read a segment of size `length` from position `offset` in the file.
370    /// When offset is [None], reads from the current cursor location.
371    /// when length is [None] uses the maximum length for the receiving Engine.
372    /// This size is set by the `file_size_segment` field in the [TransactionConfig].
373    fn get_file_segment(
374        &mut self,
375        offset: Option<u64>,
376        length: Option<u16>,
377    ) -> TransactionResult<(u64, Vec<u8>)> {
378        // use the maximum size for the receiever if no length is given
379        let length = length.unwrap_or(self.config.file_size_segment);
380        let handle = self.get_handle()?;
381        // if no offset given read from current cursor position
382        let offset = offset.unwrap_or(handle.stream_position().map_err(FileStoreError::IO)?);
383
384        // If the offset is not provided start at the current position
385
386        handle
387            .seek(SeekFrom::Start(offset))
388            .map_err(FileStoreError::IO)?;
389
390        // use take to limit the final segment from trying to read past the EoF.
391        let data = {
392            let mut buff = Vec::<u8>::with_capacity(length as usize);
393            handle
394                .take(length as u64)
395                .read_to_end(&mut buff)
396                .map_err(FileStoreError::IO)?;
397            buff
398        };
399
400        Ok((offset, data))
401    }
402
403    /// Send a segment of size `length` from position `offset` in the file.
404    /// When offset is [None], reads from the current cursor location.
405    /// when length is [None] uses the maximum length for the receiving Engine.
406    /// This size is set by the `file_size_segment` field in the [TransactionConfig].
407    pub fn send_file_segment(
408        &mut self,
409        offset: Option<u64>,
410        length: Option<u16>,
411        permit: Permit<(EntityID, PDU)>,
412    ) -> TransactionResult<()> {
413        let (offset, file_data) = self.get_file_segment(offset, length)?;
414
415        let (data, segmentation_control) = match self.config.segment_metadata_flag {
416            SegmentedData::NotPresent => (
417                FileDataPDU(UnsegmentedFileData { offset, file_data }),
418                SegmentationControl::NotPreserved,
419            ),
420            SegmentedData::Present => unimplemented!(),
421        };
422        let destination = self.config.destination_entity_id;
423
424        let payload = PDUPayload::FileData(data);
425
426        let payload_len: u16 = payload.encoded_len(self.config.file_size_flag);
427
428        self.file_bytes_sent += payload_len as u64;
429
430        let header = self.get_header(
431            Direction::ToReceiver,
432            PDUType::FileData,
433            payload_len,
434            segmentation_control,
435        );
436        let pdu = PDU { header, payload };
437
438        permit.send((destination, pdu));
439
440        Ok(())
441    }
442
443    pub fn send_missing_data(&mut self, permit: Permit<(EntityID, PDU)>) -> TransactionResult<()> {
444        match self.naks.pop_front() {
445            Some(request) => {
446                // only restart inactivity if we have something to do.
447                self.timer.restart_inactivity();
448                let (offset, length) = (
449                    request.start_offset,
450                    (request.end_offset - request.start_offset).try_into()?,
451                );
452
453                match offset == 0 && length == 0 {
454                    true => self.send_metadata(permit),
455                    false => {
456                        let current_pos = {
457                            let handle = self.get_handle()?;
458                            handle.stream_position().map_err(FileStoreError::IO)?
459                        };
460
461                        self.send_file_segment(Some(offset), Some(length), permit)?;
462
463                        // restore to original location in the file
464                        let handle = self.get_handle()?;
465                        handle
466                            .seek(SeekFrom::Start(current_pos))
467                            .map_err(FileStoreError::IO)?;
468                        Ok(())
469                    }
470                }
471            }
472            None => Ok(()),
473        }
474    }
475
476    fn prepare_eof(&mut self) -> TransactionResult<()> {
477        self.eof = Some((
478            EndOfFile {
479                condition: self.condition,
480                checksum: self.get_checksum()?,
481                file_size: self.metadata.file_size,
482            },
483            true,
484        ));
485
486        Ok(())
487    }
488
489    pub fn send_eof(&mut self, permit: Permit<(EntityID, PDU)>) -> TransactionResult<()> {
490        if let Some((eof, true)) = &self.eof {
491            self.timer.restart_eof();
492
493            let payload = PDUPayload::Directive(Operations::EoF(eof.clone()));
494            let payload_len = payload.encoded_len(self.config.file_size_flag);
495            let header = self.get_header(
496                Direction::ToReceiver,
497                PDUType::FileDirective,
498                payload_len,
499                SegmentationControl::NotPreserved,
500            );
501
502            let destination = header.destination_entity_id;
503            let pdu = PDU { header, payload };
504
505            permit.send((destination, pdu));
506            debug!("Transaction {0} sent EndOfFile.", self.id());
507            self.set_eof_flag(false);
508        }
509        Ok(())
510    }
511
512    // set the true flag on the eof such that it is sent at the next opportunity
513    fn set_eof_flag(&mut self, flag: bool) {
514        if let Some(x) = self.eof.as_mut() {
515            x.1 = flag;
516        }
517    }
518
519    pub fn shutdown(&mut self) {
520        debug!("Transaction {0} shutting down.", self.id());
521        self.status = TransactionStatus::Terminated;
522        self.state = TransactionState::Terminated;
523        self.timer.inactivity.pause();
524        self.timer.eof.pause();
525    }
526
527    /// Take action according to the defined handler mapping.
528    /// Returns a boolean indicating if the calling function should continue (true) or not (false.)
529    fn handle_fault(&mut self, condition: Condition) -> TransactionResult<()> {
530        self.condition = condition;
531        let action = self
532            .config
533            .fault_handler_override
534            .get(&self.condition)
535            .unwrap_or(&FaultHandlerAction::Abandon);
536        info!(
537            "Transaction {} handling fault {:?}, action: {:?} ",
538            self.id(),
539            condition,
540            action
541        );
542
543        match action {
544            FaultHandlerAction::Ignore => {}
545            FaultHandlerAction::Abandon => {
546                self.shutdown();
547            }
548        }
549        Ok(())
550    }
551
552    #[allow(clippy::single_match)]
553    pub fn process_pdu(&mut self, pdu: PDU) -> TransactionResult<()> {
554        if self.send_state == SendState::SendEof {
555            self.timer.restart_inactivity();
556            self.timer.reset_eof();
557        }
558        let PDU {
559            header: _header,
560            payload,
561        } = pdu;
562        match &self.config.transmission_mode {
563            TransmissionMode::Acknowledged => {
564                match payload {
565                    // Receiver payload type - Not expected at sender
566                    PDUPayload::FileData(_data) => Err(TransactionError::UnexpectedPDU(
567                        self.config.sequence_number,
568                        self.config.transmission_mode,
569                        "File Data".to_owned(),
570                    )),
571                    PDUPayload::Directive(operation) => match operation {
572                        // Receiver Opp - Not expected at sender
573                        Operations::EoF(_eof) => Err(TransactionError::UnexpectedPDU(
574                            self.config.sequence_number,
575                            self.config.transmission_mode,
576                            "End of File".to_owned(),
577                        )),
578                        // Receiver Opp - Not expected at sender
579                        Operations::Metadata(_metadata) => Err(TransactionError::UnexpectedPDU(
580                            self.config.sequence_number,
581                            self.config.transmission_mode,
582                            "Metadata PDU".to_owned(),
583                        )),
584                        Operations::Nak(nak) => {
585                            info!(
586                                "Received NAK with {} segment requests",
587                                nak.segment_requests.len()
588                            );
589                            // check and filter for naks where the length of the request is greater than
590                            // the maximum allowed file size
591                            // as the Receiver we don't care if the length is too big
592                            // But the Sender needs segments to fit in the expected size.
593
594                            // If received nak list is empty, close the transaction
595                            if nak.segment_requests.is_empty() {
596                                debug!(
597                                    "Transaction {} received empty NAK; moving transaction to Finished state",
598                                    self.id(),
599                                );
600
601                                self.empty_nak_received = true;
602
603                                self.send_indication(Indication::Finished(FinishedIndication {
604                                    id: self.id(),
605                                    report: self.generate_report(),
606                                    file_status: self.file_status,
607                                    delivery_code: self.delivery_code,
608                                }));
609
610                                {
611                                    self.shutdown()
612                                }
613                                return Ok(());
614                            }
615
616                            let formatted_segments =
617                                nak.segment_requests
618                                    .into_iter()
619                                    .flat_map(|form| match form {
620                                        val @ SegmentRequestForm {
621                                            start_offset: start,
622                                            end_offset: end,
623                                        } if start == end => vec![val],
624                                        SegmentRequestForm {
625                                            start_offset: start,
626                                            end_offset: end,
627                                        } => (start..end)
628                                            .step_by(self.config.file_size_segment.into())
629                                            .map(|num| {
630                                                if num
631                                                    < end.saturating_sub(
632                                                        self.config.file_size_segment.into(),
633                                                    )
634                                                {
635                                                    SegmentRequestForm {
636                                                        start_offset: num,
637                                                        end_offset: num
638                                                            + self.config.file_size_segment as u64,
639                                                    }
640                                                } else {
641                                                    SegmentRequestForm {
642                                                        start_offset: num,
643                                                        end_offset: end,
644                                                    }
645                                                }
646                                            })
647                                            .collect::<Vec<SegmentRequestForm>>(),
648                                    });
649                            self.naks.extend(formatted_segments);
650                            // filter out any duplicated NAKS
651                            let mut uniques = HashSet::new();
652                            self.naks.retain(|element| uniques.insert(element.clone()));
653                            Ok(())
654                        }
655                    },
656                }
657            }
658            TransmissionMode::Unacknowledged => match payload {
659                PDUPayload::FileData(_data) => Err(TransactionError::UnexpectedPDU(
660                    self.config.sequence_number,
661                    self.config.transmission_mode,
662                    "File Data".to_owned(),
663                )),
664                PDUPayload::Directive(operation) => match operation {
665                    Operations::EoF(_eof) => Err(TransactionError::UnexpectedPDU(
666                        self.config.sequence_number,
667                        self.config.transmission_mode,
668                        "End of File".to_owned(),
669                    )),
670                    Operations::Metadata(_metadata) => Err(TransactionError::UnexpectedPDU(
671                        self.config.sequence_number,
672                        self.config.transmission_mode,
673                        "Metadata PDU".to_owned(),
674                    )),
675                    Operations::Nak(_) => Err(TransactionError::UnexpectedPDU(
676                        self.config.sequence_number,
677                        self.config.transmission_mode,
678                        "NAK PDU".to_owned(),
679                    )),
680                },
681            },
682        }
683    }
684
685    fn send_metadata(&mut self, permit: Permit<(EntityID, PDU)>) -> TransactionResult<()> {
686        let destination = self.config.destination_entity_id;
687        let metadata = MetadataPDU {
688            checksum_type: self.metadata.checksum_type,
689            file_size: self.metadata.file_size,
690            source_filename: self.metadata.source_filename.clone(),
691            destination_filename: self.metadata.destination_filename.clone(),
692        };
693
694        let payload = PDUPayload::Directive(Operations::Metadata(metadata));
695        let payload_len: u16 = payload.encoded_len(self.config.file_size_flag);
696
697        let header = self.get_header(
698            Direction::ToReceiver,
699            PDUType::FileDirective,
700            payload_len,
701            SegmentationControl::NotPreserved,
702        );
703
704        let pdu = PDU { header, payload };
705
706        permit.send((destination, pdu));
707        debug!("Transaction {0} sent Metadata.", self.id());
708        Ok(())
709    }
710
711    //send the indication in another task, no need to wait for it
712    fn send_indication(&self, indication: Indication) {
713        let tx = self.indication_tx.clone();
714        tokio::task::spawn(async move { tx.send(indication).await });
715    }
716}
717
718#[cfg(test)]
719mod test {
720    use crate::assert_err;
721    use crate::{
722        filestore::{ChecksumType, NativeFileStore},
723        pdu::{
724            CRCFlag, FileSizeFlag, NegativeAcknowledgmentPDU, SegmentedData, UnsegmentedFileData,
725        },
726    };
727
728    use std::io::Write;
729    use std::thread::sleep;
730
731    use super::*;
732    use crate::transaction::test::default_config;
733
734    use camino::{Utf8Path, Utf8PathBuf};
735    use rstest::{fixture, rstest};
736    use tempfile::TempDir;
737    use tokio::sync::mpsc::channel;
738    #[fixture]
739    #[once]
740    fn tempdir_fixture() -> TempDir {
741        TempDir::new().unwrap()
742    }
743
744    #[rstest]
745    #[tokio::test]
746    async fn header(default_config: &TransactionConfig, tempdir_fixture: &TempDir) {
747        let config = default_config.clone();
748        let filestore = Arc::new(NativeFileStore::new(
749            Utf8Path::from_path(tempdir_fixture.path()).expect("Unable to make utf8 tempdir"),
750        ));
751
752        let (indication_tx, _indication_rx) = channel(10);
753        let metadata = test_metadata(10, Utf8PathBuf::from(""));
754        let mut transaction = SendTransaction::new(config, metadata, filestore, indication_tx)
755            .expect("unable to start transaction.");
756
757        let payload_len = 12;
758        let expected = PDUHeader {
759            version: U3::Zero,
760            pdu_type: PDUType::FileDirective,
761            direction: Direction::ToSender,
762            transmission_mode: TransmissionMode::Acknowledged,
763            crc_flag: CRCFlag::NotPresent,
764            large_file_flag: FileSizeFlag::Small,
765            pdu_data_field_length: payload_len,
766            segmentation_control: SegmentationControl::NotPreserved,
767            segment_metadata_flag: SegmentedData::NotPresent,
768            source_entity_id: default_config.source_entity_id,
769            destination_entity_id: default_config.destination_entity_id,
770            transaction_sequence_number: default_config.sequence_number,
771        };
772        assert_eq!(
773            expected,
774            transaction.get_header(
775                Direction::ToSender,
776                PDUType::FileDirective,
777                payload_len,
778                SegmentationControl::NotPreserved
779            )
780        )
781    }
782
783    #[rstest]
784    #[tokio::test]
785    async fn test_if_file_transfer(default_config: &TransactionConfig, tempdir_fixture: &TempDir) {
786        let config = default_config.clone();
787        let filestore = Arc::new(NativeFileStore::new(
788            Utf8Path::from_path(tempdir_fixture.path()).expect("Unable to make utf8 tempdir"),
789        ));
790
791        let (indication_tx, _indication_rx) = channel(10);
792        let metadata = test_metadata(600_u64, Utf8PathBuf::from("a"));
793        let transaction = SendTransaction::new(config, metadata, filestore, indication_tx).unwrap();
794
795        assert_eq!(
796            TransactionStatus::Undefined,
797            transaction.get_status().clone()
798        );
799
800        assert!(transaction.is_file_transfer())
801    }
802
803    #[rstest]
804    #[tokio::test]
805    async fn send_filedata(default_config: &TransactionConfig, tempdir_fixture: &TempDir) {
806        let (transport_tx, mut transport_rx) = channel(1);
807        let config = default_config.clone();
808        let filestore = Arc::new(NativeFileStore::new(
809            Utf8Path::from_path(tempdir_fixture.path()).expect("Unable to make utf8 tempdir"),
810        ));
811
812        let path = Utf8PathBuf::from("testfile");
813
814        let (indication_tx, _indication_rx) = channel(10);
815        let metadata = test_metadata(10, path.clone());
816        let mut transaction =
817            SendTransaction::new(config, metadata, filestore.clone(), indication_tx).unwrap();
818
819        let input = vec![0, 5, 255, 99];
820
821        let payload = PDUPayload::FileData(FileDataPDU(UnsegmentedFileData {
822            offset: 6,
823            file_data: input.clone(),
824        }));
825        let payload_len = payload.encoded_len(transaction.config.file_size_flag);
826
827        let header = transaction.get_header(
828            Direction::ToReceiver,
829            PDUType::FileData,
830            payload_len,
831            SegmentationControl::NotPreserved,
832        );
833        let pdu = PDU { header, payload };
834
835        tokio::task::spawn(async move {
836            let fname = transaction.metadata.source_filename.clone();
837
838            {
839                let mut handle = transaction
840                    .filestore
841                    .open(fname, OpenOptions::new().create_new(true).write(true))
842                    .unwrap();
843                handle
844                    .seek(SeekFrom::Start(6))
845                    .expect("Cannot seek file cursor in thread.");
846                handle
847                    .write_all(input.as_slice())
848                    .expect("Cannot write test file in thread.");
849                handle.sync_all().expect("Bad file sync.");
850            }
851
852            let offset = 6;
853            let length = 4;
854
855            transaction
856                .send_file_segment(
857                    Some(offset),
858                    Some(length as u16),
859                    transport_tx.reserve().await.unwrap(),
860                )
861                .unwrap();
862        });
863        let (destination_id, received_pdu) = transport_rx.recv().await.unwrap();
864        let expected_id = default_config.destination_entity_id;
865        assert_eq!(expected_id, destination_id);
866        assert_eq!(pdu, received_pdu);
867        filestore.delete_file(path).expect("cannot remove file");
868    }
869
870    #[rstest]
871    #[case(SegmentRequestForm { start_offset: 6, end_offset: 10 })]
872    #[case(SegmentRequestForm { start_offset: 0, end_offset: 0 })]
873    #[tokio::test]
874    async fn send_missing(
875        #[case] nak: SegmentRequestForm,
876        default_config: &TransactionConfig,
877        tempdir_fixture: &TempDir,
878    ) {
879        let (transport_tx, mut transport_rx) = channel(1);
880        let config = default_config.clone();
881        let filestore = Arc::new(NativeFileStore::new(
882            Utf8Path::from_path(tempdir_fixture.path()).expect("Unable to make utf8 tempdir"),
883        ));
884
885        let path = Utf8PathBuf::from("testfile_missing");
886        let metadata = test_metadata(10, path.clone());
887
888        let (indication_tx, _indication_rx) = channel(10);
889        let mut transaction =
890            SendTransaction::new(config, metadata, filestore, indication_tx).unwrap();
891
892        let input = vec![0, 5, 255, 99];
893        let pdu = match &nak {
894            SegmentRequestForm {
895                start_offset: 6,
896                end_offset: 10,
897            } => {
898                let payload = PDUPayload::FileData(FileDataPDU(UnsegmentedFileData {
899                    offset: 6,
900                    file_data: input.clone(),
901                }));
902                let payload_len = payload.encoded_len(transaction.config.file_size_flag);
903
904                let header = transaction.get_header(
905                    Direction::ToReceiver,
906                    PDUType::FileData,
907                    payload_len,
908                    SegmentationControl::NotPreserved,
909                );
910                PDU { header, payload }
911            }
912            _ => {
913                let payload = PDUPayload::Directive(Operations::Metadata(MetadataPDU {
914                    file_size: 10,
915                    checksum_type: ChecksumType::Modular,
916                    source_filename: path.clone(),
917                    destination_filename: path.clone(),
918                }));
919                let payload_len = payload.encoded_len(transaction.config.file_size_flag);
920
921                let header = transaction.get_header(
922                    Direction::ToReceiver,
923                    PDUType::FileDirective,
924                    payload_len,
925                    SegmentationControl::NotPreserved,
926                );
927                PDU { header, payload }
928            }
929        };
930
931        tokio::task::spawn(async move {
932            let fname = transaction.metadata.source_filename.clone();
933
934            {
935                let mut handle = transaction
936                    .filestore
937                    .open(
938                        fname,
939                        OpenOptions::new().create(true).truncate(true).write(true),
940                    )
941                    .unwrap();
942                handle
943                    .seek(SeekFrom::Start(6))
944                    .expect("Cannot seek file cursor in thread.");
945                handle
946                    .write_all(input.as_slice())
947                    .expect("Cannot write test file in thread.");
948                handle.sync_all().expect("Bad file sync.");
949            }
950
951            transaction.naks.push_back(nak);
952            transaction
953                .send_missing_data(transport_tx.reserve().await.unwrap())
954                .unwrap();
955
956            transaction
957                .filestore
958                .delete_file(path.clone())
959                .expect("cannot remove file");
960        });
961        sleep(Duration::from_millis(500));
962        let (destination_id, received_pdu) = transport_rx.recv().await.unwrap();
963        let expected_id = default_config.destination_entity_id;
964        assert_eq!(expected_id, destination_id);
965        assert_eq!(pdu, received_pdu);
966    }
967
968    #[rstest]
969    #[tokio::test]
970    async fn checksum_cache(default_config: &TransactionConfig, tempdir_fixture: &TempDir) {
971        let config = default_config.clone();
972        let filestore = Arc::new(NativeFileStore::new(
973            Utf8Path::from_path(tempdir_fixture.path()).expect("Unable to make utf8 tempdir"),
974        ));
975
976        let (indication_tx, _indication_rx) = channel(1);
977        let metadata = test_metadata(10, Utf8PathBuf::from(""));
978        let mut transaction =
979            SendTransaction::new(config, metadata, filestore, indication_tx).unwrap();
980
981        let result = transaction
982            .get_checksum()
983            .expect("Cannot get file checksum");
984        assert_eq!(0, result);
985        assert!(transaction.checksum.is_some());
986        let result = transaction
987            .get_checksum()
988            .expect("Cannot get file checksum");
989        assert_eq!(0, result);
990    }
991
992    #[rstest]
993    #[tokio::test]
994    async fn send_eof(default_config: &TransactionConfig, tempdir_fixture: &TempDir) {
995        let (transport_tx, mut transport_rx) = channel(1);
996        let config = default_config.clone();
997        let filestore = Arc::new(NativeFileStore::new(
998            Utf8Path::from_path(tempdir_fixture.path()).expect("Unable to make utf8 tempdir"),
999        ));
1000
1001        let input = "Here is some test data to write!$*#*.\n";
1002
1003        let (indication_tx, _indication_rx) = channel(10);
1004        let path = Utf8PathBuf::from("test_eof.dat");
1005        let metadata = test_metadata(input.as_bytes().len() as u64, path.clone());
1006        let mut transaction =
1007            SendTransaction::new(config, metadata, filestore.clone(), indication_tx).unwrap();
1008
1009        let (checksum, _overflow) =
1010            input
1011                .as_bytes()
1012                .chunks(4)
1013                .fold((0_u32, false), |accum: (u32, bool), chunk| {
1014                    let mut vec = vec![0_u8; 4];
1015                    vec[..chunk.len()].copy_from_slice(chunk);
1016                    accum
1017                        .0
1018                        .overflowing_add(u32::from_be_bytes(vec.try_into().unwrap()))
1019                });
1020
1021        let payload = PDUPayload::Directive(Operations::EoF(EndOfFile {
1022            condition: Condition::NoError,
1023            checksum,
1024            file_size: input.as_bytes().len() as u64,
1025        }));
1026        let payload_len = payload.encoded_len(transaction.config.file_size_flag);
1027
1028        let header = transaction.get_header(
1029            Direction::ToReceiver,
1030            PDUType::FileDirective,
1031            payload_len,
1032            SegmentationControl::NotPreserved,
1033        );
1034        let pdu = PDU { header, payload };
1035
1036        tokio::task::spawn(async move {
1037            let fname = transaction.metadata.source_filename.clone();
1038
1039            {
1040                let mut handle = transaction
1041                    .filestore
1042                    .open(fname, OpenOptions::new().create_new(true).write(true))
1043                    .unwrap();
1044                handle
1045                    .write_all(input.as_bytes())
1046                    .expect("Cannot write test file in thread.");
1047                handle.sync_all().expect("Bad file sync.");
1048            }
1049            transaction.prepare_eof().unwrap();
1050            transaction
1051                .send_eof(transport_tx.reserve().await.unwrap())
1052                .unwrap()
1053        });
1054
1055        let (destination_id, received_pdu) = transport_rx.recv().await.unwrap();
1056        let expected_id = default_config.destination_entity_id;
1057
1058        assert_eq!(expected_id, destination_id);
1059        assert_eq!(pdu, received_pdu);
1060
1061        filestore.delete_file(path).expect("cannot remove file");
1062    }
1063
1064    #[rstest]
1065    #[tokio::test]
1066    async fn pdu_error_unack_send(
1067        default_config: &TransactionConfig,
1068        tempdir_fixture: &TempDir,
1069        #[values(
1070            Operations::EoF(EndOfFile{
1071                condition: Condition::CancelReceived,
1072                checksum: 12_u32,
1073                file_size: 1022_u64,
1074            }),
1075            Operations::Metadata(MetadataPDU{
1076                checksum_type: ChecksumType::Modular,
1077                file_size: 1022_u64,
1078                source_filename: "test_filename".into(),
1079                destination_filename: "test_filename".into(),
1080            }),
1081            Operations::Nak(NegativeAcknowledgmentPDU{
1082                start_of_scope: 0_u64,
1083                end_of_scope: 1022_u64,
1084                segment_requests: vec![SegmentRequestForm::from((0_u32, 0_u32)), SegmentRequestForm::from((0_u32, 1022_u32))],
1085            }),
1086        )]
1087        operation: Operations,
1088    ) {
1089        let mut config = default_config.clone();
1090        config.transmission_mode = TransmissionMode::Unacknowledged;
1091
1092        let filestore = Arc::new(NativeFileStore::new(
1093            Utf8Path::from_path(tempdir_fixture.path()).expect("Unable to make utf8 tempdir"),
1094        ));
1095
1096        let (indication_tx, _indication_rx) = channel(10);
1097        let metadata = test_metadata(600, Utf8PathBuf::from("Test_file.txt"));
1098        let mut transaction =
1099            SendTransaction::new(config, metadata, filestore, indication_tx).unwrap();
1100
1101        let payload = PDUPayload::Directive(operation);
1102        let payload_len = payload.encoded_len(transaction.config.file_size_flag);
1103        let header = transaction.get_header(
1104            Direction::ToSender,
1105            PDUType::FileDirective,
1106            payload_len,
1107            SegmentationControl::NotPreserved,
1108        );
1109        let pdu = PDU { header, payload };
1110
1111        let result = transaction.process_pdu(pdu);
1112
1113        assert_err!(
1114            result,
1115            Err(TransactionError::UnexpectedPDU(
1116                _,
1117                TransmissionMode::Unacknowledged,
1118                _
1119            ))
1120        )
1121    }
1122
1123    #[rstest]
1124    #[tokio::test]
1125    async fn pdu_error_ack_send(
1126        default_config: &TransactionConfig,
1127        tempdir_fixture: &TempDir,
1128        #[values(
1129            Operations::EoF(EndOfFile{
1130                condition: Condition::CancelReceived,
1131                checksum: 12_u32,
1132                file_size: 1022_u64,
1133            }),
1134            Operations::Metadata(MetadataPDU{
1135                checksum_type: ChecksumType::Modular,
1136                file_size: 1022_u64,
1137                source_filename: "test_filename".into(),
1138                destination_filename: "test_filename".into(),
1139            }),
1140        )]
1141        operation: Operations,
1142    ) {
1143        let mut config = default_config.clone();
1144        config.transmission_mode = TransmissionMode::Acknowledged;
1145
1146        let filestore = Arc::new(NativeFileStore::new(
1147            Utf8Path::from_path(tempdir_fixture.path()).expect("Unable to make utf8 tempdir"),
1148        ));
1149
1150        let (indication_tx, _indication_rx) = channel(10);
1151        let metadata = test_metadata(600, Utf8PathBuf::from("Test_file.txt"));
1152        let mut transaction =
1153            SendTransaction::new(config, metadata, filestore, indication_tx).unwrap();
1154
1155        let payload = PDUPayload::Directive(operation);
1156        let payload_len = payload.encoded_len(transaction.config.file_size_flag);
1157        let header = transaction.get_header(
1158            Direction::ToSender,
1159            PDUType::FileDirective,
1160            payload_len,
1161            SegmentationControl::NotPreserved,
1162        );
1163        let pdu = PDU { header, payload };
1164
1165        let result = transaction.process_pdu(pdu);
1166
1167        assert_err!(
1168            result,
1169            Err(TransactionError::UnexpectedPDU(
1170                _,
1171                TransmissionMode::Acknowledged,
1172                _
1173            ))
1174        )
1175    }
1176
1177    #[rstest]
1178    #[tokio::test]
1179    async fn pdu_error_send_data(
1180        default_config: &TransactionConfig,
1181        tempdir_fixture: &TempDir,
1182        #[values(TransmissionMode::Unacknowledged, TransmissionMode::Acknowledged)]
1183        transmission_mode: TransmissionMode,
1184    ) {
1185        let mut config = default_config.clone();
1186        config.transmission_mode = transmission_mode;
1187
1188        let filestore = Arc::new(NativeFileStore::new(
1189            Utf8Path::from_path(tempdir_fixture.path()).expect("Unable to make utf8 tempdir"),
1190        ));
1191
1192        let (indication_tx, _indication_rx) = channel(10);
1193        let metadata = test_metadata(600, Utf8PathBuf::from("Test_file.txt"));
1194        let mut transaction =
1195            SendTransaction::new(config, metadata, filestore, indication_tx).unwrap();
1196
1197        let payload = PDUPayload::FileData(FileDataPDU(UnsegmentedFileData {
1198            offset: 12_u64,
1199            file_data: (0..12_u8).collect::<Vec<u8>>(),
1200        }));
1201        let payload_len = payload.encoded_len(transaction.config.file_size_flag);
1202        let header = transaction.get_header(
1203            Direction::ToSender,
1204            PDUType::FileData,
1205            payload_len,
1206            SegmentationControl::NotPreserved,
1207        );
1208        let pdu = PDU { header, payload };
1209
1210        let result = transaction.process_pdu(pdu);
1211
1212        assert_err!(result, Err(TransactionError::UnexpectedPDU(_, _, _)))
1213    }
1214
1215    #[rstest]
1216    #[tokio::test]
1217    async fn recv_nak(
1218        default_config: &TransactionConfig,
1219        tempdir_fixture: &TempDir,
1220        #[values(2_u64, u32::MAX as u64 + 100_u64)] total_size: u64,
1221    ) {
1222        let file_size_flag = match total_size > u32::MAX.into() {
1223            true => FileSizeFlag::Large,
1224            false => FileSizeFlag::Small,
1225        };
1226
1227        let mut config = default_config.clone();
1228        config.transmission_mode = TransmissionMode::Acknowledged;
1229        config.file_size_flag = file_size_flag;
1230
1231        let filestore = Arc::new(NativeFileStore::new(
1232            Utf8Path::from_path(tempdir_fixture.path()).expect("Unable to make utf8 tempdir"),
1233        ));
1234
1235        let (indication_tx, _indication_rx) = channel(10);
1236        let metadata = test_metadata(total_size, Utf8PathBuf::from("test_file"));
1237        let mut transaction =
1238            SendTransaction::new(config, metadata, filestore, indication_tx).unwrap();
1239
1240        let nak_pdu = {
1241            let payload = PDUPayload::Directive(Operations::Nak(NegativeAcknowledgmentPDU {
1242                start_of_scope: 0,
1243                end_of_scope: total_size,
1244                segment_requests: vec![SegmentRequestForm {
1245                    start_offset: 0,
1246                    end_offset: total_size,
1247                }],
1248            }));
1249            let payload_len = payload.encoded_len(transaction.config.file_size_flag);
1250
1251            let header = transaction.get_header(
1252                Direction::ToSender,
1253                PDUType::FileDirective,
1254                payload_len,
1255                SegmentationControl::NotPreserved,
1256            );
1257
1258            PDU { header, payload }
1259        };
1260
1261        let expected_naks: VecDeque<SegmentRequestForm> = (0..total_size)
1262            .step_by(transaction.config.file_size_segment.into())
1263            .map(|num| {
1264                if num < total_size.saturating_sub(transaction.config.file_size_segment.into()) {
1265                    SegmentRequestForm {
1266                        start_offset: num,
1267                        end_offset: num + transaction.config.file_size_segment as u64,
1268                    }
1269                } else {
1270                    SegmentRequestForm {
1271                        start_offset: num,
1272                        end_offset: total_size,
1273                    }
1274                }
1275            })
1276            .collect();
1277
1278        transaction.process_pdu(nak_pdu).unwrap();
1279        assert_eq!(expected_naks, transaction.naks);
1280    }
1281
1282    fn test_metadata(file_size: u64, path: Utf8PathBuf) -> Metadata {
1283        Metadata {
1284            file_size,
1285            source_filename: path.clone(),
1286            destination_filename: path,
1287            checksum_type: ChecksumType::Modular,
1288        }
1289    }
1290}
cfdp_simplified/transaction/send.rs

cfdp_simplified/transaction/
send.rs
