// This thread reads indefinitely from a UDP socket, creating link packets from
// the UDP packet payload and then writes the link packets to a gateway.
This thread reads indefinitely from a UDP socket, creating link packets from
the UDP packet payload and then writes the link packets to a gateway.
This thread reads indefinitely from a UDP socket creating link packets from the UDP packet payload and then writes the link packets to a gateway .
fn downlink_endpoint<ReadConnection: Clone, WriteConnection: Clone, Packet: LinkPacket>(
    data: &Arc<Mutex<CommsTelemetry>>,
    port: DownlinkPort,
    write_conn: WriteConnection,
    write: &Arc<WriteFn<WriteConnection>>,
    sat_ip: Ipv4Addr,
) {
    // Bind the downlink endpoint to a UDP socket.
    // let socket = match UdpSocket::bind((sat_ip, port)) {
    //     Ok(sock) => sock,
    //     Err(e) => return log_error(&data, e.to_string()).unwrap(),
    // };
    debug!("Starting downlink endpoint {:?}", &port);
    let (packet_tx, packet_rx) = mpsc::channel();
    let (return_tx, return_rx) = mpsc::channel();
    let num_packets = Arc::new(AtomicU32::new(0));
    let max = 32;
    let data_c = data.clone();
    let num_packets_c = num_packets.clone();
    // This thread receives data for downlink, buffers it and puts it in a fifo.
    // The number of buffers is limited, the thread will loop/wait for buffers to be released then
    // continue.
    let port_c = port.clone();
    thread::Builder::new()
        .stack_size(4 * 1024)
        .spawn(move || {
            let buf_size = port_c.buf_size.unwrap_or(8 * 1024);
            let port = port_c.port;
            info!(
                "Starting UDP receiving thread for {}, buf_size: {}",
                &port, &buf_size
            );
            let data = data_c;
            let num_packets = num_packets_c;
            // Bind the downlink endpoint to a UDP socket.
            let socket = match UdpSocket::bind((sat_ip, port)) {
                Ok(sock) => sock,
                Err(e) => return log_error(&data, e.to_string()).unwrap(),
            };
            let mut buf: Option<Vec<u8>> = None;
            loop {
                if let None = &buf {
                    buf = Some(match return_rx.try_recv() {
                        Ok(buf) => buf,
                        Err(_) => {
                            let num_pkts = num_packets.load(Ordering::SeqCst);
                            if num_pkts >= max {
                                std::thread::yield_now();
                                continue;
                            } else {
                                debug!("Created new buffer for {}", &port);
                                vec![0; buf_size]
                            }
                        }
                    });
                }
                if let Some(mut mut_buf) = buf.take() {
                    // Indefinitely wait for a message from any application or service.
                    let (size, address) = match socket.recv_from(&mut mut_buf) {
                        Ok(tuple) => tuple,
                        Err(e) => {
                            log_error(&data, e.to_string()).unwrap();
                            buf = Some(mut_buf);
                            continue;
                        }
                    };
                    if let Err(SendError((_size, _address, bad_buf))) =
                        packet_tx.send((size, address, mut_buf))
                    {
                        error!("Failed to send packet to channel");
                        buf = Some(bad_buf);
                        continue;
                    }
                    num_packets.fetch_add(1, Ordering::SeqCst);
                }
            }
        })
        .unwrap();
    // This socket is used specifically for sending backpreassure to the client
    let socket = UdpSocket::bind("0.0.0.0:0").unwrap();
    // Take the packets from the FIFO and downlink them.
    // Also tell the sender how many packets we want from them.
    while let Ok((size, address, buf)) = packet_rx.recv() {
        if let Some(num_pkts) = num_packets
            .fetch_update(
                |x| match x {
                    x if x > 0 => Some(x - 1),
                    _ => None,
                },
                Ordering::SeqCst,
                Ordering::SeqCst,
            )
            .ok()
        {
            // tell the sender how many packets they're allowed to send us.
            let msg = &[max as u8 - std::cmp::min(num_pkts, max) as u8];
            if let Err(e) = socket.send_to(msg, address) {
                debug!("Could not send backpreassure: {:?}", e);
            }
        }
        // Take received message and wrap it in a Link packet.
        // Setting port to 0 because we don't know the ground port...
        // That is known by the ground comms service
        let packet = match Packet::build(0, PayloadType::UDP, port.port, &buf[0..size])
            .and_then(|packet| packet.to_bytes())
        {
            Ok(packet) => packet,
            Err(e) => {
                log_error(&data, e.to_string()).unwrap();
                continue;
            }
        };
        // Write packet to the gateway and update telemetry.
        match write(&write_conn.clone(), &packet) {
            Ok(_) => {
                log_telemetry(&data, &TelemType::Down).unwrap();
                // info!("Packet successfully downlinked");
            }
            Err(e) => {
                log_telemetry(&data, &TelemType::DownFailed).unwrap();
                log_error(&data, e.to_string()).unwrap();
                error!("Packet failed to downlink");
            }
        };
        if let Err(_) = return_tx.send(buf) {
            error!("Dropping packet as failed to send back to udp thread");
        }
    }
}
fn downlink_endpoint < ReadConnection : Clone , WriteConnection : Clone , Packet : LinkPacket > ( data : & Arc < Mutex < CommsTelemetry > > , port : DownlinkPort , write_conn : WriteConnection , write : & Arc < WriteFn < WriteConnection > > , sat_ip : Ipv4Addr , ) { // Bind the downlink endpoint to a UDP socket. // let socket = match UdpSocket::bind((sat_ip, port)) { //     Ok(sock) => sock, //     Err(e) => return log_error(&data, e.to_string()).unwrap(), // }; debug ! ( " " port ) ; let ( packet_tx , packet_rx ) = mpsc :: channel ( ) ; let ( return_tx , return_rx ) = mpsc :: channel ( ) ; let num_packets = Arc :: new ( AtomicU32 :: new ( 0 ) ) ; let max = 32 ; let data_c = data . clone ( ) ; let num_packets_c = num_packets . clone ( ) ; // This thread receives data for downlink, buffers it and puts it in a fifo. // The number of buffers is limited, the thread will loop/wait for buffers to be released then // continue. let port_c = port . clone ( ) ; thread :: Builder :: new ( ) . stack_size ( 4 * 1024 ) . spawn ( move | | { let buf_size = port_c . buf_size . unwrap_or ( 8 * 1024 ) ; let port = port_c . port ; info ! ( " " port buf_size ) ; let data = data_c ; let num_packets = num_packets_c ; // Bind the downlink endpoint to a UDP socket. let socket = match UdpSocket :: bind ( ( sat_ip , port ) ) { Ok ( sock ) => sock , Err ( e ) => return log_error ( & data , e . to_string ( ) ) . unwrap ( ) , } ; let mut buf : Option < Vec < u8 > > = None ; loop { if let None = & buf { buf = Some ( match return_rx . try_recv ( ) { Ok ( buf ) => buf , Err ( _ ) => { let num_pkts = num_packets . load ( Ordering :: SeqCst ) ; if num_pkts >= max { std :: thread :: yield_now ( ) ; continue ; } else { debug ! ( " " port ) ; vec ! [ 0 buf_size ] } } } ) ; } if let Some ( mut mut_buf ) = buf . take ( ) { // Indefinitely wait for a message from any application or service. let ( size , address ) = match socket . recv_from ( & mut mut_buf ) { Ok ( tuple ) => tuple , Err ( e ) => { log_error ( & data , e . to_string ( ) ) . unwrap ( ) ; buf = Some ( mut_buf ) ; continue ; } } ; if let Err ( SendError ( ( _size , _address , bad_buf ) ) ) = packet_tx . send ( ( size , address , mut_buf ) ) { error ! ( " " ) ; buf = Some ( bad_buf ) ; continue ; } num_packets . fetch_add ( 1 , Ordering :: SeqCst ) ; } } } ) . unwrap ( ) ; // This socket is used specifically for sending backpreassure to the client let socket = UdpSocket :: bind ( " " ) . unwrap ( ) ; // Take the packets from the FIFO and downlink them. // Also tell the sender how many packets we want from them. while let Ok ( ( size , address , buf ) ) = packet_rx . recv ( ) { if let Some ( num_pkts ) = num_packets . fetch_update ( | x | match x { x if x > 0 => Some ( x - 1 ) , _ => None , } , Ordering :: SeqCst , Ordering :: SeqCst , ) . ok ( ) { // tell the sender how many packets they're allowed to send us. let msg = & [ max as u8 - std :: cmp :: min ( num_pkts , max ) as u8 ] ; if let Err ( e ) = socket . send_to ( msg , address ) { debug ! ( " " e ) ; } } // Take received message and wrap it in a Link packet. // Setting port to 0 because we don't know the ground port... // That is known by the ground comms service let packet = match Packet :: build ( 0 , PayloadType :: UDP , port . port , & buf [ 0 .. size ] ) . and_then ( | packet | packet . to_bytes ( ) ) { Ok ( packet ) => packet , Err ( e ) => { log_error ( & data , e . to_string ( ) ) . unwrap ( ) ; continue ; } } ; // Write packet to the gateway and update telemetry. match write ( & write_conn . clone ( ) , & packet ) { Ok ( _ ) => { log_telemetry ( & data , & TelemType :: Down ) . unwrap ( ) ; // info!("Packet successfully downlinked"); } Err ( e ) => { log_telemetry ( & data , & TelemType :: DownFailed ) . unwrap ( ) ; log_error ( & data , e . to_string ( ) ) . unwrap ( ) ; error ! ( " " ) ; } } ; if let Err ( _ ) = return_tx . send ( buf ) { error ! ( " " ) ; } } }
This thread reads indefinitely from a UDP socket, creating link packets from
the UDP packet payload and then writes the link packets to a gateway.
This thread reads indefinitely from a UDP socket creating link packets from the UDP packet payload and then writes the link packets to a gateway .
data
&Arc<Mutex<CommsTelemetry>>
port
DownlinkPort
write_conn
WriteConnection
write
&Arc<WriteFn<WriteConnection>>
sat_ip
Ipv4Addr