589 lines
22 KiB
Rust
589 lines
22 KiB
Rust
use crate::command::Command;
|
|
use crate::serial::SerialIO;
|
|
use crate::types::*;
|
|
|
|
use anyhow;
|
|
use std::{
|
|
error::Error,
|
|
io::{Read, Write},
|
|
thread,
|
|
time::Duration,
|
|
sync::mpsc::Receiver,
|
|
};
|
|
|
|
use esp_idf_hal::prelude::*;
|
|
use esp_idf_hal::serial::{self, Rx, Tx};
|
|
|
|
use embedded_hal::digital::v2::OutputPin;
|
|
|
|
use mqtt::packet::{ConnectPacket, PublishPacket, QoSWithPacketIdentifier, VariablePacket};
|
|
use mqtt::{Encodable, Decodable, TopicName};
|
|
|
|
pub type Result<T> = std::result::Result<T, ModemError>;
|
|
|
|
pub struct Modem<UART: serial::Uart> {
|
|
serial: SerialIO<UART>,
|
|
}
|
|
|
|
#[derive(Debug)]
|
|
pub enum ModemError {
|
|
CommandError(String),
|
|
SetupError(String),
|
|
SendDataError(String),
|
|
ReadError(String),
|
|
TimeoutError,
|
|
}
|
|
|
|
impl Error for ModemError {}
|
|
|
|
impl std::fmt::Display for ModemError {
|
|
fn fmt(&self, f: &mut std::fmt::Formatter) -> std::fmt::Result {
|
|
write!(f, "{:?}", self)
|
|
}
|
|
}
|
|
|
|
impl<UART: serial::Uart> Modem<UART> {
|
|
pub fn new(tx: Tx<UART>, rx: Rx<UART>) -> Self {
|
|
Self {
|
|
serial: SerialIO::new(tx, rx),
|
|
}
|
|
}
|
|
|
|
/// Initialize the modem (sim800l in this case). The initialization process sets all pins in the
|
|
/// required state so that the modem is turned on, then resets it a couple of times (beats me) and
|
|
/// sleeps for 3 seconds, which is enough for the modem to come online.
|
|
///
|
|
/// Below is an example for sim800l pins on a LilyGo TTGO T-Call.
|
|
///
|
|
/// # Examples
|
|
///
|
|
/// ```
|
|
/// let modem_pwrkey = dp.pins.gpio4.into_output().unwrap();
|
|
/// let modem_rst = dp.pins.gpio5.into_output().unwrap();
|
|
/// let modem_power = dp.pins.gpio23.into_output().unwrap();
|
|
///
|
|
/// modem::init(modem_pwrkey, modem_rst, modem_power);
|
|
/// ```
|
|
pub fn init(&mut self, mut pwrkey: impl OutputPin, mut rst: impl OutputPin, mut power: impl OutputPin) -> Result<()> {
|
|
println!("Turning SIM800L on ...");
|
|
power.set_high().map_err(|_| ModemError::SetupError("Error setting POWER to high.".to_string()))?;
|
|
//rst.set_high().map_err(|_| ModemError::SetupError("Error setting RST to high.".to_string()))?;
|
|
// Pull down PWRKEY for more than 1 second according to manual requirements
|
|
pwrkey.set_high().map_err(|_| ModemError::SetupError("Error setting PWRKEY to high.".to_string()))?;
|
|
thread::sleep(Duration::from_millis(1500));
|
|
pwrkey.set_low().map_err(|_| ModemError::SetupError("Error setting PWRKEY to low.".to_string()))?;
|
|
thread::sleep(Duration::from_millis(1000));
|
|
pwrkey.set_high().map_err(|_| ModemError::SetupError("Error setting PWRKEY to high.".to_string()))?;
|
|
println!("Waiting for sim module to come online ...");
|
|
thread::sleep(Duration::from_millis(3000));
|
|
loop {
|
|
match self.send_command(Command::probe()) {
|
|
Ok(_) => break,
|
|
_ => {
|
|
thread::sleep(Duration::from_millis(2000));
|
|
continue
|
|
},
|
|
}
|
|
}
|
|
Ok(())
|
|
}
|
|
|
|
/// Reads the serial RX until the `contains` string is encoutered if `contains` is Some(s), if
|
|
/// None, then the first line is returned. If a timeout is reached. The timeout is provided on
|
|
/// input via the `timeout` argument. The first argument `contains` is checked against every
|
|
/// line in the response.
|
|
fn command_read_response(&mut self, contains: Option<String>) -> Result<String> {
|
|
let mut response = String::new();
|
|
|
|
loop {
|
|
let mut buf = vec![0; 1024];
|
|
let num_bytes = self.serial
|
|
.read(buf.as_mut_slice())
|
|
.map_err(|err| ModemError::ReadError(format!("Error in serial.read(buf) ({:?})", err)))?;
|
|
|
|
response.push_str(std::str::from_utf8(&buf[0..num_bytes])
|
|
.map_err(|err| ModemError::ReadError(format!("Error in str::from_utf8 ({:?})", err)))?);
|
|
|
|
if num_bytes < buf.len() {
|
|
break
|
|
}
|
|
}
|
|
|
|
print!("Read {} bytes from serial: {}", response.len(), response);
|
|
if let Some(c) = contains {
|
|
if response.contains(&c) {
|
|
Ok(response)
|
|
} else {
|
|
Err(ModemError::CommandError(format!("Didn't get expected ({}) from modem.", c)))
|
|
}
|
|
} else {
|
|
Ok(response)
|
|
}
|
|
}
|
|
|
|
fn send_command(&mut self, cmd: Command) -> Result<String> {
|
|
if let Some(contains) = cmd.contains {
|
|
self.send(&cmd.text, &contains)
|
|
} else {
|
|
self.send(&cmd.text, "")
|
|
}
|
|
}
|
|
|
|
fn send(&mut self, at_command: &str, contains: &str) -> Result<String> {
|
|
println!("-----------------------------------------------------------");
|
|
println!("Sending {} ...", at_command);
|
|
|
|
let _ = nb::block!(self.serial
|
|
.write_bytes(&[at_command.as_bytes(), &['\r' as u8]].concat()))
|
|
.map_err(|_| ModemError::SendDataError(format!("Error in send_command({})", at_command)))?;
|
|
|
|
let contains_opt = if contains == "" { None } else { Some(contains.to_string()) };
|
|
|
|
self.command_read_response(contains_opt)
|
|
}
|
|
|
|
fn handle_prompt(&mut self) -> Result<()> {
|
|
let mut prompt_buf = vec![0; 256];
|
|
let prompt_len = self.serial.read(&mut prompt_buf)
|
|
.map_err(|err| ModemError::ReadError(format!("Error in handle_prompt() ({:?})", err)))?;
|
|
|
|
let prompt = String::from_utf8(prompt_buf[0..prompt_len].to_vec())
|
|
.unwrap_or("".to_string())
|
|
.trim()
|
|
.to_string();
|
|
|
|
println!("Prompt is: ({})", prompt);
|
|
|
|
if prompt != ">" {
|
|
let msg = format!("Prompt error, expected (>), got ({})", prompt);
|
|
Err(ModemError::SendDataError(msg))
|
|
} else {
|
|
Ok(())
|
|
}
|
|
}
|
|
|
|
fn tcp_manual_send_data(&mut self, buf: &[u8]) -> Result<String> {
|
|
println!("Sending AT+CIPSEND to serial TX!");
|
|
let _ = self.serial
|
|
.write("AT+CIPSEND\r".as_bytes())
|
|
.map_err(|_| ModemError::SendDataError("Error in tcp_manual_send_data ... AT_CIPSEND\\r".to_string()))?;
|
|
|
|
let _ = self.handle_prompt()?;
|
|
println!("Handled prompt OK!!");
|
|
|
|
println!("Writing bytes in serial TX! ({:?})", String::from_utf8(buf.to_vec()));
|
|
self.serial
|
|
.write_bytes(buf)
|
|
.map_err(|err| ModemError::SendDataError(format!("{:?}", err)))?;
|
|
self.serial
|
|
.write(&[26_u8]) // 26_u8 = Ctrl+z - to end sending data
|
|
.map_err(|err| ModemError::SendDataError(format!("{:?}", err)))?;
|
|
println!("DONE Writing bytes in serial TX!");
|
|
|
|
thread::sleep(Duration::from_millis(500));
|
|
|
|
println!("Reading bytes in serial RX!");
|
|
for _ in 0..3 {
|
|
let res = self.command_read_response(Some("SEND OK".into()));
|
|
if res.is_ok() {
|
|
return res;
|
|
}
|
|
thread::sleep(Duration::from_millis(1000))
|
|
}
|
|
Err(ModemError::ReadError(format!("ReadError: cannot read serial RX!")))
|
|
}
|
|
|
|
pub fn gprs_status(&mut self) -> Result<String> {
|
|
self.send_command(Command::gprs_bearer_status())
|
|
}
|
|
|
|
pub fn gprs_attach_ap(&mut self, apn: &str, username: &str, password: &str)-> Result<()> {
|
|
println!("init gprs ...");
|
|
let _ = self.send_command(Command::gprs_init())?;
|
|
|
|
println!("setting up gprs credentials for apn {}, {}:{})", apn, username, password);
|
|
|
|
let _ = self.send_command(Command::gprs_set_apn(apn))?;
|
|
let _ = self.send_command(Command::gprs_set_user(username))?;
|
|
let _ = self.send_command(Command::gprs_set_pwd(password))?;
|
|
|
|
Ok(())
|
|
}
|
|
|
|
pub fn gprs_connect(&mut self)-> Result<()> {
|
|
println!("open gprs ...");
|
|
self.send_command(Command::gprs_bearer_open())
|
|
.map(|_| ())
|
|
}
|
|
|
|
pub fn is_gprs_attached(&mut self)-> Result<bool> {
|
|
let res = self.send_command(Command::is_gprs_attached())?;
|
|
Ok(res.contains("+CGATT: 1"))
|
|
}
|
|
|
|
fn try_connect_gprs(&mut self) -> Result<()> {
|
|
let mut retries = 0;
|
|
loop {
|
|
if self.is_gprs_attached()? {
|
|
let _ = self.gprs_connect()?;
|
|
thread::sleep(Duration::from_millis(1000));
|
|
let ip_addr = self.gprs_status()?;
|
|
if ip_addr.contains("0.0.0.0") && retries < 5 {
|
|
thread::sleep(Duration::from_millis(1000));
|
|
retries += 1;
|
|
continue
|
|
} else if retries < 5 {
|
|
break Ok(())
|
|
} else {
|
|
break Err(ModemError::SetupError(format!("Cannot connect to GPRS after {} retries ... bailing!", retries)));
|
|
}
|
|
}
|
|
}
|
|
}
|
|
|
|
pub fn tcp_is_ssl_enabled(&mut self) -> Result<bool> {
|
|
let res = self.send_command(Command::tcp_ssl_check())?;
|
|
Ok(res.contains("+CIPSSL: (1)"))
|
|
}
|
|
|
|
pub fn tcp_ssl_disable(&mut self) -> Result<()> {
|
|
self.send_command(Command::tcp_ssl_set(false))
|
|
.map(|_| ())
|
|
}
|
|
|
|
pub fn tcp_ssl_enable(&mut self) -> Result<()> {
|
|
self.send_command(Command::tcp_ssl_set(true))
|
|
.map(|_| ())
|
|
}
|
|
|
|
pub fn tcp_connect(&mut self, addr: &str, port: u16) -> Result<()> {
|
|
let at_command = format!("AT+CIPSTART=\"TCP\",\"{}\",\"{}\"", addr, port);
|
|
let _ = self.send(&at_command, "CONNECT OK");
|
|
for _ in 0..3 {
|
|
if let Ok(reply) = self.command_read_response(Some("CONNECT OK".to_string())) {
|
|
println!("TCP connect replied with {}", reply);
|
|
break
|
|
}
|
|
thread::sleep(Duration::from_millis(500));
|
|
}
|
|
Ok(())
|
|
}
|
|
|
|
pub fn tcp_set_quick_mode(&mut self, mode: bool) -> Result<()> {
|
|
self.send_command(Command::tcp_set_quick_mode(mode))
|
|
.map(|_| ())
|
|
}
|
|
|
|
pub fn tcp_set_manual_receive(&mut self, is_manual: bool) -> Result<()> {
|
|
self.send_command(Command::tcp_set_manual_receive(is_manual))
|
|
.map(|_| ())
|
|
}
|
|
|
|
pub fn tcp_manual_send(&mut self, buf: &[u8]) -> Result<()> {
|
|
self.tcp_manual_send_data(buf)
|
|
.map(|_| ())
|
|
}
|
|
|
|
fn tcp_parse_response_size(&mut self, reply_line: &str) -> Result<usize> {
|
|
reply_line.split(',')
|
|
.into_iter()
|
|
.last()
|
|
.unwrap_or("x")
|
|
.parse::<usize>()
|
|
.map_err(|_| ModemError::CommandError(format!("response size should be a number, got {}", reply_line)))
|
|
}
|
|
|
|
pub fn tcp_receive_reply_len(&mut self) -> Result<usize> {
|
|
let reply = self.send_command(Command::tcp_receive_reply_len())?;
|
|
println!("Receiving TCP reply length!");
|
|
let res = reply.lines()
|
|
.filter(|line| line.contains("+CIPRXGET: 4"))
|
|
.next()
|
|
.ok_or(ModemError::CommandError("reply body missing :/".to_string()))
|
|
.and_then(|line| self.tcp_parse_response_size(line))
|
|
.map_err(|_| ModemError::CommandError(format!("received 0 elements from parsing")));
|
|
println!("Received ({:?})", res);
|
|
res
|
|
}
|
|
|
|
pub fn tcp_receive(&mut self, buf: &mut [u8]) -> Result<usize> {
|
|
let mut size = 0;
|
|
loop {
|
|
let reply = self.send_command(Command::tcp_receive(buf.len()))
|
|
.map(|reply| {
|
|
// TODO: parse the response properly
|
|
// 1. the first line is \r\n
|
|
// 2. next is the +CIPRXGET: 2,X,Y where X is the number of bytes read and Y is
|
|
// the number of bytes left to be read
|
|
// 3. immediately after this the payload is returned (with size X)
|
|
// 4. OK
|
|
reply
|
|
.split("\r\n")
|
|
.filter(|line| line.len() > 2 && !line.contains("+CIPRXGET: 2,"))
|
|
.next()
|
|
.map(|line| line.chars().enumerate().map(|(idx, c)| buf[size + idx] = c as u8).count())
|
|
})?;
|
|
match reply {
|
|
Some(0) | None => {
|
|
break Ok(size)
|
|
},
|
|
Some(x) => {
|
|
size += x;
|
|
continue
|
|
},
|
|
}
|
|
}
|
|
}
|
|
|
|
pub fn tcp_close_connection(&mut self) -> Result<String> {
|
|
self.send_command(Command::tcp_close())
|
|
}
|
|
|
|
pub fn http_post(&mut self, url: &str, token: &str, content: &[u8]) -> Result<String> {
|
|
let _ = self.send_command(Command::http_init());
|
|
let _ = self.send_command(Command::http_set_cid());
|
|
let _ = self.send_command(Command::http_set_url(url));
|
|
let _ = self.send_command(Command::http_set_header("X-Secret", token));
|
|
let _ = self.send_command(Command::http_set_header("X-Topic", "device-dev"));
|
|
let _ = self.send_command(Command::http_set_content("application/json"));
|
|
let _ = self.send_command(Command::http_set_ssl(true));
|
|
let _ = self.send_command(Command::http_post_len(content.len(), 100000));
|
|
let _ = self.serial.write_bytes(content);
|
|
let _ = self.serial.write(&[26_u8]);
|
|
let _ = self.send_command(Command::http_post());
|
|
self.send_command(Command::http_read_response())
|
|
}
|
|
|
|
pub fn http_get(&mut self, url: &str) -> Result<String> {
|
|
let _ = self.send_command(Command::http_init());
|
|
let _ = self.send_command(Command::http_set_cid());
|
|
let _ = self.send_command(Command::http_set_url(url));
|
|
let _ = self.send_command(Command::http_set_redirect(true));
|
|
let _ = self.send_command(Command::http_set_ssl(true));
|
|
let _ = self.send_command(Command::http_get());
|
|
self.send_command(Command::http_read_response())
|
|
}
|
|
|
|
pub fn http_close(&mut self) -> Result<()> {
|
|
self.send_command(Command::http_close())
|
|
.map(|_| ())
|
|
}
|
|
|
|
pub fn chip_info(&mut self) -> Result<()> {
|
|
let _ = self.send_command(Command::manufacturer_id())?;
|
|
thread::sleep(Duration::from_millis(1000));
|
|
let _ = self.send_command(Command::model_id())?;
|
|
thread::sleep(Duration::from_millis(1000));
|
|
let _ = self.send_command(Command::release_id())?;
|
|
Ok(())
|
|
}
|
|
|
|
pub fn location(&mut self) -> Result<()> {
|
|
self.send_command(Command::get_location())
|
|
.map(|_| ())
|
|
}
|
|
|
|
pub fn ssl_opt(&mut self) -> Result<()> {
|
|
self.send_command(Command::ssl_opt())
|
|
.map(|_| ())
|
|
}
|
|
|
|
fn file_write(&mut self, buf: &[u8], path: &str, append: bool, input_time_sec: usize) -> Result<()> {
|
|
let cmd = Command::fs_file_write(path, append, buf.len(), input_time_sec);
|
|
let _ = self.serial
|
|
.write(cmd.text.as_bytes())
|
|
.map_err(|err| ModemError::SendDataError(format!("File write error ({:?})", err)))?;
|
|
|
|
let _ = self.handle_prompt()?;
|
|
|
|
self.serial
|
|
.write(buf)
|
|
.map_err(|err| ModemError::SendDataError(format!("Error sending bytes via serial ({:?})", err)))?;
|
|
let _ = self.command_read_response(None);
|
|
|
|
Ok(())
|
|
}
|
|
|
|
pub fn upload_cert(&mut self, path: &str, cert: &[u8]) -> Result<()> {
|
|
let _ = self.send_command(Command::fs_file_create(path))?;
|
|
let _ = self.file_write(cert, path, false, 20000)?;
|
|
Ok(())
|
|
}
|
|
|
|
pub fn fs_list(&mut self, path: &str) -> Result<()> {
|
|
self.send_command(Command::fs_list(path))
|
|
.map(|_| ())
|
|
}
|
|
|
|
pub fn fs_free_space(&mut self) -> Result<()> {
|
|
self.send_command(Command::fs_free_size())
|
|
.map(|_| ())
|
|
}
|
|
|
|
pub fn ssl_set_client_cert(&mut self, path: &str, password: &str) -> Result<()> {
|
|
self.send_command(Command::ssl_set_client_cert(path, password))
|
|
.map(|_| ())
|
|
}
|
|
|
|
pub fn ssl_set_root_cert(&mut self, path: &str, filesize: usize) -> Result<()> {
|
|
self.send_command(Command::ssl_set_root_cert(path, filesize))
|
|
.map(|_| ())
|
|
}
|
|
|
|
fn mqtt_receive_reply(&mut self) -> Result<VariablePacket> {
|
|
for _ in 0..3 {
|
|
let size = self.tcp_receive_reply_len()?;
|
|
println!("received reply len({}) ...", size);
|
|
if size == 0 {
|
|
println!("retrying ...");
|
|
continue
|
|
} else {
|
|
let mut reply = vec![0 as u8; size];
|
|
println!("receiving mqtt reply ...");
|
|
let _ = self.tcp_receive(&mut reply);
|
|
let reply = std::str::from_utf8(&reply).unwrap_or("");
|
|
println!("received mqtt reply ({})", reply);
|
|
return VariablePacket::decode(&mut reply.as_bytes())
|
|
.map_err(|err| ModemError::CommandError(format!("Undecodable MQTT message. ({:?})", err)));
|
|
}
|
|
}
|
|
Err(ModemError::ReadError("TCP server didn't respond!".into()))
|
|
}
|
|
|
|
fn mqtt_connect(&mut self, device_id: &str) -> anyhow::Result<()> {
|
|
let mut buf = Vec::new();
|
|
let mut conn = ConnectPacket::new(device_id);
|
|
conn.set_clean_session(true);
|
|
conn.set_keep_alive(100);
|
|
let _ = conn.encode(&mut buf)?;
|
|
self.tcp_manual_send(&mut buf).ok();
|
|
|
|
let reply = self.mqtt_receive_reply()?;
|
|
println!("mqtt decoded packet: ({:?})", reply);
|
|
|
|
match reply {
|
|
VariablePacket::ConnackPacket(_) => Ok(()),
|
|
_ => Err(anyhow::Error::msg("Invalid MQTT reply ... expected CONNACK!"))
|
|
}
|
|
}
|
|
|
|
fn mqtt_publish(&mut self, _device_id: &str, message: &str) -> anyhow::Result<()> {
|
|
println!("entered mqtt publish ...");
|
|
let mut buf = Vec::new();
|
|
let packet = PublishPacket::new(
|
|
TopicName::new(format!("bajsevi/location")).unwrap(),
|
|
QoSWithPacketIdentifier::Level0,
|
|
message.as_bytes(),
|
|
);
|
|
let _ = packet.encode(&mut buf)?;
|
|
println!("created mqtt publish packet ... ({})",
|
|
std::str::from_utf8(buf.as_slice()).unwrap_or(""));
|
|
self.tcp_manual_send(&mut buf).ok();
|
|
Ok(())
|
|
}
|
|
}
|
|
|
|
impl<UART: serial::Uart> std::io::Read for Modem<UART> {
|
|
fn read(&mut self, buf: &mut [u8]) -> std::io::Result<usize> {
|
|
self.tcp_receive(buf).map_err(|_| std::io::Error::from(std::io::ErrorKind::ConnectionAborted))
|
|
}
|
|
}
|
|
|
|
pub fn main<T: Sync + Send>(
|
|
rx: esp_idf_hal::gpio::Gpio26<T>,
|
|
tx: esp_idf_hal::gpio::Gpio27<T>,
|
|
uart: serial::UART1,
|
|
pwrkey: esp_idf_hal::gpio::Gpio4<esp_idf_hal::gpio::Output>,
|
|
rst: esp_idf_hal::gpio::Gpio5<esp_idf_hal::gpio::Output>,
|
|
power: esp_idf_hal::gpio::Gpio23<esp_idf_hal::gpio::Output>,
|
|
receiver: Receiver<Msg>,
|
|
) -> std::result::Result<(), anyhow::Error> {
|
|
let serial_pins = serial::Pins {
|
|
tx,
|
|
rx,
|
|
cts: None,
|
|
rts: None,
|
|
};
|
|
|
|
let serial: serial::Serial<serial::UART1, _, _> = serial::Serial::new(
|
|
uart,
|
|
serial_pins,
|
|
serial::config::Config::default().baudrate(Hertz(115200)),
|
|
)?;
|
|
|
|
let (tx, rx) = serial.split();
|
|
let mut mdm = Modem::new(tx, rx);
|
|
|
|
mdm.init(pwrkey, rst, power)?;
|
|
|
|
// thread::sleep(Duration::from_millis(500));
|
|
|
|
//println!("setting up client TLS cert");
|
|
//let client_cert = include_bytes!("../certs/full-bin.p12");
|
|
//let client_cert_path = "C:\\USER\\fullchain.pem";
|
|
|
|
//let _ = mdm.upload_cert(client_cert_path, client_cert)?;
|
|
//let _ = mdm.ssl_set_client_cert(client_cert_path, "t")?;
|
|
//let _ = mdm.fs_list("C:\\USER\\")?;
|
|
|
|
// Retry 5 times to open a TCP connection, otherwise fail and wait for reboot.
|
|
let mut retries = 0;
|
|
|
|
loop {
|
|
if !mdm.is_gprs_attached()? {
|
|
let _ = mdm.gprs_attach_ap(
|
|
crate::config::A1_GPRS_AP.apn,
|
|
crate::config::A1_GPRS_AP.username,
|
|
crate::config::A1_GPRS_AP.password,
|
|
)?;
|
|
}
|
|
if let Ok(()) = mdm.try_connect_gprs() {
|
|
let device_id = "c36a72df-5bd6-4f9b-995d-059433bc3267";
|
|
|
|
// When command AT+CIPQSEND=0, it is in normal sending mode. In this mode, after user
|
|
// sends data by AT+CIPSEND, if the server receives TCP data, it will give ACK message
|
|
// to module, and the module will respond SEND OK.
|
|
let _ = mdm.send("AT+CIPQSEND=0", "OK");
|
|
// Enables getting data from network manually.
|
|
let _ = mdm.send("AT+CIPRXGET=1", "OK");
|
|
|
|
if let Err(_) = mdm.tcp_connect("51.158.66.64", 7887) {
|
|
if retries < 5 {
|
|
retries += 1;
|
|
continue;
|
|
}
|
|
}
|
|
thread::sleep(Duration::from_millis(500));
|
|
mdm.serial.clear();
|
|
|
|
let _ = mdm.mqtt_connect(device_id)?;
|
|
|
|
println!("entering queue receive loop ...");
|
|
let mut err_count = 0;
|
|
let _ = loop {
|
|
match receiver.recv() {
|
|
Ok(Msg::Gps(solution)) => {
|
|
println!("received GPS solution {:?} | sending to mqtt ...", solution);
|
|
let _ = mdm.mqtt_publish(device_id, &format!("{:?}", solution))?;
|
|
err_count = 0;
|
|
},
|
|
Ok(Msg::Accelerometer(acc)) => {
|
|
println!("received accel {} | sending to mqtt ...", acc);
|
|
let _ = mdm.mqtt_publish(device_id, &format!("{:?}", acc))?;
|
|
err_count = 0;
|
|
}
|
|
Err(e) => {
|
|
if err_count < 5 {
|
|
err_count += 1;
|
|
println!("received error {} | NOT sending to mqtt ...", e);
|
|
}
|
|
else {
|
|
break
|
|
}
|
|
}
|
|
}
|
|
};
|
|
let _ = mdm.tcp_close_connection()?;
|
|
}
|
|
}
|
|
}
|