spotflow 0.8.1

Device SDK for Spotflow IoT Platform
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
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148

use spotflow::{DeviceClientBuilder, MessageContext};
use std::{
    fs::{self, File},
    path::Path,
};

use log::{debug, info, warn};
use uuid::Uuid;

mod common;

fn main() {
    env_logger::Builder::from_env(
        env_logger::Env::default()
            .default_filter_or("sqlx=warn,ureq=warn,rumqtt=warn,spotflow=debug,info"),
    )
    .init();

    let env_ctx = common::EnvironmentContext::try_load()
        .expect("Unable to load settings from environment variables.");

    let platform_caller = common::PlatformCaller::try_new(&env_ctx)
        .map_err(|e| {
            warn!("Unable to call the Platform automatically, continuing manually. Error: {e}")
        })
        .ok();

    let device_id = Uuid::new_v4().to_string();
    let stream_group = "device-sdk";
    let stream = "rust";
    let site = "test-site";

    let batch_count = 10u32;
    let message_count = 100u32;

    let path = Path::new("./test.db");

    info!("Using device ID {}", device_id);

    info!("Initiating connection tests");

    if path.exists() {
        info!("Removing the old local database file");
        fs::remove_file(path).expect("Unable to delete the old local database file");
    }
    info!("Creating a new local database file");
    File::create(path).expect("Unable to create a new local database file");

    {
        info!("Creating ingress client");

        let client = DeviceClientBuilder::new(
            Some(device_id.clone()),
            env_ctx.provisioning_token.clone(),
            path,
        )
        .with_instance(env_ctx.instance_url.to_string())
        .with_site_id(site.to_owned())
        .with_display_provisioning_operation_callback(Box::new(
            common::ProvisioningOperationApprovalHandler::new(platform_caller.clone()),
        ))
        .build()
        .expect("Unable to build ingress connection");

        let message_context =
            MessageContext::new(Some(stream_group.to_owned()), Some(stream.to_owned()));

        for batch_id in 0..batch_count {
            let batch_id = format!("{batch_id:0>2}");

            for message_id in 0..message_count {
                let message_id = format!("{message_id:0>2}");

                debug!("Publishing message {message_id}");
                client
                    .enqueue_message(
                        &message_context,
                        Some(batch_id.clone()),
                        Some(message_id),
                        vec![b'a'; 1000],
                    )
                    .expect("Unable to send message");
            }

            info!("Completing batch {batch_id}");
            client
                .enqueue_batch_completion(&message_context, batch_id)
                .expect("Unable to complete batch");
        }

        info!("Dropping original ingress");
    }

    {
        info!("Starting new ingress client with configuration");

        let client = DeviceClientBuilder::new(
            Some(device_id.clone()),
            env_ctx.provisioning_token.clone(),
            path,
        )
        .with_instance(env_ctx.instance_url.to_string())
        .with_display_provisioning_operation_callback(Box::new(
            common::ProvisioningOperationApprovalHandler::new(platform_caller.clone()),
        ))
        .build()
        .expect("Unable to build ingress connection");

        loop {
            let pending = client
                .pending_messages_count()
                .expect("Unable to obtain number of pending messages");
            if pending < 200 {
                break;
            }
            warn!("Waiting for {} more messages to be sent.", pending);
            std::thread::sleep(std::time::Duration::from_millis(500));
        }

        info!("Dropping ingress");
    }

    {
        info!("Starting new ingress client");

        let client =
            DeviceClientBuilder::new(Some(device_id.clone()), env_ctx.provisioning_token, path)
                .with_instance(env_ctx.instance_url.to_string())
                .with_display_provisioning_operation_callback(Box::new(
                    common::ProvisioningOperationApprovalHandler::new(platform_caller),
                ))
                .build()
                .expect("Unable to build ingress connection");

        loop {
            let pending = client
                .pending_messages_count()
                .expect("Unable to obtain number of pending messages");
            if pending == 0 {
                break;
            }
            warn!("Waiting for {} more messages to be sent.", pending);
            std::thread::sleep(std::time::Duration::from_millis(500));
        }

        info!("Dropping ingress");
    }
}