Merge pull request #1659 from shogo4405/feature/fix-srtstream-memory-…

…leak

fixed MemoryLeak issue SRTConnection, SRTStream.

HaishinKit/Sources/HKStream/HKIncomingStream.swift

@@ -63,18 +63,18 @@ extension HKIncomingStream: AsyncRunner {
         isRunning = true
         Task {
             await mediaLink.startRunning()
-            for await video in await mediaLink.dequeue where await mediaLink.isRunning {
+            for await video in await mediaLink.dequeue {
                 await stream?.append(video)
             }
         }
         Task {
-            for await video in videoCodec.outputStream where videoCodec.isRunning {
+            for await video in videoCodec.outputStream {
                 await mediaLink.enqueue(video)
             }
         }
         Task {
             await audioPlayerNode?.startRunning()
-            for await audio in audioCodec.outputStream where audioCodec.isRunning {
+            for await audio in audioCodec.outputStream {
                 await audioPlayerNode?.enqueue(audio.0, when: audio.1)
                 await stream?.append(audio.0, when: audio.1)
             }

HaishinKit/Sources/HKStream/MediaLink.swift

@@ -6,7 +6,7 @@ final actor MediaLink {
     static let duration: TimeInterval = 0.0
 
     var dequeue: AsyncStream<CMSampleBuffer> {
-        AsyncStream<CMSampleBuffer> { continutation in
+        AsyncStream { continutation in
             self.continutation = continutation
         }
     }
@@ -62,11 +62,11 @@ extension MediaLink: AsyncRunner {
         guard !isRunning else {
             return
         }
-        duration = 0.0
         isRunning = true
+        duration = 0.0
         displayLink.startRunning()
         Task {
-            for await currentTime in displayLink.updateFrames where isRunning {
+            for await currentTime in displayLink.updateFrames {
                 guard let storage else {
                     continue
                 }

HaishinKit/Sources/ISO/TSReader.swift

@@ -5,7 +5,7 @@ import Foundation
 public final class TSReader {
     /// An asynchronous sequence for reading data.
     public var output: AsyncStream<(UInt16, CMSampleBuffer)> {
-        AsyncStream<(UInt16, CMSampleBuffer)> { continuation in
+        AsyncStream { continuation in
             self.continuation = continuation
         }
     }

HaishinKit/Sources/ISO/TSWriter.swift

@@ -11,7 +11,7 @@ public final class TSWriter {
     static let defaultSegmentDuration: Double = 2
     /// An asynchronous sequence for writing data.
     public var output: AsyncStream<Data> {
-        return AsyncStream<Data> { continuation in
+        AsyncStream { continuation in
             self.continuation = continuation
         }
     }
@@ -152,6 +152,7 @@ public final class TSWriter {
         clockTimeStamp = .zero
         rotatedTimeStamp = .zero
         expectedMedias.removeAll()
+        continuation = nil
     }
 
     private func writePacketizedElementaryStream(_ PID: UInt16, PES: PacketizedElementaryStream, timeStamp: CMTime, randomAccessIndicator: Bool) {

SRTHaishinKit/Sources/SRT/SRTConnection.swift

@@ -45,6 +45,7 @@ public actor SRTConnection: NetworkConnection {
     }
 
     deinit {
+        streams.removeAll()
         srt_cleanup()
     }
 

SRTHaishinKit/Sources/SRT/SRTStream.swift

@@ -43,30 +43,29 @@ public actor SRTStream {
         if await connection?.connected == true {
             readyState = .publishing
             outgoing.startRunning()
-            writer.clear()
             if outgoing.videoInputFormat != nil {
                 writer.expectedMedias.insert(.video)
             }
             if outgoing.audioInputFormat != nil {
                 writer.expectedMedias.insert(.audio)
             }
             Task {
-                for await buffer in outgoing.videoOutputStream where outgoing.isRunning {
+                for await buffer in outgoing.videoOutputStream {
                     append(buffer)
                 }
             }
             Task {
-                for await buffer in outgoing.audioOutputStream where outgoing.isRunning {
+                for await buffer in outgoing.audioOutputStream {
                     append(buffer.0, when: buffer.1)
                 }
             }
             Task {
-                for await buffer in outgoing.videoInputStream where outgoing.isRunning {
+                for await buffer in outgoing.videoInputStream {
                     outgoing.append(video: buffer)
                 }
             }
             Task {
-                for await data in writer.output where outgoing.isRunning {
+                for await data in writer.output {
                     await connection?.send(data)
                 }
             }
@@ -87,11 +86,10 @@ public actor SRTStream {
             return
         }
         if await connection?.connected == true {
-            reader.clear()
             await connection?.recv()
             Task {
                 await incoming.startRunning()
-                for await buffer in reader.output where await incoming.isRunning {
+                for await buffer in reader.output {
                     await incoming.append(buffer.1)
                 }
             }
@@ -106,6 +104,8 @@ public actor SRTStream {
         guard readyState != .idle else {
             return
         }
+        writer.clear()
+        reader.clear()
         outgoing.stopRunning()
         Task { await incoming.stopRunning() }
         readyState = .idle