TCP support working

satrs-core/src/encoding/ccsds.rs

@@ -30,6 +30,12 @@ impl PacketIdLookup for [u16] {
     }
 }
 
+impl PacketIdLookup for &[u16] {
+    fn validate(&self, packet_id: u16) -> bool {
+        self.binary_search(&packet_id).is_ok()
+    }
+}
+
 #[cfg(feature = "alloc")]
 impl PacketIdLookup for Vec<PacketId> {
     fn validate(&self, packet_id: u16) -> bool {
@@ -49,6 +55,12 @@ impl PacketIdLookup for [PacketId] {
     }
 }
 
+impl PacketIdLookup for &[PacketId] {
+    fn validate(&self, packet_id: u16) -> bool {
+        self.binary_search(&PacketId::from(packet_id)).is_ok()
+    }
+}
+
 /// This function parses a given buffer for tightly packed CCSDS space packets. It uses the
 /// [PacketId] field of the CCSDS packets to detect the start of a CCSDS space packet and then
 /// uses the length field of the packet to extract CCSDS packets.

satrs-core/src/hal/std/tcp_cobs_server.rs

@@ -130,7 +130,7 @@ impl<TmError: 'static, TcError: 'static> TcpTmtcInCobsServer<TmError, TcError> {
         cfg: ServerConfig,
         tm_source: Box<dyn TmPacketSource<Error = TmError>>,
         tc_receiver: Box<dyn ReceivesTc<Error = TcError>>,
-    ) -> Result<Self, TcpTmtcError<TmError, TcError>> {
+    ) -> Result<Self, std::io::Error> {
         Ok(Self {
             generic_server: TcpTmtcGenericServer::new(
                 cfg,

satrs-core/src/hal/std/tcp_spacepackets_server.rs

@@ -113,7 +113,7 @@ impl<TmError: 'static, TcError: 'static> TcpSpacepacketsServer<TmError, TcError>
         tm_source: Box<dyn TmPacketSource<Error = TmError>>,
         tc_receiver: Box<dyn ReceivesTc<Error = TcError>>,
         packet_id_lookup: Box<dyn PacketIdLookup + Send>,
-    ) -> Result<Self, TcpTmtcError<TmError, TcError>> {
+    ) -> Result<Self, std::io::Error> {
         Ok(Self {
             generic_server: TcpTmtcGenericServer::new(
                 cfg,

satrs-example/Cargo.toml

@@ -19,8 +19,8 @@ num_enum = "0.7"
 thiserror = "1"
 
 [dependencies.satrs-core]
-version = "0.1.0-alpha.1"
+# version = "0.1.0-alpha.1"
-# path = "../satrs-core"
+path = "../satrs-core"
 
 [dependencies.satrs-mib]
 version = "0.1.0-alpha.1"

satrs-example/pyclient/common.py

@@ -4,7 +4,11 @@ import dataclasses
 import enum
 import struct
 
+from spacepackets.ecss.tc import PacketId, PacketType
+
 EXAMPLE_PUS_APID = 0x02
+EXAMPLE_PUS_PACKET_ID_TM = PacketId(PacketType.TM, True, EXAMPLE_PUS_APID)
+TM_PACKET_IDS = [EXAMPLE_PUS_PACKET_ID_TM]
 
 
 class EventSeverity(enum.IntEnum):

satrs-example/pyclient/main.py

@@ -45,7 +45,7 @@ from tmtccmd.util.obj_id import ObjectIdDictT
 
 import pus_tc
 import tc_definitions
-from common import EXAMPLE_PUS_APID, EventU32
+from common import EXAMPLE_PUS_APID, TM_PACKET_IDS, EventU32
 
 _LOGGER = logging.getLogger()
 
@@ -63,7 +63,7 @@ class SatRsConfigHook(HookBase):
         cfg = create_com_interface_cfg_default(
             com_if_key=com_if_key,
             json_cfg_path=self.cfg_path,
-            space_packet_ids=None,
+            space_packet_ids=TM_PACKET_IDS,
         )
         return create_com_interface_default(cfg)
 

satrs-example/pyclient/tmtc_conf.json

@@ -1,6 +1,8 @@
 {
-    "com_if": "udp",
+    "com_if": "tcp",
     "tcpip_udp_ip_addr": "127.0.0.1",
     "tcpip_udp_port": 7301,
-    "tcpip_udp_recv_max_size": 1500
+    "tcpip_udp_recv_max_size": 1500,
-}
+    "tcpip_tcp_ip_addr": "127.0.0.1",
+    "tcpip_tcp_port": 7301
+}

satrs-example/src/ccsds.rs

@@ -3,6 +3,7 @@ use satrs_core::spacepackets::{CcsdsPacket, SpHeader};
 use satrs_core::tmtc::{CcsdsPacketHandler, ReceivesCcsdsTc};
 use satrs_example::PUS_APID;
 
+#[derive(Clone)]
 pub struct CcsdsReceiver {
     pub tc_source: PusTcSource,
 }

satrs-example/src/main.rs

@@ -3,10 +3,12 @@ mod hk;
 mod logging;
 mod pus;
 mod requests;
+mod tcp;
 mod tmtc;
 mod udp;
 
 use log::{info, warn};
+use satrs_core::hal::std::tcp_server::ServerConfig;
 use satrs_core::hal::std::udp_server::UdpTcServer;
 
 use crate::ccsds::CcsdsReceiver;
@@ -19,6 +21,7 @@ use crate::pus::scheduler::Pus11Wrapper;
 use crate::pus::test::Service17CustomWrapper;
 use crate::pus::{PusReceiver, PusTcMpscRouter};
 use crate::requests::{Request, RequestWithToken};
+use crate::tcp::{SyncTcpTmSource, TcpTask};
 use crate::tmtc::{PusTcSource, TcArgs, TcStore, TmArgs, TmFunnel, TmtcTask};
 use crate::udp::UdpTmtcServer;
 use satrs_core::event_man::{
@@ -143,7 +146,7 @@ fn main() {
     let tm_args = TmArgs {
         tm_store: shared_tm_store.clone(),
         tm_sink_sender: tm_funnel_tx.clone(),
-        tm_server_rx,
+        tm_udp_server_rx: tm_server_rx,
     };
 
     let aocs_tm_funnel = tm_funnel_tx.clone();
@@ -275,22 +278,45 @@ fn main() {
     };
     let mut tmtc_task = TmtcTask::new(tc_args, PusReceiver::new(verif_reporter, pus_router));
 
-    let ccsds_distributor = CcsdsDistributor::new(Box::new(ccsds_receiver));
+    let udp_ccsds_distributor = CcsdsDistributor::new(Box::new(ccsds_receiver.clone()));
-    let udp_tc_server = UdpTcServer::new(sock_addr, 2048, Box::new(ccsds_distributor))
+    let udp_tc_server = UdpTcServer::new(sock_addr, 2048, Box::new(udp_ccsds_distributor))
         .expect("creating UDP TMTC server failed");
     let mut udp_tmtc_server = UdpTmtcServer {
         udp_tc_server,
-        tm_rx: tm_args.tm_server_rx,
+        tm_rx: tm_args.tm_udp_server_rx,
         tm_store: tm_args.tm_store.clone_backing_pool(),
     };
 
     info!("Starting TMTC and UDP task");
-    let jh0 = thread::Builder::new()
+    let jh_udp_tmtc = thread::Builder::new()
-        .name("TMTC_UDP".to_string())
+        .name("TMTC and UDP".to_string())
-        .spawn(move || loop {
+        .spawn(move || {
-            udp_tmtc_server.periodic_operation();
+            info!("Running UDP server on port {SERVER_PORT}");
-            tmtc_task.periodic_operation();
+            loop {
-            thread::sleep(Duration::from_millis(400));
+                udp_tmtc_server.periodic_operation();
+                tmtc_task.periodic_operation();
+                thread::sleep(Duration::from_millis(400));
+            }
+        })
+        .unwrap();
+
+    let tcp_ccsds_distributor = CcsdsDistributor::new(Box::new(ccsds_receiver));
+    let tcp_server_cfg = ServerConfig::new(sock_addr, Duration::from_millis(400), 4096, 8192);
+    let mut sync_tm_tcp_source = SyncTcpTmSource::new(200);
+    let mut tcp_server = TcpTask::new(
+        tcp_server_cfg,
+        sync_tm_tcp_source.clone(),
+        tcp_ccsds_distributor,
+    )
+    .expect("tcp server creation failed");
+    info!("Starting TCP task");
+    let jh_tcp = thread::Builder::new()
+        .name("TCP".to_string())
+        .spawn(move || {
+            info!("Running TCP server on port {SERVER_PORT}");
+            loop {
+                tcp_server.periodic_operation();
+            }
         })
         .unwrap();
 
@@ -331,6 +357,7 @@ fn main() {
                         .tm_server_tx
                         .send(addr)
                         .expect("Sending TM to server failed");
+                    sync_tm_tcp_source.add_tm(tm_raw);
                 }
             }
         })
@@ -502,7 +529,12 @@ fn main() {
             thread::sleep(Duration::from_millis(200));
         })
         .unwrap();
-    jh0.join().expect("Joining UDP TMTC server thread failed");
+    jh_udp_tmtc
+        .join()
+        .expect("Joining UDP TMTC server thread failed");
+    jh_tcp
+        .join()
+        .expect("Joining TCP TMTC server thread failed");
     jh1.join().expect("Joining TM Funnel thread failed");
     jh2.join().expect("Joining Event Manager thread failed");
     jh3.join().expect("Joining AOCS thread failed");

satrs-example/src/tcp.rs

@@ -0,0 +1,110 @@
+use std::{
+    collections::VecDeque,
+    sync::{Arc, Mutex},
+};
+
+use log::{info, warn};
+use satrs_core::{
+    hal::std::tcp_server::{ServerConfig, TcpSpacepacketsServer},
+    spacepackets::PacketId,
+    tmtc::{CcsdsDistributor, CcsdsError, TmPacketSourceCore},
+};
+use satrs_example::PUS_APID;
+
+use crate::tmtc::MpscStoreAndSendError;
+
+pub const PACKET_ID_LOOKUP: &[PacketId] = &[PacketId::const_tc(true, PUS_APID)];
+
+#[derive(Default, Clone)]
+pub struct SyncTcpTmSource {
+    tm_queue: Arc<Mutex<VecDeque<Vec<u8>>>>,
+    max_packets_stored: usize,
+    pub silent_packet_overwrite: bool,
+}
+
+impl SyncTcpTmSource {
+    pub fn new(max_packets_stored: usize) -> Self {
+        Self {
+            tm_queue: Arc::default(),
+            max_packets_stored,
+            silent_packet_overwrite: true,
+        }
+    }
+
+    pub fn add_tm(&mut self, tm: &[u8]) {
+        let mut tm_queue = self.tm_queue.lock().expect("locking tm queue failec");
+        if tm_queue.len() > self.max_packets_stored {
+            if !self.silent_packet_overwrite {
+                warn!("TPC TM source is full, deleting oldest packet");
+            }
+            tm_queue.pop_front();
+        }
+        tm_queue.push_back(tm.to_vec());
+    }
+}
+
+impl TmPacketSourceCore for SyncTcpTmSource {
+    type Error = ();
+
+    fn retrieve_packet(&mut self, buffer: &mut [u8]) -> Result<usize, Self::Error> {
+        let mut tm_queue = self.tm_queue.lock().expect("locking tm queue failed");
+        if !tm_queue.is_empty() {
+            let next_vec = tm_queue.front().unwrap();
+            if buffer.len() < next_vec.len() {
+                panic!(
+                    "provided buffer too small, must be at least {} bytes",
+                    next_vec.len()
+                );
+            }
+            let next_vec = tm_queue.pop_front().unwrap();
+            buffer[0..next_vec.len()].copy_from_slice(&next_vec);
+            if next_vec.len() > 9 {
+                let service = next_vec[7];
+                let subservice = next_vec[8];
+                info!("Sending PUS TM[{service},{subservice}]")
+            } else {
+                info!("Sending PUS TM");
+            }
+            return Ok(next_vec.len());
+        }
+        Ok(0)
+    }
+}
+
+pub struct TcpTask {
+    server: TcpSpacepacketsServer<(), CcsdsError<MpscStoreAndSendError>>,
+}
+
+impl TcpTask {
+    pub fn new(
+        cfg: ServerConfig,
+        tm_source: SyncTcpTmSource,
+        tc_receiver: CcsdsDistributor<MpscStoreAndSendError>,
+    ) -> Result<Self, std::io::Error> {
+        Ok(Self {
+            server: TcpSpacepacketsServer::new(
+                cfg,
+                Box::new(tm_source),
+                Box::new(tc_receiver),
+                Box::new(PACKET_ID_LOOKUP),
+            )?,
+        })
+    }
+
+    pub fn periodic_operation(&mut self) {
+        loop {
+            let result = self.server.handle_next_connection();
+            match result {
+                Ok(conn_result) => {
+                    info!(
+                        "Served {} TMs and {} TCs for client {:?}",
+                        conn_result.num_sent_tms, conn_result.num_received_tcs, conn_result.addr
+                    );
+                }
+                Err(e) => {
+                    warn!("TCP server error: {e:?}");
+                }
+            }
+        }
+    }
+}

satrs-example/src/tmtc.rs

@@ -1,8 +1,6 @@
-use log::{info, warn};
+use log::warn;
-use satrs_core::hal::std::udp_server::{ReceiveResult, UdpTcServer};
 use satrs_core::pus::ReceivesEcssPusTc;
 use satrs_core::spacepackets::SpHeader;
-use std::net::SocketAddr;
 use std::sync::mpsc::{Receiver, SendError, Sender, TryRecvError};
 use thiserror::Error;
 
@@ -12,12 +10,12 @@ use satrs_core::pus::TcAddrWithToken;
 use satrs_core::spacepackets::ecss::tc::PusTcReader;
 use satrs_core::spacepackets::ecss::PusPacket;
 use satrs_core::tmtc::tm_helper::SharedTmStore;
-use satrs_core::tmtc::{CcsdsError, ReceivesCcsdsTc};
+use satrs_core::tmtc::ReceivesCcsdsTc;
 
 pub struct TmArgs {
     pub tm_store: SharedTmStore,
     pub tm_sink_sender: Sender<StoreAddr>,
-    pub tm_server_rx: Receiver<StoreAddr>,
+    pub tm_udp_server_rx: Receiver<StoreAddr>,
 }
 
 pub struct TcArgs {
@@ -61,7 +59,6 @@ pub struct TmFunnel {
     pub tm_server_tx: Sender<StoreAddr>,
 }
 
-
 #[derive(Clone)]
 pub struct PusTcSource {
     pub tc_source: Sender<StoreAddr>,