Handle reflector packet traffic

[m17rt] / m17app / src / reflector.rs

diff --git a/m17app/src/reflector.rs b/m17app/src/reflector.rs
index 9735298131bd510132fb21b41bd7311d2e8f0a22..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::{
     io::{self, Read, Write},
     net::{Ipv4Addr, Ipv6Addr, SocketAddr, ToSocketAddrs, UdpSocket},
     sync::{

+        Arc, Mutex,

         atomic::{AtomicBool, Ordering},
         mpsc::{self, Receiver, Sender},
         atomic::{AtomicBool, Ordering},
         mpsc::{self, Receiver, Sender},

-        Arc, Mutex,

     },
     thread,
     time::Duration,
     },
     thread,
     time::Duration,


@@ -12,19 +12,20 @@ use std::{
 
 use crate::{link_setup::M17Address, tnc::Tnc, util::out_buffer::OutBuffer};
 use m17core::{
 
 use crate::{link_setup::M17Address, tnc::Tnc, util::out_buffer::OutBuffer};
 use m17core::{

-    kiss::KissFrame,
+    kiss::{KissBuffer, KissCommand, KissFrame, PORT_PACKET_BASIC, PORT_PACKET_FULL, PORT_STREAM},
+    protocol::{LsfFrame, StreamFrame},

     reflector::{
     reflector::{

-        convert::VoiceToRf,
-        packet::{Connect, Pong, ServerMessage},
+        convert::{RfToVoice, VoiceToRf},
+        packet::{Connect, Packet, Pong, ServerMessage, Voice},

     },
 };
 
 #[derive(Debug, PartialEq, Eq, Clone)]
 pub struct ReflectorClientConfig {
     },
 };
 
 #[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>>;
 }
 
 type WrappedStatusHandler = Arc<Mutex<dyn StatusHandler + Send + 'static>>;


@@ -40,6 +41,8 @@ pub struct ReflectorClientTnc {
     kiss_out: OutBuffer,
     event_tx: Arc<Mutex<Option<Sender<TncEvent>>>>,
     is_closed: Arc<AtomicBool>,
     kiss_out: OutBuffer,
     event_tx: Arc<Mutex<Option<Sender<TncEvent>>>>,
     is_closed: Arc<AtomicBool>,

+    kiss_buffer: Arc<Mutex<KissBuffer>>,
+    rf_to_voice: Arc<Mutex<Option<RfToVoice>>>,

 }
 
 impl ReflectorClientTnc {
 }
 
 impl ReflectorClientTnc {


@@ -60,6 +63,8 @@ impl ReflectorClientTnc {
             kiss_out: OutBuffer::new(rx),
             event_tx: Arc::new(Mutex::new(None)),
             is_closed: Arc::new(AtomicBool::new(false)),
             kiss_out: OutBuffer::new(rx),
             event_tx: Arc::new(Mutex::new(None)),
             is_closed: Arc::new(AtomicBool::new(false)),

+            kiss_buffer: Arc::new(Mutex::new(KissBuffer::new())),
+            rf_to_voice: Arc::new(Mutex::new(None)),

         }
     }
 }
         }
     }
 }


@@ -72,7 +77,65 @@ impl Read for ReflectorClientTnc {
 
 impl Write for ReflectorClientTnc {
     fn write(&mut self, buf: &[u8]) -> std::io::Result<usize> {
 
 impl Write for ReflectorClientTnc {
     fn write(&mut self, buf: &[u8]) -> std::io::Result<usize> {

-        Ok(buf.len())
+        let mut kiss = self.kiss_buffer.lock().unwrap();
+        let rem = kiss.buf_remaining();
+        let sz = buf.len().min(rem.len());
+        rem[0..sz].copy_from_slice(&buf[0..sz]);
+        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) {
+                    if len == 30 {
+                        let lsf = LsfFrame(payload);
+                        let mut to_voice = self.rf_to_voice.lock().unwrap();
+                        match &mut *to_voice {
+                            Some(to_voice) => to_voice.process_lsf(lsf),
+                            None => *to_voice = Some(RfToVoice::new(lsf)),
+                        }
+                    } else if len == 26 {
+                        let frame_num_part = u16::from_be_bytes([payload[6], payload[7]]);
+                        let frame = StreamFrame {
+                            lich_idx: payload[5] >> 5,
+                            lich_part: payload[0..5].try_into().unwrap(),
+                            frame_number: frame_num_part & 0x7fff,
+                            end_of_stream: frame_num_part & 0x8000 > 0,
+                            stream_data: payload[8..24].try_into().unwrap(),
+                        };
+                        let to_voice = self.rf_to_voice.lock().unwrap();
+                        if let Some(to_voice) = &*to_voice {
+                            let voice = to_voice.process_stream(&frame);
+                            if let Some(tx) = self.event_tx.lock().unwrap().as_ref() {
+                                let _ = tx.send(TncEvent::TransmitVoice(voice));
+                            }
+                        }
+                    }
+                };
+            } 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)

     }
 
     fn flush(&mut self) -> std::io::Result<()> {
     }
 
     fn flush(&mut self) -> std::io::Result<()> {


@@ -107,6 +170,7 @@ impl Tnc for ReflectorClientTnc {
 enum TncEvent {
     Close,
     Received(ServerMessage),
 enum TncEvent {
     Close,
     Received(ServerMessage),

+    TransmitVoice(Voice),

 }
 
 fn spawn_runner(
 }
 
 fn spawn_runner(


@@ -155,8 +219,9 @@ fn spawn_runner(
                     config.clone(),
                     status.clone(),
                 );
                     config.clone(),
                     status.clone(),
                 );

+                // Cool off a bit if connect rejected, etc.
+                thread::sleep(Duration::from_secs(10));

             }
             }

-            println!("single conn ended");

         }
         status.lock().unwrap().status_changed(TncStatus::Closed);
     });
         }
         status.lock().unwrap().status_changed(TncStatus::Closed);
     });


@@ -177,9 +242,9 @@ fn run_single_conn(
     };
 
     let mut connect = Connect::new();
     };
 
     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);
     connect.set_module(config.module);

-    socket.send_to(connect.as_bytes(), dest).unwrap();
+    let _ = socket.send_to(connect.as_bytes(), dest);

     let mut converter = VoiceToRf::new();
     let single_conn_ended = Arc::new(AtomicBool::new(false));
     // TODO: unwrap
     let mut converter = VoiceToRf::new();
     let single_conn_ended = Arc::new(AtomicBool::new(false));
     // TODO: unwrap


@@ -192,7 +257,6 @@ fn run_single_conn(
     while let Ok(ev) = event_rx.recv_timeout(Duration::from_secs(30)) {
         match ev {
             TncEvent::Close => {
     while let Ok(ev) = event_rx.recv_timeout(Duration::from_secs(30)) {
         match ev {
             TncEvent::Close => {

-                println!("writer: close");

                 break;
             }
             TncEvent::Received(server_msg) => match server_msg {
                 break;
             }
             TncEvent::Received(server_msg) => match server_msg {


@@ -222,18 +286,27 @@ fn run_single_conn(
                     let kiss = KissFrame::new_stream_data(&stream).unwrap();
                     let _ = kiss_out_tx.send(kiss.as_bytes().into());
                 }
                     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();
                 ServerMessage::Ping(_ping) => {
                     let mut pong = Pong::new();

-                    pong.set_address(
-                        M17Address::from_callsign("VK7XT")
-                            .unwrap()
-                            .address()
-                            .clone(),
-                    );
-                    socket.send_to(pong.as_bytes(), dest).unwrap();
+                    pong.set_address(config.local_callsign.address());
+                    if socket.send_to(pong.as_bytes(), dest).is_err() {
+                        break;
+                    }

                 }
                 _ => {}
             },
                 }
                 _ => {}
             },

+            TncEvent::TransmitVoice(voice) => {
+                if socket.send_to(voice.as_bytes(), dest).is_err() {
+                    break;
+                };
+            }

         }
     }
     single_conn_ended.store(true, Ordering::Release);
         }
     }
     single_conn_ended.store(true, Ordering::Release);


@@ -241,7 +314,6 @@ fn run_single_conn(
         .lock()
         .unwrap()
         .status_changed(TncStatus::Disconnected);
         .lock()
         .unwrap()
         .status_changed(TncStatus::Disconnected);

-    println!("write thread terminating");

 }
 
 fn spawn_reader(socket: UdpSocket, event_tx: Sender<TncEvent>, cancel: Arc<AtomicBool>) {
 }
 
 fn spawn_reader(socket: UdpSocket, event_tx: Sender<TncEvent>, cancel: Arc<AtomicBool>) {


@@ -257,7 +329,6 @@ fn spawn_reader(socket: UdpSocket, event_tx: Sender<TncEvent>, cancel: Arc<Atomi
                 }
             }
         }
                 }
             }
         }

-        println!("read thread terminating");

     });
 }
 
     });
 }