diff --git a/docs/guide/client-api.md b/docs/guide/client-api.md index 865dedbe..78e29315 100644 --- a/docs/guide/client-api.md +++ b/docs/guide/client-api.md @@ -655,7 +655,8 @@ peer.onStats(Duration.ofSeconds(1), stats -> System.out.printf( | `candidateType()` | The type of the local candidate in use: `host`, `srflx`, `prflx` or `relay`. Null while not connected. | | `roundTripTime()` | The current round trip time, or null while not measured. | | `bytesSent()`, `bytesReceived()` | The bytes sent and received on the connection so far. | -| `outgoingBitrate()`, `incomingBitrate()` | Bits per second since the previous sample, or -1 for a peer's first sample. | +| `timestamp()` | When WebRTC measured the statistics. | +| `outgoingBitrate()`, `incomingBitrate()` | Bits per second between the measurement times of the previous sample and this one, or -1 when there is none to compare with: for a peer's first sample, after the clock went back, or after the byte counter started again. A report WebRTC hands out again within its short cache lifetime repeats the previous bitrates. | | `packetLoss()` | The share of received packets lost so far, 0 to 1. | | `outboundVideo()`, `inboundVideo()` | Codec, resolution and frame rate of each video stream, and what limits a sent one. | | `report()` | The whole `RTCStatsReport`. | diff --git a/webrtc/src/main/java/dev/onvoid/webrtc/client/PeerImpl.java b/webrtc/src/main/java/dev/onvoid/webrtc/client/PeerImpl.java index 21f75d03..03e88af7 100644 --- a/webrtc/src/main/java/dev/onvoid/webrtc/client/PeerImpl.java +++ b/webrtc/src/main/java/dev/onvoid/webrtc/client/PeerImpl.java @@ -481,7 +481,7 @@ public Subscription onStats(Duration interval, Consumer handler) { return; } if (error == null) { - handler.accept(stats); + call(handler, stats); } else if (!(unwrap(error) instanceof PeerClosedException)) { deliverError(asWebRtcException(error, "Reading the statistics failed")); diff --git a/webrtc/src/main/java/dev/onvoid/webrtc/client/PeerStats.java b/webrtc/src/main/java/dev/onvoid/webrtc/client/PeerStats.java index d0580d7f..13859f5e 100644 --- a/webrtc/src/main/java/dev/onvoid/webrtc/client/PeerStats.java +++ b/webrtc/src/main/java/dev/onvoid/webrtc/client/PeerStats.java @@ -24,8 +24,8 @@ /** * The statistics people look for first, taken from an {@link RTCStatsReport}. - * Bitrates are computed between two samples, so the first sample a peer takes - * has none. + * Bitrates are computed between the measurement times of two samples, so the + * first sample a peer takes has none. */ public final class PeerStats { @@ -71,7 +71,8 @@ public final class PeerStats { } /** - * @return when the statistics were taken. + * @return when WebRTC measured the statistics, the timestamp of the + * report. */ public Instant timestamp() { return timestamp; @@ -107,8 +108,9 @@ public long bytesReceived() { } /** - * @return the bits sent per second since the previous sample, or -1 for - * the first sample. + * @return the bits sent per second since the previous sample, or -1 when + * there is none to compare with: for the first sample, after the clock + * went back, or after the byte counter started again. */ public long outgoingBitrate() { return outgoingBitrate; @@ -116,7 +118,8 @@ public long outgoingBitrate() { /** * @return the bits received per second since the previous sample, or -1 - * for the first sample. + * when there is none to compare with: for the first sample, after the + * clock went back, or after the byte counter started again. */ public long incomingBitrate() { return incomingBitrate; diff --git a/webrtc/src/main/java/dev/onvoid/webrtc/client/StatsReader.java b/webrtc/src/main/java/dev/onvoid/webrtc/client/StatsReader.java index 0e4ba125..aef8e496 100644 --- a/webrtc/src/main/java/dev/onvoid/webrtc/client/StatsReader.java +++ b/webrtc/src/main/java/dev/onvoid/webrtc/client/StatsReader.java @@ -18,6 +18,7 @@ import java.time.Duration; import java.time.Instant; +import java.time.temporal.ChronoUnit; import java.util.ArrayList; import java.util.Collections; import java.util.List; @@ -29,16 +30,22 @@ /** * Takes {@link PeerStats} from the reports of one peer. It keeps the previous - * sample to compute bitrates. Used on the peer's control executor. + * sample to compute bitrates over the time between the two measurements, not + * between their arrival here, which a busy control executor may delay. Used on + * the peer's control executor. */ final class StatsReader { - private long lastNanos = -1; + private long lastMicros = -1; private long lastBytesSent; private long lastBytesReceived; + private long lastOutgoingBitrate = -1; + + private long lastIncomingBitrate = -1; + PeerStats read(RTCStatsReport report) { Map stats = report.getStats(); @@ -93,29 +100,49 @@ else if (entry.getType() == RTCStatsType.OUTBOUND_RTP) { } } - long now = System.nanoTime(); + long micros = report.getTimestamp(); long outgoingBitrate = -1; long incomingBitrate = -1; - if (lastNanos >= 0 && now > lastNanos) { - double seconds = (now - lastNanos) / 1e9; - - outgoingBitrate = (long) (Math.max(0, bytesSent - lastBytesSent) * 8 / seconds); - incomingBitrate = (long) (Math.max(0, bytesReceived - lastBytesReceived) * 8 / seconds); + if (lastMicros >= 0 && micros == lastMicros) { + // WebRTC hands out a report again for a short while: the same + // measurement, with the same bitrates, measured from the same + // sample. + outgoingBitrate = lastOutgoingBitrate; + incomingBitrate = lastIncomingBitrate; } + else { + // A clock that went back leaves no interval to measure, and the + // next sample is measured from this one. + if (lastMicros >= 0 && micros > lastMicros) { + double seconds = (micros - lastMicros) / 1e6; + + outgoingBitrate = bitrate(bytesSent, lastBytesSent, seconds); + incomingBitrate = bitrate(bytesReceived, lastBytesReceived, seconds); + } - lastNanos = now; - lastBytesSent = bytesSent; - lastBytesReceived = bytesReceived; + lastMicros = micros; + lastBytesSent = bytesSent; + lastBytesReceived = bytesReceived; + lastOutgoingBitrate = outgoingBitrate; + lastIncomingBitrate = incomingBitrate; + } long packetsExpected = packetsReceived + packetsLost; double packetLoss = packetsExpected == 0 ? 0 : (double) packetsLost / packetsExpected; - return new PeerStats(Instant.now(), candidateType, roundTripTime, bytesSent, bytesReceived, + return new PeerStats(Instant.EPOCH.plus(micros, ChronoUnit.MICROS), candidateType, roundTripTime, bytesSent, bytesReceived, outgoingBitrate, incomingBitrate, packetLoss, Collections.unmodifiableList(outbound), Collections.unmodifiableList(inbound), report); } + private static long bitrate(long bytes, long lastBytes, double seconds) { + // Fewer bytes than before: the counter started again, such as on a + // transport that replaced another, and what was sent meanwhile is + // unknown. + return bytes < lastBytes ? -1 : (long) ((bytes - lastBytes) * 8 / seconds); + } + private static VideoStats video(Map stats, Map values, String trackId, String qualityLimitation) { RTCStats codec = lookup(stats, values.get("codecId")); diff --git a/webrtc/src/test/java/dev/onvoid/webrtc/client/PeerTests.java b/webrtc/src/test/java/dev/onvoid/webrtc/client/PeerTests.java index 685f5cf9..7fd31d28 100644 --- a/webrtc/src/test/java/dev/onvoid/webrtc/client/PeerTests.java +++ b/webrtc/src/test/java/dev/onvoid/webrtc/client/PeerTests.java @@ -24,6 +24,8 @@ import static org.junit.jupiter.api.Assertions.assertTrue; import java.time.Duration; +import java.time.Instant; +import java.time.temporal.ChronoUnit; import java.util.ArrayList; import java.util.Arrays; import java.util.Collections; @@ -551,8 +553,54 @@ void statsAfterConnecting() throws Exception { assertEquals("host", first.candidateType()); assertTrue(first.bytesSent() > 10_000, () -> "Bytes sent: " + first.bytesSent()); + assertEquals(first.report().getTimestamp(), ChronoUnit.MICROS.between(Instant.EPOCH, first.timestamp())); assertEquals(-1, first.outgoingBitrate()); - assertTrue(second.outgoingBitrate() >= 0); + // Asked for right after the first, the second may be the report WebRTC + // still caches, which repeats the first one's bitrate. + if (second.timestamp().equals(first.timestamp())) { + assertEquals(-1, second.outgoingBitrate()); + } + else { + assertTrue(second.outgoingBitrate() >= 0); + } + } + + @Test + void statsHandlerErrorReported() throws Exception { + // A stats handler that throws reaches the runtime's error handler, and + // the samples after it still come. + ManualTimer timer = new ManualTimer(); + CompletableFuture reported = new CompletableFuture<>(); + Duration interval = Duration.ofSeconds(7); + + try (WebRtc runtime = WebRtc.builder() + .audio(AudioMode.HEADLESS) + .errorHandler(error -> { + errors.add(error); + reported.complete(error); + }) + .timer(timer) + .build()) { + Peer peer = runtime.createPeer(manualOptions()); + AtomicInteger calls = new AtomicInteger(); + CompletableFuture later = new CompletableFuture<>(); + + peer.onStats(interval, stats -> { + if (calls.incrementAndGet() == 1) { + throw new IllegalStateException("Handler failed"); + } + later.complete(stats); + }); + + assertEquals(1, timer.fire(interval)); + Throwable error = await(reported); + + assertInstanceOf(IllegalStateException.class, error); + errors.remove(error); + + assertEquals(1, timer.fire(interval)); + await(later); + } } @Test diff --git a/webrtc/src/test/java/dev/onvoid/webrtc/client/StatsReaderTests.java b/webrtc/src/test/java/dev/onvoid/webrtc/client/StatsReaderTests.java new file mode 100644 index 00000000..f6f62f8c --- /dev/null +++ b/webrtc/src/test/java/dev/onvoid/webrtc/client/StatsReaderTests.java @@ -0,0 +1,120 @@ +/* + * Copyright 2026 Alex Andres + * + * Licensed under the Apache License, Version 2.0 (the "License"); + * you may not use this file except in compliance with the License. + * You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +package dev.onvoid.webrtc.client; + +import static org.junit.jupiter.api.Assertions.assertEquals; + +import java.time.Instant; +import java.util.HashMap; +import java.util.Map; + +import dev.onvoid.webrtc.RTCStats; +import dev.onvoid.webrtc.RTCStatsReport; +import dev.onvoid.webrtc.RTCStatsType; + +import org.junit.jupiter.api.Test; + +class StatsReaderTests { + + private static final long START = 1_700_000_000_000_000L; + + private static final long SECOND = 1_000_000; + + private final StatsReader reader = new StatsReader(); + + + @Test + void bitrateFromReportTimes() { + // Read one right after the other, as a busy control executor hands + // them on: the reports were measured a second apart. + PeerStats first = reader.read(report(START, 5_000, 2_000)); + PeerStats second = reader.read(report(START + SECOND, 6_000, 2_500)); + + assertEquals(-1, first.outgoingBitrate()); + assertEquals(-1, first.incomingBitrate()); + assertEquals(8_000, second.outgoingBitrate()); + assertEquals(4_000, second.incomingBitrate()); + } + + @Test + void timestampOfReport() { + PeerStats stats = reader.read(report(START + 250, 0, 0)); + + assertEquals(Instant.ofEpochSecond(1_700_000_000, 250_000), stats.timestamp()); + } + + @Test + void repeatedReportKeepsBitrates() { + // WebRTC hands out the report it cached again. The one after it is + // measured from the sample before. + reader.read(report(START, 0, 0)); + reader.read(report(START + SECOND, 1_000, 1_000)); + PeerStats repeated = reader.read(report(START + SECOND, 1_000, 1_000)); + PeerStats next = reader.read(report(START + 2 * SECOND, 3_000, 1_000)); + + assertEquals(8_000, repeated.outgoingBitrate()); + assertEquals(8_000, repeated.incomingBitrate()); + assertEquals(16_000, next.outgoingBitrate()); + assertEquals(0, next.incomingBitrate()); + } + + @Test + void repeatedFirstReport() { + reader.read(report(START, 1_000, 1_000)); + PeerStats repeated = reader.read(report(START, 1_000, 1_000)); + + assertEquals(-1, repeated.outgoingBitrate()); + assertEquals(-1, repeated.incomingBitrate()); + } + + @Test + void clockWentBack() { + reader.read(report(START, 0, 0)); + PeerStats back = reader.read(report(START - SECOND, 1_000, 1_000)); + PeerStats next = reader.read(report(START, 2_000, 3_000)); + + assertEquals(-1, back.outgoingBitrate()); + assertEquals(-1, back.incomingBitrate()); + assertEquals(8_000, next.outgoingBitrate()); + assertEquals(16_000, next.incomingBitrate()); + } + + @Test + void counterStartedAgain() { + // Fewer bytes sent than before. Received counts on. + reader.read(report(START, 10_000, 1_000)); + PeerStats reset = reader.read(report(START + SECOND, 500, 2_000)); + PeerStats next = reader.read(report(START + 2 * SECOND, 1_500, 2_000)); + + assertEquals(-1, reset.outgoingBitrate()); + assertEquals(8_000, reset.incomingBitrate()); + assertEquals(8_000, next.outgoingBitrate()); + assertEquals(0, next.incomingBitrate()); + } + + private static RTCStatsReport report(long micros, long bytesSent, long bytesReceived) { + Map values = new HashMap<>(); + values.put("bytesSent", bytesSent); + values.put("bytesReceived", bytesReceived); + + Map stats = new HashMap<>(); + stats.put("T01", new RTCStats(micros, RTCStatsType.TRANSPORT, "T01", values) { }); + + return new RTCStatsReport(stats, micros) { }; + } + +}