Skip to content

Commit b84fe75

Browse files
committed
Merge remote-tracking branch 'origin/main' into fix/client-robustness
# Conflicts: # webrtc/src/main/java/dev/onvoid/webrtc/client/WebRtc.java
2 parents 500f87b + 299805a commit b84fe75

11 files changed

Lines changed: 504 additions & 101 deletions

File tree

‎docs/guide/client-api.md‎

Lines changed: 5 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -203,6 +203,8 @@ Every method of the client API may be called from any thread, including from ins
203203

204204
The handlers of one peer never run at the same time, and they run in the order of the events. The handlers of different peers may run at the same time. With the default executor, the event thread is named after its peer, such as `webrtc-client-events-alice`, which shows up in thread dumps.
205205

206+
An executor of your own that rejects work, such as a bounded pool that is full, loses the events it rejects, and the rejection goes to the error handler. The futures those events would have completed complete all the same, on the thread that found them rejected. Once the runtime is closed, every future of this API that is not done fails with `PeerClosedException`, including those of calls made after the close.
207+
206208
### Handlers on the UI Thread
207209

208210
Pass the UI toolkit's executor, and every handler may touch UI components directly:
@@ -540,7 +542,9 @@ viewer.peer().onTrack(track -> System.out.println("Receiving " + track.kind()));
540542

541543
The offer goes to the endpoint with the local candidates, as WHIP and WHEP servers expect, and the future completes once the peer is connected. If the endpoint refuses the offer, or the peer does not connect within the timeout (15 seconds by default), the future fails, and the session is deleted on the server again. `onTrack` hands over tracks that already arrived, so a handler added once the future completed misses none.
542544

543-
`session.close()` deletes the session on the server, which frees what the server holds for it, and closes the peer. The tracks a WHIP session published stay open. `session.peer()` is a regular peer, for stats and anything else.
545+
`session.close()` deletes the session on the server, which frees what the server holds for it, and closes the peer. The tracks a WHIP session published stay open. `session.peer()` is a regular peer, for stats and anything else. A deletion the server refuses or that fails goes to the runtime's error handler.
546+
547+
The bearer token goes to the endpoint, and along with the deletion to the session the endpoint names in its `Location` header. That session may be on another host, but an `https` endpoint cannot name a plain `http` one. Redirects are not followed, since they would take the token elsewhere: a redirected request fails with its status.
544548

545549
| Builder method | What it sets |
546550
|---|---|

‎webrtc/src/main/java/dev/onvoid/webrtc/client/DataChannelImpl.java‎

Lines changed: 21 additions & 9 deletions
Original file line numberDiff line numberDiff line change
@@ -100,6 +100,8 @@ final class DataChannelImpl implements DataChannel {
100100
this.sendLimit = peer.channelSendLimit();
101101
this.messageHandlers = new Handlers<>(peer.rtc::reportError);
102102
this.stateHandlers = new Handlers<>(peer.rtc::reportError);
103+
104+
peer.rtc.track(opened, peer.events);
103105
}
104106

105107
/**
@@ -260,7 +262,8 @@ void closeFromPeer(WebRtcException error) {
260262
state = ChannelState.CLOSED;
261263
peer.channelClosed(this);
262264

263-
peer.events.execute(() -> deliverState(ChannelState.CLOSED, error));
265+
peer.events.execute(() -> deliverState(ChannelState.CLOSED, error),
266+
() -> settleOpened(ChannelState.CLOSED, error));
264267
}
265268

266269
private CompletableFuture<Void> enqueue(byte[] data, boolean binary) {
@@ -364,7 +367,7 @@ else if (newState == ChannelState.CLOSING) {
364367
failPending(new PeerClosedException("The data channel " + label + " is closing"));
365368
}
366369

367-
peer.events.execute(() -> deliverState(newState, null));
370+
peer.events.execute(() -> deliverState(newState, null), () -> settleOpened(newState, null));
368371
}
369372

370373
private void bufferChanged() {
@@ -375,21 +378,30 @@ private void bufferChanged() {
375378

376379
// On the event executor.
377380

381+
/**
382+
* Completes the opened future for the state, which is all that is left of
383+
* delivering it when the event executor rejects that.
384+
*/
385+
private void settleOpened(ChannelState newState, WebRtcException error) {
386+
if (newState == ChannelState.OPEN) {
387+
opened.complete(this);
388+
}
389+
else if (newState == ChannelState.CLOSED) {
390+
opened.completeExceptionally(error != null ? error
391+
: new PeerClosedException("The data channel " + label + " is closed"));
392+
}
393+
}
394+
378395
private void deliverState(ChannelState newState, WebRtcException error) {
379396
if (newState == deliveredState || deliveredState == ChannelState.CLOSED) {
380397
return;
381398
}
382399

383400
deliveredState = newState;
384401
stateHandlers.accept(newState);
402+
settleOpened(newState, error);
385403

386-
if (newState == ChannelState.OPEN) {
387-
opened.complete(this);
388-
}
389-
else if (newState == ChannelState.CLOSED) {
390-
opened.completeExceptionally(error != null ? error
391-
: new PeerClosedException("The data channel " + label + " is closed"));
392-
404+
if (newState == ChannelState.CLOSED) {
393405
messageHandlers.clear();
394406
stateHandlers.clear();
395407
early.clear();

‎webrtc/src/main/java/dev/onvoid/webrtc/client/DevicesImpl.java‎

Lines changed: 1 addition & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -227,6 +227,7 @@ private static List<DesktopSource> desktopSources(DesktopCapturer capturer) {
227227

228228
private <T> CompletableFuture<T> onMediaExecutor(Supplier<T> action) {
229229
CompletableFuture<T> result = new CompletableFuture<>();
230+
media.rtc.track(result, null);
230231

231232
media.executor.execute(() -> {
232233
try {

‎webrtc/src/main/java/dev/onvoid/webrtc/client/HttpSession.java‎

Lines changed: 141 additions & 54 deletions
Original file line numberDiff line numberDiff line change
@@ -27,6 +27,8 @@
2727
import java.nio.charset.StandardCharsets;
2828
import java.time.Duration;
2929
import java.util.concurrent.CompletableFuture;
30+
import java.util.concurrent.RejectedExecutionException;
31+
import java.util.concurrent.ScheduledFuture;
3032
import java.util.concurrent.TimeUnit;
3133
import java.util.concurrent.atomic.AtomicBoolean;
3234
import java.util.concurrent.atomic.AtomicReference;
@@ -42,6 +44,12 @@
4244
*/
4345
abstract class HttpSession implements AutoCloseable {
4446

47+
/** The largest answer read; an SDP answer is a few kilobytes. */
48+
private static final int MAX_BODY = 1 << 20;
49+
50+
/** How much of an error response goes into a message. */
51+
private static final int MAX_ERROR_BODY = 1024;
52+
4553
private final Peer peer;
4654

4755
private final URI resource;
@@ -88,17 +96,18 @@ public URI resource() {
8896
*/
8997
public CompletableFuture<Void> closeAsync() {
9098
if (closing.compareAndSet(false, true)) {
91-
CompletableFuture.runAsync(() -> {
92-
try {
93-
if (resource != null) {
94-
delete(resource, bearerToken, timeout);
95-
}
96-
}
97-
catch (Exception e) {
98-
rtc.reportError(new WebRtcException("Deleting the session " + resource + " failed", e));
99-
}
100-
}, rtc.controlPool)
101-
.thenCompose(ignored -> peer.closeAsync())
99+
CompletableFuture<Void> deleted;
100+
101+
try {
102+
deleted = CompletableFuture.runAsync(() -> discard(rtc, resource, bearerToken, timeout),
103+
rtc.controlPool);
104+
}
105+
catch (RejectedExecutionException e) {
106+
leftOnServer(rtc, resource);
107+
deleted = CompletableFuture.completedFuture(null);
108+
}
109+
110+
deleted.thenCompose(ignored -> peer.closeAsync())
102111
.whenComplete((ignored, error) -> closed.complete(null));
103112
}
104113

@@ -122,6 +131,8 @@ public void close() {
122131
static CompletableFuture<URI> negotiate(WebRtc rtc, Peer peer, URI endpoint, String bearerToken,
123132
Duration timeout) {
124133
CompletableFuture<URI> result = new CompletableFuture<>();
134+
// The resource the server created. Whoever takes it out owns it: the
135+
// session that connected, or whoever deletes it.
125136
AtomicReference<URI> resource = new AtomicReference<>();
126137
AtomicBoolean finished = new AtomicBoolean();
127138

@@ -131,50 +142,66 @@ static CompletableFuture<URI> negotiate(WebRtc rtc, Peer peer, URI endpoint, Str
131142
return;
132143
}
133144

134-
// Leaves nothing behind, on the server or here.
135-
CompletableFuture.runAsync(() -> {
136-
try {
137-
URI created = resource.get();
145+
// Reported before the resource is deleted, which may take as long
146+
// as the timeout again.
147+
peer.close();
148+
result.completeExceptionally(PeerImpl.asWebRtcException(error,
149+
"The session with " + endpoint + " could not be set up"));
138150

139-
if (created != null) {
140-
delete(created, bearerToken, timeout);
141-
}
151+
URI created = resource.getAndSet(null);
152+
153+
if (created != null) {
154+
try {
155+
CompletableFuture.runAsync(() -> discard(rtc, created, bearerToken, timeout), rtc.controlPool);
142156
}
143-
catch (Exception e) {
144-
// The failure that ended the session is the one to report.
157+
catch (RejectedExecutionException e) {
158+
leftOnServer(rtc, created);
145159
}
146-
147-
peer.close();
148-
result.completeExceptionally(PeerImpl.asWebRtcException(error,
149-
"The session with " + endpoint + " could not be set up"));
150-
}, rtc.controlPool);
160+
}
151161
};
152162

163+
ScheduledFuture<?> deadline = rtc.timer.schedule(() -> fail.accept(new WebRtcException(
164+
"The session with " + endpoint + " was not connected within " + timeout)),
165+
timeout.toMillis(), TimeUnit.MILLISECONDS);
166+
153167
peer.negotiator().createOffer(Gathering.COMPLETE)
154168
.thenApplyAsync(offer -> {
155-
Response response = post(endpoint, offer.sdp, bearerToken, timeout);
156-
resource.set(response.location);
157-
return response.body;
169+
try {
170+
return post(endpoint, offer.sdp, bearerToken, timeout, resource);
171+
}
172+
finally {
173+
if (finished.get()) {
174+
// The session ended while the offer was out, so what
175+
// the server created meanwhile has no one else to
176+
// delete it. The answer then meets a closed peer.
177+
discard(rtc, resource.getAndSet(null), bearerToken, timeout);
178+
}
179+
}
158180
}, rtc.controlPool)
159181
.thenCompose(answer -> peer.negotiator()
160182
.applyAnswer(new RTCSessionDescription(RTCSdpType.ANSWER, answer)))
161183
.thenCompose(applied -> peer.whenConnected())
162184
.whenComplete((connected, error) -> {
185+
deadline.cancel(false);
186+
163187
if (error != null) {
164188
fail.accept(error);
165189
}
166190
else if (finished.compareAndSet(false, true)) {
167-
result.complete(resource.get());
191+
result.complete(resource.getAndSet(null));
168192
}
169193
});
170194

171-
rtc.timer.schedule(() -> fail.accept(new WebRtcException("The session with " + endpoint
172-
+ " was not connected within " + timeout)), timeout.toMillis(), TimeUnit.MILLISECONDS);
173-
174195
return result;
175196
}
176197

177-
private static Response post(URI endpoint, String sdp, String bearerToken, Duration timeout) {
198+
/**
199+
* Posts the offer and returns the answer. The resource the server names is
200+
* set as soon as it is known, so that it is deleted even if reading the
201+
* answer fails.
202+
*/
203+
private static String post(URI endpoint, String sdp, String bearerToken, Duration timeout,
204+
AtomicReference<URI> resource) {
178205
HttpURLConnection connection = null;
179206

180207
try {
@@ -194,13 +221,12 @@ private static Response post(URI endpoint, String sdp, String bearerToken, Durat
194221

195222
if (status != HttpURLConnection.HTTP_CREATED && status != HttpURLConnection.HTTP_OK) {
196223
throw new WebRtcException("The endpoint " + endpoint + " answered the offer with status " + status
197-
+ ": " + read(connection.getErrorStream()));
224+
+ ": " + read(connection.getErrorStream(), MAX_ERROR_BODY, true));
198225
}
199226

200-
String location = connection.getHeaderField("Location");
227+
resource.set(resourceOf(endpoint, connection.getHeaderField("Location")));
201228

202-
return new Response(read(connection.getInputStream()),
203-
location == null ? null : endpoint.resolve(location));
229+
return read(connection.getInputStream(), MAX_BODY, false);
204230
}
205231
catch (IOException e) {
206232
throw new WebRtcException("Sending the offer to " + endpoint + " failed: " + e.getMessage(), e);
@@ -216,19 +242,82 @@ private static void delete(URI resource, String bearerToken, Duration timeout) t
216242
HttpURLConnection connection = open(resource, "DELETE", bearerToken, timeout);
217243

218244
try {
219-
connection.getResponseCode();
245+
int status = connection.getResponseCode();
246+
247+
if (status / 100 != 2) {
248+
throw new WebRtcException("The endpoint answered the deletion of " + resource + " with status "
249+
+ status + ": " + read(connection.getErrorStream(), MAX_ERROR_BODY, true));
250+
}
220251
}
221252
finally {
222253
connection.disconnect();
223254
}
224255
}
225256

257+
/**
258+
* Deletes a resource that nothing else will, and reports rather than
259+
* throws if that fails.
260+
*/
261+
private static void discard(WebRtc rtc, URI resource, String bearerToken, Duration timeout) {
262+
if (resource == null) {
263+
return;
264+
}
265+
266+
try {
267+
delete(resource, bearerToken, timeout);
268+
}
269+
catch (Exception e) {
270+
rtc.reportError(new WebRtcException("Deleting the session " + resource + " failed", e));
271+
}
272+
}
273+
274+
private static void leftOnServer(WebRtc rtc, URI resource) {
275+
if (resource != null) {
276+
rtc.reportError(new WebRtcException("The session " + resource
277+
+ " was left on the server, since the runtime is closed"));
278+
}
279+
}
280+
281+
/**
282+
* Resolves the resource the server named for a session. It may be on
283+
* another host, as WHIP and WHEP allow, but the bearer token goes along to
284+
* delete it, so an https endpoint cannot name a plain http resource.
285+
*/
286+
static URI resourceOf(URI endpoint, String location) {
287+
if (location == null) {
288+
return null;
289+
}
290+
291+
URI resource;
292+
293+
try {
294+
resource = endpoint.resolve(location.trim());
295+
}
296+
catch (IllegalArgumentException e) {
297+
throw new WebRtcException("The endpoint " + endpoint + " named an invalid resource: " + location, e);
298+
}
299+
300+
String scheme = resource.getScheme();
301+
302+
if (!"http".equalsIgnoreCase(scheme) && !"https".equalsIgnoreCase(scheme)) {
303+
throw new WebRtcException("The endpoint " + endpoint + " named a resource that is not http: " + resource);
304+
}
305+
if ("https".equalsIgnoreCase(endpoint.getScheme()) && !"https".equalsIgnoreCase(scheme)) {
306+
throw new WebRtcException("The endpoint " + endpoint + " named a resource without TLS: " + resource);
307+
}
308+
309+
return resource;
310+
}
311+
226312
private static HttpURLConnection open(URI uri, String method, String bearerToken, Duration timeout)
227313
throws IOException {
228314
HttpURLConnection connection = (HttpURLConnection) uri.toURL().openConnection();
229315
connection.setRequestMethod(method);
230316
connection.setConnectTimeout((int) Math.min(Integer.MAX_VALUE, timeout.toMillis()));
231317
connection.setReadTimeout((int) Math.min(Integer.MAX_VALUE, timeout.toMillis()));
318+
// A redirect would take the bearer token to wherever it points, so it
319+
// fails the request instead.
320+
connection.setInstanceFollowRedirects(false);
232321

233322
if (bearerToken != null) {
234323
connection.setRequestProperty("Authorization", "Bearer " + bearerToken);
@@ -237,7 +326,11 @@ private static HttpURLConnection open(URI uri, String method, String bearerToken
237326
return connection;
238327
}
239328

240-
private static String read(InputStream in) throws IOException {
329+
/**
330+
* Reads a response body of at most the given size. A larger one fails, or
331+
* is cut off when only a message is made of it.
332+
*/
333+
private static String read(InputStream in, int limit, boolean truncate) throws IOException {
241334
if (in == null) {
242335
return "";
243336
}
@@ -248,6 +341,15 @@ private static String read(InputStream in) throws IOException {
248341
int count;
249342

250343
while ((count = stream.read(buffer)) != -1) {
344+
if (out.size() + count > limit) {
345+
if (!truncate) {
346+
throw new IOException("The response is larger than " + limit + " bytes");
347+
}
348+
349+
out.write(buffer, 0, limit - out.size());
350+
break;
351+
}
352+
251353
out.write(buffer, 0, count);
252354
}
253355

@@ -266,19 +368,4 @@ static URI checkEndpoint(URI endpoint) {
266368

267369
return endpoint;
268370
}
269-
270-
271-
272-
private static final class Response {
273-
274-
final String body;
275-
276-
final URI location;
277-
278-
279-
Response(String body, URI location) {
280-
this.body = body;
281-
this.location = location;
282-
}
283-
}
284371
}

0 commit comments

Comments
 (0)