在 Rust 中使用 Fute 协议进行多播 UDP 本地传输失败

Question

我正在尝试在本地运行 frude in rust 演示代码。 我将演示代码从 read me 修改为下面

use flute::sender::Sender;
use flute::sender::ObjectDesc;
use flute::sender::Cenc;
use flute::core::UDPEndpoint;
use std::net::UdpSocket;
use std::time::SystemTime;
use flute::receiver::{writer, MultiReceiver};
use std::rc::Rc;
use std::thread;
use std::time::Duration; 
use std::net::Ipv4Addr;  


fn main() {
    println!("Begin");
    let receiver_thread = thread::Builder::new()  
        .name("ReceiverThread".to_string())  
        .spawn(|| fn_set_receiver())  
        .expect("Failed to create receiver thread"); 
    thread::sleep(Duration::from_secs(5));  
    let sender_thread = thread::Builder::new()  
        .name("SenderThread".to_string())  
        .spawn(|| fn_set_sender())  
        .expect("Failed to create sender thread"); 
    sender_thread.join().unwrap(); 
    receiver_thread.join().unwrap();  
    println!("End");
}

fn fn_set_sender() {
    println!("Begin sender");
        // Create UDP Socket
    let udp_socket = UdpSocket::bind("0.0.0.0:0").unwrap();
    udp_socket.connect("224.0.1.201:3400").expect("Connection failed");

    // Create FLUTE Sender
    let tsi = 1;
    let oti = Default::default();
    let config = Default::default();
    let endpoint = UDPEndpoint::new(None, "224.0.1.201".to_string(), 3400);
    let mut sender = Sender::new(endpoint, tsi, &oti, &config);

    // Add object(s) (files) to the FLUTE sender (priority queue 0)
    let obj = ObjectDesc::create_from_buffer(b"hello world", "text/plain",
    &url::Url::parse(r#"C:\Users\myname\Documents\_\Multicast\src\data.txt"#).unwrap(), 1, None, None, None, Cenc::Null, true, None, true).unwrap();
    sender.add_object(0, obj);

    // Always call publish after adding objects
    sender.publish(SystemTime::now());

    // Send FLUTE packets over UDP/IP
    while let Some(pkt) = sender.read(SystemTime::now()) {
        udp_socket.send(&pkt).unwrap();
        std::thread::sleep(std::time::Duration::from_millis(1));
    }
    println!("End sender");
}

fn fn_set_receiver() {
    println!("Begin Receiver");
    // Create UDP/IP socket to receive FLUTE pkt
    let any = Ipv4Addr::new(0,0,0,0);
    let endpoint = UDPEndpoint::new(None, "224.0.1.201".to_string(), 3400);
    let udp_socket = UdpSocket::bind((any, endpoint.port)).expect("Fail to bind");

    // Create a writer able to write received files to the filesystem
    let writer = match writer::ObjectWriterFSBuilder::new(&std::path::Path::new(r#"C:\Users\myname\Documents\_\Multicast\dest\"#)) {  
        Ok(writer) => Rc::new(writer),  
        Err(err) => {  
            println!("Failed to create object writer: {:?}", err);  
            return;  
        }  
    }; 

    // Create a multi-receiver capable of de-multiplexing several FLUTE sessions
    let mut receiver = MultiReceiver::new(writer, None, false);

    // Receive pkt from UDP/IP socket and push it to the FLUTE receiver
    let mut buf = [0; 2048];
    loop {
        let (n, _src) = udp_socket.recv_from(&mut buf).expect("Failed to receive data");
        let now = SystemTime::now();
        receiver.push(&endpoint, &buf[..n], now).unwrap();
        receiver.cleanup(now);
    }
}

在调试模式下运行时，我可以看到它毫无例外地通过接收器和发送器。
然而，

udp_socket.recv_from

似乎没有得到任何东西。最终代码陷入了

receiver_thread.join().unwrap();

等待获取内容的状态。

我对所有这些（UDP、多播、铁锈等）都很陌生，我想我可能会弄乱多个部分。但无一例外，我需要一些关于要检查什么的线索。

Answer 1

我需要进行一些调整，但主要是

UDPEndpoint

应该使用多播IP（如224.0.0.1）进行初始化。
但

UdpSocket

应绑定到（任何或）常规 IP，例如 192.168.8.8。 更新功能如下

fn fn_set_sender() {
    println!("Begin sender");
        // Create UDP Socket
    let udp_socket = UdpSocket::bind("0.0.0.0:0").unwrap();
    //  udp_socket.connect("15.38.89.164:3400").expect("Connection failed");
    udp_socket.connect("192.168.8.8:3400").expect("Connection failed");

    // Create FLUTE Sender
    let tsi = 1;
    let oti = Default::default();
    let config = Default::default();
    let endpoint = UDPEndpoint::new(None, "224.0.1.201".to_string(), 3400);
    let mut sender = Sender::new(endpoint, tsi, &oti, &config);

    // Add object(s) (files) to the FLUTE sender (priority queue 0)    
    let obj = ObjectDesc::create_from_file(
        Path::new("C:/Users/myname/Documents/_/Multicast/src/data.txt"),
        None,
        "application/octet-stream",
        true,
        1,
        None,
        None,
        None,
        Cenc::Null,
        true,
        None,
        true,
    )
    .unwrap();
    sender.add_object(0, obj).unwrap();

    // Always call publish after adding objects
    sender.publish(SystemTime::now());

    // Send FLUTE packets over UDP/IP
    while let Some(pkt) = sender.read(SystemTime::now()) {
        udp_socket.send(&pkt).unwrap();
        std::thread::sleep(std::time::Duration::from_millis(5));
    }
    println!("End sender");
}

fn fn_set_receiver() {

    let endpoint = UDPEndpoint::new(None, "224.0.1.201".to_string(), 3400);
    // let udp_socket = UdpSocket::bind(("15.38.89.164:3400")).expect("Fail to bind");
    let udp_socket = UdpSocket::bind(("192.168.8.8:3400")).expect("Fail to bind");

    // Create a writer able to write received files to the filesystem
    let writer = match writer::ObjectWriterFSBuilder::new(&std::path::Path::new(r#"C:\Users\myname\Documents\_\Multicast\dest\"#)) {  
        Ok(writer) => Rc::new(writer),  
        Err(err) => {  
            println!("Failed to create object writer: {:?}", err);  
            return;  
        }   
    }; 

    // Create a multi-receiver capable of de-multiplexing several FLUTE sessions
    let mut receiver = MultiReceiver::new(writer, None, false);

    // Receive pkt from UDP/IP socket and push it to the FLUTE receiver
    let mut buf = [0; 2048];
    loop {
        let (n, _src) = udp_socket.recv_from(&mut buf).expect("Failed to receive data");
        let now = SystemTime::now();
        receiver.push(&endpoint, &buf[..n], now).unwrap();
        receiver.cleanup(now);
    }
}