tcp client stack and some improvements

.gitignore

@@ -2,4 +2,3 @@
 /.embuild
 /target
 /Cargo.lock
-/secret

Cargo.toml

@@ -17,11 +17,9 @@ anyhow = "1.0.57"
 embedded-hal = "0.2.7"
 esp-idf-hal = "0.37.4"
 esp-idf-sys = { version = "0.31.5", features = ["binstart", "native"] }
-mqtt-protocol = "0.11.2"
+minimq = "0.5.3"
 nb = "1.0.0"
-nmea0183 = "0.3.0"
+std-embedded-time = "0.1.0"
-serde-json-core = "0.5.0"
-serde = "*"
 
 [build-dependencies]
 embuild = "0.29"

rust-toolchain.toml

@@ -1,2 +1,2 @@
 [toolchain]
-channel = "esp-1.69.0.0"
+channel = "esp"

sdkconfig.defaults

@@ -1,13 +1,10 @@
 # Rust often needs a bit of an extra main task stack size compared to C (the default is 3K)
-CONFIG_ESP_MAIN_TASK_STACK_SIZE=128000
+CONFIG_ESP_MAIN_TASK_STACK_SIZE=17000
 
 # Use this to set FreeRTOS kernel tick frequency to 1000 Hz (100 Hz by default).
 # This allows to use 1 ms granuality for thread sleeps (10 ms by default).
 #CONFIG_FREERTOS_HZ=1000
 
-# Explicitly specify UART0 for console debugging.
-CONFIG_CONSOLE_UART_NUM=0
-
 # Workaround for https://github.com/espressif/esp-idf/issues/7631
 CONFIG_MBEDTLS_CERTIFICATE_BUNDLE=n
 CONFIG_MBEDTLS_CERTIFICATE_BUNDLE_DEFAULT_FULL=n

src/accel.rs

@@ -1,18 +0,0 @@
-use anyhow;
-
-use std::sync::mpsc::SyncSender;
-use std::thread;
-use std::time::Duration;
-
-use crate::types::*;
-
-pub fn main(sender: SyncSender<Msg>) -> anyhow::Result<()> {
-    let mut c = 1_usize;
-    println!("entering ACCELERATOR sender loop ...");
-    loop {
-        println!("sending ACCELERATOR message No. {}", c);
-        let _ = sender.send(Msg::Accelerometer("{\"velocity\": 21.43, \"altitude\": 367}".to_string()))?;
-        thread::sleep(Duration::from_secs(5));
-        c += 1;
-    }
-}

src/command.rs

@@ -1,6 +1,5 @@
 use std::time::Duration;
 
-#[derive(Debug)]
 pub struct Command {
     pub text: String,
     pub timeout: Duration,
@@ -12,7 +11,7 @@ impl Command {
         Command {
             text: "ATI".to_string(),
             timeout: Duration::from_millis(6000),
-            contains: Some("OK".to_string()),
+            contains: Some("+CIEV".to_string()),
         }
     }
 
@@ -72,6 +71,14 @@ impl Command {
         }
     }
 
+    pub fn gprs_open() -> Command {
+        Command {
+            text: "AT+SAPBR=1,1".to_string(),
+            timeout: Duration::from_millis(3000),
+            contains: Some("OK".to_string()),
+        }
+    }
+
     pub fn gprs_set_apn(apn: &str) -> Command {
         Command {
             text: format!("AT+SAPBR=3,1,\"APN\",\"{}\"", apn),
@@ -96,7 +103,7 @@ impl Command {
         }
     }
 
-    pub fn gprs_bearer_status() -> Command {
+    pub fn getbear() -> Command {
         Command {
             text: "AT+SAPBR=2,1".to_string(),
             timeout: Duration::from_millis(3000),
@@ -104,22 +111,6 @@ impl Command {
         }
     }
 
-    pub fn gprs_bearer_open() -> Command {
-        Command {
-            text: "AT+SAPBR=1,1".to_string(),
-            timeout: Duration::from_millis(3000),
-            contains: Some("OK".to_string()),
-        }
-    }
-
-    pub fn gprs_bearer_close() -> Command {
-        Command {
-            text: "AT+SAPBR=0,1".to_string(),
-            timeout: Duration::from_millis(3000),
-            contains: Some("OK".to_string()),
-        }
-    }
-
     pub fn get_local_ip_addr() -> Command {
         Command {
             text: "AT+CIFSR".to_string(),
@@ -144,7 +135,7 @@ impl Command {
         }
     }
 
-    pub fn http_set_cid() -> Command {
+    pub fn http_set() -> Command {
         Command {
             text: "AT+HTTPPARA=\"CID\",1".to_string(),
             timeout: Duration::from_millis(3000),
@@ -152,25 +143,25 @@ impl Command {
         }
     }
 
-    pub fn http_set_url(url: &str) -> Command {
+    pub fn http_enable_ssl() -> Command {
         Command {
-            text: format!("AT+HTTPPARA=\"URL\",\"{}\"", url),
+            text: "AT+HTTPSSL=1".to_string(),
             timeout: Duration::from_millis(3000),
             contains: Some("OK".to_string()),
         }
     }
 
-    pub fn http_set_ssl(enabled: bool) -> Command {
+    pub fn http_disable_ssl() -> Command {
         Command {
-            text: format!("AT+HTTPSSL={}", enabled as u8),
+            text: "AT+HTTPSSL=0".to_string(),
-            timeout: Duration::from_millis(1000),
+            timeout: Duration::from_millis(3000),
             contains: Some("OK".to_string()),
         }
     }
 
-    pub fn http_set_header(header: &str, value: &str) -> Command {
+    pub fn http_init_url() -> Command {
         Command {
-            text: format!("AT+HTTPPARA=\"USERDATA\",\"{}: {}\"", header, value),
+            text: "AT+HTTPPARA=\"URL\",\"{}\"".to_string(),
             timeout: Duration::from_millis(3000),
             contains: Some("OK".to_string()),
         }
@@ -184,39 +175,31 @@ impl Command {
         }
     }
 
+    pub fn http_set_content() -> Command {
+        Command {
+            text: "AT+HTTPPARA=\"CONTENT\",\"{}\"".to_string(),
+            timeout: Duration::from_millis(3000),
+            contains: Some("OK".to_string()),
+        }
+    }
+
+    pub fn http_post_len() -> Command {
+        Command {
+            text: "AT+HTTPDATA={}5000".to_string(),
+            timeout: Duration::from_millis(3000),
+            contains: Some("DOWNLOAD".to_string()),
+        }
+    }
+
     pub fn http_post() -> Command {
         Command {
             text: "AT+HTTPACTION=1".to_string(),
-            timeout: Duration::from_millis(10000),
-            contains: Some("HTTPACTION".to_string()),
-        }
-    }
-
-    pub fn http_set_content(content: &str) -> Command {
-        Command {
-            text: format!("AT+HTTPPARA=\"CONTENT\",\"{}\"", content),
             timeout: Duration::from_millis(3000),
-            contains: Some("OK".to_string()),
+            contains: Some("+HTTPACTION".to_string()),
         }
     }
 
-    pub fn http_set_redirect(redirect: bool) -> Command {
+    pub fn http_get_data() -> Command {
-        Command {
-            text: format!("AT+HTTPPARA=\"REDIR\",\"{}\"", redirect as u8),
-            timeout: Duration::from_millis(3000),
-            contains: Some("OK".to_string()),
-        }
-    }
-
-    pub fn http_post_len(size: usize, time: usize) -> Command {
-        Command {
-            text: format!("AT+HTTPDATA={},{}", size, time),
-            timeout: Duration::from_millis(5000),
-            contains: Some("OK".to_string()),
-        }
-    }
-
-    pub fn http_read_response() -> Command {
         Command {
             text: "AT+HTTPREAD".to_string(),
             timeout: Duration::from_millis(3000),
@@ -224,7 +207,7 @@ impl Command {
         }
     }
 
-    pub fn http_close() -> Command {
+    pub fn closehttp() -> Command {
         Command {
             text: "AT+HTTPTERM".to_string(),
             timeout: Duration::from_millis(3000),
@@ -232,11 +215,19 @@ impl Command {
         }
     }
 
+    pub fn closebear() -> Command {
+        Command {
+            text: "AT+SAPBR=0,1".to_string(),
+            timeout: Duration::from_millis(3000),
+            contains: Some("OK".to_string()),
+        }
+    }
+
     pub fn probe() -> Command {
         Command {
             text: "AT".to_string(),
             timeout: Duration::from_millis(3000),
-            contains: Some("OK".to_string()),
+            contains: Some("+CIEV".to_string()),
         }
     }
 
@@ -256,6 +247,14 @@ impl Command {
         }
     }
 
+    pub fn tcp_ssl_enable() -> Command {
+        Command {
+            text: "AT+CIPSSL=1".to_string(),
+            timeout: Duration::from_millis(3000),
+            contains: Some("OK".to_string()),
+        }
+    }
+
     pub fn tcp_ssl_check() -> Command {
         Command {
             text: "AT+CIPSSL=?".to_string(),
@@ -268,7 +267,7 @@ impl Command {
         Command {
             text: format!("AT+CIPSTART=\"TCP\",\"{}\",\"{}\"", addr, port),
             timeout: Duration::from_millis(5000),
-            contains: Some("OK".to_string()),
+            contains: Some("CONNECT OK".to_string()),
         }
     }
 
@@ -344,91 +343,11 @@ impl Command {
         }
     }
 
-    pub fn manufacturer_id() -> Command {
+    pub fn tcp_connection_state() -> Command {
         Command {
-            text: "AT+GMI".to_string(),
+            text: "AT+CIPSTATUS".to_string(),
-            timeout: Duration::from_millis(3000),
-            contains: Some("OK".to_string()),
-        }
-    }
-
-    pub fn model_id() -> Command {
-        Command {
-            text: "AT+GMM".to_string(),
-            timeout: Duration::from_millis(3000),
-            contains: Some("OK".to_string()),
-        }
-    }
-
-    pub fn release_id() -> Command {
-        Command {
-            text: "AT+GMR".to_string(),
-            timeout: Duration::from_millis(3000),
-            contains: Some("OK".to_string()),
-        }
-    }
-
-    pub fn get_location() -> Command {
-        Command {
-            text: "AT+CLBS=1,1".to_string(),
-            timeout: Duration::from_millis(10000),
-            contains: Some("OK".to_string()),
-        }
-    }
-
-    pub fn ssl_opt() -> Command {
-        Command {
-            text: "AT+SSLOPT=1,1".to_string(),
-            timeout: Duration::from_millis(3000),
-            contains: Some("OK".to_string()),
-        }
-    }
-
-    pub fn ssl_set_client_cert(path: &str, password: &str) -> Command {
-        Command {
-            text: format!("AT+SSLSETCERT={},{}", path, password),
             timeout: Duration::from_millis(2000),
-            contains: Some("SSLSETCERT".to_string()),
+            contains: Some("STATE".to_string()),
-        }
-    }
-
-    pub fn ssl_set_root_cert(path: &str, size: usize) -> Command {
-        Command {
-            text: format!("AT+SSLSETROOT={},{}", path, size),
-            timeout: Duration::from_millis(2000),
-            contains: Some("SSLSETCERT".to_string()),
-        }
-    }
-
-    pub fn fs_file_create(path: &str) -> Command {
-        Command {
-            text: format!("AT+FSCREATE={}", path),
-            timeout: Duration::from_millis(2000),
-            contains: Some("OK".to_string()),
-        }
-    }
-
-    pub fn fs_file_write(path: &str, append: bool, size: usize, input_time_sec: usize) -> Command {
-        Command {
-            text: format!("AT+FSWRITE={},{},{},{}", path, append as u8, size, input_time_sec),
-            timeout: Duration::from_millis(20000),
-            contains: Some("OK".to_string()),
-        }
-    }
-
-    pub fn fs_list(path: &str) -> Command {
-        Command {
-            text: format!("AT+FSLS={}", path),
-            timeout: Duration::from_millis(2000),
-            contains: Some("OK".to_string()),
-        }
-    }
-
-    pub fn fs_free_size() -> Command {
-        Command {
-            text: "AT+FSMEM".to_string(),
-            timeout: Duration::from_millis(2000),
-            contains: Some("OK".to_string()),
         }
     }
 }

src/config.rs

@@ -1,19 +1,11 @@
-#![allow(dead_code)]
-
 pub struct GprsAp<'a> {
     pub apn: &'a str,
     pub username: &'a str,
     pub password: &'a str,
 }
 
-pub const A1: GprsAp = GprsAp {
+pub const A1_GPRS_AP: GprsAp = GprsAp {
     apn: "internet",
     username: "internet",
     password: "internet",
 };
-
-pub const MTS: GprsAp = GprsAp {
-    apn: "gprswap",
-    username: "mts",
-    password: "064",
-};

src/gps.rs

@@ -1,90 +0,0 @@
-use anyhow;
-
-use std::{
-    sync::mpsc::SyncSender,
-    thread,
-    time::Duration,
-    io::Read,
-};
-
-use esp_idf_hal::prelude::*;
-use esp_idf_hal::serial::{self, Rx, Tx};
-
-use nmea0183::{Parser, ParseResult, Sentence, Source};
-
-use crate::types::Msg;
-use crate::serial::SerialIO;
-
-struct GpsModule<UART: serial::Uart> {
-    port: SerialIO<UART>,
-}
-
-impl<UART: serial::Uart> GpsModule<UART> {
-    pub fn new(tx: Tx<UART>, rx: Rx<UART>) -> Self {
-        GpsModule {
-            port: SerialIO::new(tx, rx),
-        }
-    }
-}
-
-pub fn main<PRx,PTx>
-(
-    tx: PTx,
-    rx: PRx,
-    uart: serial::UART2,
-    sender: SyncSender<Msg>,
-) -> std::result::Result<(), anyhow::Error>
-where
-    PRx: esp_idf_hal::gpio::Pin + esp_idf_hal::gpio::InputPin + esp_idf_hal::gpio::OutputPin,
-    PTx: esp_idf_hal::gpio::Pin + esp_idf_hal::gpio::InputPin + esp_idf_hal::gpio::OutputPin,
-{
-    let serial_pins = serial::Pins {
-        tx,
-        rx,
-        cts: None,
-        rts: None,
-    };
-
-    let serial: serial::Serial<serial::UART2, _, _> = serial::Serial::new(
-        uart,
-        serial_pins,
-        serial::config::Config::default().baudrate(Hertz(9600)),
-    )?;
-
-    let (tx, rx) = serial.split();
-    let mut device = GpsModule::new(tx, rx);
-
-    let mut parser = Parser::new()
-        .source_only(Source::GPS)
-        .sentence_filter(Sentence::GLL | Sentence::GGA);
-
-    let mut c = 0;
-    let mut nmea = [0_u8; 1024];
-    loop {
-        if let Ok(_) = device.port.read(nmea.as_mut_slice()) {
-            println!("\r\n\r\n\r\n\r\n");
-            for result in parser.parse_from_bytes(&nmea[..]) {
-                match result {
-                    Ok(ParseResult::GLL(Some(gll))) => {
-                        sender.send(Msg::Gps(gll.into()))?;
-                    },
-                    Ok(ParseResult::GGA(Some(gga))) => {
-                        sender.send(Msg::Gps(gga.into()))?;
-                    }
-                    _ => { }
-                }
-            }
-            c = 0;
-        } else {
-            println!("nothing to read after {} tries ...", c);
-            if c > 100 {
-                println!("reached {} retries ... bailing!", c);
-                break;
-            }
-        }
-        thread::sleep(Duration::from_millis(5000));
-        c += 1;
-    }
-
-    Ok(())
-}

src/main.rs

@@ -1,89 +1,91 @@
-mod accel;
 mod config;
+mod modem;
 #[allow(dead_code)]
 mod command;
-mod modem;
-mod serial;
-mod types;
-mod gps;
 
 use anyhow;
-use std::{thread::{self, JoinHandle}, time::Duration};
-use esp_idf_hal::peripherals::Peripherals;
 use esp_idf_hal::prelude::*;
-use esp_idf_hal::serial::{Pins, config::Config, Serial, UART1, Uart};
+use esp_idf_hal::peripherals::Peripherals;
-use embedded_hal::digital::v2::OutputPin;
+use esp_idf_hal::serial;
 
-use types::*;
+use minimq::{Minimq, QoS, Retain};
 
 fn main() -> anyhow::Result<()> {
     esp_idf_sys::link_patches();
 
     let dp = Peripherals::take().expect("error taking peripherals");
 
-    println!("Rust main thread: {:?}", thread::current());
-
-    let mut threads: Vec<JoinHandle<anyhow::Result<_>>> = vec![];
-
-    // // Rx/Tx pins for the GPS modem
-    let gps_rx = dp.pins.gpio32;
-    let gps_tx = dp.pins.gpio33;
-    // // UART interface for the GPS modem
-    let gps_uart = dp.uart2;
-
-
-    let (gps_sender, receiver) = std::sync::mpsc::sync_channel::<Msg>(3);
-    //let accel_sender = gps_sender.clone();
-
-    //let _ = gps::main(gps_tx, gps_rx, gps_uart, gps_sender)?;
-    //threads.push(thread::spawn(move || accel::main(accel_sender)));
-
-    // ==================================
-    // MODEM INITIALIZATION AND MAIN LOOP
-    // ==================================
-
     // LilyGo TTGO T-Call sim800l board serial pins.
-    let modem_rx = dp.pins.gpio26;
+    let serial_rx = dp.pins.gpio26;
-    let modem_tx = dp.pins.gpio27;
+    let serial_tx = dp.pins.gpio27;
-    // LilyGo TTGO T-Call sim800l board power / reset pins.
-    let modem_pwrkey = dp.pins.gpio4.into_output()?;
-    let modem_rst = dp.pins.gpio5.into_output()?;
-    let modem_power = dp.pins.gpio23.into_output()?;
-    // UART interface for the GSM modem
-    let modem_uart = dp.uart1;
 
-    let serial_pins = Pins {
+    let serial_pins = serial::Pins {
-        tx: modem_tx,
+        tx: serial_tx,
-        rx: modem_rx,
+        rx: serial_rx,
         cts: None,
         rts: None,
     };
 
-    let serial: Serial<UART1, _, _> = Serial::new(
+    // Create the serial and panic with a message ... if we can't create the serial port, then we
-        modem_uart,
+    // can't communicate with the sim800l module, hence we don't run anymore.
+    let serial: serial::Serial<serial::UART1, _, _> = serial::Serial::new(
+        dp.uart1,
         serial_pins,
-        Config::default().baudrate(Hertz(115200)),
+        serial::config::Config::default().baudrate(Hertz(115200)),
     )?;
 
     let (tx, rx) = serial.split();
-    type PwrkeyOutput = esp_idf_hal::gpio::Gpio4<esp_idf_hal::gpio::Output>;
+    let mut mdm = modem::Modem::new(tx, rx);
-    type ResetOutput = esp_idf_hal::gpio::Gpio5<esp_idf_hal::gpio::Output>;
-    type PowerOutput = esp_idf_hal::gpio::Gpio23<esp_idf_hal::gpio::Output>;
 
-    let mut mdm: modem::Modem<UART1, PwrkeyOutput, ResetOutput, PowerOutput> = modem::Modem::new(tx, rx, modem_pwrkey, modem_rst, modem_power, receiver);
+    let modem_pwrkey = dp.pins.gpio4.into_output()?;
+    let modem_rst = dp.pins.gpio5.into_output()?;
+    let modem_power = dp.pins.gpio23.into_output()?;
 
-    let mqtt_username = include_str!("../secret/username").trim();
+    mdm.init(modem_pwrkey, modem_rst, modem_power)?;
-    let mqtt_password = include_str!("../secret/password").trim();
 
-    threads.push(thread::spawn(move || gps::main(gps_tx, gps_rx, gps_uart, gps_sender.clone())));
+    if !mdm.is_gprs_attached()? {
+        let _ = mdm.connect_to_gprs_ap(
+            config::A1_GPRS_AP.apn,
+            config::A1_GPRS_AP.username,
+            config::A1_GPRS_AP.password,
+        )?;
+    }
 
-    println!("======================= MAIN =======================");
+    if mdm.is_gprs_attached()? {
-    mdm.init().unwrap_or(());
+        let _ = mdm.get_ip_addr()?;
-    let _ = mdm.echo(false).unwrap_or(());
+
-    println!("resetting modem ... ");
+        //println!("connecting to server!");
-    println!("======================= MODEM =======================");
+        //if !mdm.tcp_is_ssl_enabled()? {
-    let _ = mdm.mqtt_send_position_loop("51.158.66.64", 7887, mqtt_username, mqtt_password).unwrap_or(());
+        //    let _ = mdm.tcp_ssl_enable()?;
-    let _ = mdm.tcp_close_connection().unwrap_or(());
+        //}
-    thread::sleep(Duration::from_millis(1500));
+        if mdm.tcp_is_ssl_enabled()? {
-    panic!("rebooting");
+            let _ = mdm.tcp_ssl_disable()?;
+        }
+        let _ = mdm.tcp_set_quick_mode(false);
+        let _ = mdm.tcp_set_manual_receive(true)?;
+
+        let mut mqtt: Minimq<_, _, 256, 16> = Minimq::new(
+            "51.158.66.64".parse().unwrap(),
+            "e-bike-tracker",
+            mdm,
+            std_embedded_time::StandardClock::default()).unwrap();
+
+        mqtt.client
+            .set_will(
+                "exit",
+                "Test complete".as_bytes(),
+                QoS::AtMostOnce,
+                Retain::NotRetained,
+                &[],
+            )
+            .unwrap();
+
+        println!("created mqtt client ... ");
+        let message = "{\"lat\": 20, \"long\": 44}";
+        println!("message = {}", message);
+        mqtt.client.publish("devices/location", message.as_bytes(), QoS::AtMostOnce, Retain::NotRetained, &[]).unwrap();
+        println!("published message ... ");
+    }
+
+    Ok(())
 }

src/modem.rs

@@ -1,43 +1,29 @@
-#![allow(dead_code)]
-
 use crate::command::Command;
-use crate::serial::SerialIO;
-use crate::types::*;
 
-use anyhow;
+use std::thread;
-use std::{
+use std::error::Error;
-    error::Error,
+use std::time::{Duration, Instant};
-    io::{Read, Write},
-    thread,
-    time::Duration,
-    sync::mpsc::Receiver,
-};
-
-use esp_idf_hal::serial::{self, Rx, Tx};
 
+use embedded_hal::serial::{Read, Write};
 use embedded_hal::digital::v2::OutputPin;
+use esp_idf_hal::serial::{self, Rx, Tx};
+use minimq::embedded_nal::{TcpClientStack, SocketAddr};
 
-use mqtt::{
+const MAX_TCP_MANUAL_REPLY_SIZE: usize = 300;
-    Encodable,
-    Decodable,
-    TopicName,
-    packet::{
-        ConnectPacket,
-        PublishPacket,
-        QoSWithPacketIdentifier,
-        VariablePacket,
-    },
-};
-use serde_json_core;
 
 pub type Result<T> = std::result::Result<T, ModemError>;
 
+pub struct Modem<UART: serial::Uart> {
+    rx: RxIter<UART>,
+    tx: Tx<UART>,
+}
+
 #[derive(Debug)]
 pub enum ModemError {
     CommandError(String),
     SetupError(String),
-    SendDataError(String),
+    SendDataError,
-    ReadError(String),
+    ReadError,
     TimeoutError,
 }
 
@@ -49,22 +35,74 @@ impl std::fmt::Display for ModemError {
     }
 }
 
-pub struct Modem<UART: serial::Uart, PWK: OutputPin, RST: OutputPin, PW: OutputPin> {
+pub struct RxIter<UART: serial::Uart> {
-    serial: SerialIO<UART>,
+    inner: Rx<UART>,
-    reset: RST,
+    timeout: Duration,
-    power: PW,
-    power_key: PWK,
-    receiver: Receiver<Msg>,
 }
 
-impl<UART: serial::Uart, PWK: OutputPin, RST: OutputPin, PW: OutputPin> Modem<UART, PWK, RST, PW> {
+impl<UART: serial::Uart> RxIter<UART> {
-    pub fn new(tx: Tx<UART>, rx: Rx<UART>, mut pwrkey: PWK, mut rst: RST, mut power: PW, receiver: Receiver<Msg>) -> Self {
+    fn reset(&mut self, timeout: Duration) -> &mut Self {
+        self.timeout = timeout;
+        self
+    }
+
+    fn clear(&mut self) -> () {
+        println!("clearing serial rx");
+        self.reset(Duration::from_millis(500)).for_each(drop);
+    }
+
+    /// Reads a whole line (that ends with \\n) within the given `timeout` passed on input.
+    fn read_line(&mut self, timeout: Duration) -> Result<String> {
+        let mut line: String = self.reset(timeout)
+            .map(|b| char::from(b))
+            .take_while(|c| *c != '\n')
+            .collect();
+
+        // \r must come right before \n on read; take_while excludes the matched element.
+        if line.ends_with('\r') {
+            line.push('\n');
+            Ok(line)
+        }
+        else if self.timeout.as_millis() == 0 {
+            Err(ModemError::TimeoutError)
+        }
+        else {
+            Err(ModemError::ReadError)
+        }
+    }
+}
+
+impl<UART: serial::Uart> Iterator for RxIter<UART> {
+    type Item = u8;
+
+    /// `nb` returns Ok(byte), or one of Err(WouldBlock) and Err(Other) which isn't of anyone's
+    /// interest, so the retry mechanism is triggered on _any_ error every 200ms until a byte is
+    /// received, or the timeout is reached.
+    fn next(&mut self) -> Option<Self::Item> {
+        let start = Instant::now();
+        loop {
+            match self.inner.read() {
+                Ok(b) => {
+                    self.timeout = self.timeout.saturating_sub(start.elapsed());
+                    break Some(b)
+                },
+                Err(_) => {
+                    if start.elapsed() > self.timeout {
+                        self.timeout = Duration::ZERO;
+                        break None
+                    }
+                    thread::sleep(Duration::from_millis(200));
+                }
+            }
+        }
+    }
+}
+
+impl<UART: serial::Uart> Modem<UART> {
+    pub fn new(tx: Tx<UART>, rx: Rx<UART>) -> Self {
         Self {
-            serial: SerialIO::new(tx, rx),
+            rx: RxIter { inner: rx, timeout: Duration::from_millis(0) },
-            reset: rst,
+            tx,
-            power,
-            power_key: pwrkey,
-            receiver,
         }
     }
 
@@ -83,241 +121,141 @@ impl<UART: serial::Uart, PWK: OutputPin, RST: OutputPin, PW: OutputPin> Modem<UA
     ///
     /// modem::init(modem_pwrkey, modem_rst, modem_power);
     /// ```
-    pub fn init(&mut self) -> Result<()> {
+    pub fn init(&mut self, mut pwrkey: impl OutputPin, mut rst: impl OutputPin, mut power: impl OutputPin) -> Result<()> {
         println!("Turning SIM800L on ...");
-        self.power.set_high().map_err(|_| ModemError::SetupError("Error setting POWER to high.".to_string()))?;
+        power.set_high().map_err(|_| ModemError::SetupError("Error setting POWER to high.".to_string()))?;
-        self.reset.set_high().map_err(|_| ModemError::SetupError("Error setting RST 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
-        self.power_key.set_high().map_err(|_| ModemError::SetupError("Error setting PWRKEY to high.".to_string()))?;
+        pwrkey.set_high().map_err(|_| ModemError::SetupError("Error setting PWRKEY to high.".to_string()))?;
-        thread::sleep(Duration::from_millis(1500));
+        thread::sleep(Duration::from_millis(100));
-        self.power_key.set_low().map_err(|_| ModemError::SetupError("Error setting PWRKEY to low.".to_string()))?;
+        pwrkey.set_low().map_err(|_| ModemError::SetupError("Error setting PWRKEY to low.".to_string()))?;
         thread::sleep(Duration::from_millis(1000));
-        self.power_key.set_high().map_err(|_| ModemError::SetupError("Error setting PWRKEY to high.".to_string()))?;
+        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 {
-        for _ in 0..10 {
+            match self.send_command(Command::probe()) {
-            let _ = self.send_command(Command::probe()).unwrap_or("".to_string());
+                Ok(_) => break,
-            thread::sleep(Duration::from_millis(1000));
+                _ => continue,
+            }
         }
-        self.serial.clear();
         Ok(())
     }
 
-    pub fn echo(&mut self, enabled: bool) -> Result<()> {
+    /// Reads the serial RX until a \\n char is encoutered, or a timeout is reached. The timeout is
-        let cmd = format!("ATE{}", if enabled { 1 } else { 0 });
+    /// provided on input via the `timeout` argument. The first argument `contains` is checked
-        self.send(&cmd, "OK").map(|_| ())
+    /// against a line in the response, if it's there the reading stops.
-    }
+    ///
-
+    /// If `contains` is `None`, the first line only is returned in the response. If it's
-    /// Reads the serial RX until the `contains` string is encoutered if `contains` is Some(s), if
+    /// `Some(match_txt)`, then the end of the response is matched against `match_txt`.
-    /// None, then the first line is returned. If a timeout is reached. The timeout is provided on
+    fn read_response(&mut self, contains: Option<String>, timeout: Duration) -> Result<String> {
-    /// 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();
+        let start = Instant::now();
+        let match_text: String = contains.unwrap_or("\n".to_string());
 
         loop {
-            let mut buf = vec![0; 1024];
+            let timeout = timeout.saturating_sub(start.elapsed());
-            let num_bytes = self.serial
+            let line = self.rx.read_line(timeout)?;
-                .read(buf.as_mut_slice())
+            print!("Read {} bytes from serial: {}", line.len(), line);
-                .map_err(|err| ModemError::ReadError(format!("Error in serial.read(buf) ({:?})", err)))?;
+            response.push_str(&line);
-
+            if line.contains("ERROR") || line.contains(&match_text) {
-            response.push_str(std::str::from_utf8(&buf[0..num_bytes])
+                println!("Found match {} for line {} ; exiting response reader now ...", match_text, line);
-                .map_err(|err| ModemError::ReadError(format!("Error in str::from_utf8 ({:?})", err)))?);
+                println!("-----------------------------------------------------------");
-
+                break Ok(response.to_string())
-            if num_bytes < buf.len() {
-                break
             }
         }
+    }
 
-        print!("Read {} bytes from serial: {}", response.len(), response);
+    #[inline(always)]
-        if let Some(c) = contains {
+    fn send_bytes(&mut self, payload: &[u8], eos: char) -> Result<()> {
-            if response.contains(&c) {
+        for b in payload.iter() {
-                Ok(response)
+            nb::block!(self.tx.write(*b))
-            } else {
+                .map_err(|_| ModemError::CommandError(format!("error writing {} to serial", b)))?;
-                Err(ModemError::CommandError(format!("Didn't get expected ({}) from modem. Got: {}", c, response)))
-            }
-        } else {
-            Ok(response)
         }
+        nb::block!(self.tx.write(eos as u8))
+            .map_err(|_| ModemError::CommandError(format!("error writing {} to serial", eos)))?;
+        Ok(())
     }
 
     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> {
-        self.serial.clear();
         println!("-----------------------------------------------------------");
-        println!("Sending {} ...", at_command);
+        println!("Sending {} ...", cmd.text);
-
+        let _ = self.send_bytes(cmd.text.as_bytes(), '\r')?;
-        let _ = nb::block!(self.serial
+        self.read_response(cmd.contains, cmd.timeout)
-            .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<()> {
+    fn send_data(&mut self, buf: &[u8]) -> Result<usize> {
-        let mut prompt_buf = vec![0; 256];
+        self.rx.clear();
-        let prompt_len = self.serial.read(&mut prompt_buf)
+        let _ = self.send_bytes("AT+CIPSEND".as_bytes(), '\r')?;
-            .map_err(|err| ModemError::ReadError(format!("Error in handle_prompt() ({:?})", err)))?;
+        let send_request: String = self.rx.reset(Duration::from_millis(3000))
+            .map(char::from)
+            .take_while(|c| *c != '>').collect();
 
-        let prompt = String::from_utf8(prompt_buf[0..prompt_len].to_vec())
+        if send_request == "" {
-            .unwrap_or("".to_string())
+            return Err(ModemError::SendDataError);
-            .trim()
-            .to_string();
-
-        println!("Prompt is: ({})", prompt);
-
-        if prompt != ">" {
-            let msg = format!("Prompt error, expected (>), got ({})", prompt);
-            Err(ModemError::SendDataError(msg))
-        } else {
-            Ok(())
         }
+
+        self.send_bytes(buf, 26 as char)?; // 26_u8 = Ctrl+z - to end sending data
+        let _ = self.read_response(Some("SEND OK".to_string()), Duration::from_millis(1000))?;
+        Ok(buf.len())
     }
 
-    fn tcp_manual_send_data(&mut self, buf: &[u8]) -> Result<String> {
+    pub fn get_ip_addr(&mut self) -> Result<String> {
-        println!("Sending AT+CIPSEND to serial TX!");
+        self.send_command(Command::getbear())
-        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! ({:?})", buf.into_iter().map(|b| char::from(*b)).collect::<String>());
-        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> {
+    pub fn connect_to_gprs_ap(&mut self, apn: &str, username: &str, password: &str)-> Result<()> {
-        self.send_command(Command::gprs_bearer_status())
-    }
-
-    pub fn gprs_attach_ap(&mut self, config: crate::config::GprsAp)-> Result<()> {
         println!("init gprs ...");
         let _ = self.send_command(Command::gprs_init())?;
 
-        println!("setting up gprs credentials for apn {}, {}:{})", config.apn, config.username, config.password);
+        println!("setting up gprs credentials for apn {}, {}:{})", apn, username, password);
 
-        let _ = self.send_command(Command::gprs_set_apn(config.apn))?;
+        let _ = self.send_command(Command::gprs_set_apn(apn))?;
-        let _ = self.send_command(Command::gprs_set_user(config.username))?;
+        let _ = self.send_command(Command::gprs_set_user(username))?;
-        let _ = self.send_command(Command::gprs_set_pwd(config.password))?;
+        let _ = self.send_command(Command::gprs_set_pwd(password))?;
+
+        println!("open gprs ...");
+        let _ = self.send_command(Command::gprs_open())?;
 
         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;
-        println!("TRYING TO CONNECT TO GPRS");
-        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))
+        let _ = 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 mut reply_result = self.send(&at_command, "CONNECT OK");
-        for _ in 0..3 {
-            if let Ok(reply) = reply_result {
-                println!("TCP connect replied with {}", reply);
-                break
-            } else {
-                reply_result = self.command_read_response(Some("CONNECT OK".to_string()));
-            }
-            thread::sleep(Duration::from_millis(1000));
-        }
         Ok(())
     }
 
-    pub fn tls_connect(&mut self, addr: &str, port: u16) -> Result<()> {
+    pub fn tcp_ssl_enable(&mut self) -> Result<()> {
-        let _ = self.tcp_connect(addr, port)?;
+        let _ = self.send_command(Command::tcp_ssl_set(true))?;
-
+        Ok(())
-        // ------------------------
+    }
-        // TLS handshake goes here.
-        // ------------------------
 
+    pub fn tcp_connect(&mut self, addr: &str, port: u16) -> Result<()> {
+        self.send_command(Command::tcp_connect(addr, port))?;
         Ok(())
     }
 
     pub fn tcp_set_quick_mode(&mut self, mode: bool) -> Result<()> {
-        self.send_command(Command::tcp_set_quick_mode(mode))
+        self.send_command(Command::tcp_set_quick_mode(mode))?;
-            .map(|_| ())
+        Ok(())
     }
 
     pub fn tcp_set_manual_receive(&mut self, is_manual: bool) -> Result<()> {
-        self.send_command(Command::tcp_set_manual_receive(is_manual))
+        self.send_command(Command::tcp_set_manual_receive(is_manual))?;
-            .map(|_| ())
+        Ok(())
     }
 
-    pub fn tcp_manual_send(&mut self, buf: &[u8]) -> Result<()> {
+    pub fn tcp_send(&mut self, buf: &[u8]) -> Result<usize> {
-        thread::sleep(Duration::from_millis(200));
+        self.send_data(buf)
-        // self.serial.clear();
-        self.tcp_manual_send_data(buf)
-            .map(|_| ())
     }
 
     fn tcp_parse_response_size(&mut self, reply_line: &str) -> Result<usize> {
@@ -331,249 +269,117 @@ impl<UART: serial::Uart, PWK: OutputPin, RST: OutputPin, PW: OutputPin> Modem<UA
 
     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!");
+        reply.lines()
-        let res = reply.lines()
             .filter(|line| line.contains("+CIPRXGET: 4"))
             .next()
-            .ok_or(ModemError::CommandError("reply body missing :/".to_string()))
+            .ok_or(ModemError::CommandError("reply not found :/".to_string()))
-            .and_then(|line| self.tcp_parse_response_size(line))
+            .map(|line| self.tcp_parse_response_size(line))
-            .map_err(|_| ModemError::CommandError(format!("received 0 elements from parsing")));
+            .unwrap_or(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()))
+            let reply_len: usize = self.send_command(Command::tcp_receive(MAX_TCP_MANUAL_REPLY_SIZE))
-                .map(|reply| {
+                .map(|reply: String| {
-                    // TODO: parse the response properly
+                    reply.lines()
-                    // 1. the first line is \r\n
+                        .fold(0, |acc, line| {
-                    // 2. next is the +CIPRXGET: 2,X,Y where X is the number of bytes read and Y is
+                            if !line.starts_with("\r") && !line.contains("+CIPRXGET: 2,") && !line.contains("OK") {
-                    //    the number of bytes left to be read
+                                acc += line.chars().enumerate().map(|(idx, c)| { buf[size + idx] = c as u8; }).count()
-                    // 3. immediately after this the payload is returned (with size X)
+                            }
-                    // 4. OK
+                            else {
-                    reply
+                                0
-                        .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 {
+            if reply_len == 0 {
-                Some(0) | None => {
+                break Ok(size)
-                    break Ok(size)
+            }
-                },
+            else {
-                Some(x) => {
+                size += reply_len;
-                    size += x;
+                continue
-                    continue
-                },
             }
         }
     }
 
     pub fn tcp_close_connection(&mut self) -> Result<()> {
-        self.send_command(Command::tcp_close()).map(|_| ())
+        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<()> {
+    pub fn tcp_is_connected(&mut self) -> Result<bool> {
-        self.send_command(Command::get_location())
+        let response = self.send_command(Command::tcp_connection_state())?;
-            .map(|_| ())
+        let state = response.lines().last().and_then(|line| line.split(",").last());
-    }
+        Ok(state.unwrap_or("CLOSED") == "CONNECTED")
-
-    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, username: &str, password: &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);
-        conn.set_user_name(Some(username.to_string()));
-        conn.set_password(Some(password.to_string()));
-        let _ = conn.encode(&mut buf)?;
-        let _ = self.tcp_manual_send(&mut buf)?;
-
-        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)?;
-        self.tcp_manual_send(&mut buf)?;
-        Ok(())
-    }
-
-    pub fn mqtt_send_position_loop(&mut self, host: &str, port: u16, username: &str, password: &str) -> anyhow::Result<()> {
-        if !self.is_gprs_attached()? {
-            let _ = self.gprs_attach_ap(crate::config::MTS)?;
-            let _ = self.try_connect_gprs()?;
-        }
-        // 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 _ = self.send("AT+CIPQSEND=0", "OK");
-        // Enables getting data from network manually.
-        let _ = self.send("AT+CIPRXGET=1", "OK");
-
-        for _ in 0..5 {
-            if let Ok(_) = self.tcp_connect(host, port) {
-                break
-            }
-        }
-
-        let device_id = "c36a72df-5bd6-4f9b-995d-059433bc3267";
-        println!("connecting to MQTT with ({}:{})", username, password);
-        let _ = self.mqtt_connect(device_id, username, password)?;
-
-        println!("entering queue receive loop ...");
-        let mut err_count = 0;
-        let _ = loop {
-            match self.receiver.recv() {
-                Ok(Msg::Gps(solution)) => {
-                    println!("received GPS solution {:?} | sending to mqtt ...", solution);
-                    serde_json_core::ser::to_string::<Solution, 512>(&solution)
-                        .map_err(|e| anyhow::Error::new(e))
-                        .and_then(|sol| self.mqtt_publish(device_id, &sol))?;
-                    err_count = 0;
-                },
-                Ok(Msg::Accelerometer(acc)) => {
-                    println!("received accel {} | sending to mqtt ...", acc);
-                    let _ = self.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
-                    }
-                }
-            }
-        };
-
-        Ok(())
     }
 }
 
-impl<UART: serial::Uart, PWK: OutputPin, RST: OutputPin, PW: OutputPin> std::io::Read for Modem<UART, PWK, RST, PW> {
+pub struct ModemTcpStack;
-    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 struct ModemSocket<T> {
+    state: SocketState<T>,
+}
+
+pub enum SocketState<UART: serial::Uart> {
+    Building,
+    Connected(Modem<UART>),
+}
+
+impl<T: serial::Uart> SocketState<T> {
+    fn new() -> Self {
+        Self::Building
+    }
+
+    fn get_running(&mut self) -> std::io::Result<&mut T> {
+        match self {
+            SocketState::Connected(ref mut s) => Ok(s),
+            _ => OutOfOrder.into(),
+        }
+    }
+}
+
+impl<T: serial::Uart> ModemSocket<T> {
+    fn new() -> Self {
+        Self {
+            state: SocketState::new(),
+        }
+    }
+
+    fn get_running(s: Modem<T>) -> Self {
+        Self {
+            state: SocketState::Connected(s)
+        }
+    }
+}
+
+impl<UART: serial::Uart> TcpClientStack for ModemTcpStack {
+    type Error = ModemError;
+    type TcpSocket = ModemSocket<UART>;
+
+    fn socket(&mut self) -> Result<Self::TcpSocket> {
+        Ok(self.modem)
+    }
+
+    fn connect(&mut self, socket: &mut Self::TcpSocket, remote: SocketAddr) ->  std::result::Result<(), nb::Error<Self::Error>> {
+        self.tcp_connect(&format!("{}", remote.ip()), remote.port())
+            .map_err(|err| nb::Error::Other(err))
+    }
+
+    fn is_connected(&mut self, _socket: &Self::TcpSocket) -> Result<bool> {
+        self.tcp_is_connected()
+    }
+
+    fn send(&mut self, _socket: &mut Self::TcpSocket, buffer: &[u8]) -> std::result::Result<usize, nb::Error<Self::Error>> {
+        self.tcp_send(buffer)
+            .map_err(|err| nb::Error::Other(err))
+    }
+
+    fn receive( &mut self, _socket: &mut Self::TcpSocket, buffer: &mut [u8]) -> std::result::Result<usize, nb::Error<Self::Error>> {
+        self.tcp_receive(buffer)
+            .map_err(|err| nb::Error::Other(err))
+    }
+
+    fn close(&mut self, _socket: Self::TcpSocket) -> Result<()> {
+        self.tcp_close_connection()
     }
 }

src/serial.rs

@@ -1,129 +0,0 @@
-use std::error::Error;
-use std::io;
-use std::thread;
-use std::time::Duration;
-use embedded_hal::serial::{Read, Write};
-use esp_idf_hal::serial::{self, Rx, Tx};
-
-const READ_MAX_RETRIES: usize = 5;
-const READ_WAIT_TIME: u64 = 10;
-
-#[derive(Debug)]
-pub enum SerialError {
-    ReadError(String),
-    WriteError(String),
-}
-
-impl Error for SerialError {}
-
-impl std::fmt::Display for SerialError {
-    fn fmt(&self, f: &mut std::fmt::Formatter) -> std::fmt::Result {
-        write!(f, "{:?}", self)
-    }
-}
-
-pub type Result<T> = nb::Result<T, SerialError>;
-
-pub struct SerialIO<UART: serial::Uart> {
-    pub rx: Rx<UART>,
-    pub tx: Tx<UART>,
-}
-
-impl<UART: serial::Uart> SerialIO<UART> {
-    pub fn new(tx: Tx<UART>, rx: Rx<UART>) -> Self {
-        Self { rx, tx }
-    }
-
-    pub fn write_bytes(&mut self, payload: &[u8]) -> Result<usize> {
-        let mut num_bytes = 0;
-        for b in payload.iter() {
-            self.tx.write(*b)
-                .map_err(|err| SerialError::WriteError(
-                        format!("Error writing in serial port ({:?})", err)))?;
-            num_bytes += 1;
-        }
-        if num_bytes == payload.len() {
-            Ok(num_bytes)
-        }
-        else {
-            Err(nb::Error::Other(
-                SerialError::WriteError(
-                    "Written bytes shorter than payload length (write_bytes)".to_string()
-                )
-            ))
-        }
-    }
-
-    pub fn read_bytes(&mut self, buf: &mut [u8]) -> Result<usize> {
-        let mut started_reading = false;
-        let mut count = 0;
-        let mut retries = 0;
-
-        loop {
-            match self.rx.read() {
-                Ok(b) => {
-                    started_reading = true;
-                    if count < buf.len() {
-                        buf[count] = b;
-                        count += 1;
-                    }
-                    else { break }
-                },
-                Err(nb::Error::WouldBlock) => {
-                    if started_reading || retries > READ_MAX_RETRIES { break }
-                    else {
-                        thread::sleep(Duration::from_millis(READ_WAIT_TIME));
-                        retries += 1;
-                    }
-                },
-                Err(nb::Error::Other(err)) => println!("Serial read error :: {:?}", err),
-            }
-        };
-        if count > 0 {
-            Ok(count)
-        } else {
-            Err(nb::Error::Other(SerialError::ReadError("Rx buffer empty.".to_string())))
-        }
-    }
-
-    pub fn clear(&mut self) {
-        let mut started_reading = false;
-        let mut retries = 0;
-
-        loop {
-            match self.rx.read() {
-                Ok(_) => {
-                    started_reading = true;
-                },
-                Err(nb::Error::WouldBlock) => {
-                    if started_reading || retries > READ_MAX_RETRIES { break; }
-                    else {
-                        thread::sleep(Duration::from_millis(READ_WAIT_TIME));
-                        retries += 1;
-                    }
-                },
-                Err(nb::Error::Other(err)) => println!("Serial read error :: {:?}", err),
-            }
-        }
-    }
-}
-
-impl<UART: serial::Uart> io::Read for SerialIO<UART> {
-    fn read(&mut self, buf: &mut [u8]) -> io::Result<usize> {
-        let count = nb::block!(self.read_bytes(buf))
-            .map_err(|err| io::Error::new(io::ErrorKind::Other, err))?;
-        Ok(count)
-    }
-}
-
-impl<UART: serial::Uart> io::Write for SerialIO<UART> {
-    fn write(&mut self, buf: &[u8]) -> io::Result<usize> {
-        nb::block!(self.write_bytes(buf))
-            .map_err(|err| io::Error::new(io::ErrorKind::Other, err))
-    }
-
-    fn flush(&mut self) -> io::Result<()> {
-        self.tx.flush()
-            .map_err(|err| io::Error::new(io::ErrorKind::Other, SerialError::ReadError(format!("Flush error ({:?})", err))))
-    }
-}

src/types.rs

@@ -1,106 +0,0 @@
-#![allow(dead_code)]
-use nmea0183::{
-    GGA,
-    GLL,
-    coords::{
-        Latitude as NMEALatitude,
-        Longitude as NMEALongitude,
-        Hemisphere as NMEAHemisphere,
-    },
-};
-use serde::Serialize;
-
-#[derive(Debug, Serialize)]
-pub enum Hemisphere {
-    North,
-    South,
-    East,
-    West,
-}
-
-#[derive(Debug, Serialize)]
-pub struct Latitude {
-    degrees: u8,
-    minutes: u8,
-    seconds: f32,
-    hemisphere: Hemisphere,
-}
-
-#[derive(Debug, Serialize)]
-pub struct Longitude {
-    degrees: u8,
-    minutes: u8,
-    seconds: f32,
-    hemisphere: Hemisphere,
-}
-
-impl From<NMEALatitude> for Latitude {
-    fn from(lat: NMEALatitude) -> Self {
-        Self {
-            degrees: lat.degrees,
-            minutes: lat.minutes,
-            seconds: lat.seconds,
-            hemisphere: lat.hemisphere.into(),
-        }
-    }
-}
-
-impl From<NMEALongitude> for Longitude {
-    fn from(lon: NMEALongitude) -> Self {
-        Self {
-            degrees: lon.degrees,
-            minutes: lon.minutes,
-            seconds: lon.seconds,
-            hemisphere: lon.hemisphere.into(),
-        }
-    }
-}
-
-impl From<NMEAHemisphere> for Hemisphere {
-    fn from(hem: NMEAHemisphere) -> Self {
-        match hem {
-            NMEAHemisphere::North => Self::North,
-            NMEAHemisphere::South => Self::South,
-            NMEAHemisphere::East => Self::East,
-            NMEAHemisphere::West => Self::West,
-        }
-    }
-}
-
-#[derive(Debug, Serialize)]
-pub struct Solution {
-    pub latitude: Latitude,
-    pub longitude: Longitude,
-    pub altitude: Option<f32>,
-    pub speed: Option<f32>,
-    pub direction: Option<f32>,
-}
-
-impl From<GGA> for Solution {
-    fn from(gga: GGA) -> Self {
-        Self {
-            latitude: gga.latitude.into(),
-            longitude: gga.longitude.into(),
-            altitude: Some(gga.altitude.meters),
-            speed: None,
-            direction: None,
-        }
-    }
-}
-
-impl From<GLL> for Solution {
-    fn from(gll: GLL) -> Self {
-        Self {
-            latitude: gll.latitude.into(),
-            longitude: gll.longitude.into(),
-            altitude: None,
-            speed: None,
-            direction: None,
-        }
-    }
-}
-
-pub enum Msg {
-    Gps(Solution),
-    Accelerometer(String),
-}