rustot 0.4.1

AWS IoT
Documentation 
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101

use core::fmt::Write;

use mqttrust::QoS;

use super::ControlInterface;
use crate::jobs::data_types::JobStatus;
use crate::jobs::subscribe::Topic;
use crate::jobs::Jobs;
use crate::ota::config::Config;
use crate::ota::encoding::json::JobStatusReason;
use crate::ota::encoding::FileContext;
use crate::ota::error::OtaError;

impl<T: mqttrust::Mqtt> ControlInterface for T {
    /// Initialize the control interface by subscribing to the OTA job
    /// notification topics.
    fn init(&self) -> Result<(), OtaError> {
        Jobs::subscribe::<1>()
            .topic(Topic::NotifyNext, QoS::AtLeastOnce)
            .send(self)?;
        Ok(())
    }

    /// Check for next available OTA job from the job service by publishing a
    /// "get next job" message to the job service.
    fn request_job(&self) -> Result<(), OtaError> {
        Jobs::describe().send(self, QoS::AtLeastOnce)?;

        Ok(())
    }

    /// Update the job status on the service side with progress or completion
    /// info
    fn update_job_status(
        &self,
        file_ctx: &mut FileContext,
        config: &Config,
        status: JobStatus,
        reason: JobStatusReason,
    ) -> Result<(), OtaError> {
        file_ctx
            .status_details
            .insert(
                heapless::String::from("self_test"),
                heapless::String::from(reason.as_str()),
            )
            .map_err(|_| OtaError::Overflow)?;

        let mut qos = QoS::AtLeastOnce;

        if let (JobStatus::InProgress, _) | (JobStatus::Succeeded, _) = (status, reason) {
            let total_blocks =
                ((file_ctx.filesize + config.block_size - 1) / config.block_size) as u32;
            let received_blocks = total_blocks - file_ctx.blocks_remaining as u32;

            // Output a status update once in a while. Always update first and
            // last status
            if file_ctx.blocks_remaining != 0
                && received_blocks != 0
                && received_blocks % config.status_update_frequency != 0
            {
                return Ok(());
            }

            // Don't override the progress on succeeded, nor on self-test
            // active. (Cases where progess counter is lost due to device
            // restarts)
            if status != JobStatus::Succeeded && reason != JobStatusReason::SelfTestActive {
                let mut progress = heapless::String::new();
                progress
                    .write_fmt(format_args!("{}/{}", received_blocks, total_blocks))
                    .map_err(|_| OtaError::Overflow)?;

                file_ctx
                    .status_details
                    .insert(heapless::String::from("progress"), progress)
                    .map_err(|_| OtaError::Overflow)?;
            }

            // Downgrade progress updates to QOS 0 to avoid overloading MQTT
            // buffers during active streaming
            if status == JobStatus::InProgress {
                qos = QoS::AtMostOnce;
            }
        }

        Jobs::update(file_ctx.job_name.as_str(), status)
            .status_details(&file_ctx.status_details)
            .send(self, qos)?;

        Ok(())
    }

    /// Perform any cleanup operations required for control plane
    fn cleanup(&self) -> Result<(), OtaError> {
        Jobs::unsubscribe::<1>()
            .topic(Topic::NotifyNext)
            .send(self)?;
        Ok(())
    }
}