sctp-async 0.0.0

Async implementation for the SCTP protocol
Documentation 
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110

use sctp_async as sctp;

use sctp::association::*;
use sctp::stream::*;
use sctp::Error;

use bytes::Bytes;
use clap::{App, AppSettings, Arg};
use std::sync::Arc;
use std::time::Duration;
use tokio::net::UdpSocket;
use tokio::signal;
use tokio::sync::mpsc;
use util::{conn::conn_disconnected_packet::DisconnectedPacketConn, Conn};

// RUST_LOG=trace cargo run --color=always --package webrtc-sctp --example pong -- --host 0.0.0.0:5678

#[tokio::main]
async fn main() -> Result<(), Error> {
    /*env_logger::Builder::new()
    .format(|buf, record| {
        writeln!(
            buf,
            "{}:{} [{}] {} - {}",
            record.file().unwrap_or("unknown"),
            record.line().unwrap_or(0),
            record.level(),
            chrono::Local::now().format("%H:%M:%S.%6f"),
            record.args()
        )
    })
    .filter(None, log::LevelFilter::Trace)
    .init();*/

    let mut app = App::new("SCTP Pong")
        .version("0.1.0")
        .author("Rain Liu <yliu@webrtc.rs>")
        .about("An example of SCTP Server")
        .setting(AppSettings::DeriveDisplayOrder)
        .setting(AppSettings::SubcommandsNegateReqs)
        .arg(
            Arg::with_name("FULLHELP")
                .help("Prints more detailed help information")
                .long("fullhelp"),
        )
        .arg(
            Arg::with_name("host")
                .required_unless("FULLHELP")
                .takes_value(true)
                .long("host")
                .help("SCTP host name."),
        );

    let matches = app.clone().get_matches();

    if matches.is_present("FULLHELP") {
        app.print_long_help().unwrap();
        std::process::exit(0);
    }

    let host = matches.value_of("host").unwrap();
    let conn = DisconnectedPacketConn::new(Arc::new(UdpSocket::bind(host).await.unwrap()));
    println!("listening {}...", conn.local_addr().await.unwrap());

    let config = Config {
        net_conn: Arc::new(conn),
        max_receive_buffer_size: 0,
        max_message_size: 0,
        name: "server".to_owned(),
    };
    let a = Association::server(config).await?;
    println!("created a server");

    let stream = a.accept_stream().await.unwrap();
    println!("accepted a stream");

    // set unordered = true and 10ms treshold for dropping packets
    stream.set_reliability_params(true, ReliabilityType::Timed, 10);

    let (done_tx, mut done_rx) = mpsc::channel::<()>(1);
    let stream2 = Arc::clone(&stream);
    tokio::spawn(async move {
        let mut buff = vec![0u8; 1024];
        while let Ok(n) = stream2.read(&mut buff).await {
            let ping_msg = String::from_utf8(buff[..n].to_vec()).unwrap();
            println!("received: {}", ping_msg);

            let pong_msg = format!("pong [{}]", ping_msg);
            println!("sent: {}", pong_msg);
            stream2.write(&Bytes::from(pong_msg)).await?;

            tokio::time::sleep(Duration::from_secs(1)).await;
        }
        println!("finished ping-pong");
        drop(done_tx);

        Result::<(), Error>::Ok(())
    });

    println!("Waiting for Ctrl-C...");
    signal::ctrl_c().await.expect("failed to listen for event");
    println!("Closing stream and association...");

    stream.close().await?;
    a.close().await?;

    let _ = done_rx.recv().await;

    Ok(())
}