From e6e06f711b93de2d611e8b9479c4022406459256 Mon Sep 17 00:00:00 2001 From: Alex Andres Date: Thu, 29 Oct 2026 08:17:21 +0100 Subject: [PATCH] feat: encoded frame transforms in the client API PeerOptions take a send and a receive transform, set on each sender before negotiation and on each receiver before onTrack. Sender.transform and RemoteTrack.transform replace them for one sender or track. EncryptedRecordingExample now uses the client API. --- docs/guide/advanced/encoded-transforms.md | 16 +- docs/guide/client-api.md | 26 ++ docs/guide/examples.md | 6 +- .../examples/EncryptedRecordingExample.java | 379 ++++++------------ .../dev/onvoid/webrtc/client/PeerImpl.java | 48 ++- .../dev/onvoid/webrtc/client/PeerOptions.java | 55 +++ .../dev/onvoid/webrtc/client/RemoteTrack.java | 23 ++ .../onvoid/webrtc/client/RemoteTrackImpl.java | 26 ++ .../java/dev/onvoid/webrtc/client/Sender.java | 21 + .../dev/onvoid/webrtc/client/SenderImpl.java | 22 +- .../onvoid/webrtc/client/TransformTests.java | 304 ++++++++++++++ 11 files changed, 662 insertions(+), 264 deletions(-) create mode 100644 webrtc/src/test/java/dev/onvoid/webrtc/client/TransformTests.java diff --git a/docs/guide/advanced/encoded-transforms.md b/docs/guide/advanced/encoded-transforms.md index a9f0d26b3..5285a7991 100644 --- a/docs/guide/advanced/encoded-transforms.md +++ b/docs/guide/advanced/encoded-transforms.md @@ -41,6 +41,20 @@ Set the transform of a video sender before negotiating when you can. The first t The transform belongs to the native sender or receiver, not to the Java object: every `RTCRtpSender` instance for the same sender shares it, setting it through one replaces what was set through another, and disposing of an instance leaves it in place. +### With the Client API + +The [Client API](/guide/client-api#encoded-transforms) takes the same transforms. Those of the peer options apply to every sender and remote track of the peer, and are set before the first frame passes either way: + +```java +PeerOptions.builder() + .signaling(signaling) + .sendTransform(frame -> encrypt(frame)) + .receiveTransform(frame -> decrypt(frame)) + .build(); +``` + +`Sender.transform(...)` and `RemoteTrack.transform(...)` replace them for one sender or track. + ## Reading and Changing the Payload `getData()` returns the payload in a `ByteBuffer`, from position zero to its size. The buffer holds a copy, which may be changed freely; `setData(...)` then writes it back: @@ -91,7 +105,7 @@ Both do nothing for audio. ## End-to-End Encryption -A complete example with AES-GCM is in [`EncryptedRecordingExample`](https://github.com/devopvoid/webrtc-java/blob/master/webrtc-examples/src/main/java/dev/onvoid/webrtc/examples/EncryptedRecordingExample.java). Two points matter for any encrypting transform: +A complete example with AES-GCM, on the transforms of the client API's peer options, is in [`EncryptedRecordingExample`](https://github.com/devopvoid/webrtc-java/blob/master/webrtc-examples/src/main/java/dev/onvoid/webrtc/examples/EncryptedRecordingExample.java). Two points matter for any encrypting transform: - **Keep the codec header in the clear.** The packetizer and depacketizer read the start of each frame: a VP8 receiver finds key frames and the frame size in the first bytes. Browsers leave 10 bytes of a VP8 key frame, 3 of a VP8 delta frame and 1 of an Opus frame unencrypted, and authenticate them as associated data instead. H.264 is split at its NAL unit start codes before it is sent, so encrypting it whole breaks the packetizer; prefer VP8, VP9 or AV1 for encrypted calls, or leave the NAL unit headers in the clear. - **Make each frame self-contained.** Frames are lost, so each needs what it takes to be decrypted, such as its IV, and a receiver must drop a frame that fails to authenticate rather than pass it on. diff --git a/docs/guide/client-api.md b/docs/guide/client-api.md index 974769d32..4a1ab00b5 100644 --- a/docs/guide/client-api.md +++ b/docs/guide/client-api.md @@ -204,6 +204,7 @@ Every method of the client API may be called from any thread, including from ins | Your calls into the API | your thread | They don't block. The exceptions are `WebRtc.createPeer`, which waits briefly for the connection, and `WebRtc.close()`, which waits up to the close timeout. | | Peer, track and channel handlers, future callbacks, `Signaling.send` | the peer's event thread | Yes, but that delays only the events of this peer. They must not wait for a future of this API, see below. | | `onFrame` and `onAudio` handlers | a thread of each subscription | Better not. A slow video handler gets the newest frame next, and the frames in between are dropped; a slow audio handler loses the oldest audio beyond its buffer. | +| [Encoded transforms](#encoded-transforms) | a thread of each sender and receiver | Better not. A slow transform delays only its own frames, and one that falls several seconds behind has frames dropped. | | `LocalTrack.onEnded` and `Devices.onChange` handlers, callbacks of the futures of local tracks and devices | the event thread of the runtime's local media | Yes, but that delays the other events of local media. They must not wait for a future of this API either. A track is released before its `onEnded` handlers run, so they never hold up a release. | | WebRTC's own callbacks | WebRTC's threads | Not your code. The library only copies what they report and queues it. | @@ -548,6 +549,31 @@ try (MediaRecorder recorder = new MediaRecorder(Paths.get("call.mkv"))) { Closing the peer, or removing a sender, during a recording ends that track in the file, and `stop()` still finishes it. The [Media Recording](/guide/media/media-recording) guide covers containers and the recorder's events. +### Encoded Transforms + +An [encoded transform](/guide/advanced/encoded-transforms) sees every encoded frame between the encoder and the network, or between the network and the decoder, which is where end-to-end encryption goes. The transforms of the peer options apply to every sender and every remote track of the peer: + +```java +Peer peer = rtc.createPeer(PeerOptions.builder() + .signaling(signaling) + .sendTransform(frame -> encrypt(frame)) + .receiveTransform(frame -> decrypt(frame)) + .build()); +``` + +The peer sets the send transform on each sender as its track is added, before negotiation, so no frame leaves without it and the encoder needs no restart. It sets the receive transform on each receiver before `onTrack` hands out the track, so the transform is in place before the first frame arrives. + +A single sender or remote track takes a transform of its own, in place of the peer's, and `null` removes it: + +```java +videoSender.transform(frame -> addMetadata(frame)); +remoteTrack.transform(null); // frames reach the decoder unchanged +``` + +The first transform set on a video sender that already sends restarts its encoder, which costs a key frame. A transform set on a remote track misses the frames that arrive before its future completes, so a transform that every frame needs, such as decryption, belongs in the peer options. A recording of a sender holds its frames before the transform, a recording of a remote track after it. + +Transforms run on a thread of each sender and receiver, not on the peer's event thread. A frame a transform throws on is dropped rather than passed on, so a failed encryption never sends a frame in the clear. The first such failure of each sender or receiver goes to the peer's `onError` handlers. The [`EncryptedRecordingExample`](https://github.com/devopvoid/webrtc-java/blob/master/webrtc-examples/src/main/java/dev/onvoid/webrtc/examples/EncryptedRecordingExample.java) encrypts a call with AES-GCM this way. + ## Data Channels ```java diff --git a/docs/guide/examples.md b/docs/guide/examples.md index c775b14cb..8b4445ee5 100644 --- a/docs/guide/examples.md +++ b/docs/guide/examples.md @@ -149,13 +149,13 @@ mvn exec:java -D"exec.mainClass=dev.onvoid.webrtc.examples.JavaFxCameraExample" ## Encrypted Recording -The [`EncryptedRecordingExample`](https://github.com/devopvoid/webrtc-java/blob/master/webrtc-examples/src/main/java/dev/onvoid/webrtc/examples/EncryptedRecordingExample.java) encrypts a call end to end with encoded frame transforms, and records what the receiving side decrypted into a media file. See the [Encoded Transforms](/guide/advanced/encoded-transforms) and [Media Recording](/guide/media/media-recording) guides for the APIs it uses. +The [`EncryptedRecordingExample`](https://github.com/devopvoid/webrtc-java/blob/master/webrtc-examples/src/main/java/dev/onvoid/webrtc/examples/EncryptedRecordingExample.java) encrypts a call end to end with encoded frame transforms, with the [Client API](/guide/client-api#encoded-transforms), and records what the receiving side decrypted into a media file. See the [Encoded Transforms](/guide/advanced/encoded-transforms) and [Media Recording](/guide/media/media-recording) guides for the APIs it uses. **Key features demonstrated:** -- Encrypting every encoded audio and video frame with AES-GCM in an `RTCEncodedFrameTransformer` on the senders, and decrypting it on the receivers +- Encrypting every encoded audio and video frame with AES-GCM in the send transform of the peer options, and decrypting it in the receive transform, from the first frame on - Keeping the codec header of each frame in the clear for the packetizer, and authenticating it instead - Dropping frames that fail to encrypt or to authenticate, rather than passing them on -- Recording the receivers with a `MediaRecorder`, without decoding or re-encoding +- Recording the remote tracks with a `MediaRecorder`, without decoding or re-encoding ::: info Like the [Media File](#media-file) example, this one needs the `webrtc-java-media` module. The output file is optional; its extension picks the container. diff --git a/webrtc-examples/src/main/java/dev/onvoid/webrtc/examples/EncryptedRecordingExample.java b/webrtc-examples/src/main/java/dev/onvoid/webrtc/examples/EncryptedRecordingExample.java index 1a7f7066e..6b7bd950e 100644 --- a/webrtc-examples/src/main/java/dev/onvoid/webrtc/examples/EncryptedRecordingExample.java +++ b/webrtc-examples/src/main/java/dev/onvoid/webrtc/examples/EncryptedRecordingExample.java @@ -21,10 +21,11 @@ import java.nio.file.Paths; import java.security.GeneralSecurityException; import java.security.SecureRandom; -import java.util.List; import java.util.concurrent.CompletableFuture; -import java.util.concurrent.CountDownLatch; +import java.util.concurrent.Executors; +import java.util.concurrent.ScheduledExecutorService; import java.util.concurrent.TimeUnit; +import java.util.concurrent.atomic.AtomicInteger; import java.util.concurrent.atomic.AtomicLong; import java.util.logging.Level; import java.util.logging.Logger; @@ -34,34 +35,23 @@ import javax.crypto.spec.GCMParameterSpec; import javax.crypto.spec.SecretKeySpec; -import dev.onvoid.webrtc.CreateSessionDescriptionObserver; -import dev.onvoid.webrtc.PeerConnectionFactory; -import dev.onvoid.webrtc.PeerConnectionObserver; -import dev.onvoid.webrtc.RTCAnswerOptions; -import dev.onvoid.webrtc.RTCConfiguration; import dev.onvoid.webrtc.RTCEncodedFrame; -import dev.onvoid.webrtc.RTCEncodedFrameTransformer; import dev.onvoid.webrtc.RTCEncodedVideoFrame; -import dev.onvoid.webrtc.RTCIceCandidate; -import dev.onvoid.webrtc.RTCOfferOptions; -import dev.onvoid.webrtc.RTCPeerConnection; -import dev.onvoid.webrtc.RTCPeerConnectionState; -import dev.onvoid.webrtc.RTCRtpReceiver; -import dev.onvoid.webrtc.RTCRtpSender; -import dev.onvoid.webrtc.RTCRtpTransceiver; -import dev.onvoid.webrtc.RTCSessionDescription; -import dev.onvoid.webrtc.SetSessionDescriptionObserver; -import dev.onvoid.webrtc.media.audio.AudioDeviceModule; -import dev.onvoid.webrtc.media.audio.AudioLayer; -import dev.onvoid.webrtc.media.audio.AudioTrack; -import dev.onvoid.webrtc.media.audio.CustomAudioSource; +import dev.onvoid.webrtc.client.AudioMode; +import dev.onvoid.webrtc.client.CustomAudio; +import dev.onvoid.webrtc.client.CustomVideo; +import dev.onvoid.webrtc.client.PcmFormat; +import dev.onvoid.webrtc.client.Peer; +import dev.onvoid.webrtc.client.PeerOptions; +import dev.onvoid.webrtc.client.RemoteTrack; +import dev.onvoid.webrtc.client.Signaling; +import dev.onvoid.webrtc.client.SignalingPair; +import dev.onvoid.webrtc.client.WebRtc; import dev.onvoid.webrtc.media.recorder.MediaRecorder; import dev.onvoid.webrtc.media.recorder.MediaRecorderListener; import dev.onvoid.webrtc.media.recorder.MediaRecorderResult; -import dev.onvoid.webrtc.media.video.CustomVideoSource; import dev.onvoid.webrtc.media.video.NativeI420Buffer; import dev.onvoid.webrtc.media.video.VideoFrame; -import dev.onvoid.webrtc.media.video.VideoTrack; /** * Encrypts a call end to end with encoded frame transforms, and records what @@ -69,17 +59,21 @@ *

* This example shows how to: *

*

+ * The peer options put the transforms on each sender before negotiation and + * on each receiver before its track is handed out, so no frame leaves + * unencrypted and none reaches the decoder still encrypted. + *

* Both peers live in this application and the key is simply shared between * them; a real application exchanges it over its signaling channel, or better * derives it with a key agreement such as MLS. @@ -100,50 +94,67 @@ public class EncryptedRecordingExample { private static final int WIDTH = 640; private static final int HEIGHT = 480; + private static final PcmFormat AUDIO_FORMAT = PcmFormat.of(48000, 1); + public static void main(String[] args) throws Exception { Path file = Paths.get(args.length > 0 ? args[0] : "encrypted-call.mkv"); - // Pushed audio needs a factory whose audio layer does not capture. - AudioDeviceModule audioModule = new AudioDeviceModule(AudioLayer.kDummyAudio); - PeerConnectionFactory factory = new PeerConnectionFactory(audioModule); - - CustomVideoSource videoSource = new CustomVideoSource(); - CustomAudioSource audioSource = new CustomAudioSource(); - VideoTrack videoTrack = factory.createVideoTrack("video", videoSource); - AudioTrack audioTrack = factory.createAudioTrack("audio", audioSource); - - Peer caller = new Peer(factory); - Peer callee = new Peer(factory); - caller.remote = callee; - callee.remote = caller; - FrameCipher cipher = new FrameCipher(newKey()); - RTCRtpSender videoSender = caller.connection.addTrack(videoTrack, List.of("stream")); - RTCRtpSender audioSender = caller.connection.addTrack(audioTrack, List.of("stream")); - - // Set before negotiating, so the encoder needs no restart. - videoSender.setTransform(cipher::encrypt); - audioSender.setTransform(cipher::encrypt); - - callee.setRemoteDescription(caller.createOffer()); - caller.setRemoteDescription(callee.createAnswer()); + // A headless runtime sends pushed audio and needs no microphone or speakers. + try (WebRtc rtc = WebRtc.builder().audio(AudioMode.HEADLESS).build()) { + SignalingPair link = Signaling.loopback(); + + // Every frame a peer sends is encrypted, and every frame it receives decrypted. + Peer caller = rtc.createPeer(PeerOptions.builder() + .name("caller") + .signaling(link.first()) + .sendTransform(cipher::encrypt) + .receiveTransform(cipher::decrypt) + .build()); + Peer callee = rtc.createPeer(PeerOptions.builder() + .name("callee") + .polite(true) + .signaling(link.second()) + .sendTransform(cipher::encrypt) + .receiveTransform(cipher::decrypt) + .build()); + + CompletableFuture remoteVideo = new CompletableFuture<>(); + CompletableFuture remoteAudio = new CompletableFuture<>(); + + callee.onTrack(track -> (track.isVideo() ? remoteVideo : remoteAudio).complete(track)); + + CustomVideo video = rtc.media().customVideo("video"); + CustomAudio audio = rtc.media().customAudio("audio"); + caller.addTrack(video); + caller.addTrack(audio); + + ScheduledExecutorService feeder = Executors.newSingleThreadScheduledExecutor(); - // The callee's transceivers follow the order of the offer. - RTCRtpTransceiver[] transceivers = callee.connection.getTransceivers(); - RTCRtpReceiver videoReceiver = transceivers[0].getReceiver(); - RTCRtpReceiver audioReceiver = transceivers[1].getReceiver(); + try { + feed(feeder, video, audio); - videoReceiver.setTransform(cipher::decrypt); - audioReceiver.setTransform(cipher::decrypt); + caller.whenConnected().get(10, TimeUnit.SECONDS); - caller.connected.await(10, TimeUnit.SECONDS); - callee.connected.await(10, TimeUnit.SECONDS); + record(file, remoteVideo.get(10, TimeUnit.SECONDS), remoteAudio.get(10, TimeUnit.SECONDS)); + } + finally { + feeder.shutdownNow(); - MediaGenerator generator = new MediaGenerator(videoSource, audioSource); - generator.start(); + LOG.info(String.format("Encrypted %d frames, decrypted %d, rejected %d", + cipher.encrypted.get(), cipher.decrypted.get(), cipher.rejected.get())); + } + } + // The runtime closed the peers, the tracks and the factory. + } + /** + * Records the remote tracks for a while. They hand the recorder what + * they decrypted. + */ + private static void record(Path file, RemoteTrack video, RemoteTrack audio) throws Exception { try (MediaRecorder recorder = new MediaRecorder(file)) { recorder.setListener(new MediaRecorderListener() { @@ -163,9 +174,8 @@ public void onError(String message) { } }); - // The receivers hand the recorder what they decrypted. - recorder.addTrack(videoReceiver); - recorder.addTrack(audioReceiver); + recorder.addTrack(video); + recorder.addTrack(audio); recorder.start(); Thread.sleep(TimeUnit.SECONDS.toMillis(RECORD_SECONDS)); @@ -175,46 +185,67 @@ public void onError(String message) { + (recorder.getResult() == MediaRecorderResult.PARTIAL ? ", incomplete" : "")); } } - finally { - generator.stop(); + } + + private static SecretKey newKey() { + byte[] key = new byte[16]; + new SecureRandom().nextBytes(key); - LOG.info(String.format("Encrypted %d frames, decrypted %d, rejected %d", - cipher.encrypted.get(), cipher.decrypted.get(), cipher.rejected.get())); + return new SecretKeySpec(key, "AES"); + } - videoReceiver.dispose(); - audioReceiver.dispose(); - for (RTCRtpTransceiver transceiver : transceivers) { - transceiver.dispose(); + /** + * Pushes a moving test picture at 30 frames per second, and a 440 Hz tone + * a second at a time. + */ + private static void feed(ScheduledExecutorService feeder, CustomVideo video, CustomAudio audio) { + AtomicInteger frames = new AtomicInteger(); + + feeder.scheduleAtFixedRate(() -> { + NativeI420Buffer buffer = NativeI420Buffer.allocate(WIDTH, HEIGHT); + ByteBuffer y = buffer.getDataY(); + int stride = buffer.getStrideY(); + int offset = frames.getAndIncrement() * 4; + + // Diagonal stripes that move, so every frame differs. + for (int row = 0; row < HEIGHT; row++) { + for (int col = 0; col < WIDTH; col++) { + y.put(row * stride + col, (byte) (((row + col + offset) / 16 % 2) * 180 + 40)); + } } - videoSender.dispose(); - audioSender.dispose(); - caller.connection.close(); - callee.connection.close(); + VideoFrame frame = new VideoFrame(buffer, System.nanoTime()); + video.push(frame); + frame.release(); + }, 0, 33, TimeUnit.MILLISECONDS); - videoTrack.dispose(); - audioTrack.dispose(); - videoSource.dispose(); - audioSource.dispose(); + // Audio may be pushed in pieces of any length; the track sends it in 10 ms chunks. + byte[] second = tone(AUDIO_FORMAT.sampleRate()); - factory.dispose(); - audioModule.dispose(); - } + feeder.scheduleAtFixedRate(() -> audio.push(second, AUDIO_FORMAT), 0, 1, TimeUnit.SECONDS); } - private static SecretKey newKey() { - byte[] key = new byte[16]; - new SecureRandom().nextBytes(key); + private static byte[] tone(int sampleRate) { + byte[] pcm = new byte[sampleRate * 2]; - return new SecretKeySpec(key, "AES"); + for (int i = 0; i < sampleRate; i++) { + short sample = (short) (Math.sin(2 * Math.PI * 440 * i / sampleRate) * 6000); + + pcm[2 * i] = (byte) sample; + pcm[2 * i + 1] = (byte) (sample >> 8); + } + + return pcm; } + /** * AES-GCM over the payload of each frame. An encrypted frame is laid out * as the clear header, the ciphertext with its tag, then the 12-byte IV. - * Each transform runs on a thread of its own, one per sender or receiver, - * so each call makes its own Cipher rather than share one. + * One instance serves every sender and receiver, each of which runs its + * transform on a thread of its own, so each call makes its own Cipher + * rather than share one. */ private static class FrameCipher { @@ -224,7 +255,7 @@ private static class FrameCipher { private final SecretKey key; private final SecureRandom random = new SecureRandom(); - // The audio and the video transform each run on a thread of their own. + // Counted from the threads of every sender and receiver. final AtomicLong encrypted = new AtomicLong(); final AtomicLong decrypted = new AtomicLong(); final AtomicLong rejected = new AtomicLong(); @@ -331,172 +362,4 @@ private static int clearBytes(RTCEncodedFrame frame, int size) { return Math.min(clear, size); } } - - - /** - * Pushes a moving test picture at 30 frames and a tone in 10 ms chunks - * per second, in real time. - */ - private static class MediaGenerator { - - private final CustomVideoSource videoSource; - private final CustomAudioSource audioSource; - - private volatile boolean running; - private Thread thread; - - - MediaGenerator(CustomVideoSource videoSource, CustomAudioSource audioSource) { - this.videoSource = videoSource; - this.audioSource = audioSource; - } - - void start() { - running = true; - thread = new Thread(this::run, "media-generator"); - thread.setDaemon(true); - thread.start(); - } - - void stop() throws InterruptedException { - running = false; - thread.join(); - } - - private void run() { - byte[] audio = new byte[480 * 2]; - long startNs = System.nanoTime(); - long chunks = 0; - long frames = 0; - - while (running) { - long elapsedMs = (System.nanoTime() - startNs) / 1_000_000; - - while (chunks * 10 <= elapsedMs) { - for (int i = 0; i < 480; i++) { - double t = (chunks * 480 + i) / 48000.0; - short sample = (short) (Math.sin(2 * Math.PI * 440 * t) * 6000); - audio[2 * i] = (byte) sample; - audio[2 * i + 1] = (byte) (sample >> 8); - } - - audioSource.pushAudio(audio, 16, 48000, 1, 480); - chunks++; - } - - if (frames * 1000 / 30 <= elapsedMs) { - NativeI420Buffer buffer = NativeI420Buffer.allocate(WIDTH, HEIGHT); - ByteBuffer y = buffer.getDataY(); - int stride = buffer.getStrideY(); - int offset = (int) (frames * 4); - - // Diagonal stripes that move, so every frame differs. - for (int row = 0; row < HEIGHT; row++) { - for (int col = 0; col < WIDTH; col++) { - y.put(row * stride + col, (byte) (((row + col + offset) / 16 % 2) * 180 + 40)); - } - } - - VideoFrame frame = new VideoFrame(buffer, 0); - videoSource.pushFrame(frame); - frame.release(); - frames++; - } - - try { - Thread.sleep(2); - } - catch (InterruptedException e) { - return; - } - } - } - } - - - /** - * One end of the call, exchanging candidates with the other directly. - */ - private static class Peer implements PeerConnectionObserver { - - final RTCPeerConnection connection; - final CountDownLatch connected = new CountDownLatch(1); - - volatile Peer remote; - - - Peer(PeerConnectionFactory factory) { - connection = factory.createPeerConnection(new RTCConfiguration(), this); - } - - @Override - public void onIceCandidate(RTCIceCandidate candidate) { - remote.connection.addIceCandidate(candidate); - } - - @Override - public void onConnectionChange(RTCPeerConnectionState state) { - if (state == RTCPeerConnectionState.CONNECTED) { - connected.countDown(); - } - } - - RTCSessionDescription createOffer() throws Exception { - CompletableFuture created = new CompletableFuture<>(); - connection.createOffer(new RTCOfferOptions(), created(created)); - - return setLocalDescription(created.get(10, TimeUnit.SECONDS)); - } - - RTCSessionDescription createAnswer() throws Exception { - CompletableFuture created = new CompletableFuture<>(); - connection.createAnswer(new RTCAnswerOptions(), created(created)); - - return setLocalDescription(created.get(10, TimeUnit.SECONDS)); - } - - void setRemoteDescription(RTCSessionDescription description) throws Exception { - CompletableFuture set = new CompletableFuture<>(); - connection.setRemoteDescription(description, set(set)); - set.get(10, TimeUnit.SECONDS); - } - - private RTCSessionDescription setLocalDescription(RTCSessionDescription description) throws Exception { - CompletableFuture set = new CompletableFuture<>(); - connection.setLocalDescription(description, set(set)); - set.get(10, TimeUnit.SECONDS); - - return description; - } - - private static CreateSessionDescriptionObserver created(CompletableFuture future) { - return new CreateSessionDescriptionObserver() { - - @Override - public void onSuccess(RTCSessionDescription description) { - future.complete(description); - } - - @Override - public void onFailure(String error) { - future.completeExceptionally(new IllegalStateException(error)); - } - }; - } - - private static SetSessionDescriptionObserver set(CompletableFuture future) { - return new SetSessionDescriptionObserver() { - - @Override - public void onSuccess() { - future.complete(null); - } - - @Override - public void onFailure(String error) { - future.completeExceptionally(new IllegalStateException(error)); - } - }; - } - } } 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 89ee92427..2e619bde5 100644 --- a/webrtc/src/main/java/dev/onvoid/webrtc/client/PeerImpl.java +++ b/webrtc/src/main/java/dev/onvoid/webrtc/client/PeerImpl.java @@ -55,6 +55,7 @@ import dev.onvoid.webrtc.RTCAnswerOptions; import dev.onvoid.webrtc.RTCConfiguration; import dev.onvoid.webrtc.RTCDataChannel; +import dev.onvoid.webrtc.RTCEncodedFrameTransformer; import dev.onvoid.webrtc.RTCIceCandidate; import dev.onvoid.webrtc.RTCIceGatheringState; import dev.onvoid.webrtc.RTCIceServer; @@ -360,7 +361,7 @@ private SenderImpl addSender(LocalTrackImpl local, List ids, List encodedFrames(Supplier { + try { + transform.transform(frame); + } + catch (Throwable e) { + frame.drop(); + + if (reported.compareAndSet(false, true)) { + deliverError(new WebRtcException("The transform of " + owner + + " failed, and drops each frame it fails on", e)); + } + } + }; + } + void deliverError(WebRtcException error) { events.execute(() -> { if (errorHandlers.isEmpty()) { @@ -1621,6 +1654,19 @@ private void trackAdded(RTCRtpReceiver receiver, MediaStream[] streams) { return; } + if (options.receiveTransform != null) { + String owner = "the track " + id + " of " + name; + + try { + // Before the application hears of the track, and so before + // its first frame arrives. + receiver.setTransform(guard(options.receiveTransform, owner)); + } + catch (Throwable e) { + deliverError(asWebRtcException(e, "Setting the receive transform of " + owner + " failed")); + } + } + RemoteTrackImpl remote = track instanceof VideoTrack ? new RemoteVideoImpl(this, receiver, (VideoTrack) track, id, streamIds) : new RemoteAudioImpl(this, receiver, (AudioTrack) track, id, streamIds); diff --git a/webrtc/src/main/java/dev/onvoid/webrtc/client/PeerOptions.java b/webrtc/src/main/java/dev/onvoid/webrtc/client/PeerOptions.java index 6bbeebc6b..5c6ba2296 100644 --- a/webrtc/src/main/java/dev/onvoid/webrtc/client/PeerOptions.java +++ b/webrtc/src/main/java/dev/onvoid/webrtc/client/PeerOptions.java @@ -28,6 +28,7 @@ import java.util.function.Consumer; import dev.onvoid.webrtc.RTCConfiguration; +import dev.onvoid.webrtc.RTCEncodedFrameTransformer; import dev.onvoid.webrtc.RTCIceServer; import dev.onvoid.webrtc.RTCIceTransportPolicy; import dev.onvoid.webrtc.internal.Timeouts; @@ -66,6 +67,10 @@ public final class PeerOptions { final long channelReceiveLimit; + final RTCEncodedFrameTransformer sendTransform; + + final RTCEncodedFrameTransformer receiveTransform; + private PeerOptions(Builder builder) { this.name = builder.name; @@ -80,6 +85,8 @@ private PeerOptions(Builder builder) { this.gatheringTimeout = builder.gatheringTimeout; this.channelSendLimit = builder.channelSendLimit; this.channelReceiveLimit = builder.channelReceiveLimit; + this.sendTransform = builder.sendTransform; + this.receiveTransform = builder.receiveTransform; } /** @@ -143,6 +150,10 @@ public static final class Builder { private long channelReceiveLimit = 16 * 1024 * 1024; + private RTCEncodedFrameTransformer sendTransform; + + private RTCEncodedFrameTransformer receiveTransform; + private Builder() { } @@ -303,6 +314,50 @@ public Builder channelReceiveLimit(long bytes) { return this; } + /** + * Sets the transform that every encoded frame the peer sends passes + * through, after the encoder and before the packetizer, as end-to-end + * encryption needs. It is set on each sender as the track is added, + * before negotiation, so no frame leaves without it and the encoder + * needs no restart. {@link Sender#transform} replaces it for one + * sender. + *

+ * The transform runs on a thread of each sender, not on the peer's + * event thread; see {@link RTCEncodedFrameTransformer}. A frame it + * throws on is dropped, and the first such failure of each sender + * reaches the peer's {@code onError} handlers. + * + * @param transform the transform of every sender. + * + * @return this builder. + */ + public Builder sendTransform(RTCEncodedFrameTransformer transform) { + this.sendTransform = requireNonNull(transform, "transform"); + return this; + } + + /** + * Sets the transform that every encoded frame the peer receives passes + * through, after the depacketizer and before the decoder, as + * end-to-end encryption needs. It is set on each receiver before + * {@link Peer#onTrack} hands its track out, so it is in place before + * the first frame arrives. {@link RemoteTrack#transform} replaces it + * for one track. + *

+ * The transform runs on a thread of each receiver, not on the peer's + * event thread; see {@link RTCEncodedFrameTransformer}. A frame it + * throws on is dropped, and the first such failure of each receiver + * reaches the peer's {@code onError} handlers. + * + * @param transform the transform of every receiver. + * + * @return this builder. + */ + public Builder receiveTransform(RTCEncodedFrameTransformer transform) { + this.receiveTransform = requireNonNull(transform, "transform"); + return this; + } + /** * Changes anything else of the {@link RTCConfiguration} the peer is * created with, after this builder applied its own values. diff --git a/webrtc/src/main/java/dev/onvoid/webrtc/client/RemoteTrack.java b/webrtc/src/main/java/dev/onvoid/webrtc/client/RemoteTrack.java index 14ba60e3a..9b230c49c 100644 --- a/webrtc/src/main/java/dev/onvoid/webrtc/client/RemoteTrack.java +++ b/webrtc/src/main/java/dev/onvoid/webrtc/client/RemoteTrack.java @@ -17,8 +17,11 @@ package dev.onvoid.webrtc.client; import java.util.List; +import java.util.concurrent.CompletableFuture; import java.util.function.Consumer; +import dev.onvoid.webrtc.RTCEncodedFrameTransformer; + /** * A track the remote peer sends, handed to {@link Peer#onTrack}. It ends when * the remote peer stops sending it or the peer closes. @@ -68,6 +71,26 @@ default boolean isAudio() { */ RemoteAudio asAudio(); + /** + * Sets the transform that every encoded frame of this track passes + * through, after the depacketizer and before the decoder, in place of the + * one set before, such as the {@link PeerOptions.Builder#receiveTransform + * receive transform} of the peer. Frames that arrive before the future + * completes pass the one before; set the peer's receive transform to have + * it in place from the first frame. + *

+ * The transform runs on a thread of this track's receiver, not on the + * peer's event thread; see {@link RTCEncodedFrameTransformer}. A frame it + * throws on is dropped, and the first such failure reaches the peer's + * {@code onError} handlers. + * + * @param transform the transform, or null to receive frames unchanged. + * + * @return a future that completes once the transform is set, and fails + * with an {@link IllegalStateException} once the track ended. + */ + CompletableFuture transform(RTCEncodedFrameTransformer transform); + /** * @return true when the remote peer sends no media on the track for now. */ diff --git a/webrtc/src/main/java/dev/onvoid/webrtc/client/RemoteTrackImpl.java b/webrtc/src/main/java/dev/onvoid/webrtc/client/RemoteTrackImpl.java index c8f160f7e..1a2d42c44 100644 --- a/webrtc/src/main/java/dev/onvoid/webrtc/client/RemoteTrackImpl.java +++ b/webrtc/src/main/java/dev/onvoid/webrtc/client/RemoteTrackImpl.java @@ -23,6 +23,7 @@ import java.util.concurrent.CompletableFuture; import java.util.function.Consumer; +import dev.onvoid.webrtc.RTCEncodedFrameTransformer; import dev.onvoid.webrtc.RTCRtpReceiver; import dev.onvoid.webrtc.client.internal.EncodedFrames; import dev.onvoid.webrtc.internal.NativeApi; @@ -134,6 +135,31 @@ public Subscription onEnded(Runnable handler) { }); } + @Override + public CompletableFuture transform(RTCEncodedFrameTransformer transform) { + CompletableFuture result = new CompletableFuture<>(); + + peer.control.execute(() -> { + try { + if (peer.isClosed()) { + throw peer.closedError(); + } + if (ended) { + throw new IllegalStateException("The track " + id + " ended"); + } + + receiver.setTransform(peer.guard(transform, "the track " + id + " of " + peer.name)); + result.complete(null); + } + catch (Throwable e) { + result.completeExceptionally(e instanceof IllegalStateException + ? e : PeerImpl.asWebRtcException(e, "Setting the transform of the track " + id + " failed")); + } + }, () -> result.completeExceptionally(new PeerClosedException("The runtime is closed"))); + + return peer.toPublic(result); + } + @Override public String toString() { return "RemoteTrack[" + kind + ", " + id + "]"; diff --git a/webrtc/src/main/java/dev/onvoid/webrtc/client/Sender.java b/webrtc/src/main/java/dev/onvoid/webrtc/client/Sender.java index dbcc91589..e9e60538a 100644 --- a/webrtc/src/main/java/dev/onvoid/webrtc/client/Sender.java +++ b/webrtc/src/main/java/dev/onvoid/webrtc/client/Sender.java @@ -20,6 +20,8 @@ import java.util.concurrent.CompletableFuture; import java.util.function.Consumer; +import dev.onvoid.webrtc.RTCEncodedFrameTransformer; + /** * Sends one track on one peer, made by {@link Peer#addTrack}. */ @@ -48,6 +50,25 @@ public interface Sender { */ CompletableFuture configure(Encoding encoding); + /** + * Sets the transform that every encoded frame of this sender passes + * through, after the encoder and before the packetizer, in place of the + * one set before, such as the {@link PeerOptions.Builder#sendTransform + * send transform} of the peer. The first transform set on a video sender + * that already sends restarts its encoder, which costs a key frame; set + * the peer's send transform to have it in place from the start. + *

+ * The transform stays with the sender when its track is replaced. It runs + * on a thread of this sender, not on the peer's event thread; see {@link + * RTCEncodedFrameTransformer}. A frame it throws on is dropped, and the + * first such failure reaches the peer's {@code onError} handlers. + * + * @param transform the transform, or null to send frames unchanged. + * + * @return a future that completes once the transform is set. + */ + CompletableFuture transform(RTCEncodedFrameTransformer transform); + /** * Sends DTMF tones on an audio sender, 100 ms each with 70 ms between * them. See {@link #sendTones(String, Duration, Duration)}. diff --git a/webrtc/src/main/java/dev/onvoid/webrtc/client/SenderImpl.java b/webrtc/src/main/java/dev/onvoid/webrtc/client/SenderImpl.java index dcbe3a818..ff5fe74a1 100644 --- a/webrtc/src/main/java/dev/onvoid/webrtc/client/SenderImpl.java +++ b/webrtc/src/main/java/dev/onvoid/webrtc/client/SenderImpl.java @@ -25,6 +25,7 @@ import java.util.function.Consumer; import dev.onvoid.webrtc.RTCDtmfSender; +import dev.onvoid.webrtc.RTCEncodedFrameTransformer; import dev.onvoid.webrtc.RTCPeerConnection; import dev.onvoid.webrtc.RTCRtpEncodingParameters; import dev.onvoid.webrtc.RTCRtpSendParameters; @@ -76,8 +77,10 @@ final class SenderImpl implements Sender { /** * @param encodings the encodings to send the track in, as one per layer * of simulcast, or none for WebRTC's single one. + * @param transform the transform of the frames, or null for none. */ - void attach(RTCPeerConnection connection, List streamIds, List encodings) { + void attach(RTCPeerConnection connection, List streamIds, List encodings, + RTCEncodedFrameTransformer transform) { LocalTrackImpl local = track; if (!local.register(this)) { @@ -109,6 +112,18 @@ void attach(RTCPeerConnection connection, List streamIds, List configure(Encoding encoding) { }); } + @Override + public CompletableFuture transform(RTCEncodedFrameTransformer transform) { + return run(() -> sender.setTransform(peer.guard(transform, "a sender of " + peer.name))); + } + @Override public CompletableFuture sendTones(String tones) { return sendTones(tones, Duration.ofMillis(100), Duration.ofMillis(70)); diff --git a/webrtc/src/test/java/dev/onvoid/webrtc/client/TransformTests.java b/webrtc/src/test/java/dev/onvoid/webrtc/client/TransformTests.java new file mode 100644 index 000000000..4d2701dc9 --- /dev/null +++ b/webrtc/src/test/java/dev/onvoid/webrtc/client/TransformTests.java @@ -0,0 +1,304 @@ +/* + * 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 static org.junit.jupiter.api.Assertions.assertInstanceOf; +import static org.junit.jupiter.api.Assertions.assertThrows; +import static org.junit.jupiter.api.Assertions.assertTrue; + +import java.nio.ByteBuffer; +import java.util.List; +import java.util.concurrent.CompletableFuture; +import java.util.concurrent.CopyOnWriteArrayList; +import java.util.concurrent.ExecutionException; +import java.util.concurrent.Executors; +import java.util.concurrent.ScheduledExecutorService; +import java.util.concurrent.TimeUnit; +import java.util.concurrent.atomic.AtomicInteger; +import java.util.function.BooleanSupplier; + +import dev.onvoid.webrtc.RTCEncodedFrame; +import dev.onvoid.webrtc.RTCStats; +import dev.onvoid.webrtc.RTCStatsType; +import dev.onvoid.webrtc.media.video.NativeI420Buffer; +import dev.onvoid.webrtc.media.video.VideoFrame; + +import org.junit.jupiter.api.AfterEach; +import org.junit.jupiter.api.BeforeEach; +import org.junit.jupiter.api.Test; + +/** + * Transforms the encoded frames of senders and remote tracks: the transforms + * of the peer options are in place from the first frame, those of a sender or + * track replace them, and a transform that throws drops its frames. + */ +class TransformTests extends ClientTestBase { + + /** What the send transforms append and the receive transforms strip. */ + private static final byte[] MARK = { 'w', 'r', 't', 'c' }; + + private ScheduledExecutorService feeder; + + + @BeforeEach + void createFeeder() { + feeder = Executors.newSingleThreadScheduledExecutor(); + } + + @AfterEach + void stopFeeder() throws InterruptedException { + feeder.shutdownNow(); + feeder.awaitTermination(5, TimeUnit.SECONDS); + } + + @Test + void peerTransformsEveryFrame() throws Exception { + AtomicInteger marked = new AtomicInteger(); + AtomicInteger unmarked = new AtomicInteger(); + Peer[] peers = pair(options -> options + .sendTransform(TransformTests::mark) + .receiveTransform(frame -> (unmark(frame) ? marked : unmarked).incrementAndGet())); + CustomVideo video = rtc.media().customVideo("video"); + CompletableFuture decoded = new CompletableFuture<>(); + + peers[1].onTrack(track -> track.asVideo().onFrame(frame -> decoded.complete(null))); + peers[0].addTrack(video); + + feedVideo(video); + await(decoded); + waitFor(() -> marked.get() >= 10); + + // Every frame decoded passed the receive transform first, so it was + // in place before the first one arrived. The count of the decoder is + // read first, as the transform runs ahead of it. + long decodedFrames = framesDecoded(peers[1]); + int transformed = marked.get(); + + assertTrue(decodedFrames > 0 && decodedFrames <= transformed, + () -> "Decoded " + decodedFrames + " frames, of which " + transformed + " were transformed"); + // And every frame sent passed the send transform. + assertEquals(0, unmarked.get()); + } + + @Test + void senderAndTrackTransformsReplacePeers() throws Exception { + AtomicInteger byPeer = new AtomicInteger(); + AtomicInteger byTrack = new AtomicInteger(); + Peer[] peers = pair(options -> options + .sendTransform(TransformTests::mark) + .receiveTransform(frame -> { + unmark(frame); + byPeer.incrementAndGet(); + })); + CustomVideo video = rtc.media().customVideo("video"); + CompletableFuture received = new CompletableFuture<>(); + + peers[1].onTrack(received::complete); + Sender sender = peers[0].addTrack(video); + + feedVideo(video); + RemoteTrack track = await(received); + waitFor(() -> byPeer.get() > 0); + + // Frames sent unmarked from now on, and received as they are. + await(sender.transform(null)); + await(track.transform(frame -> { + if (!hasMark(frame)) { + byTrack.incrementAndGet(); + } + })); + + waitFor(() -> byTrack.get() >= 10); + } + + @Test + void trackTransformWithoutPeerTransform() throws Exception { + Peer[] peers = pair(); + CustomVideo video = rtc.media().customVideo("video"); + AtomicInteger seen = new AtomicInteger(); + CompletableFuture> set = new CompletableFuture<>(); + + // Set from the handler, as an application would. + peers[1].onTrack(track -> set.complete(track.transform(frame -> seen.incrementAndGet()))); + Sender sender = peers[0].addTrack(video); + await(sender.transform(frame -> { })); + + feedVideo(video); + await(await(set)); + + waitFor(() -> seen.get() >= 10); + } + + @Test + void throwingTransformDropsFrames() throws Exception { + AtomicInteger thrown = new AtomicInteger(); + Peer[] peers = pair(options -> options.receiveTransform(frame -> { + thrown.incrementAndGet(); + throw new IllegalStateException("No key"); + })); + List failures = new CopyOnWriteArrayList<>(); + CustomVideo video = rtc.media().customVideo("video"); + + peers[1].onError(failures::add); + peers[0].addTrack(video); + + feedVideo(video); + waitFor(() -> thrown.get() >= 10 && !failures.isEmpty()); + + // Reported once, not for every frame. + assertEquals(1, failures.size(), () -> "Failures: " + failures); + assertInstanceOf(IllegalStateException.class, failures.get(0).getCause()); + // No frame got past the transform to the decoder. + assertEquals(0, framesDecoded(peers[1])); + } + + @Test + void transformOfRemovedSenderFails() throws Exception { + Peer[] peers = pair(); + Sender sender = peers[0].addTrack(rtc.media().customVideo("video")); + + await(sender.remove()); + + assertFailure(IllegalStateException.class, sender.transform(frame -> { })); + } + + @Test + void transformOfEndedTrackFails() throws Exception { + Peer[] peers = pair(); + CompletableFuture received = new CompletableFuture<>(); + CompletableFuture ended = new CompletableFuture<>(); + + peers[1].onTrack(track -> { + received.complete(track); + track.onEnded(() -> ended.complete(null)); + }); + Sender sender = peers[0].addTrack(rtc.media().customVideo("video")); + RemoteTrack track = await(received); + + await(sender.remove()); + await(ended); + + assertFailure(IllegalStateException.class, track.transform(frame -> { })); + } + + @Test + void transformAfterPeerClosedFails() throws Exception { + Peer[] peers = pair(); + CompletableFuture received = new CompletableFuture<>(); + + peers[1].onTrack(received::complete); + Sender sender = peers[0].addTrack(rtc.media().customVideo("video")); + RemoteTrack track = await(received); + + await(peers[0].closeAsync()); + await(peers[1].closeAsync()); + + assertFailure(PeerClosedException.class, sender.transform(frame -> { })); + assertFailure(PeerClosedException.class, track.transform(frame -> { })); + } + + private static void mark(RTCEncodedFrame frame) { + ByteBuffer data = frame.getData(); + ByteBuffer marked = ByteBuffer.allocate(data.remaining() + MARK.length); + + marked.put(data).put(MARK).flip(); + frame.setData(marked); + } + + /** + * @return true if the frame was marked, which it no longer is. + */ + private static boolean unmark(RTCEncodedFrame frame) { + if (!hasMark(frame)) { + return false; + } + + ByteBuffer data = frame.getData(); + data.limit(data.limit() - MARK.length); + frame.setData(data); + + return true; + } + + private static boolean hasMark(RTCEncodedFrame frame) { + ByteBuffer data = frame.getData(); + int size = data.limit(); + + if (size < MARK.length) { + return false; + } + + for (int i = 0; i < MARK.length; i++) { + if (data.get(size - MARK.length + i) != MARK[i]) { + return false; + } + } + + return true; + } + + private void feedVideo(CustomVideo video) { + AtomicInteger shade = new AtomicInteger(); + + feeder.scheduleAtFixedRate(() -> { + NativeI420Buffer buffer = NativeI420Buffer.allocate(160, 120); + ByteBuffer y = buffer.getDataY(); + int offset = shade.addAndGet(3); + + for (int i = 0; i < y.capacity(); i++) { + y.put(i, (byte) (i + offset)); + } + + VideoFrame frame = new VideoFrame(buffer, System.nanoTime()); + video.push(frame); + frame.release(); + }, 0, 33, TimeUnit.MILLISECONDS); + } + + private static long framesDecoded(Peer peer) throws Exception { + long frames = 0; + + for (RTCStats entry : await(peer.stats()).report().getStats().values()) { + Object decoded = entry.getAttributes().get("framesDecoded"); + + if (entry.getType() == RTCStatsType.INBOUND_RTP && decoded instanceof Number) { + frames += ((Number) decoded).longValue(); + } + } + + return frames; + } + + /** Waits for frames to arrive, which only time brings. */ + private static void waitFor(BooleanSupplier condition) throws InterruptedException { + long deadline = System.nanoTime() + TimeUnit.SECONDS.toNanos(TIMEOUT_SECONDS); + + while (!condition.getAsBoolean()) { + assertTrue(System.nanoTime() < deadline, "Timed out"); + Thread.sleep(20); + } + } + + private static void assertFailure(Class type, CompletableFuture future) { + ExecutionException e = assertThrows(ExecutionException.class, + () -> future.get(TIMEOUT_SECONDS, TimeUnit.SECONDS)); + + assertInstanceOf(type, e.getCause()); + } + +}