code.octet-stream.net Git - m17rt/blob - m17app/src/app.rs

   1 use crate::tnc::Tnc;
   2 use m17core::kiss::KissFrame;
   3 use m17core::protocol::PacketType;
   4 use m17core::traits::{PacketListener, StreamListener};
   5
   6 use log::debug;
   7 use std::collections::HashMap;
   8 use std::io::{Read, Write};
   9 use std::sync::mpsc;
  10 use std::sync::{Arc, RwLock};
  11
  12 pub struct M17App {
  13     listeners: Arc<RwLock<Listeners>>,
  14     event_tx: mpsc::SyncSender<TncControlEvent>,
  15 }
  16
  17 impl M17App {
  18     pub fn new<T: Tnc + Send + 'static>(mut tnc: T) -> Self {
  19         let write_tnc = tnc.try_clone().unwrap();
  20         let (event_tx, event_rx) = mpsc::sync_channel(128);
  21         let listeners = Arc::new(RwLock::new(Listeners::new()));
  22         spawn_reader(tnc, listeners.clone());
  23         spawn_writer(write_tnc, event_rx);
  24         Self {
  25             listeners,
  26             event_tx,
  27         }
  28     }
  29
  30     pub fn add_packet_listener<P: PacketListener + 'static>(&self, listener: P) -> usize {
  31         let mut listeners = self.listeners.write().unwrap();
  32         let id = listeners.next;
  33         listeners.next += 1;
  34         listeners.packet.insert(id, Box::new(listener));
  35         id
  36     }
  37
  38     pub fn add_stream_listener<S: StreamListener + 'static>(&self, listener: S) -> usize {
  39         let mut listeners = self.listeners.write().unwrap();
  40         let id = listeners.next;
  41         listeners.next += 1;
  42         listeners.stream.insert(id, Box::new(listener));
  43         id
  44     }
  45
  46     pub fn remove_packet_listener(&self, id: usize) {
  47         self.listeners.write().unwrap().packet.remove(&id);
  48     }
  49
  50     pub fn remove_stream_listener(&self, id: usize) {
  51         self.listeners.write().unwrap().stream.remove(&id);
  52     }
  53
  54     pub fn transmit_packet(&self, type_code: PacketType, payload: &[u8]) {
  55         // hang on where do we get the LSF details from? We need a destination obviously
  56         // our source address needs to be configured here too
  57         // also there is possible CAN, encryption, meta payload
  58
  59         // we will immediately convert this into a KISS payload before sending into channel so we only need borrow on data
  60     }
  61
  62     // add more methods here for stream outgoing
  63
  64     pub fn transmit_stream_start(&self /* lsf?, payload? what needs to be configured ?! */) {}
  65
  66     // as long as there is only one TNC it is implied there is only ever one stream transmission in flight
  67
  68     pub fn transmit_stream_next(&self, /* next payload,  */ end_of_stream: bool) {}
  69
  70     pub fn start(&self) {
  71         let _ = self.event_tx.send(TncControlEvent::Start);
  72     }
  73
  74     pub fn close(&self) {
  75         let _ = self.event_tx.send(TncControlEvent::Close);
  76     }
  77 }
  78
  79 /// Synchronised structure for listeners subscribing to packets and streams.
  80 ///
  81 /// Each listener will be notified in turn of each event.
  82 struct Listeners {
  83     /// Identifier to be assigned to the next listener, starting from 0
  84     next: usize,
  85     packet: HashMap<usize, Box<dyn PacketListener>>,
  86     stream: HashMap<usize, Box<dyn StreamListener>>,
  87 }
  88
  89 impl Listeners {
  90     fn new() -> Self {
  91         Self {
  92             next: 0,
  93             packet: HashMap::new(),
  94             stream: HashMap::new(),
  95         }
  96     }
  97 }
  98
  99 /// Carries a request from a method on M17App to the TNC's writer thread, which will execute it.
 100 enum TncControlEvent {
 101     Kiss(KissFrame),
 102     Start,
 103     Close,
 104 }
 105
 106 fn spawn_reader<T: Tnc + Send + 'static>(mut tnc: T, listeners: Arc<RwLock<Listeners>>) {
 107     std::thread::spawn(move || {
 108         let mut buf = [0u8; 1713];
 109         let mut n = 0;
 110         loop {
 111             // I want to call tnc.read() here
 112             // Probably these needs a helper in m17core::kiss? It will be common to both TNC and host
 113
 114             // After a read...
 115             // if this does not start with FEND, forget all data up until first FEND
 116             // if we start with a FEND, see if there is another FEND with at least one byte between
 117             // for each such case, turn that FEND..=FEND slice into a KissFrame and attempt to parse it
 118             // once all such pairs have been handled...
 119             // move the last FEND onwards back to the start of the buffer
 120             //   - if there is no room to do so, this is an oversize frame. purge everything and carry on.
 121             // perform next read from end
 122         }
 123     });
 124 }
 125
 126 fn spawn_writer<T: Tnc + Send + 'static>(mut tnc: T, event_rx: mpsc::Receiver<TncControlEvent>) {
 127     std::thread::spawn(move || {
 128         while let Ok(ev) = event_rx.recv() {
 129             match ev {
 130                 TncControlEvent::Kiss(k) => {
 131                     if let Err(e) = tnc.write_all(&k.as_bytes()) {
 132                         debug!("kiss send err: {:?}", e);
 133                         return;
 134                     }
 135                 }
 136                 TncControlEvent::Start => {
 137                     if let Err(e) = tnc.start() {
 138                         debug!("tnc start err: {:?}", e);
 139                         return;
 140                     }
 141                 }
 142                 TncControlEvent::Close => {
 143                     if let Err(e) = tnc.close() {
 144                         debug!("tnc close err: {:?}", e);
 145                         return;
 146                     }
 147                 }
 148             }
 149         }
 150     });
 151 }