fix: call network thread observers on a callback thread

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
 

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 <jni.h>
+
+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<void(JNIEnv *) &&> callback);
+	};
+}
+
+#endif

webrtc-jni/src/main/cpp/include/api/RTCDtlsTransportObserver.h

@@ -24,14 +24,20 @@
 
 #include <jni.h>
 #include <memory>
+#include <mutex>
 
 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<jobject> & 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<jobject> observer;
+			// Shared with the queued calls, which may outlive this observer.
+			struct Target
+			{
+				Target(const JavaGlobalRef<jobject> & observer,
+					std::shared_ptr<JavaRTCDtlsTransportObserverClass> javaClass);
 
-			const std::shared_ptr<JavaRTCDtlsTransportObserverClass> javaClass;
+				// Recursive, so that a Java observer may unregister itself.
+				std::recursive_mutex mutex;
+				bool active;
+				const JavaGlobalRef<jobject> observer;
+				const std::shared_ptr<JavaRTCDtlsTransportObserverClass> javaClass;
+			};
+
+		private:
+			const std::shared_ptr<Target> target;
 	};
 }
 

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 <condition_variable>
+#include <deque>
+#include <mutex>
+#include <thread>
+#include <utility>
+
+namespace jni
+{
+	namespace
+	{
+		struct Queue
+		{
+			std::mutex mutex;
+			std::condition_variable condition;
+			std::deque<absl::AnyInvocable<void(JNIEnv *) &&>> callbacks;
+		};
+
+		void Run(Queue * queue)
+		{
+			JavaVM * vm = javaContext->getVM();
+			JNIEnv * env = nullptr;
+
+			JavaVMAttachArgs args;
+			args.version = JNI_VERSION_1_6;
+			args.name = const_cast<char *>("WebRTC-Callbacks");
+			args.group = nullptr;
+
+			if (vm->AttachCurrentThreadAsDaemon(reinterpret_cast<void **>(&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<void(JNIEnv *) &&> callback;
+
+				{
+					std::unique_lock<std::mutex> 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<void(JNIEnv *) &&> callback)
+	{
+		Queue & queue = GetQueue();
+
+		{
+			std::lock_guard<std::mutex> lock(queue.mutex);
+
+			queue.callbacks.push_back(std::move(callback));
+		}
+
+		queue.condition.notify_one();
+	}
+}

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 <optional>
+#include <string>
+
 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<jstring> message(env, env->NewStringUTF(error));
-			if (!env->ExceptionCheck()) {
-				env->CallVoidMethod(callback.get(), javaClass->onFailure, message.get());
+		try {
+			std::optional<std::string> 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<jstring> 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)

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<jobject> & observer) :
+		target(std::make_shared<Target>(observer, JavaClasses::get<JavaRTCDtlsTransportObserverClass>(env)))
+	{
+	}
+
+	RTCDtlsTransportObserver::Target::Target(const JavaGlobalRef<jobject> & observer,
+		std::shared_ptr<JavaRTCDtlsTransportObserverClass> javaClass) :
+		active(true),
 		observer(observer),
-		javaClass(JavaClasses::get<JavaRTCDtlsTransportObserverClass>(env))
+		javaClass(std::move(javaClass))
 	{
 	}
 
+	RTCDtlsTransportObserver::~RTCDtlsTransportObserver()
+	{
+		std::lock_guard<std::recursive_mutex> 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<std::recursive_mutex> 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<std::recursive_mutex> lock(target->mutex);
 
-		if (env == nullptr) {
-			return;
-		}
+			if (!target->active) {
+				return;
+			}
 
-		try {
-			JavaLocalRef<jstring> errorMessage = JavaString::toJava(env, RTCErrorToString(error));
+			try {
+				JavaLocalRef<jstring> 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)

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.
 	 * <p>
 	 * 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.
 	 * <p>
 	 * 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.