1 use anyhow::Result;
2 use clap::{AppSettings, Arg, Command};
3 use serde::{Deserialize, Serialize};
4 use std::io::Write;
5 use std::sync::Arc;
6 use tokio::sync::Notify;
7 use tokio::time::Duration;
8 use webrtc::api::APIBuilder;
9 use webrtc::data_channel::data_channel_message::DataChannelMessage;
10 use webrtc::data_channel::data_channel_parameters::DataChannelParameters;
11 use webrtc::data_channel::RTCDataChannel;
12 use webrtc::dtls_transport::dtls_parameters::DTLSParameters;
13 use webrtc::ice_transport::ice_candidate::RTCIceCandidate;
14 use webrtc::ice_transport::ice_gatherer::RTCIceGatherOptions;
15 use webrtc::ice_transport::ice_parameters::RTCIceParameters;
16 use webrtc::ice_transport::ice_role::RTCIceRole;
17 use webrtc::ice_transport::ice_server::RTCIceServer;
18 use webrtc::peer_connection::math_rand_alpha;
19 use webrtc::sctp_transport::sctp_transport_capabilities::SCTPTransportCapabilities;
20
21 #[tokio::main]
main() -> Result<()>22 async fn main() -> Result<()> {
23 let mut app = Command::new("ortc")
24 .version("0.1.0")
25 .author("Rain Liu <[email protected]>")
26 .about("An example of ORTC.")
27 .setting(AppSettings::DeriveDisplayOrder)
28 .subcommand_negates_reqs(true)
29 .arg(
30 Arg::new("FULLHELP")
31 .help("Prints more detailed help information")
32 .long("fullhelp"),
33 )
34 .arg(
35 Arg::new("debug")
36 .long("debug")
37 .short('d')
38 .help("Prints debug log information"),
39 )
40 .arg(
41 Arg::new("offer")
42 .long("offer")
43 .help("Act as the offerer if set."),
44 );
45
46 let matches = app.clone().get_matches();
47
48 if matches.is_present("FULLHELP") {
49 app.print_long_help().unwrap();
50 std::process::exit(0);
51 }
52
53 let is_offer = matches.is_present("offer");
54 let debug = matches.is_present("debug");
55 if debug {
56 env_logger::Builder::new()
57 .format(|buf, record| {
58 writeln!(
59 buf,
60 "{}:{} [{}] {} - {}",
61 record.file().unwrap_or("unknown"),
62 record.line().unwrap_or(0),
63 record.level(),
64 chrono::Local::now().format("%H:%M:%S.%6f"),
65 record.args()
66 )
67 })
68 .filter(None, log::LevelFilter::Trace)
69 .init();
70 }
71
72 // Everything below is the Pion WebRTC (ORTC) API! Thanks for using it ❤️.
73
74 // Prepare ICE gathering options
75 let ice_options = RTCIceGatherOptions {
76 ice_servers: vec![RTCIceServer {
77 urls: vec!["stun:stun.l.google.com:19302".to_owned()],
78 ..Default::default()
79 }],
80 ..Default::default()
81 };
82
83 // Create an API object
84 let api = APIBuilder::new().build();
85
86 // Create the ICE gatherer
87 let gatherer = Arc::new(api.new_ice_gatherer(ice_options)?);
88
89 // Construct the ICE transport
90 let ice = Arc::new(api.new_ice_transport(Arc::clone(&gatherer)));
91
92 // Construct the DTLS transport
93 let dtls = Arc::new(api.new_dtls_transport(Arc::clone(&ice), vec![])?);
94
95 // Construct the SCTP transport
96 let sctp = Arc::new(api.new_sctp_transport(Arc::clone(&dtls))?);
97
98 let done = Arc::new(Notify::new());
99 let done_answer = done.clone();
100 let done_offer = done.clone();
101
102 // Handle incoming data channels
103 sctp.on_data_channel(Box::new(move |d: Arc<RTCDataChannel>| {
104 let d_label = d.label().to_owned();
105 let d_id = d.id();
106 println!("New DataChannel {d_label} {d_id}");
107
108 let done_answer1 = done_answer.clone();
109 // Register the handlers
110 Box::pin(async move {
111 // no need to downgrade this to Weak, since on_open is FnOnce callback
112 let d2 = Arc::clone(&d);
113 let done_answer2 = done_answer1.clone();
114 d.on_open(Box::new(move || {
115 Box::pin(async move {
116 tokio::select! {
117 _ = done_answer2.notified() => {
118 println!("received done_answer signal!");
119 }
120 _ = handle_on_open(d2) => {}
121 };
122
123 println!("exit data answer");
124 })
125 }));
126
127 // Register text message handling
128 d.on_message(Box::new(move |msg: DataChannelMessage| {
129 let msg_str = String::from_utf8(msg.data.to_vec()).unwrap();
130 println!("Message from DataChannel '{d_label}': '{msg_str}'");
131 Box::pin(async {})
132 }));
133 })
134 }));
135
136 let (gather_finished_tx, mut gather_finished_rx) = tokio::sync::mpsc::channel::<()>(1);
137 let mut gather_finished_tx = Some(gather_finished_tx);
138 gatherer.on_local_candidate(Box::new(move |c: Option<RTCIceCandidate>| {
139 if c.is_none() {
140 gather_finished_tx.take();
141 }
142 Box::pin(async {})
143 }));
144
145 // Gather candidates
146 gatherer.gather().await?;
147
148 let _ = gather_finished_rx.recv().await;
149
150 let ice_candidates = gatherer.get_local_candidates().await?;
151
152 let ice_parameters = gatherer.get_local_parameters().await?;
153
154 let dtls_parameters = dtls.get_local_parameters()?;
155
156 let sctp_capabilities = sctp.get_capabilities();
157
158 let local_signal = Signal {
159 ice_candidates,
160 ice_parameters,
161 dtls_parameters,
162 sctp_capabilities,
163 };
164
165 // Exchange the information
166 let json_str = serde_json::to_string(&local_signal)?;
167 let b64 = signal::encode(&json_str);
168 println!("{b64}");
169
170 let line = signal::must_read_stdin()?;
171 let json_str = signal::decode(line.as_str())?;
172 let remote_signal = serde_json::from_str::<Signal>(&json_str)?;
173
174 let ice_role = if is_offer {
175 RTCIceRole::Controlling
176 } else {
177 RTCIceRole::Controlled
178 };
179
180 ice.set_remote_candidates(&remote_signal.ice_candidates)
181 .await?;
182
183 // Start the ICE transport
184 ice.start(&remote_signal.ice_parameters, Some(ice_role))
185 .await?;
186
187 // Start the DTLS transport
188 dtls.start(remote_signal.dtls_parameters).await?;
189
190 // Start the SCTP transport
191 sctp.start(remote_signal.sctp_capabilities).await?;
192
193 // Construct the data channel as the offerer
194 if is_offer {
195 let id = 1u16;
196
197 let dc_params = DataChannelParameters {
198 label: "Foo".to_owned(),
199 negotiated: Some(id),
200 ..Default::default()
201 };
202
203 let d = Arc::new(api.new_data_channel(Arc::clone(&sctp), dc_params).await?);
204
205 // Register the handlers
206 // channel.OnOpen(handleOnOpen(channel)) // TODO: OnOpen on handle ChannelAck
207 // Temporary alternative
208
209 // no need to downgrade this to Weak
210 let d2 = Arc::clone(&d);
211 tokio::spawn(async move {
212 tokio::select! {
213 _ = done_offer.notified() => {
214 println!("received done_offer signal!");
215 }
216 _ = handle_on_open(d2) => {}
217 };
218
219 println!("exit data offer");
220 });
221
222 let d_label = d.label().to_owned();
223 d.on_message(Box::new(move |msg: DataChannelMessage| {
224 let msg_str = String::from_utf8(msg.data.to_vec()).unwrap();
225 println!("Message from DataChannel '{d_label}': '{msg_str}'");
226 Box::pin(async {})
227 }));
228 }
229
230 println!("Press ctrl-c to stop");
231 tokio::signal::ctrl_c().await.unwrap();
232 done.notify_waiters();
233
234 sctp.stop().await?;
235 dtls.stop().await?;
236 ice.stop().await?;
237
238 Ok(())
239 }
240
241 // Signal is used to exchange signaling info.
242 // This is not part of the ORTC spec. You are free
243 // to exchange this information any way you want.
244 #[derive(Debug, Clone, Serialize, Deserialize)]
245 struct Signal {
246 #[serde(rename = "iceCandidates")]
247 ice_candidates: Vec<RTCIceCandidate>, // `json:"iceCandidates"`
248
249 #[serde(rename = "iceParameters")]
250 ice_parameters: RTCIceParameters, // `json:"iceParameters"`
251
252 #[serde(rename = "dtlsParameters")]
253 dtls_parameters: DTLSParameters, // `json:"dtlsParameters"`
254
255 #[serde(rename = "sctpCapabilities")]
256 sctp_capabilities: SCTPTransportCapabilities, // `json:"sctpCapabilities"`
257 }
258
handle_on_open(d: Arc<RTCDataChannel>) -> Result<()>259 async fn handle_on_open(d: Arc<RTCDataChannel>) -> Result<()> {
260 println!("Data channel '{}'-'{}' open. Random messages will now be sent to any connected DataChannels every 5 seconds", d.label(), d.id());
261
262 let mut result = Result::<usize>::Ok(0);
263 while result.is_ok() {
264 let timeout = tokio::time::sleep(Duration::from_secs(5));
265 tokio::pin!(timeout);
266
267 tokio::select! {
268 _ = timeout.as_mut() =>{
269 let message = math_rand_alpha(15);
270 println!("Sending '{message}'");
271 result = d.send_text(message).await.map_err(Into::into);
272 }
273 };
274 }
275
276 Ok(())
277 }
278