CODEWITHSUNDEEP
socketsrustudprust-tokio
Mats Kindahl
Mats Kindahl

Reputation: 2060

Multicast UDP packets using Tokio futures

I'm playing around with Tokio and Rust and as an example, I am trying to write a simple UDP proxy that will just accept UDP packets on one socket and send it out to multiple other destinations. However, I stumble over the situation that I need to send the received packet to multiple addresses and am not sure how to do that in a idiomatic way.

Code I have this far:

extern crate bytes;
extern crate futures;

use std::net::SocketAddr;
use tokio::codec::BytesCodec;
use tokio::net::{UdpFramed, UdpSocket};
use tokio::prelude::*;

fn main() {
    let listen_address = "127.0.0.1:4711".parse::<SocketAddr>().unwrap();
    let forwarder = {
        let socket = UdpSocket::bind(&listen_address).unwrap();
        let peers = vec![
            "192.168.1.136:4711".parse::<SocketAddr>().unwrap(),
            "192.168.1.136:4712".parse::<SocketAddr>().unwrap(),
        ];
        UdpFramed::new(UdpSocket::bind(&listen_address).unwrap(), BytesCodec::new()).for_each(
            move |(bytes, _from)| {
                // These are the problematic lines
                for peer in peers.iter() {
                    socket.send_dgram(&bytes, &peer);
                }
                Ok(())
            },
        )
    };

    tokio::run({
        forwarder
            .map_err(|err| println!("Error: {}", err))
            .map(|_| ())
    });
}

The problematic lines are trying to send the received packet to multiple other addresses using a newly bound socket.

The existing examples all forward packets to single destinations, or internally use mpsc channels to communicate between internal tasks. I do not think that this is necessary and that it should be possible to do without having to spawn more than one task per listening socket.

Update: Thanks to @Ömer-erden I got this code that works. 

extern crate bytes;
extern crate futures;

use std::net::SocketAddr;
use tokio::codec::BytesCodec;
use tokio::net::{UdpFramed, UdpSocket};
use tokio::prelude::*;

fn main() -> Result<(), Box<dyn std::error::Error>> {
    let listen_address = "0.0.0.0:4711".parse::<SocketAddr>()?;
    let socket = UdpSocket::bind(&listen_address)?;
    let peers: Vec<SocketAddr> = vec!["192.168.1.136:8080".parse()?, "192.168.1.136:8081".parse()?];
    let (mut writer, reader) = UdpFramed::new(socket, BytesCodec::new()).split();
    let forwarder = reader.for_each(move |(bytes, _from)| {
        for peer in peers.iter() {
            writer.start_send((bytes.clone().into(), peer.clone()))?;
        }
        writer.poll_complete()?;
        Ok(())
    });

    tokio::run({
        forwarder
            .map_err(|err| println!("Error: {}", err))
            .map(|_| ())
    });
    Ok(())
}

Note that:

Upvotes: 5

Views: 2787

Answers (1)

&#214;mer Erden
&#214;mer Erden

Reputation: 8793

Your code has a logical mistake: you are trying to bind the same address twice, as sender and receiver respectively. Instead, you can use a stream and sink. UdpFramed has the functionality to provide that, please see Sink:

A Sink is a value into which other values can be sent, asynchronously.

let listen_address = "127.0.0.1:4711".parse::<SocketAddr>().unwrap();
let forwarder = {
    let (mut socket_sink, socket_stream) =
        UdpFramed::new(UdpSocket::bind(&listen_address).unwrap(), BytesCodec::new()).split();
    let peers = vec![
        "192.168.1.136:4711".parse::<SocketAddr>().unwrap(),
        "192.168.1.136:4712".parse::<SocketAddr>().unwrap(),
    ];

    socket_stream.for_each(move |(bytes, _from)| {
        for peer in peers.iter() {
            socket_sink.start_send((bytes.clone().into(), *peer));
            socket_sink.poll_complete();
        }
        Ok(())
    })
};

tokio::run({
    forwarder
        .map_err(|err| println!("Error: {}", err))
        .map(|_| ())
});

Upvotes: 2

Related Questions