Handle reflector packet traffic

[m17rt] / m17app / src / reflector.rs

diff --git a/m17app/src/reflector.rs b/m17app/src/reflector.rs
index 419093eb7b0dc2b41f77d4385315e4761fc58e3c..e7737681578d123d8984cf78defc50799544df8e 100644 (file)
--- a/m17app/src/reflector.rs
+++ b/m17app/src/reflector.rs
@@ -2,9 +2,9 @@ use std::{
     io::{self, Read, Write},
     net::{Ipv4Addr, Ipv6Addr, SocketAddr, ToSocketAddrs, UdpSocket},
     sync::{
+        Arc, Mutex,
         atomic::{AtomicBool, Ordering},
         mpsc::{self, Receiver, Sender},
-        Arc, Mutex,
     },
     thread,
     time::Duration,
@@ -12,20 +12,20 @@ use std::{
 
 use crate::{link_setup::M17Address, tnc::Tnc, util::out_buffer::OutBuffer};
 use m17core::{
-    kiss::{KissBuffer, KissCommand, KissFrame, PORT_STREAM},
+    kiss::{KissBuffer, KissCommand, KissFrame, PORT_PACKET_BASIC, PORT_PACKET_FULL, PORT_STREAM},
     protocol::{LsfFrame, StreamFrame},
     reflector::{
         convert::{RfToVoice, VoiceToRf},
-        packet::{Connect, Pong, ServerMessage, Voice},
+        packet::{Connect, Packet, Pong, ServerMessage, Voice},
     },
 };
 
 #[derive(Debug, PartialEq, Eq, Clone)]
 pub struct ReflectorClientConfig {
-    hostname: String,
-    port: u16,
-    module: char,
-    local_callsign: M17Address,
+    pub hostname: String,
+    pub port: u16,
+    pub module: char,
+    pub local_callsign: M17Address,
 }
 
 type WrappedStatusHandler = Arc<Mutex<dyn StatusHandler + Send + 'static>>;
@@ -79,9 +79,10 @@ impl Write for ReflectorClientTnc {
     fn write(&mut self, buf: &[u8]) -> std::io::Result<usize> {
         let mut kiss = self.kiss_buffer.lock().unwrap();
         let rem = kiss.buf_remaining();
-        let sz = buf.len().max(rem.len());
+        let sz = buf.len().min(rem.len());
         rem[0..sz].copy_from_slice(&buf[0..sz]);
-        if let Some(frame) = kiss.next_frame() {
+        kiss.did_write(sz);
+        while let Some(frame) = kiss.next_frame() {
             if Ok(KissCommand::DataFrame) == frame.command() && frame.port() == Ok(PORT_STREAM) {
                 let mut payload = [0u8; 30];
                 if let Ok(len) = frame.decode_payload(&mut payload) {
@@ -110,6 +111,28 @@ impl Write for ReflectorClientTnc {
                         }
                     }
                 };
+            } else if Ok(KissCommand::DataFrame) == frame.command()
+                && frame.port() == Ok(PORT_PACKET_BASIC)
+            {
+                // basic packets not supported for now, they will require more config
+            } else if Ok(KissCommand::DataFrame) == frame.command()
+                && frame.port() == Ok(PORT_PACKET_FULL)
+            {
+                let mut payload = [0u8; 855];
+                let Ok(len) = frame.decode_payload(&mut payload) else {
+                    continue;
+                };
+                if len < 33 {
+                    continue;
+                }
+                let mut lsf = LsfFrame([0u8; 30]);
+                lsf.0.copy_from_slice(&payload[0..30]);
+                if lsf.check_crc() != 0 {
+                    continue;
+                }
+                let mut packet = Packet::new();
+                packet.set_link_setup_frame(&lsf);
+                packet.set_payload(&payload[30..]);
             }
         }
         Ok(sz)
@@ -196,6 +219,8 @@ fn spawn_runner(
                     config.clone(),
                     status.clone(),
                 );
+                // Cool off a bit if connect rejected, etc.
+                thread::sleep(Duration::from_secs(10));
             }
         }
         status.lock().unwrap().status_changed(TncStatus::Closed);
@@ -217,7 +242,7 @@ fn run_single_conn(
     };
 
     let mut connect = Connect::new();
-    connect.set_address(config.local_callsign.address().to_owned());
+    connect.set_address(config.local_callsign.address());
     connect.set_module(config.module);
     let _ = socket.send_to(connect.as_bytes(), dest);
     let mut converter = VoiceToRf::new();
@@ -261,14 +286,16 @@ fn run_single_conn(
                     let kiss = KissFrame::new_stream_data(&stream).unwrap();
                     let _ = kiss_out_tx.send(kiss.as_bytes().into());
                 }
+                ServerMessage::Packet(packet) => {
+                    if let Ok(kiss) =
+                        KissFrame::new_full_packet(&packet.link_setup_frame().0, packet.payload())
+                    {
+                        let _ = kiss_out_tx.send(kiss.as_bytes().into());
+                    }
+                }
                 ServerMessage::Ping(_ping) => {
                     let mut pong = Pong::new();
-                    pong.set_address(
-                        M17Address::from_callsign("VK7XT")
-                            .unwrap()
-                            .address()
-                            .clone(),
-                    );
+                    pong.set_address(config.local_callsign.address());
                     if socket.send_to(pong.as_bytes(), dest).is_err() {
                         break;
                     }