| 688 | } |
| 689 | |
| 690 | async fn enqueue(&mut self) -> Result<Option<(FuotaJob, DateTime<Utc>)>> { |
| 691 | // Proceed with next step after reaching the max attempts. |
| 692 | if self.job.attempt_count > self.job.max_retry_count { |
| 693 | return Ok(Some((FuotaJob::FragStatus, Utc::now()))); |
| 694 | } |
| 695 | |
| 696 | info!("Enqueueing fragmented payload to multicast group"); |
| 697 | self.job.attempt_count += 1; |
| 698 | |
| 699 | let fuota_devices = fuota::get_devices(self.job.fuota_deployment_id.into(), -1, 0).await?; |
| 700 | |
| 701 | // Filter on devices that have completed the previous step. |
| 702 | let fuota_devices: Vec<fuota::FuotaDeploymentDevice> = fuota_devices |
| 703 | .into_iter() |
| 704 | .filter(|d| d.mc_session_completed_at.is_some()) |
| 705 | .collect(); |
| 706 | |
| 707 | if fuota_devices.is_empty() { |
| 708 | self.job.error_msg = "There are no devices available to complete this step".into(); |
| 709 | return Ok(Some((FuotaJob::DeleteMcGroup, Utc::now()))); |
| 710 | } |
| 711 | |
| 712 | let payload_length = self.fuota_deployment.payload.len(); |
| 713 | let fragment_size = self.fuota_deployment.fragmentation_fragment_size as usize; |
| 714 | let padding = (fragment_size - (payload_length % fragment_size)) % fragment_size; |
| 715 | |
| 716 | let fragments = (payload_length as f32 / fragment_size as f32).ceil() as usize; |
| 717 | let redundancy = (fragments as f32 |
| 718 | * self.fuota_deployment.fragmentation_redundancy_percentage as f32 |
| 719 | / 100.0) |
| 720 | .ceil() as usize; |
| 721 | |
| 722 | let mut payload = self.fuota_deployment.payload.clone(); |
| 723 | payload.extend_from_slice(&vec![0; padding]); |
| 724 | |
| 725 | let payloads = match self.device_profile.app_layer_params.ts004_version { |
| 726 | Some(Ts004Version::V100) => { |
| 727 | let mut payloads = Vec::new(); |
| 728 | let encoded_fragments = |
| 729 | fragmentation::v1::encode(&payload, fragment_size, redundancy)?; |
| 730 | for (i, frag) in encoded_fragments.iter().enumerate() { |
| 731 | payloads.push( |
| 732 | fragmentation::v1::Payload::DataFragment( |
| 733 | fragmentation::v1::DataFragmentPayload { |
| 734 | index_and_n: fragmentation::v1::DataFragmentPayloadIndexAndN { |
| 735 | frag_index: 0, |
| 736 | n: (i + 1) as u16, |
| 737 | }, |
| 738 | data: frag.clone(), |
| 739 | }, |
| 740 | ) |
| 741 | .to_vec()?, |
| 742 | ); |
| 743 | } |
| 744 | payloads |
| 745 | } |
| 746 | Some(Ts004Version::V200) => { |
| 747 | let mut payloads = Vec::new(); |