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
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
#[cfg(test)]
mod connection {
use webrtc_connection::utils::test_helpers::create_connection_pair;
/// Base test that creates a connection that uses in-memory `DuplexStream`
/// for the signaling channel.
#[tokio::test]
async fn must_connect() {
// test that we establish a connection
create_connection_pair().await;
}
mod on_remote_channel_stream {
use webrtc_connection::utils::test_helpers::create_connection_pair;
#[tokio::test]
async fn expose_on_remote_channel_stream() {
let (mut client_connection, mut server_connection) = create_connection_pair().await;
let client_on_remote_channel = client_connection.on_remote_channel();
let server_on_remote_channel = server_connection.on_remote_channel();
assert!(
client_on_remote_channel.is_ok(),
"Client connection must expose \"on_remote_channel\" stream.",
);
assert!(
server_on_remote_channel.is_ok(),
"Server connection must expose \"on_remote_channel\" stream.",
);
assert!(
client_connection.on_remote_channel().is_err(),
"Client connection must expose \"on_remote_channel\" stream only once.",
);
assert!(
server_connection.on_remote_channel().is_err(),
"Server connection must expose \"on_remote_channel\" stream only once.",
);
client_connection.off_remote_channel(client_on_remote_channel.unwrap()).unwrap();
assert!(
client_connection.on_remote_channel().is_ok(),
"Client connection must be able to receive \"on_remote_channel\" listener back.",
);
assert!(
server_connection.on_remote_channel().is_err(),
"Server connection must stay as \"None\".",
);
server_connection.off_remote_channel(server_on_remote_channel.unwrap()).unwrap();
assert!(
server_connection.on_remote_channel().is_ok(),
"Server connection must be able to receive \"on_remote_channel\" listener back.",
);
}
}
/// Test the `on_remote_channel` logic, for the cases when the cahnnel
/// is initialized from the server, client or both sides.
mod events {
use tokio::try_join;
use connection_utils::Connected;
use cs_utils::{random_number, random_str};
use webrtc_connection::utils::test_helpers::create_connection_pair;
#[tokio::test]
async fn must_expose_on_data_channel_event_stream_server_initialized() {
// test that we establish a connection
let (mut client_connection, mut server_connection) = create_connection_pair().await;
let channel_name = format!("test channel {}", random_str(8));
let channel_name1 = channel_name.clone();
let channel_name2 = channel_name.clone();
let mut on_data_channel = client_connection.on_remote_channel()
.expect("Cannot get \"on_remote_channel\" stream.");
let _res = try_join!(
tokio::spawn(async move {
let channel = on_data_channel
.recv()
.await
.expect("Cannot receive remote channel creation event.");
assert_eq!(
*channel.label(),
channel_name1,
);
}),
tokio::spawn(async move {
let channel = server_connection
.channel(channel_name2.clone())
.await
.expect("Cannot create channel.");
assert_eq!(
*channel.label(),
channel_name2,
);
}),
).unwrap();
}
#[tokio::test]
async fn must_expose_on_data_channel_event_stream_client_initialized() {
// test that we establish a connection
let (mut client_connection, mut server_connection) = create_connection_pair().await;
let channel_name = format!("test channel {}", random_str(8));
let channel_name1 = channel_name.clone();
let channel_name2 = channel_name.clone();
let mut on_data_channel = server_connection.on_remote_channel()
.expect("Cannot get \"on_remote_channel\" stream.");
let _res = try_join!(
tokio::spawn(async move {
loop {
let channel = on_data_channel
.recv()
.await
.expect("Cannot receive remote channel creation event.");
// During the connection initialization phase, we need to create
// data channel, otherwise the connection won't be set up. That connection
// has the default label of "default_channel" so we filter this one out
// from our tests.
if channel.label() == "default_channel" {
continue;
}
assert_eq!(
*channel.label(),
channel_name1,
);
break;
}
}),
tokio::spawn(async move {
loop {
let channel = client_connection
.channel(channel_name2.clone())
.await
.expect("Cannot create channel.");
// During the connection initialization phase, we need to create
// data channel, otherwise the connection won't be set up. That connection
// has the default label of "default_channel" so we filter this one out
// from our tests.
if channel.label() == "default_channel" {
continue;
}
assert_eq!(
*channel.label(),
channel_name2,
);
break;
}
}),
).unwrap();
}
#[tokio::test]
async fn must_expose_on_data_channel_event_stream_bidirectional() {
// test that we establish a connection
let (client_connection, server_connection) = create_connection_pair().await;
fn create_channel_labels(
prefix: String,
channels_count: usize,
) -> Vec<String> {
let mut res = vec![];
for _i in 0..channels_count {
res.push(format!("{}_{}", &prefix, random_str(6)));
}
return res;
}
let mut client_create_channels = vec!["default_channel".to_string()];
client_create_channels
.append(&mut create_channel_labels(
"client_channel".to_string(),
random_number(1..=5),
));
let client_create_channels1 = client_create_channels.clone();
let client_create_channels2 = client_create_channels.clone();
let server_create_channels = create_channel_labels(
"server_channel".to_string(),
random_number(1..=5),
);
let server_create_channels1 = server_create_channels.clone();
let server_create_channels2 = server_create_channels.clone();
async fn test_connection_channels(
mut connection: Box<dyn Connected>,
create_channel_labels: Vec<String>,
receive_channel_labels: Vec<String>,
) {
let mut on_remote_channel = connection.on_remote_channel()
.expect("Cannot get \"on_remote_channel\" stream.");
let mut received_channels = vec![];
let _res = try_join!(
tokio::spawn(async move {
loop {
let channel = on_remote_channel
.recv()
.await
.expect("Cannot receive remote channel creation event.");
received_channels.push(channel.label().clone());
if received_channels.len() == receive_channel_labels.len() {
assert_eq!(
received_channels.join(""),
receive_channel_labels.join(""),
"Received channels must match.",
);
break;
}
}
}),
tokio::spawn(async move {
for label in create_channel_labels {
let channel = connection
.channel(label.clone())
.await
.expect(
&format!("Cannot create channel [{}]", &label),
);
assert_eq!(
*channel.label(),
label,
"Created channel labels must match.",
);
}
}),
).unwrap();
}
let _res = try_join!(
tokio::spawn(async move {
test_connection_channels(
client_connection,
client_create_channels1,
server_create_channels1,
).await;
}),
tokio::spawn(async move {
test_connection_channels(
server_connection,
server_create_channels2,
client_create_channels2,
).await;
}),
).unwrap();
}
}
}