refactor: let the PeerConnectionFactory own its worker thread

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

@@ -20,6 +20,7 @@
 #include "api/audio/audio_device.h"
 #include "api/audio/audio_device_defines.h"
 #include "api/scoped_refptr.h"
+#include "rtc_base/thread.h"
 
 #include <atomic>
 #include <cstdint>
@@ -61,8 +62,12 @@ namespace jni
 			~ProxyAudioDeviceModule() override;
 
 			// Enables or disables the device capture path. May be called from any
-			// thread. Callers that must not interleave with AudioState starting a
-			// recording should run this on the factory's worker thread.
+			// thread. AudioState starts and stops the module's recording on the
+			// factory's worker thread, so the switch is made there to keep the two
+			// from interleaving. The factory creates and owns that thread; this
+			// proxy learns it when WebRTC registers its transport, which it does
+			// on the worker thread. Until then no send stream can exist, and the
+			// switch is made on the calling thread.
 			void SetCaptureEnabled(bool enabled);
 			bool CaptureEnabled() const;
 
@@ -145,6 +150,8 @@ namespace jni
 			std::optional<Stats> GetStats() const override;
 
 		private:
+			void ApplyCaptureEnabled(bool enabled);
+
 			// The AudioTransport registered with the wrapped module in place of
 			// the one WebRTC registers with this proxy. Render calls are always
 			// forwarded; recorded data is forwarded only while capture is enabled.
@@ -203,6 +210,8 @@ namespace jni
 			std::atomic<bool> captureEnabled_;
 			TransportGate gate_;
 			bool gateRegistered_;
+			// The factory's worker thread, set while WebRTC's transport is registered.
+			std::atomic<webrtc::Thread *> workerThread_;
 	};
 }
 

webrtc-jni/src/main/cpp/src/JNI_PeerConnectionFactory.cpp

@@ -32,10 +32,14 @@
 #include "JavaUtils.h"
 #include "WebRTCContext.h"
 
+#include "api/audio/builtin_audio_processing_builder.h"
 #include "api/audio/create_audio_device_module.h"
-#include "api/create_peerconnection_factory.h"
+#include "api/create_modular_peer_connection_factory.h"
+#include "api/enable_media.h"
+#include "api/environment/environment_factory.h"
 #include "api/audio_codecs/builtin_audio_decoder_factory.h"
 #include "api/audio_codecs/builtin_audio_encoder_factory.h"
+#include "api/rtc_event_log/rtc_event_log_factory.h"
 
 #include "api/ProxyAudioDeviceModule.h"
 #include "media/audio/CustomAudioSource.h"
@@ -92,23 +96,16 @@ JNIEXPORT void JNICALL Java_dev_onvoid_webrtc_PeerConnectionFactory_initialize
 		auto signalingThread = webrtc::Thread::Create();
 		signalingThread->SetName("webrtc_jni_signaling_thread", nullptr);
 
-		auto workerThread = webrtc::Thread::Create();
-		workerThread->SetName("webrtc_jni_worker_thread", nullptr);
-
 		if (!networkThread->Start()) {
 			throw jni::Exception("Start network thread failed");
 		}
 		if (!signalingThread->Start()) {
 			throw jni::Exception("Start signaling thread failed");
 		}
-		if (!workerThread->Start()) {
-			throw jni::Exception("Start worker thread failed");
-		}
 
 		webrtc::AudioProcessing * processing = (audioProcessing != nullptr)
 			? GetHandle<webrtc::AudioProcessing>(env, audioProcessing)
 			: nullptr;
-		webrtc::scoped_refptr<webrtc::AudioProcessing> apm(processing);
 		webrtc::scoped_refptr<webrtc::AudioDeviceModule> adm(audioDevModule);
 
 		if (!adm) {
@@ -136,25 +133,31 @@ JNIEXPORT void JNICALL Java_dev_onvoid_webrtc_PeerConnectionFactory_initialize
 		// send stream.
 		auto proxy = jni::ProxyAudioDeviceModule::Create(adm);
 
-		auto factory = webrtc::CreatePeerConnectionFactory(
-			networkThread.get(),
-			workerThread.get(),
-			signalingThread.get(),
-			proxy,
-			webrtc::CreateBuiltinAudioEncoderFactory(),
-			webrtc::CreateBuiltinAudioDecoderFactory(),
-			std::move(videoEncoderFactory),
-			std::move(videoDecoderFactory),
-			nullptr,
-			apm,
-			nullptr,
-			std::move(fieldTrials));
+		// The worker thread is left to the factory, which creates and owns one
+		// when PeerConnectionFactoryDependencies::worker_thread is not set.
+		webrtc::PeerConnectionFactoryDependencies dependencies;
+		dependencies.network_thread = networkThread.get();
+		dependencies.signaling_thread = signalingThread.get();
+		dependencies.socket_factory = networkThread->socketserver();
+		dependencies.event_log_factory = std::make_unique<webrtc::RtcEventLogFactory>();
+		dependencies.env = webrtc::CreateEnvironment(std::move(fieldTrials));
+		dependencies.adm = proxy;
+		dependencies.audio_encoder_factory = webrtc::CreateBuiltinAudioEncoderFactory();
+		dependencies.audio_decoder_factory = webrtc::CreateBuiltinAudioDecoderFactory();
+		dependencies.audio_processing_builder = (processing != nullptr)
+			? webrtc::CustomAudioProcessing(webrtc::scoped_refptr<webrtc::AudioProcessing>(processing))
+			: std::make_unique<webrtc::BuiltinAudioProcessingBuilder>();
+		dependencies.video_encoder_factory = std::move(videoEncoderFactory);
+		dependencies.video_decoder_factory = std::move(videoDecoderFactory);
+
+		webrtc::EnableMedia(dependencies);
+
+		auto factory = webrtc::CreateModularPeerConnectionFactory(std::move(dependencies));
 
 		if (factory != nullptr) {
 			SetHandle(env, caller, factory.release());
 			SetHandle(env, caller, "networkThreadHandle", networkThread.release());
 			SetHandle(env, caller, "signalingThreadHandle", signalingThread.release());
-			SetHandle(env, caller, "workerThreadHandle", workerThread.release());
 			SetHandle(env, caller, "audioModuleHandle", proxy.release());
 		}
 		else {
@@ -174,7 +177,6 @@ JNIEXPORT void JNICALL Java_dev_onvoid_webrtc_PeerConnectionFactory_dispose
 
 	webrtc::Thread * networkThread = GetHandle<webrtc::Thread>(env, caller, "networkThreadHandle");
 	webrtc::Thread * signalingThread = GetHandle<webrtc::Thread>(env, caller, "signalingThreadHandle");
-	webrtc::Thread * workerThread = GetHandle<webrtc::Thread>(env, caller, "workerThreadHandle");
 	jni::ProxyAudioDeviceModule * audioModule = GetHandle<jni::ProxyAudioDeviceModule>(env, caller, "audioModuleHandle");
 
 	webrtc::RefCountReleaseStatus status = factory->Release();
@@ -196,10 +198,6 @@ JNIEXPORT void JNICALL Java_dev_onvoid_webrtc_PeerConnectionFactory_dispose
 			signalingThread->Stop();
 			delete signalingThread;
 		}
-		if (workerThread) {
-			workerThread->Stop();
-			delete workerThread;
-		}
 		if (audioModule) {
 			// The factory dropped its reference above; this releases the last one
 			// and, with it, the proxy's reference on the wrapped module.
@@ -223,15 +221,8 @@ JNIEXPORT void JNICALL Java_dev_onvoid_webrtc_PeerConnectionFactory_setDeviceCap
 	jni::ProxyAudioDeviceModule * audioModule = GetHandle<jni::ProxyAudioDeviceModule>(env, caller, "audioModuleHandle");
 	CHECK_HANDLE(audioModule);
 
-	webrtc::Thread * workerThread = GetHandle<webrtc::Thread>(env, caller, "workerThreadHandle");
-	CHECK_HANDLE(workerThread);
-
 	try {
-		// AudioState starts and stops the module's recording on the worker thread.
-		// Switching the capture path there keeps the two from interleaving.
-		workerThread->BlockingCall([audioModule, enabled]() {
-			audioModule->SetCaptureEnabled(enabled == JNI_TRUE);
-		});
+		audioModule->SetCaptureEnabled(enabled == JNI_TRUE);
 	}
 	catch (...) {
 		ThrowCxxJavaException(env);

webrtc-jni/src/main/cpp/src/api/ProxyAudioDeviceModule.cpp

@@ -34,7 +34,8 @@ namespace jni
 			delegate_(std::move(delegate)),
 			captureEnabled_(false),
 			gate_(captureEnabled_),
-			gateRegistered_(false)
+			gateRegistered_(false),
+			workerThread_(nullptr)
 	{
 	}
 
@@ -52,6 +53,20 @@ namespace jni
 	}
 
 	void ProxyAudioDeviceModule::SetCaptureEnabled(bool enabled)
+	{
+		webrtc::Thread * workerThread = workerThread_.load(std::memory_order_acquire);
+
+		if (workerThread) {
+			workerThread->BlockingCall([this, enabled]() {
+				ApplyCaptureEnabled(enabled);
+			});
+		}
+		else {
+			ApplyCaptureEnabled(enabled);
+		}
+	}
+
+	void ProxyAudioDeviceModule::ApplyCaptureEnabled(bool enabled)
 	{
 		captureEnabled_.store(enabled, std::memory_order_release);
 
@@ -85,6 +100,10 @@ namespace jni
 
 		if (result == 0) {
 			gateRegistered_ = audioCallback != nullptr;
+
+			// WebRtcVoiceEngine registers and unregisters its transport on the
+			// factory's worker thread.
+			workerThread_.store(audioCallback ? webrtc::Thread::Current() : nullptr, std::memory_order_release);
 		}
 
 		return result;

webrtc/src/main/java/dev/onvoid/webrtc/PeerConnectionFactory.java

@@ -93,9 +93,6 @@ private enum AudioInputMode {
 	@SuppressWarnings("unused")
 	private long signalingThreadHandle;
 
-	@SuppressWarnings("unused")
-	private long workerThreadHandle;
-
 	@SuppressWarnings("unused")
 	private long audioModuleHandle;
 

webrtc/src/test/java/dev/onvoid/webrtc/PeerConnectionFactoryTests.java

@@ -381,6 +381,37 @@ void customAudioTrackNullParamsDoNotCommitFactory() {
 		}
 	}
 
+	@Test
+	void deviceAudioSourceAfterPeerConnection() {
+		// A peer connection starts the factory's media engine on the worker
+		// thread the factory creates for itself. Device capture is then switched
+		// on that thread, which must still work once audio is being sent.
+		AudioDeviceModule audioModule = new AudioDeviceModule(AudioLayer.kDummyAudio);
+		PeerConnectionFactory audioFactory = new PeerConnectionFactory(audioModule);
+
+		try {
+			RTCPeerConnection peerConnection = audioFactory.createPeerConnection(
+					new RTCConfiguration(), candidate -> { });
+
+			AudioTrackSource deviceSource = audioFactory.createAudioSource(new AudioOptions());
+			AudioTrack deviceTrack = audioFactory.createAudioTrack("deviceTrack", deviceSource);
+
+			RTCRtpSender sender = peerConnection.addTrack(deviceTrack,
+					Collections.singletonList("stream0"));
+
+			assertNotNull(sender);
+
+			sender.dispose();
+			peerConnection.close();
+			deviceTrack.dispose();
+			deviceSource.dispose();
+		}
+		finally {
+			audioFactory.dispose();
+			audioModule.dispose();
+		}
+	}
+
 	@Test
 	void createVideoTrackNullParams() {
 		assertThrows(NullPointerException.class, () -> {