fix memory-leak RTMPConnection, RTMPStream.

HaishinKit/Sources/Codec/AudioCodec.swift

@@ -20,9 +20,9 @@ final class AudioCodec {
     }
 
     var outputStream: AsyncStream<(AVAudioBuffer, AVAudioTime)> {
-        let (stream, continuation) = AsyncStream.makeStream(of: (AVAudioBuffer, AVAudioTime).self)
-        self.continuation = continuation
-        return stream
+        AsyncStream { continuation in
+            self.continuation = continuation
+        }
     }
 
     /// This instance is running to process(true) or not(false).
@@ -45,9 +45,13 @@ final class AudioCodec {
     }
     private var cursor: Int = 0
     private var inputBuffers: [AVAudioBuffer] = []
+    private var continuation: AsyncStream<(AVAudioBuffer, AVAudioTime)>.Continuation? {
+        didSet {
+            oldValue?.finish()
+        }
+    }
     private var outputBuffers: [AVAudioBuffer] = []
     private var audioConverter: AVAudioConverter?
-    private var continuation: AsyncStream<(AVAudioBuffer, AVAudioTime)>.Continuation?
 
     func append(_ sampleBuffer: CMSampleBuffer) {
         guard isRunning else {
@@ -184,7 +188,7 @@ extension AudioCodec: Runner {
         guard isRunning else {
             return
         }
-        continuation?.finish()
         isRunning = false
+        continuation = nil
     }
 }

HaishinKit/Sources/Codec/VideoCodec.swift

@@ -26,7 +26,7 @@ final class VideoCodec {
     var needsSync = true
     var passthrough = true
     var outputStream: AsyncStream<CMSampleBuffer> {
-        AsyncStream<CMSampleBuffer> { continuation in
+        AsyncStream { continuation in
             self.continuation = continuation
         }
     }

HaishinKit/Sources/HKStream/HKOutgoingStream.swift

@@ -44,13 +44,13 @@ public final class HKOutgoingStream {
     /// The asynchronous sequence for video input buffer.
     public var videoInputStream: AsyncStream<CMSampleBuffer> {
         if 0 < videoInputBufferCounts {
-            let (stream, continuation) = AsyncStream.makeStream(of: CMSampleBuffer.self, bufferingPolicy: .bufferingNewest(videoInputBufferCounts))
-            self.videoInputContinuation = continuation
-            return stream
+            return AsyncStream(CMSampleBuffer.self, bufferingPolicy: .bufferingNewest(videoInputBufferCounts)) { continuation in
+                self.videoInputContinuation = continuation
+            }
         } else {
-            let (stream, continuation) = AsyncStream.makeStream(of: CMSampleBuffer.self)
-            self.videoInputContinuation = continuation
-            return stream
+            return AsyncStream { continuation in
+                self.videoInputContinuation = continuation
+            }
         }
     }
 
@@ -59,7 +59,11 @@ public final class HKOutgoingStream {
 
     private var audioCodec = AudioCodec()
     private var videoCodec = VideoCodec()
-    private var videoInputContinuation: AsyncStream<CMSampleBuffer>.Continuation?
+    private var videoInputContinuation: AsyncStream<CMSampleBuffer>.Continuation? {
+        didSet {
+            oldValue?.finish()
+        }
+    }
 
     /// Create a new instance.
     public init() {
@@ -106,8 +110,9 @@ extension HKOutgoingStream: Runner {
         guard isRunning else {
             return
         }
+        isRunning = false
         videoCodec.stopRunning()
         audioCodec.stopRunning()
-        isRunning = false
+        videoInputContinuation = nil
     }
 }

HaishinKit/Sources/Network/NetworkMonitor.swift

@@ -10,19 +10,28 @@ public final actor NetworkMonitor {
 
     /// An asynchronous sequence for network monitoring  event.
     public var event: AsyncStream<NetworkMonitorEvent> {
-        let (stream, continuation) = AsyncStream<NetworkMonitorEvent>.makeStream()
-        self.continuation = continuation
-        return stream
+        AsyncStream { continuation in
+            self.continuation = continuation
+        }
     }
 
     public private(set) var isRunning = false
+    private var timer: Task<Void, Never>? {
+        didSet {
+            oldValue?.cancel()
+        }
+    }
     private var measureInterval = 3
     private var currentBytesInPerSecond = 0
     private var currentBytesOutPerSecond = 0
     private var previousTotalBytesIn = 0
     private var previousTotalBytesOut = 0
     private var previousQueueBytesOut: [Int] = []
-    private var continuation: AsyncStream<NetworkMonitorEvent>.Continuation?
+    private var continuation: AsyncStream<NetworkMonitorEvent>.Continuation? {
+        didSet {
+            oldValue?.finish()
+        }
+    }
     private weak var reporter: (any NetworkTransportReporter)?
 
     /// Creates a new instance.
@@ -74,11 +83,11 @@ extension NetworkMonitor: AsyncRunner {
             return
         }
         isRunning = true
-        let timer = AsyncStream {
-            try? await Task.sleep(nanoseconds: 1_000_000_000)
-        }
-        Task {
-            for await _ in timer where isRunning {
+        timer = Task {
+            let timer = AsyncStream {
+                try? await Task.sleep(nanoseconds: 1_000_000_000)
+            }
+            for await _ in timer {
                 do {
                     let event = try await collect()
                     continuation?.yield(event)
@@ -94,5 +103,7 @@ extension NetworkMonitor: AsyncRunner {
             return
         }
         isRunning = false
+        timer = nil
+        continuation = nil
     }
 }

HaishinKit/Sources/RTMP/RTMPStream.swift

@@ -383,17 +383,17 @@ public actor RTMPStream {
             try? send("@setDataFrame", arguments: "onMetaData", metadata)
             outgoing.startRunning()
             Task {
-                for await audio in outgoing.audioOutputStream where outgoing.isRunning {
+                for await audio in outgoing.audioOutputStream {
                     append(audio.0, when: audio.1)
                 }
             }
             Task {
-                for await video in outgoing.videoOutputStream where outgoing.isRunning {
+                for await video in outgoing.videoOutputStream {
                     append(video)
                 }
             }
             Task {
-                for await video in outgoing.videoInputStream where outgoing.isRunning {
+                for await video in outgoing.videoInputStream {
                     outgoing.append(video: video)
                 }
             }