Suspected memory leak #239

import com.lsl.core.webrtc.model.WebrtcPushPeerObserver;
import dev.onvoid.webrtc.*;
import dev.onvoid.webrtc.media.audio.*;
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;
import jakarta.annotation.PreDestroy;
import lombok.extern.slf4j.Slf4j;

import java.nio.ByteBuffer;
import java.util.List;
import java.util.concurrent.*;
import java.util.concurrent.locks.ReentrantLock;

@Slf4j
public class WebrtcPushVoiceService {

    private final ReentrantLock PUSH_VOICE_LOCK = new ReentrantLock();

    private volatile boolean pushVoiceFlag = false;

    private final SignalingClient signalingClient;
    private final AudioDevice audioDevice;

    private AudioDeviceModule audioModule;
    private PeerConnectionFactory factory;
    private RTCPeerConnection peerConnection;

    private AudioTrack audioTrack;
    private VideoTrack videoTrack;

    private CustomVideoSource memoryVideoSource;
    private ScheduledExecutorService dummyVideoExecutor;

    private NativeI420Buffer dummyBlackBuffer;

    private final ExecutorService webrtcPushAsyncExecutor = Executors.newSingleThreadExecutor(
            Thread.ofPlatform().name("webrtc-push-async-thread").factory());

    public WebrtcPushVoiceService(SignalingClient signalingClient, AudioDevice audioDevice) {
        this.signalingClient = signalingClient;
        this.audioDevice = audioDevice;
    }

    public void startPushVoiceStream(String pushUrl) {
        PUSH_VOICE_LOCK.lock();
        try {
            if (pushVoiceFlag) {
                return;
            }

            stopPushInternal();

            pushVoiceFlag = true;

            audioModule = new AudioDeviceModule();
            audioModule.setRecordingDevice(audioDevice);
            audioModule.initRecording();

            factory = new PeerConnectionFactory(audioModule);

            RTCConfiguration config = new RTCConfiguration();

            WebrtcPushPeerObserver observer = new WebrtcPushPeerObserver(
                    pushUrl,
                    signalingClient,
                    () -> webrtcPushAsyncExecutor.submit(this::stopPushVoiceStream));
            peerConnection = factory.createPeerConnection(config, observer);
            observer.setPeerConnection(peerConnection);

            AudioTrackSource audioSource = factory.createAudioSource(new AudioOptions());
            audioTrack = factory.createAudioTrack("audio", audioSource);

            RTCRtpTransceiverInit audioInit = new RTCRtpTransceiverInit();
            audioInit.direction = RTCRtpTransceiverDirection.SEND_ONLY;
            audioInit.streamIds = List.of("rts");
            peerConnection.addTransceiver(audioTrack, audioInit);

            injectMemoryVideoTrackAndOffer();
        } catch (Exception e) {
            stopPushInternal();
        } finally {
            PUSH_VOICE_LOCK.unlock();
        }
    }

    private void injectMemoryVideoTrackAndOffer() {
        try {
            memoryVideoSource = new CustomVideoSource();
            videoTrack = factory.createVideoTrack("dummy_video", memoryVideoSource);

            RTCRtpTransceiverInit videoInit = new RTCRtpTransceiverInit();
            videoInit.direction = RTCRtpTransceiverDirection.SEND_ONLY;
            videoInit.streamIds = List.of("rts");
            peerConnection.addTransceiver(videoTrack, videoInit);

            startDummyVideoLoop();

            createAndSendOffer();
        } catch (Exception e) {
            createAndSendOffer();
        }
    }

    private void startDummyVideoLoop() {
        dummyVideoExecutor = new ScheduledThreadPoolExecutor(1,
                Thread.ofPlatform().name("dummy-video-loop").factory());

        int width = 16;
        int height = 16;

        dummyBlackBuffer = NativeI420Buffer.allocate(width, height);

        ByteBuffer dataY = dummyBlackBuffer.getDataY();
        ByteBuffer dataU = dummyBlackBuffer.getDataU();
        ByteBuffer dataV = dummyBlackBuffer.getDataV();

        byte blackY = 0;
        byte blackUV = (byte) 128;

        while (dataY.hasRemaining())
            dataY.put(blackY);
        while (dataU.hasRemaining())
            dataU.put(blackUV);
        while (dataV.hasRemaining())
            dataV.put(blackUV);

        dummyVideoExecutor.scheduleAtFixedRate(() -> {
            if (!pushVoiceFlag || memoryVideoSource == null || dummyBlackBuffer == null)
                return;
            try {
                dummyBlackBuffer.retain();

                long timestampNs = System.nanoTime();
                VideoFrame frame = new VideoFrame(dummyBlackBuffer, timestampNs);

                memoryVideoSource.pushFrame(frame);

                frame.release();
            } catch (Exception e) {
            }
        }, 0, 33, TimeUnit.MILLISECONDS); // 30 FPS
    }

    private void createAndSendOffer() {
        RTCOfferOptions offerOptions = new RTCOfferOptions();

        peerConnection.createOffer(offerOptions, new CreateSessionDescriptionObserver() {
            @Override
            public void onSuccess(RTCSessionDescription offerDescription) {
                if (!pushVoiceFlag) {
                    return;
                }

                RTCPeerConnection currentPc = peerConnection;
                if (currentPc == null)
                    return;

                currentPc.setLocalDescription(offerDescription, new SetSessionDescriptionObserver() {
                    @Override
                    public void onSuccess() {
                    }

                    @Override
                    public void onFailure(String error) {
                        webrtcPushAsyncExecutor.submit(WebrtcPushVoiceService.this::stopPushVoiceStream);
                    }
                });
            }

            @Override
            public void onFailure(String error) {
                webrtcPushAsyncExecutor.submit(() -> stopPushVoiceStream());
            }
        });
    }

    public void stopPushVoiceStream() {
        PUSH_VOICE_LOCK.lock();
        try {
            stopPushInternal();
        } finally {
            PUSH_VOICE_LOCK.unlock();
        }
    }

    private void stopPushInternal() {
        this.pushVoiceFlag = false;

        if (dummyVideoExecutor != null) {
            dummyVideoExecutor.shutdown();
            try {
                if (!dummyVideoExecutor.awaitTermination(100, TimeUnit.MILLISECONDS)) {
                    dummyVideoExecutor.shutdownNow();
                }
            } catch (InterruptedException e) {
                dummyVideoExecutor.shutdownNow();
                Thread.currentThread().interrupt();
            }
            dummyVideoExecutor = null;
        }

        if (peerConnection != null) {
            RTCPeerConnection tempPc = peerConnection;
            peerConnection = null;
            try {
                tempPc.close();
            } catch (Exception e) {
                log.error("error", e);
            }
        }

        videoTrack = null;
        audioTrack = null;

        if (memoryVideoSource != null) {
            CustomVideoSource tempSource = memoryVideoSource;
            memoryVideoSource = null;
            try {
                tempSource.dispose();
            } catch (Throwable e) {
            }
        }

        if (dummyBlackBuffer != null) {
            try {
                dummyBlackBuffer.release();
                dummyBlackBuffer = null;
            } catch (Exception e) {
            }
        }

        if (factory != null) {
            PeerConnectionFactory tempFactory = factory;
            factory = null;
            try {
                tempFactory.dispose();
            } catch (Throwable t) {
            }
        }

        if (audioModule != null) {
            audioModule.stopRecording();
            AudioDeviceModule tempModule = audioModule;
            audioModule = null;
            try {
                tempModule.dispose();
            } catch (Exception e) {
            }
        }

    }

    @PreDestroy
    public void destroy() {
        stopPushVoiceStream();
        if (!webrtcPushAsyncExecutor.isShutdown()) {
            webrtcPushAsyncExecutor.shutdown();
            try {
                if (!webrtcPushAsyncExecutor.awaitTermination(2, TimeUnit.SECONDS)) {
                    webrtcPushAsyncExecutor.shutdownNow();
                }
            } catch (InterruptedException e) {
                webrtcPushAsyncExecutor.shutdownNow();
                Thread.currentThread().interrupt();
            }
        }
    }
}

public class Test2 {
    static void main() {
        String startUrl = "http://localhost:6666/test/webrtc/push/start?url=?";
        String stopUrl = "http://localhost:6666/test/webrtc/push/stop";

        for (int i = 0; i < 10000; i++) {
            String startResult = HttpUtil.get(startUrl);

            ThreadUtil.sleep(5000);

            String stopResult = HttpUtil.get(stopUrl);
        }
    }
}