diff --git a/docs/guide/data/data-channels.md b/docs/guide/data/data-channels.md index 3f024590..e39b5ff4 100644 --- a/docs/guide/data/data-channels.md +++ b/docs/guide/data/data-channels.md @@ -203,7 +203,7 @@ dataChannel.sendAsync(binaryChannelBuffer, new RTCDataChannelSendObserver() { }); ``` -The observer is called exactly once, normally on the native network thread, and with a failure if the operation is discarded while the channel shuts down. Success means the message was queued locally, not that the peer received it. Do not block in the callbacks or call other WebRTC methods from them synchronously; dispatch further work to an executor of your own. Without an observer, `sendAsync` logs failures and reports nothing to the caller. +The observer is called exactly once, with a failure if the operation is discarded while the channel shuts down. Success means the message was queued locally, not that the peer received it. Callbacks run on a native callback thread shared with other send and DTLS transport observers; they may call other WebRTC methods, but should not block, so dispatch long work to an executor of your own. Without an observer, `sendAsync` logs failures and reports nothing to the caller. ### Receiving Data diff --git a/webrtc-jni/src/main/cpp/include/api/JavaCallbackThread.h b/webrtc-jni/src/main/cpp/include/api/JavaCallbackThread.h new file mode 100644 index 00000000..f13c5406 --- /dev/null +++ b/webrtc-jni/src/main/cpp/include/api/JavaCallbackThread.h @@ -0,0 +1,45 @@ +/* + * 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. + */ + +#ifndef JNI_WEBRTC_API_JAVA_CALLBACK_THREAD_H_ +#define JNI_WEBRTC_API_JAVA_CALLBACK_THREAD_H_ + +#include "absl/functional/any_invocable.h" + +#include + +namespace jni +{ + // The thread that runs the Java observers WebRTC notifies on its network + // thread. + // + // The network thread is also WebRTC's worker thread, and many peer + // connection calls block the signaling thread on the worker thread. An + // observer making such a call on the network thread would wait for itself + // and deadlock every connection of its factory, and a slow one would stall + // their media. So these observers are queued and called here instead, one + // at a time and in the order they were posted. + // + // The thread starts on first use and is attached to the JVM as a daemon, + // so that it never holds up the JVM's exit. + class JavaCallbackThread + { + public: + static void Post(absl::AnyInvocable callback); + }; +} + +#endif diff --git a/webrtc-jni/src/main/cpp/include/api/RTCDtlsTransportObserver.h b/webrtc-jni/src/main/cpp/include/api/RTCDtlsTransportObserver.h index 37f4daa9..50d897cc 100644 --- a/webrtc-jni/src/main/cpp/include/api/RTCDtlsTransportObserver.h +++ b/webrtc-jni/src/main/cpp/include/api/RTCDtlsTransportObserver.h @@ -24,14 +24,20 @@ #include #include +#include namespace jni { + // WebRTC notifies this observer on its network thread; the Java observer is + // called on the JavaCallbackThread instead. class RTCDtlsTransportObserver : public webrtc::DtlsTransportObserverInterface { public: explicit RTCDtlsTransportObserver(JNIEnv * env, const JavaGlobalRef & observer); - ~RTCDtlsTransportObserver() = default; + + // Waits for a Java call in progress, so the Java observer hears + // nothing more once this returns. + ~RTCDtlsTransportObserver(); // DtlsTransportObserverInterface implementation. void OnStateChange(webrtc::DtlsTransportInformation info) override; @@ -47,10 +53,21 @@ namespace jni jmethodID onError; }; - private: - JavaGlobalRef observer; + // Shared with the queued calls, which may outlive this observer. + struct Target + { + Target(const JavaGlobalRef & observer, + std::shared_ptr javaClass); - const std::shared_ptr javaClass; + // Recursive, so that a Java observer may unregister itself. + std::recursive_mutex mutex; + bool active; + const JavaGlobalRef observer; + const std::shared_ptr javaClass; + }; + + private: + const std::shared_ptr target; }; } diff --git a/webrtc-jni/src/main/cpp/src/api/JavaCallbackThread.cpp b/webrtc-jni/src/main/cpp/src/api/JavaCallbackThread.cpp new file mode 100644 index 00000000..d0593017 --- /dev/null +++ b/webrtc-jni/src/main/cpp/src/api/JavaCallbackThread.cpp @@ -0,0 +1,108 @@ +/* + * 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. + */ + +#include "api/JavaCallbackThread.h" +#include "JavaContext.h" + +#include "rtc_base/logging.h" + +#include +#include +#include +#include +#include + +namespace jni +{ + namespace + { + struct Queue + { + std::mutex mutex; + std::condition_variable condition; + std::deque> callbacks; + }; + + void Run(Queue * queue) + { + JavaVM * vm = javaContext->getVM(); + JNIEnv * env = nullptr; + + JavaVMAttachArgs args; + args.version = JNI_VERSION_1_6; + args.name = const_cast("WebRTC-Callbacks"); + args.group = nullptr; + + if (vm->AttachCurrentThreadAsDaemon(reinterpret_cast(&env), &args) != JNI_OK) { + env = nullptr; + + RTC_LOG(LS_ERROR) << "The Java callback thread cannot attach to the JVM, observers will not be called"; + } + + while (true) { + absl::AnyInvocable callback; + + { + std::unique_lock lock(queue->mutex); + + queue->condition.wait(lock, [queue] { + return !queue->callbacks.empty(); + }); + + callback = std::move(queue->callbacks.front()); + queue->callbacks.pop_front(); + } + + if (env != nullptr) { + try { + std::move(callback)(env); + } + catch (...) { + RTC_LOG(LS_ERROR) << "A Java callback failed with a native exception"; + } + } + } + } + + Queue & GetQueue() + { + // Never destroyed: its thread waits on it for as long as the + // process runs. + static Queue * queue = [] { + Queue * created = new Queue(); + + std::thread(Run, created).detach(); + + return created; + }(); + + return *queue; + } + } + + void JavaCallbackThread::Post(absl::AnyInvocable callback) + { + Queue & queue = GetQueue(); + + { + std::lock_guard lock(queue.mutex); + + queue.callbacks.push_back(std::move(callback)); + } + + queue.condition.notify_one(); + } +} diff --git a/webrtc-jni/src/main/cpp/src/api/RTCDataChannelSendObserver.cpp b/webrtc-jni/src/main/cpp/src/api/RTCDataChannelSendObserver.cpp index df1999e9..a517fa19 100644 --- a/webrtc-jni/src/main/cpp/src/api/RTCDataChannelSendObserver.cpp +++ b/webrtc-jni/src/main/cpp/src/api/RTCDataChannelSendObserver.cpp @@ -1,8 +1,14 @@ #include "api/RTCDataChannelSendObserver.h" +#include "api/JavaCallbackThread.h" #include "api/WebRTCUtils.h" #include "JavaString.h" #include "JNI_WebRTC.h" +#include "rtc_base/logging.h" + +#include +#include + namespace jni { RTCDataChannelSendObserver::RTCDataChannelSendObserver(JNIEnv * env, jobject observer) : @@ -43,21 +49,30 @@ namespace jni if (!callback.get()) { return; } - JNIEnv * env = AttachCurrentThread(); - if (env == nullptr) { - return; - } - if (error == nullptr) { - env->CallVoidMethod(callback.get(), javaClass->onSuccess); - } - else { - JavaLocalRef message(env, env->NewStringUTF(error)); - if (!env->ExceptionCheck()) { - env->CallVoidMethod(callback.get(), javaClass->onFailure, message.get()); + try { + std::optional message; + if (error != nullptr) { + message = error; } + // WebRTC completes sends on its network thread, where the Java observer must not run. + JavaCallbackThread::Post([callback = std::move(callback), javaClass = javaClass, + message = std::move(message)](JNIEnv * env) { + if (!message) { + env->CallVoidMethod(callback.get(), javaClass->onSuccess); + } + else { + JavaLocalRef jMessage(env, env->NewStringUTF(message->c_str())); + if (!env->ExceptionCheck()) { + env->CallVoidMethod(callback.get(), javaClass->onFailure, jMessage.get()); + } + } + // A Java exception must not escape into the next callback. + ReportPendingException(env); + }); + } + catch (...) { + RTC_LOG(LS_ERROR) << "Could not report a data channel send result"; } - // A Java exception must not escape into WebRTC's network task or a destructor. - ReportPendingException(env); } RTCDataChannelSendObserver::JavaSendObserverClass::JavaSendObserverClass(JNIEnv * env) diff --git a/webrtc-jni/src/main/cpp/src/api/RTCDtlsTransportObserver.cpp b/webrtc-jni/src/main/cpp/src/api/RTCDtlsTransportObserver.cpp index 69227b98..d356cfc1 100644 --- a/webrtc-jni/src/main/cpp/src/api/RTCDtlsTransportObserver.cpp +++ b/webrtc-jni/src/main/cpp/src/api/RTCDtlsTransportObserver.cpp @@ -15,6 +15,7 @@ */ #include "api/RTCDtlsTransportObserver.h" +#include "api/JavaCallbackThread.h" #include "api/WebRTCUtils.h" #include "JavaEnums.h" #include "JavaRef.h" @@ -25,49 +26,67 @@ namespace jni { RTCDtlsTransportObserver::RTCDtlsTransportObserver(JNIEnv * env, const JavaGlobalRef & observer) : + target(std::make_shared(observer, JavaClasses::get(env))) + { + } + + RTCDtlsTransportObserver::Target::Target(const JavaGlobalRef & observer, + std::shared_ptr javaClass) : + active(true), observer(observer), - javaClass(JavaClasses::get(env)) + javaClass(std::move(javaClass)) { } + RTCDtlsTransportObserver::~RTCDtlsTransportObserver() + { + std::lock_guard lock(target->mutex); + + target->active = false; + } + void RTCDtlsTransportObserver::OnStateChange(webrtc::DtlsTransportInformation info) { - JNIEnv * env = AttachCurrentThread(); + JavaCallbackThread::Post([target = target, state = info.state()](JNIEnv * env) { + std::lock_guard lock(target->mutex); - if (env == nullptr) { - return; - } + if (!target->active) { + return; + } - try { - auto state = JavaEnums::toJava(env, info.state()); + try { + auto jState = JavaEnums::toJava(env, state); - env->CallVoidMethod(observer, javaClass->onStateChange, state.get()); - } - catch (...) { - ThrowCxxJavaException(env); - } + env->CallVoidMethod(target->observer, target->javaClass->onStateChange, jState.get()); + } + catch (...) { + ThrowCxxJavaException(env); + } - ReportPendingException(env); + ReportPendingException(env); + }); } void RTCDtlsTransportObserver::OnError(webrtc::RTCError error) { - JNIEnv * env = AttachCurrentThread(); + JavaCallbackThread::Post([target = target, message = RTCErrorToString(error)](JNIEnv * env) { + std::lock_guard lock(target->mutex); - if (env == nullptr) { - return; - } + if (!target->active) { + return; + } - try { - JavaLocalRef errorMessage = JavaString::toJava(env, RTCErrorToString(error)); + try { + JavaLocalRef errorMessage = JavaString::toJava(env, message); - env->CallVoidMethod(observer, javaClass->onError, errorMessage.get()); - } - catch (...) { - ThrowCxxJavaException(env); - } + env->CallVoidMethod(target->observer, target->javaClass->onError, errorMessage.get()); + } + catch (...) { + ThrowCxxJavaException(env); + } - ReportPendingException(env); + ReportPendingException(env); + }); } RTCDtlsTransportObserver::JavaRTCDtlsTransportObserverClass::JavaRTCDtlsTransportObserverClass(JNIEnv * env) diff --git a/webrtc/src/main/java/dev/onvoid/webrtc/RTCDataChannel.java b/webrtc/src/main/java/dev/onvoid/webrtc/RTCDataChannel.java index 66244953..5fbbdc6d 100644 --- a/webrtc/src/main/java/dev/onvoid/webrtc/RTCDataChannel.java +++ b/webrtc/src/main/java/dev/onvoid/webrtc/RTCDataChannel.java @@ -249,15 +249,15 @@ public void sendAsync(RTCDataChannelBuffer buffer) { * returns, without changing its position or limit. *

* The observer is called once when the operation completes or is discarded - * during channel shutdown. It normally runs on the native network thread; - * a discarded operation can report failure on the thread that destroys it. - * The callback may run before this method returns. It must not block or - * call other WebRTC methods synchronously. Dispatch further work to an - * application executor. No callback is guaranteed during JVM shutdown. + * during channel shutdown. It runs on a native callback thread that all + * send and DTLS transport observers share, in the order the results arrive, + * and may run before this method returns. It may call other WebRTC methods, + * but should not block, since that holds up the other observers. No + * callback is guaranteed during JVM shutdown. *

* Invalid arguments and failures preparing the native operation are thrown * on the calling thread. If preparation fails, the observer is not called. - * Exceptions thrown by the observer are printed and cleared on its native + * Exceptions thrown by the observer are printed and cleared on the callback * thread; they do not propagate to the sender. * * @param buffer The buffer to be queued for transmission. diff --git a/webrtc/src/main/java/dev/onvoid/webrtc/RTCDtlsTransportObserver.java b/webrtc/src/main/java/dev/onvoid/webrtc/RTCDtlsTransportObserver.java index a5b054a7..7270ed78 100644 --- a/webrtc/src/main/java/dev/onvoid/webrtc/RTCDtlsTransportObserver.java +++ b/webrtc/src/main/java/dev/onvoid/webrtc/RTCDtlsTransportObserver.java @@ -18,6 +18,12 @@ /** * Receives events from an {@link RTCDtlsTransport}. + *

+ * The events arrive in order on a native callback thread that all DTLS + * transport and data channel send observers share. An observer may call other + * WebRTC methods, but should not block, since that holds up the other + * observers. Once {@link RTCDtlsTransport#unregisterObserver()} returns, the + * observer receives no more events. * * @author Alex Andres */ diff --git a/webrtc/src/test/java/dev/onvoid/webrtc/RTCDataChannelSendCompletionTests.java b/webrtc/src/test/java/dev/onvoid/webrtc/RTCDataChannelSendCompletionTests.java index 80ff23d9..3c4b9bf9 100644 --- a/webrtc/src/test/java/dev/onvoid/webrtc/RTCDataChannelSendCompletionTests.java +++ b/webrtc/src/test/java/dev/onvoid/webrtc/RTCDataChannelSendCompletionTests.java @@ -8,11 +8,16 @@ import static org.junit.jupiter.api.Assertions.assertTrue; import java.nio.ByteBuffer; +import java.util.Collections; import java.util.concurrent.CompletableFuture; import java.util.concurrent.CountDownLatch; import java.util.concurrent.TimeUnit; import java.util.concurrent.atomic.AtomicInteger; +import dev.onvoid.webrtc.media.audio.AudioOptions; +import dev.onvoid.webrtc.media.audio.AudioTrack; +import dev.onvoid.webrtc.media.audio.AudioTrackSource; + import org.junit.jupiter.api.Test; class RTCDataChannelSendCompletionTests extends TestBase { @@ -110,6 +115,48 @@ public void onFailure(String error) { } } + @Test + void observerCanUseRtpSender() throws Exception { + // WebRTC completes sends on the network thread, which is also the + // worker thread. Getting a sender's parameters blocks the signaling + // thread on the worker thread, which deadlocks should the observer + // run on that thread. + TestPeerConnection caller = new TestPeerConnection(factory); + TestPeerConnection callee = new TestPeerConnection(factory); + try (TestDataChannelPair pair = new TestDataChannelPair(factory)) { + AudioTrackSource source = factory.createAudioSource(new AudioOptions()); + AudioTrack track = factory.createAudioTrack("audioTrack", source); + RTCRtpSender audioSender = caller.getPeerConnection() + .addTrack(track, Collections.singletonList("stream1")); + + caller.setRemotePeerConnection(callee); + callee.setRemotePeerConnection(caller); + callee.setRemoteDescription(caller.createOffer()); + caller.setRemoteDescription(callee.createAnswer()); + + pair.connect(); + + CompletableFuture parameters = new CompletableFuture<>(); + RTCDataChannelBuffer buffer = new RTCDataChannelBuffer(ByteBuffer.allocate(1), true); + pair.sender.sendAsync(buffer, new RTCDataChannelSendObserver() { + @Override + public void onSuccess() { + parameters.complete(audioSender.getParameters()); + } + + @Override + public void onFailure(String error) { + parameters.completeExceptionally(new IllegalStateException(error)); + } + }); + assertNotNull(parameters.get(5, TimeUnit.SECONDS)); + } + finally { + caller.close(); + callee.close(); + } + } + @Test void pendingSendsCompleteOnceWhenPeerCloses() throws Exception { RTCPeerConnection peer = factory.createPeerConnection(new RTCConfiguration(), candidate -> { }); diff --git a/webrtc/src/test/java/dev/onvoid/webrtc/RTCDtlsTransportTests.java b/webrtc/src/test/java/dev/onvoid/webrtc/RTCDtlsTransportTests.java index 45b31931..e9e82c45 100644 --- a/webrtc/src/test/java/dev/onvoid/webrtc/RTCDtlsTransportTests.java +++ b/webrtc/src/test/java/dev/onvoid/webrtc/RTCDtlsTransportTests.java @@ -21,7 +21,10 @@ import static org.junit.jupiter.api.Assertions.assertNotNull; import java.util.ArrayList; +import java.util.Collections; import java.util.List; +import java.util.concurrent.CompletableFuture; +import java.util.concurrent.TimeUnit; import java.util.concurrent.atomic.AtomicInteger; import dev.onvoid.webrtc.media.audio.AudioOptions; @@ -124,4 +127,54 @@ public void onError(String error) { } "The replaced observer must not receive further events"); } + @Test + void observerCanUseRtpSender() throws Exception { + // WebRTC reports DTLS states on the network thread, which is also the + // worker thread. Getting a sender's parameters blocks the signaling + // thread on the worker thread, which deadlocks should the observer + // run on that thread. + TestPeerConnection offerer = new TestPeerConnection(factory); + TestPeerConnection answerer = new TestPeerConnection(factory); + + try { + AudioTrackSource audioSource = factory.createAudioSource(new AudioOptions()); + AudioTrack audioTrack = factory.createAudioTrack("audioTrack", audioSource); + RTCRtpSender audioSender = offerer.getPeerConnection() + .addTrack(audioTrack, Collections.singletonList("stream1")); + + offerer.setRemotePeerConnection(answerer); + answerer.setRemotePeerConnection(offerer); + + RTCSessionDescription offer = offerer.createOffer(); + RTCDtlsTransport dtlsTransport = audioSender.getTransport(); + assertNotNull(dtlsTransport); + + CompletableFuture parameters = new CompletableFuture<>(); + dtlsTransport.registerObserver(new RTCDtlsTransportObserver() { + @Override + public void onStateChange(RTCDtlsTransportState state) { + if (state == RTCDtlsTransportState.CONNECTED) { + parameters.complete(audioSender.getParameters()); + } + } + + @Override + public void onError(String error) { + parameters.completeExceptionally(new IllegalStateException(error)); + } + }); + + answerer.setRemoteDescription(offer); + offerer.setRemoteDescription(answerer.createAnswer()); + + assertNotNull(parameters.get(10, TimeUnit.SECONDS)); + + dtlsTransport.unregisterObserver(); + } + finally { + offerer.close(); + answerer.close(); + } + } + }