Skip to content

fix(android): serialize factory teardown and share callback guards #71

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Merged
greenfrvr merged 1 commit into from
Sep 22, 2026
Merged

fix(android): serialize factory teardown and share callback guards #71

Changes from all commits
Commits
Show all changes
1 commit
Select commit
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
166 changes: 82 additions & 84 deletions android/src/main/java/com/oney/WebRTCModule/WebRTCModule.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -27,6 +27,7 @@
import org.webrtc.*;
import org.webrtc.audio.AudioDeviceModule;

import java.util.ArrayDeque;
import java.util.ArrayList;
import java.util.Arrays;
import java.util.HashMap;
Expand Down Expand Up @@ -55,6 +56,10 @@ public class WebRTCModule extends ReactContextBaseJavaModule {
private final SparseArray<PeerConnectionObserver> mPeerConnectionObservers;
final Map<String, MediaStream> localStreams;

// Accessed only on ThreadUtils' executor. Lifecycle requests wait for both teardown phases.
private final ArrayDeque<Runnable> pendingFactoryOperations = new ArrayDeque<>();
private boolean factoryDisposalPending;

// Store generated certificates by ID to avoid exposing private keys to JS
private static final Map<String, RtcCertificatePem> mCertificates = new HashMap<>();

Expand Down Expand Up @@ -129,7 +134,7 @@ public WebRTCModule(ReactApplicationContext reactContext) {

@ReactMethod
public void createCallFactory(ReadableMap options, Promise promise) {
ThreadUtils.runOnExecutor(() -> {
runFactoryOperation(() -> {
try {
boolean bypassVoiceProcessing = options != null && options.hasKey("bypassVoiceProcessing")
&& options.getBoolean("bypassVoiceProcessing");
Expand Down Expand Up @@ -165,7 +170,20 @@ public void createCallFactory(ReadableMap options, Promise promise) {

@ReactMethod
public void disposeCallFactory(Promise promise) {
ThreadUtils.runOnExecutor(() -> disposeCurrentFactoryOrdered(promise::resolve));
runFactoryOperation(() -> disposeCurrentFactoryOrdered(promise::resolve));
}

private void runFactoryOperation(Runnable operation) {
ThreadUtils.runOnExecutor(() -> {
pendingFactoryOperations.addLast(operation);
drainFactoryOperations();
});
}

private void drainFactoryOperations() {
while (!factoryDisposalPending && !pendingFactoryOperations.isEmpty()) {
pendingFactoryOperations.removeFirst().run();
}
}

/**
Expand All @@ -190,6 +208,7 @@ private void disposeCurrentFactoryOrdered(Consumer<Boolean> onDisposed) {
return;
}

factoryDisposalPending = true;
for (int pcId : factoryRegistry.currentOwnedPcIds()) {
try {
PeerConnectionObserver pco = mPeerConnectionObservers.get(pcId);
Expand All @@ -202,41 +221,40 @@ private void disposeCurrentFactoryOrdered(Consumer<Boolean> onDisposed) {
}

ThreadUtils.runOnExecutor(() -> {
for (Map.Entry<String, MediaStream> entry : localStreams.entrySet()) {
try {
MediaStream stream = entry.getValue();
for (AudioTrack t : new ArrayList<>(stream.audioTracks)) stream.removeTrack(t);
for (VideoTrack t : new ArrayList<>(stream.videoTracks)) stream.removeTrack(t);
stream.dispose();
} catch (Exception e) {
Log.w(TAG, "disposeCurrentFactoryOrdered(): error disposing stream " + entry.getKey(), e);
try {
for (Map.Entry<String, MediaStream> entry : localStreams.entrySet()) {
try {
MediaStream stream = entry.getValue();
for (AudioTrack t : new ArrayList<>(stream.audioTracks)) stream.removeTrack(t);
for (VideoTrack t : new ArrayList<>(stream.videoTracks)) stream.removeTrack(t);
stream.dispose();
} catch (Exception e) {
Log.w(TAG, "disposeCurrentFactoryOrdered(): error disposing stream " + entry.getKey(), e);
}
}
}
localStreams.clear();
localStreams.clear();

for (int pcId : factoryRegistry.currentOwnedPcIds()) {
try {
PeerConnectionObserver pco = mPeerConnectionObservers.get(pcId);
if (pco != null) {
pco.dispose();
for (int pcId : factoryRegistry.currentOwnedPcIds()) {
try {
disposePeerConnection(pcId);
} catch (Exception e) {
Log.w(TAG, "disposeCurrentFactoryOrdered(): error disposing pc " + pcId, e);
}
} catch (Exception e) {
Log.w(TAG, "disposeCurrentFactoryOrdered(): error disposing pc " + pcId, e);
} finally {
mPeerConnectionObservers.remove(pcId);
factoryRegistry.unbindPeerConnection(pcId);
}
}

for (String trackId : factoryRegistry.currentOwnedTrackIds()) {
try {
getUserMediaImpl.disposeTrack(trackId);
} catch (Exception e) {
Log.w(TAG, "disposeCurrentFactoryOrdered(): error disposing track " + trackId, e);
for (String trackId : factoryRegistry.currentOwnedTrackIds()) {
try {
getUserMediaImpl.disposeTrack(trackId);
} catch (Exception e) {
Log.w(TAG, "disposeCurrentFactoryOrdered(): error disposing track " + trackId, e);
}
}
}

onDisposed.accept(factoryRegistry.disposeCurrent());
onDisposed.accept(factoryRegistry.disposeCurrent());
} finally {
factoryDisposalPending = false;
drainFactoryOperations();
}
});
}

Expand Down Expand Up @@ -350,6 +368,19 @@ private PeerConnection getPeerConnection(int id) {
return (pco == null) ? null : pco.getPeerConnection();
}

private void runWithPeerConnection(
PeerConnectionObserver pco, Promise promise, Consumer<PeerConnection> action) {
ThreadUtils.runOnExecutor(() -> {
PeerConnection pc = pco.getPeerConnection();
if (pc == null) {
Log.d(TAG, "PeerConnection disposed before callback ran");
promise.reject("E_PC_DISPOSED", "PeerConnection disposed");
return;
}
action.accept(pc);
});
}

void sendEvent(String eventName, @Nullable ReadableMap params) {
if (getReactApplicationContext().hasActiveReactInstance()) {
getReactApplicationContext()
Expand Down Expand Up @@ -1294,14 +1325,7 @@ public void onCreateFailure(String s) {

@Override
public void onCreateSuccess(SessionDescription sdp) {
ThreadUtils.runOnExecutor(() -> {
final PeerConnection livePc = pco.getPeerConnection();
if (livePc == null) {
Log.d(TAG, "onCreateSuccess: pc " + id + " disposed before the callback ran");
promise.reject("E_PC_DISPOSED", "PeerConnection disposed");
return;
}

runWithPeerConnection(pco, promise, livePc -> {
WritableMap params = Arguments.createMap();
WritableMap sdpInfo = Arguments.createMap();

Expand All @@ -1312,7 +1336,7 @@ public void onCreateSuccess(SessionDescription sdp) {
params.putMap("sdpInfo", sdpInfo);

WritableArray newTransceivers = Arguments.createArray();
for (RtpTransceiver transceiver : peerConnection.getTransceivers()) {
for (RtpTransceiver transceiver : livePc.getTransceivers()) {
if (!receiversIds.contains(transceiver.getReceiver().id())) {
WritableMap newTransceiver = Arguments.createMap();
newTransceiver.putInt("transceiverOrder", pco.getNextTransceiverId());
Expand Down Expand Up @@ -1359,14 +1383,7 @@ public void onCreateFailure(String s) {

@Override
public void onCreateSuccess(SessionDescription sdp) {
ThreadUtils.runOnExecutor(() -> {
final PeerConnection livePc = pco.getPeerConnection();
if (livePc == null) {
Log.d(TAG, "onCreateSuccess: pc " + id + " disposed before the callback ran");
promise.reject("E_PC_DISPOSED", "PeerConnection disposed");
return;
}

runWithPeerConnection(pco, promise, livePc -> {
WritableMap params = Arguments.createMap();
WritableMap sdpInfo = Arguments.createMap();

Expand Down Expand Up @@ -1408,14 +1425,7 @@ public void onCreateSuccess(SessionDescription sdp) {}

@Override
public void onSetSuccess() {
ThreadUtils.runOnExecutor(() -> {
final PeerConnection livePc = pco.getPeerConnection();
if (livePc == null) {
Log.d(TAG, "peerConnectionSetLocalDescription() onSetSuccess: pc " + pcId
+ " disposed before the callback ran");
promise.reject("E_PC_DISPOSED", "PeerConnection disposed");
return;
}
runWithPeerConnection(pco, promise, livePc -> {
WritableMap newSdpMap = Arguments.createMap();
WritableMap params = Arguments.createMap();

Expand Down Expand Up @@ -1480,14 +1490,7 @@ public void onCreateSuccess(final SessionDescription sdp) {}

@Override
public void onSetSuccess() {
ThreadUtils.runOnExecutor(() -> {
final PeerConnection livePc = pco.getPeerConnection();
if (livePc == null) {
Log.d(TAG, "peerConnectionSetRemoteDescription() onSetSuccess: pc " + id
+ " disposed before the callback ran");
promise.reject("E_PC_DISPOSED", "PeerConnection disposed");
return;
}
runWithPeerConnection(pco, promise, livePc -> {
WritableMap newSdpMap = Arguments.createMap();
WritableMap params = Arguments.createMap();

Expand All @@ -1502,7 +1505,7 @@ public void onSetSuccess() {
params.putMap("sdpInfo", newSdpMap);

WritableArray newTransceivers = Arguments.createArray();
for (RtpTransceiver transceiver : peerConnection.getTransceivers()) {
for (RtpTransceiver transceiver : livePc.getTransceivers()) {
if (!receiversIds.contains(transceiver.getReceiver().id())) {
WritableMap newTransceiver = Arguments.createMap();
newTransceiver.putInt("transceiverOrder", pco.getNextTransceiverId());
Expand Down Expand Up @@ -1634,14 +1637,7 @@ public void peerConnectionAddICECandidate(int pcId, ReadableMap candidateMap, Pr
peerConnection.addIceCandidate(candidate, new AddIceObserver() {
@Override
public void onAddSuccess() {
ThreadUtils.runOnExecutor(() -> {
final PeerConnection livePc = pco.getPeerConnection();
if (livePc == null) {
Log.d(TAG, "peerConnectionAddICECandidate() onAddSuccess: pc " + pcId
+ " disposed before the callback ran");
promise.reject("E_PC_DISPOSED", "PeerConnection disposed");
return;
}
runWithPeerConnection(pco, promise, livePc -> {
WritableMap newSdpMap = Arguments.createMap();
SessionDescription newSdp = livePc.getRemoteDescription();
newSdpMap.putString("type", newSdp.type.canonicalForm());
Expand Down Expand Up @@ -1685,19 +1681,21 @@ public void peerConnectionClose(int id) {

@ReactMethod
public void peerConnectionDispose(int id) {
ThreadUtils.runOnExecutor(() -> {
PeerConnectionObserver pco = mPeerConnectionObservers.get(id);
try {
if (pco == null) {
Log.d(TAG, "peerConnectionDispose() peerConnection observer is null");
} else {
pco.dispose();
}
} finally {
mPeerConnectionObservers.remove(id);
factoryRegistry.unbindPeerConnection(id);
ThreadUtils.runOnExecutor(() -> disposePeerConnection(id));
}

private void disposePeerConnection(int id) {
PeerConnectionObserver pco = mPeerConnectionObservers.get(id);
try {
if (pco == null) {
Log.d(TAG, "peerConnectionDispose() peerConnection observer is null");
} else {
pco.dispose();
}
});
} finally {
mPeerConnectionObservers.remove(id);
factoryRegistry.unbindPeerConnection(id);
}
}

@ReactMethod
Expand Down
Loading