1 use std::{
2 sync::{
3 atomic::{AtomicUsize, Ordering},
4 Arc,
5 },
6 time::{Duration, SystemTime},
7 };
8
9 use bytes::Bytes;
10
11 use webrtc::{
12 api::{
13 interceptor_registry::register_default_interceptors, media_engine::MediaEngine, APIBuilder,
14 },
15 data_channel::data_channel_init::RTCDataChannelInit,
16 ice_transport::{ice_candidate::RTCIceCandidate, ice_server::RTCIceServer},
17 interceptor::registry::Registry,
18 peer_connection::{
19 configuration::RTCConfiguration, peer_connection_state::RTCPeerConnectionState,
20 RTCPeerConnection,
21 },
22 };
23
24 const BUFFERED_AMOUNT_LOW_THRESHOLD: usize = 512 * 1024; // 512 KB
25 const MAX_BUFFERED_AMOUNT: usize = 1024 * 1024; // 1 MB
26
create_peer_connection() -> anyhow::Result<RTCPeerConnection>27 async fn create_peer_connection() -> anyhow::Result<RTCPeerConnection> {
28 // Create unique MediaEngine,
29 // as MediaEngine must not be shared between PeerConnections
30 let mut media_engine = MediaEngine::default();
31
32 media_engine.register_default_codecs()?;
33
34 let mut interceptor_registry = Registry::new();
35
36 interceptor_registry = register_default_interceptors(interceptor_registry, &mut media_engine)?;
37
38 // Create API that bundles the global functions of the WebRTC API
39 let api = APIBuilder::new()
40 .with_media_engine(media_engine)
41 .with_interceptor_registry(interceptor_registry)
42 .build();
43
44 let ice_servers = vec![RTCIceServer {
45 ..Default::default()
46 }];
47
48 let config = RTCConfiguration {
49 ice_servers,
50 ..Default::default()
51 };
52
53 Ok(api.new_peer_connection(config).await?)
54 }
55
create_requester() -> anyhow::Result<RTCPeerConnection>56 async fn create_requester() -> anyhow::Result<RTCPeerConnection> {
57 // Create a peer connection first
58 let pc = create_peer_connection().await?;
59
60 // Data transmission requires a data channel, so prepare to create one
61 let options = Some(RTCDataChannelInit {
62 ordered: Some(false),
63 max_retransmits: Some(0u16),
64 ..Default::default()
65 });
66
67 // Create a data channel to send data over a peer connection
68 let dc = pc.create_data_channel("data", options).await?;
69
70 // Use mpsc channel to send and receive a signal when more data can be sent
71 let (more_can_be_sent, mut maybe_more_can_be_sent) = tokio::sync::mpsc::channel(1);
72
73 // Get a shared pointer to the data channel
74 let shared_dc = dc.clone();
75 dc.on_open(Box::new(|| {
76 Box::pin(async move {
77 // This callback shouldn't be blocked for a long time, so we spawn our handler
78 tokio::spawn(async move {
79 let buf = Bytes::from_static(&[0u8; 1024]);
80
81 loop {
82 if shared_dc.send(&buf).await.is_err() {
83 break;
84 }
85
86 let buffered_amount = shared_dc.buffered_amount().await;
87
88 if buffered_amount + buf.len() > MAX_BUFFERED_AMOUNT {
89 // Wait for the signal that more can be sent
90 let _ = maybe_more_can_be_sent.recv().await;
91 }
92 }
93 });
94 })
95 }));
96
97 dc.set_buffered_amount_low_threshold(BUFFERED_AMOUNT_LOW_THRESHOLD)
98 .await;
99
100 dc.on_buffered_amount_low(Box::new(move || {
101 let more_can_be_sent = more_can_be_sent.clone();
102
103 Box::pin(async move {
104 // Send a signal that more can be sent
105 more_can_be_sent.send(()).await.unwrap();
106 })
107 }))
108 .await;
109
110 Ok(pc)
111 }
112
create_responder() -> anyhow::Result<RTCPeerConnection>113 async fn create_responder() -> anyhow::Result<RTCPeerConnection> {
114 // Create a peer connection first
115 let pc = create_peer_connection().await?;
116
117 // Set a data channel handler so that we can receive data
118 pc.on_data_channel(Box::new(move |dc| {
119 Box::pin(async move {
120 let total_bytes_received = Arc::new(AtomicUsize::new(0));
121
122 let shared_total_bytes_received = total_bytes_received.clone();
123 dc.on_open(Box::new(move || {
124 Box::pin(async {
125 // This callback shouldn't be blocked for a long time, so we spawn our handler
126 tokio::spawn(async move {
127 let start = SystemTime::now();
128
129 tokio::time::sleep(Duration::from_secs(1)).await;
130 println!();
131
132 loop {
133 let total_bytes_received =
134 shared_total_bytes_received.load(Ordering::Relaxed);
135
136 let elapsed = SystemTime::now().duration_since(start);
137 let bps =
138 (total_bytes_received * 8) as f64 / elapsed.unwrap().as_secs_f64();
139
140 println!(
141 "Throughput is about {:.03} Mbps",
142 bps / (1024 * 1024) as f64
143 );
144 tokio::time::sleep(Duration::from_secs(1)).await;
145 }
146 });
147 })
148 }));
149
150 dc.on_message(Box::new(move |msg| {
151 let total_bytes_received = total_bytes_received.clone();
152
153 Box::pin(async move {
154 total_bytes_received.fetch_add(msg.data.len(), Ordering::Relaxed);
155 })
156 }));
157 })
158 }));
159
160 Ok(pc)
161 }
162
163 #[tokio::main]
main() -> anyhow::Result<()>164 async fn main() -> anyhow::Result<()> {
165 env_logger::init();
166
167 let requester = Arc::new(create_requester().await?);
168 let responder = Arc::new(create_responder().await?);
169
170 let maybe_requester = Arc::downgrade(&requester);
171 responder.on_ice_candidate(Box::new(move |candidate: Option<RTCIceCandidate>| {
172 let maybe_requester = maybe_requester.clone();
173
174 Box::pin(async move {
175 if let Some(candidate) = candidate {
176 if let Ok(candidate) = candidate.to_json() {
177 if let Some(requester) = maybe_requester.upgrade() {
178 if let Err(err) = requester.add_ice_candidate(candidate).await {
179 log::warn!("{}", err);
180 }
181 }
182 }
183 }
184 })
185 }));
186
187 let maybe_responder = Arc::downgrade(&responder);
188 requester.on_ice_candidate(Box::new(move |candidate: Option<RTCIceCandidate>| {
189 let maybe_responder = maybe_responder.clone();
190
191 Box::pin(async move {
192 if let Some(candidate) = candidate {
193 if let Ok(candidate) = candidate.to_json() {
194 if let Some(responder) = maybe_responder.upgrade() {
195 if let Err(err) = responder.add_ice_candidate(candidate).await {
196 log::warn!("{}", err);
197 }
198 }
199 }
200 }
201 })
202 }));
203
204 let (fault, mut reqs_fault) = tokio::sync::mpsc::channel(1);
205 requester.on_peer_connection_state_change(Box::new(move |s: RTCPeerConnectionState| {
206 let fault = fault.clone();
207
208 Box::pin(async move {
209 if s == RTCPeerConnectionState::Failed {
210 fault.send(()).await.unwrap();
211 }
212 })
213 }));
214
215 let (fault, mut resp_fault) = tokio::sync::mpsc::channel(1);
216 responder.on_peer_connection_state_change(Box::new(move |s: RTCPeerConnectionState| {
217 let fault = fault.clone();
218
219 Box::pin(async move {
220 if s == RTCPeerConnectionState::Failed {
221 fault.send(()).await.unwrap();
222 }
223 })
224 }));
225
226 let reqs = requester.create_offer(None).await?;
227
228 requester.set_local_description(reqs.clone()).await?;
229 responder.set_remote_description(reqs).await?;
230
231 let resp = responder.create_answer(None).await?;
232
233 responder.set_local_description(resp.clone()).await?;
234 requester.set_remote_description(resp).await?;
235
236 tokio::select! {
237 _ = tokio::signal::ctrl_c() => {}
238 _ = reqs_fault.recv() => {
239 log::error!("Requester's peer connection failed...")
240 }
241 _ = resp_fault.recv() => {
242 log::error!("Responder's peer connection failed...");
243 }
244 }
245
246 requester.close().await?;
247 responder.close().await?;
248
249 println!();
250
251 Ok(())
252 }
253