Address some clippy lints

[m17rt] / m17app / src / app.rs

diff --git a/m17app/src/app.rs b/m17app/src/app.rs
index 9c2815de50c4fbe53cc6280ef2a47ebe2630094c..7d363dd1808522b986e06e687e39b82c3c8bd544 100644 (file)
--- a/m17app/src/app.rs
+++ b/m17app/src/app.rs
@@ -1,16 +1,16 @@
+use crate::adapter::{PacketAdapter, StreamAdapter};
+use crate::link_setup::LinkSetup;

 use crate::tnc::Tnc;
 use m17core::kiss::{KissBuffer, KissCommand, KissFrame};
 use crate::tnc::Tnc;
 use m17core::kiss::{KissBuffer, KissCommand, KissFrame};

-use m17core::protocol::PacketType;
-use m17core::traits::{PacketListener, StreamListener};
+use m17core::protocol::{EncryptionType, LsfFrame, PacketType, StreamFrame};

 
 use log::debug;
 use std::collections::HashMap;
 
 use log::debug;
 use std::collections::HashMap;

-use std::io::{Read, Write};

 use std::sync::mpsc;
 use std::sync::{Arc, RwLock};
 
 pub struct M17App {
 use std::sync::mpsc;
 use std::sync::{Arc, RwLock};
 
 pub struct M17App {

-    listeners: Arc<RwLock<Listeners>>,
+    adapters: Arc<RwLock<Adapters>>,

     event_tx: mpsc::SyncSender<TncControlEvent>,
 }
 
     event_tx: mpsc::SyncSender<TncControlEvent>,
 }
 


@@ -18,75 +18,111 @@ impl M17App {
     pub fn new<T: Tnc + Send + 'static>(mut tnc: T) -> Self {
         let write_tnc = tnc.try_clone().unwrap();
         let (event_tx, event_rx) = mpsc::sync_channel(128);
     pub fn new<T: Tnc + Send + 'static>(mut tnc: T) -> Self {
         let write_tnc = tnc.try_clone().unwrap();
         let (event_tx, event_rx) = mpsc::sync_channel(128);

-        let listeners = Arc::new(RwLock::new(Listeners::new()));
+        let listeners = Arc::new(RwLock::new(Adapters::new()));

         spawn_reader(tnc, listeners.clone());
         spawn_writer(write_tnc, event_rx);
         Self {
         spawn_reader(tnc, listeners.clone());
         spawn_writer(write_tnc, event_rx);
         Self {

-            listeners,
+            adapters: listeners,

             event_tx,
         }
     }
 
             event_tx,
         }
     }
 

-    pub fn add_packet_listener<P: PacketListener + 'static>(&self, listener: P) -> usize {
-        let mut listeners = self.listeners.write().unwrap();
-        let id = listeners.next;
-        listeners.next += 1;
-        listeners.packet.insert(id, Box::new(listener));
+    pub fn add_packet_adapter<P: PacketAdapter + 'static>(&self, adapter: P) -> usize {
+        let adapter = Arc::new(adapter);
+        let mut adapters = self.adapters.write().unwrap();
+        let id = adapters.next;
+        adapters.next += 1;
+        adapters.packet.insert(id, adapter.clone());
+        drop(adapters);
+        adapter.adapter_registered(id, self.tx());

         id
     }
 
         id
     }
 

-    pub fn add_stream_listener<S: StreamListener + 'static>(&self, listener: S) -> usize {
-        let mut listeners = self.listeners.write().unwrap();
-        let id = listeners.next;
-        listeners.next += 1;
-        listeners.stream.insert(id, Box::new(listener));
+    pub fn add_stream_adapter<S: StreamAdapter + 'static>(&self, adapter: S) -> usize {
+        let adapter = Arc::new(adapter);
+        let mut adapters = self.adapters.write().unwrap();
+        let id = adapters.next;
+        adapters.next += 1;
+        adapters.stream.insert(id, adapter.clone());
+        drop(adapters);
+        adapter.adapter_registered(id, self.tx());

         id
     }
 
         id
     }
 

-    pub fn remove_packet_listener(&self, id: usize) {
-        self.listeners.write().unwrap().packet.remove(&id);
+    pub fn remove_packet_adapter(&self, id: usize) {
+        if let Some(a) = self.adapters.write().unwrap().packet.remove(&id) {
+            a.adapter_removed();
+        }

     }
 
     }
 

-    pub fn remove_stream_listener(&self, id: usize) {
-        self.listeners.write().unwrap().stream.remove(&id);
+    pub fn remove_stream_adapter(&self, id: usize) {
+        if let Some(a) = self.adapters.write().unwrap().stream.remove(&id) {
+            a.adapter_removed();
+        }

     }
 
     }
 

-    pub fn transmit_packet(&self, type_code: PacketType, payload: &[u8]) {
-        // hang on where do we get the LSF details from? We need a destination obviously
-        // our source address needs to be configured here too
-        // also there is possible CAN, encryption, meta payload
-
-        // we will immediately convert this into a KISS payload before sending into channel so we only need borrow on data
+    /// Create a handle that can be used to transmit data on the TNC
+    pub fn tx(&self) -> TxHandle {
+        TxHandle {
+            event_tx: self.event_tx.clone(),
+        }

     }
 
     }
 

-    // add more methods here for stream outgoing
-
-    pub fn transmit_stream_start(&self /* lsf?, payload? what needs to be configured ?! */) {}
-
-    // as long as there is only one TNC it is implied there is only ever one stream transmission in flight
-
-    pub fn transmit_stream_next(&self, /* next payload,  */ end_of_stream: bool) {}
-

     pub fn start(&self) {
         let _ = self.event_tx.send(TncControlEvent::Start);
     }
 
     pub fn close(&self) {
     pub fn start(&self) {
         let _ = self.event_tx.send(TncControlEvent::Start);
     }
 
     pub fn close(&self) {

+        // TODO: blocking function to indicate TNC has finished closing
+        // then we could call this in a signal handler to ensure PTT is dropped before quit

         let _ = self.event_tx.send(TncControlEvent::Close);
     }
 }
 
         let _ = self.event_tx.send(TncControlEvent::Close);
     }
 }
 

+pub struct TxHandle {
+    event_tx: mpsc::SyncSender<TncControlEvent>,
+}
+
+impl TxHandle {
+    pub fn transmit_packet(&self, link_setup: &LinkSetup, packet_type: PacketType, payload: &[u8]) {
+        let (pack_type, pack_type_len) = packet_type.as_proto();
+        if pack_type_len + payload.len() > 823 {
+            // TODO: error for invalid transmission type
+            return;
+        }
+        let mut full_payload = vec![];
+        full_payload.extend_from_slice(&pack_type[0..pack_type_len]);
+        full_payload.extend_from_slice(&payload);
+        let crc = m17core::crc::m17_crc(&full_payload);
+        full_payload.extend_from_slice(&crc.to_be_bytes());
+        let kiss_frame = KissFrame::new_full_packet(&link_setup.raw.0, &full_payload).unwrap();
+        let _ = self.event_tx.send(TncControlEvent::Kiss(kiss_frame));
+    }
+
+    pub fn transmit_stream_start(&self, link_setup: &LinkSetup) {
+        let kiss_frame = KissFrame::new_stream_setup(&link_setup.raw.0).unwrap();
+        let _ = self.event_tx.send(TncControlEvent::Kiss(kiss_frame));
+    }
+
+    // as long as there is only one TNC it is implied there is only ever one stream transmission in flight
+
+    pub fn transmit_stream_next(&self, stream: &StreamFrame) {
+        let kiss_frame = KissFrame::new_stream_data(&stream).unwrap();
+        let _ = self.event_tx.send(TncControlEvent::Kiss(kiss_frame));
+    }
+}
+

 /// Synchronised structure for listeners subscribing to packets and streams.
 ///
 /// Each listener will be notified in turn of each event.
 /// Synchronised structure for listeners subscribing to packets and streams.
 ///
 /// Each listener will be notified in turn of each event.

-struct Listeners {
+struct Adapters {

     /// Identifier to be assigned to the next listener, starting from 0
     next: usize,
     /// Identifier to be assigned to the next listener, starting from 0
     next: usize,

-    packet: HashMap<usize, Box<dyn PacketListener>>,
-    stream: HashMap<usize, Box<dyn StreamListener>>,
+    packet: HashMap<usize, Arc<dyn PacketAdapter>>,
+    stream: HashMap<usize, Arc<dyn StreamAdapter>>,

 }
 
 }
 

-impl Listeners {
+impl Adapters {

     fn new() -> Self {
         Self {
             next: 0,
     fn new() -> Self {
         Self {
             next: 0,


@@ -103,9 +139,10 @@ enum TncControlEvent {
     Close,
 }
 
     Close,
 }
 

-fn spawn_reader<T: Tnc + Send + 'static>(mut tnc: T, listeners: Arc<RwLock<Listeners>>) {
+fn spawn_reader<T: Tnc>(mut tnc: T, adapters: Arc<RwLock<Adapters>>) {

     std::thread::spawn(move || {
         let mut kiss_buffer = KissBuffer::new();
     std::thread::spawn(move || {
         let mut kiss_buffer = KissBuffer::new();

+        let mut stream_running = false;

         loop {
             let mut buf = kiss_buffer.buf_remaining();
             let n = match tnc.read(&mut buf) {
         loop {
             let mut buf = kiss_buffer.buf_remaining();
             let n = match tnc.read(&mut buf) {


@@ -118,14 +155,98 @@ fn spawn_reader<T: Tnc + Send + 'static>(mut tnc: T, listeners: Arc<RwLock<Liste
                     continue;
                 }
                 match frame.port() {
                     continue;
                 }
                 match frame.port() {

-                    Ok(0) => {
-                        // handle basic frame and send it to subscribers
+                    Ok(m17core::kiss::PORT_PACKET_BASIC) => {
+                        // no action
+                        // we will handle the more full-featured version from from port 1

                     }
                     }

-                    Ok(1) => {
-                        // handle full frame and send it to subscribers - I guess they need to know the type, probably not the CRC
+                    Ok(m17core::kiss::PORT_PACKET_FULL) => {
+                        let mut payload = [0u8; 855]; // 30 byte LSF + 825 byte packet including CRC
+                        let Ok(n) = frame.decode_payload(&mut payload) else {
+                            debug!("failed to decode payload from KISS frame");
+                            continue;
+                        };
+                        if n < 33 {
+                            debug!("unusually short full packet frame");
+                            continue;
+                        }
+                        let lsf = LsfFrame(payload[0..30].try_into().unwrap());
+                        if lsf.check_crc() != 0 {
+                            debug!("LSF in full packet frame did not pass CRC");
+                            continue;
+                        }
+                        if lsf.encryption_type() != EncryptionType::None {
+                            debug!("we only understand None encryption for now - skipping packet");
+                            continue;
+                        }
+                        let Some((packet_type, type_len)) = PacketType::from_proto(&payload[30..n])
+                        else {
+                            debug!("failed to decode packet type");
+                            continue;
+                        };
+                        if (n - 30 - type_len) < 2 {
+                            debug!("packet payload too small to provide CRC");
+                            continue;
+                        }
+                        let packet_crc = m17core::crc::m17_crc(&payload[30..n]);
+                        if packet_crc != 0 {
+                            debug!("packet CRC does not pass");
+                            continue;
+                        }
+                        let packet_payload: Arc<[u8]> =
+                            Arc::from(&payload[(30 + type_len)..(n - 2)]);
+
+                        let subs: Vec<_> =
+                            adapters.read().unwrap().packet.values().cloned().collect();
+                        for s in subs {
+                            s.packet_received(
+                                LinkSetup::new_raw(lsf.clone()),
+                                packet_type.clone(),
+                                packet_payload.clone(),
+                            );
+                        }

                     }
                     }

-                    Ok(2) => {
-                        // handle stream and send it to subscribers
+                    Ok(m17core::kiss::PORT_STREAM) => {
+                        let mut payload = [0u8; 32];
+                        let Ok(n) = frame.decode_payload(&mut payload) else {
+                            debug!("failed to decode stream payload from KISS frame");
+                            continue;
+                        };
+                        if n == 30 {
+                            let lsf = LsfFrame(payload[0..30].try_into().unwrap());
+                            if lsf.check_crc() != 0 {
+                                debug!("initial LSF in stream did not pass CRC");
+                                continue;
+                            }
+                            stream_running = true;
+                            let subs: Vec<_> =
+                                adapters.read().unwrap().stream.values().cloned().collect();
+                            for s in subs {
+                                s.stream_began(LinkSetup::new_raw(lsf.clone()));
+                            }
+                        } else if n == 26 {
+                            if !stream_running {
+                                debug!("ignoring stream data as we didn't get a valid LSF first");
+                                continue;
+                            }
+                            // TODO: parse LICH and handle the different changing subvalues META could have
+                            if m17core::crc::m17_crc(&payload[6..n]) != 0 {
+                                debug!("stream data CRC mismatch");
+                                continue;
+                            }
+                            let mut frame_number = u16::from_be_bytes([payload[6], payload[7]]);
+                            let is_final = (frame_number & 0x8000) > 0;
+                            frame_number &= 0x7fff;
+                            let data: [u8; 16] = payload[8..24].try_into().unwrap();
+                            let data = Arc::new(data);
+                            if is_final {
+                                stream_running = false;
+                            }
+                            let subs: Vec<_> =
+                                adapters.read().unwrap().stream.values().cloned().collect();
+                            for s in subs {
+                                s.stream_data(frame_number, is_final, data.clone());
+                            }
+                        }

                     }
                     _ => (),
                 }
                     }
                     _ => (),
                 }


@@ -134,7 +255,7 @@ fn spawn_reader<T: Tnc + Send + 'static>(mut tnc: T, listeners: Arc<RwLock<Liste
     });
 }
 
     });
 }
 

-fn spawn_writer<T: Tnc + Send + 'static>(mut tnc: T, event_rx: mpsc::Receiver<TncControlEvent>) {
+fn spawn_writer<T: Tnc>(mut tnc: T, event_rx: mpsc::Receiver<TncControlEvent>) {

     std::thread::spawn(move || {
         while let Ok(ev) = event_rx.recv() {
             match ev {
     std::thread::spawn(move || {
         while let Ok(ev) = event_rx.recv() {
             match ev {