fn sending_part(
    config: &Config,
    queue: &Arc<Mutex<BinaryHeap<PacketWrapper>>>,
    condvar: &Arc<Condvar>,
    socket: &Arc<UdpSocket>,
    send_addr: SocketAddrV4,
    thread_name: &str,
    brk: Arc<AtomicBool>,
) -> JoinHandle<()> {
    let config = config.clone();
    let queue = queue.clone();
    let condvar = condvar.clone();
    let socket = socket.clone();

    thread::Builder::new()
        .name(String::from(format!("{}_send", thread_name)))
        .spawn(move || {
            while !brk.load(Ordering::SeqCst) {
                // get packet to send
                let to_send = {
                    // lock queue to get data
                    let mut queue_guard = queue.lock().expect("Can't lock mutex from the sender part");
                    // loop waiting for the packet to be send
                    while !brk.load(Ordering::SeqCst) {
                        // get the wait time based on the first packet that should be send
                        let wait_time = queue_guard.peek().map_or(Duration::from_secs(u64::MAX), |wrapper| { wrapper.send_in() });
                        // if it should be already send break the loop waiting for the packet
                        if wait_time.as_millis() == 0 {
                            break;
                        }
                        // wait time cannot be longer than 1s because of the termination
                        let wait_time = Duration::min(wait_time, Duration::from_secs(1));
                        // else wait specified time or until new packet (possibly with earlier sending time) is inserted
                        let result = condvar.wait_timeout(
                            queue_guard,
                            wait_time,
                        ).expect("Can't lock mutex from the sender part");
                        queue_guard = result.0;
                    };
                    // packet in the queue, pop it
                    let packet = match queue_guard.pop() {
                        Some(x) => x,
                        None => continue,
                    };
                    // validate once more it should be send already
                    if !packet.should_be_send() {
                        continue;
                    };
                    // return the packet from the loop
                    packet
                };

                // send packet
                match socket.send_to(to_send.content(), send_addr) {
                    Ok(send_size) => config.vlog(&format!("Send data of size {}b to {}", send_size, send_addr)),
                    Err(e) => eprintln!("Error sending data {}", e),
                };
            };
        }).expect(&format!("Can't create sender part of the {}", thread_name))
}
/// Handles sending part of the communication.
/// It pulls packets from the `queue` (after the required amount of time passed) and
/// send them to `sendaddr` using `socket`.
/// When new packet arrive into the `queue` it should be signaled using `condvar`.
Handles sending part of the communication.
Handles sending part of the communication .
fn sending_part(
    config: &Config,
    queue: &Arc<Mutex<BinaryHeap<PacketWrapper>>>,
    condvar: &Arc<Condvar>,
    socket: &Arc<UdpSocket>,
    send_addr: SocketAddrV4,
    thread_name: &str,
    brk: Arc<AtomicBool>,
) -> JoinHandle<()> {
    let config = config.clone();
    let queue = queue.clone();
    let condvar = condvar.clone();
    let socket = socket.clone();
    thread::Builder::new()
        .name(String::from(format!("{}_send", thread_name)))
        .spawn(move || {
            while !brk.load(Ordering::SeqCst) {
                // get packet to send
                let to_send = {
                    // lock queue to get data
                    let mut queue_guard = queue.lock().expect("Can't lock mutex from the sender part");
                    // loop waiting for the packet to be send
                    while !brk.load(Ordering::SeqCst) {
                        // get the wait time based on the first packet that should be send
                        let wait_time = queue_guard.peek().map_or(Duration::from_secs(u64::MAX), |wrapper| { wrapper.send_in() });
                        // if it should be already send break the loop waiting for the packet
                        if wait_time.as_millis() == 0 {
                            break;
                        }
                        // wait time cannot be longer than 1s because of the termination
                        let wait_time = Duration::min(wait_time, Duration::from_secs(1));
                        // else wait specified time or until new packet (possibly with earlier sending time) is inserted
                        let result = condvar.wait_timeout(
                            queue_guard,
                            wait_time,
                        ).expect("Can't lock mutex from the sender part");
                        queue_guard = result.0;
                    };
                    // packet in the queue, pop it
                    let packet = match queue_guard.pop() {
                        Some(x) => x,
                        None => continue,
                    };
                    // validate once more it should be send already
                    if !packet.should_be_send() {
                        continue;
                    };
                    // return the packet from the loop
                    packet
                };
                // send packet
                match socket.send_to(to_send.content(), send_addr) {
                    Ok(send_size) => config.vlog(&format!("Send data of size {}b to {}", send_size, send_addr)),
                    Err(e) => eprintln!("Error sending data {}", e),
                };
            };
        }).expect(&format!("Can't create sender part of the {}", thread_name))
}
fn sending_part ( config : & Config , queue : & Arc < Mutex < BinaryHeap < PacketWrapper > > > , condvar : & Arc < Condvar > , socket : & Arc < UdpSocket > , send_addr : SocketAddrV4 , thread_name : & str , brk : Arc < AtomicBool > , ) -> JoinHandle < ( ) > { let config = config . clone ( ) ; let queue = queue . clone ( ) ; let condvar = condvar . clone ( ) ; let socket = socket . clone ( ) ; thread :: Builder :: new ( ) . name ( String :: from ( format ! ( " " thread_name ) ) ) . spawn ( move | | { while ! brk . load ( Ordering :: SeqCst ) { // get packet to send let to_send = { // lock queue to get data let mut queue_guard = queue . lock ( ) . expect ( " " ) ; // loop waiting for the packet to be send while ! brk . load ( Ordering :: SeqCst ) { // get the wait time based on the first packet that should be send let wait_time = queue_guard . peek ( ) . map_or ( Duration :: from_secs ( u64 :: MAX ) , | wrapper | { wrapper . send_in ( ) } ) ; // if it should be already send break the loop waiting for the packet if wait_time . as_millis ( ) == 0 { break ; } // wait time cannot be longer than 1s because of the termination let wait_time = Duration :: min ( wait_time , Duration :: from_secs ( 1 ) ) ; // else wait specified time or until new packet (possibly with earlier sending time) is inserted let result = condvar . wait_timeout ( queue_guard , wait_time , ) . expect ( " " ) ; queue_guard = result . 0 ; } ; // packet in the queue, pop it let packet = match queue_guard . pop ( ) { Some ( x ) => x , None => continue , } ; // validate once more it should be send already if ! packet . should_be_send ( ) { continue ; } ; // return the packet from the loop packet } ; // send packet match socket . send_to ( to_send . content ( ) , send_addr ) { Ok ( send_size ) => config . vlog ( & format ! ( " " send_size send_addr ) ) , Err ( e ) => eprintln ! ( " " e ) , } ; } ; } ) . expect ( & format ! ( " " thread_name ) ) }
Handles sending part of the communication.
Handles sending part of the communication .
config
&Config
queue
&Arc<Mutex<BinaryHeap<PacketWrapper>>>
condvar
&Arc<Condvar>
socket
&Arc<UdpSocket>
send_addr
SocketAddrV4
thread_name
&str
brk
Arc<AtomicBool>
config
&Config
queue
&Arc<Mutex<BinaryHeap<PacketWrapper>>>
condvar
&Arc<Condvar>
socket
&Arc<UdpSocket>
send_addr
SocketAddrV4
thread_name
&str
brk
Arc<AtomicBool>