Merge remote-tracking branch 'lachenmayer/fix/switch-to-latest-leak-2' into linux-leak

Author: lhoward

Sources/AsyncChannels/AsyncBufferedChannel.swift

@@ -35,7 +35,7 @@ public final class AsyncBufferedChannel<Element>: AsyncSequence, Sendable {
 
   struct Awaiting: Hashable {
     let id: Int
-    let continuation: UnsafeContinuation<Element?, Never>?
+    let continuation: CheckedContinuation<Element?, Never>?
 
     static func placeHolder(id: Int) -> Awaiting {
       Awaiting(id: id, continuation: nil)
@@ -178,7 +178,7 @@ public final class AsyncBufferedChannel<Element>: AsyncSequence, Sendable {
 
       awaiting?.continuation?.resume(returning: nil)
     } operation: {
-      await withUnsafeContinuation { [state] (continuation: UnsafeContinuation<Element?, Never>) in
+      await withCheckedContinuation { [state] (continuation: CheckedContinuation<Element?, Never>) in
         let decision = state.withCriticalRegion { state -> AwaitingDecision in
           let isCancelled = cancellation.withCriticalRegion { $0 }
           guard !isCancelled else { return .resume(nil) }

Sources/AsyncSubjects/AsyncCurrentValueSubject.swift

@@ -67,24 +67,28 @@ public final class AsyncCurrentValueSubject<Element>: AsyncSubject where Element
   /// Sends a value to all consumers
   /// - Parameter element: the value to send
   public func send(_ element: Element) {
-    self.state.withCriticalRegion { state in
+    let channels = self.state.withCriticalRegion { state -> [AsyncBufferedChannel<Element>] in
       state.current = element
-      for channel in state.channels.values {
-        channel.send(element)
-      }
+      return Array(state.channels.values)
+    }
+
+    for channel in channels {
+      channel.send(element)
     }
   }
 
   /// Finishes the async sequences with a normal ending.
   /// - Parameter termination: The termination to finish the subject.
   public func send(_ termination: Termination<Failure>) {
-    self.state.withCriticalRegion { state in
+    let channels = self.state.withCriticalRegion { state -> [AsyncBufferedChannel<Element>] in
       state.terminalState = termination
       let channels = Array(state.channels.values)
       state.channels.removeAll()
-      for channel in channels {
-        channel.finish()
-      }
+      return channels
+    }
+
+    for channel in channels {
+      channel.finish()
     }
   }
 
@@ -133,10 +137,10 @@ public final class AsyncCurrentValueSubject<Element>: AsyncSubject where Element
     }
 
     public mutating func next() async -> Element? {
-      await withTaskCancellationHandler {
-        await self.iterator.next()
-      } onCancel: { [unregister] in
+      await withTaskCancellationHandler { [unregister] in
         unregister()
+      } operation: {
+        await self.iterator.next()
       }
     }
   }

Sources/AsyncSubjects/AsyncPassthroughSubject.swift

@@ -52,23 +52,27 @@ public final class AsyncPassthroughSubject<Element>: AsyncSubject {
   /// Sends a value to all consumers
   /// - Parameter element: the value to send
   public func send(_ element: Element) {
-    self.state.withCriticalRegion { state in
-      for channel in state.channels.values {
-        channel.send(element)
-      }
+    let channels = self.state.withCriticalRegion { state in
+      state.channels.values
+    }
+
+    for channel in channels {
+      channel.send(element)
     }
   }
 
   /// Finishes the subject with a normal ending.
   /// - Parameter termination: The termination to finish the subject
   public func send(_ termination: Termination<Failure>) {
-    self.state.withCriticalRegion { state in
+    let channels = self.state.withCriticalRegion { state -> [AsyncBufferedChannel<Element>] in
       state.terminalState = termination
       let channels = Array(state.channels.values)
       state.channels.removeAll()
-      for channel in channels {
-        channel.finish()
-      }
+      return channels
+    }
+
+    for channel in channels {
+      channel.finish()
     }
   }
 
@@ -116,10 +120,10 @@ public final class AsyncPassthroughSubject<Element>: AsyncSubject {
     }
 
     public mutating func next() async -> Element? {
-      await withTaskCancellationHandler {
-        await self.iterator.next()
-      } onCancel: { [unregister] in
+      await withTaskCancellationHandler { [unregister] in
         unregister()
+      } operation: {
+        await self.iterator.next()
       }
     }
   }

Sources/AsyncSubjects/AsyncReplaySubject.swift

@@ -46,29 +46,33 @@ public final class AsyncReplaySubject<Element>: AsyncSubject where Element: Send
   /// Sends a value to all consumers
   /// - Parameter element: the value to send
   public func send(_ element: Element) {
-    self.state.withCriticalRegion { state in
+    let channels = self.state.withCriticalRegion { state -> [AsyncBufferedChannel<Element>] in
       if state.buffer.count >= state.bufferSize && !state.buffer.isEmpty {
         state.buffer.removeFirst()
       }
       state.buffer.append(element)
-      for channel in state.channels.values {
-        channel.send(element)
-      }
+      return Array(state.channels.values)
+    }
+
+    for channel in channels {
+      channel.send(element)
     }
   }
 
   /// Finishes the subject with a normal ending.
   /// - Parameter termination: The termination to finish the subject.
   public func send(_ termination: Termination<Failure>) {
-    self.state.withCriticalRegion { state in
+    let channels = self.state.withCriticalRegion { state -> [AsyncBufferedChannel<Element>] in
       state.terminalState = termination
       let channels = Array(state.channels.values)
       state.channels.removeAll()
       state.buffer.removeAll()
       state.bufferSize = 0
-      for channel in channels {
-        channel.finish()
-      }
+      return channels
+    }
+
+    for channel in channels {
+      channel.finish()
     }
   }
 
@@ -120,10 +124,10 @@ public final class AsyncReplaySubject<Element>: AsyncSubject where Element: Send
     }
 
     public mutating func next() async -> Element? {
-      await withTaskCancellationHandler {
-        await self.iterator.next()
-      } onCancel: { [unregister] in
+      await withTaskCancellationHandler { [unregister] in
         unregister()
+      } operation: {
+        await self.iterator.next()
       }
     }
   }

Sources/AsyncSubjects/AsyncThrowingCurrentValueSubject.swift

@@ -67,28 +67,32 @@ public final class AsyncThrowingCurrentValueSubject<Element, Failure: Error>: As
   /// Sends a value to all consumers
   /// - Parameter element: the value to send
   public func send(_ element: Element) {
-    self.state.withCriticalRegion { state in
+    let channels = self.state.withCriticalRegion { state -> [AsyncThrowingBufferedChannel<Element, Error>] in
       state.current = element
-      for channel in state.channels.values {
-        channel.send(element)
-      }
+      return Array(state.channels.values)
+    }
+
+    for channel in channels {
+      channel.send(element)
     }
   }
 
   /// Finishes the subject with either a normal ending or an error.
   /// - Parameter termination: The termination to finish the subject.
   public func send(_ termination: Termination<Failure>) {
-    self.state.withCriticalRegion { state in
+    let channels = self.state.withCriticalRegion { state -> [AsyncThrowingBufferedChannel<Element, Error>] in
       state.terminalState = termination
       let channels = Array(state.channels.values)
       state.channels.removeAll()
-      for channel in channels {
-        switch termination {
-          case .finished:
-            channel.finish()
-          case .failure(let error):
-            channel.fail(error)
-        }
+      return channels
+    }
+
+    for channel in channels {
+      switch termination {
+        case .finished:
+          channel.finish()
+        case .failure(let error):
+          channel.fail(error)
       }
     }
   }
@@ -144,10 +148,10 @@ public final class AsyncThrowingCurrentValueSubject<Element, Failure: Error>: As
     }
 
     public mutating func next() async throws -> Element? {
-      try await withTaskCancellationHandler {
-        try await self.iterator.next()
-      } onCancel: { [unregister] in
+      try await withTaskCancellationHandler { [unregister] in
         unregister()
+      } operation: {
+        try await self.iterator.next()
       }
     }
   }

Sources/AsyncSubjects/AsyncThrowingPassthroughSubject.swift

@@ -53,28 +53,31 @@ public final class AsyncThrowingPassthroughSubject<Element, Failure: Error>: Asy
   /// Sends a value to all consumers
   /// - Parameter element: the value to send
   public func send(_ element: Element) {
-    self.state.withCriticalRegion { state in
-      for channel in state.channels.values {
-        channel.send(element)
-      }
+    let channels = self.state.withCriticalRegion { state in
+      state.channels.values
+    }
+
+    for channel in channels {
+      channel.send(element)
     }
   }
 
   /// Finishes the subject with either a normal ending or an error.
   /// - Parameter termination: The termination to finish the subject
   public func send(_ termination: Termination<Failure>) {
-    self.state.withCriticalRegion { state in
+    let channels = self.state.withCriticalRegion { state -> [AsyncThrowingBufferedChannel<Element, Error>] in
       state.terminalState = termination
       let channels = Array(state.channels.values)
       state.channels.removeAll()
+      return channels
+    }
 
-      for channel in channels {
-        switch termination {
-          case .finished:
-            channel.finish()
-          case .failure(let error):
-            channel.fail(error)
-        }
+    for channel in channels {
+      switch termination {
+        case .finished:
+          channel.finish()
+        case .failure(let error):
+          channel.fail(error)
       }
     }
   }
@@ -129,10 +132,10 @@ public final class AsyncThrowingPassthroughSubject<Element, Failure: Error>: Asy
     }
 
     public mutating func next() async throws -> Element? {
-      try await withTaskCancellationHandler {
-        try await self.iterator.next()
-      } onCancel: { [unregister] in
+      try await withTaskCancellationHandler { [unregister] in
         unregister()
+      } operation: {
+        try await self.iterator.next()
       }
     }
   }

Sources/AsyncSubjects/AsyncThrowingReplaySubject.swift

@@ -45,33 +45,37 @@ public final class AsyncThrowingReplaySubject<Element, Failure: Error>: AsyncSub
   /// Sends a value to all consumers
   /// - Parameter element: the value to send
   public func send(_ element: Element) {
-    self.state.withCriticalRegion { state in
+    let channels = self.state.withCriticalRegion { state -> [AsyncThrowingBufferedChannel<Element, Error>] in
       if state.buffer.count >= state.bufferSize && !state.buffer.isEmpty {
         state.buffer.removeFirst()
       }
       state.buffer.append(element)
-      for channel in state.channels.values {
-        channel.send(element)
-      }
+      return Array(state.channels.values)
+    }
+
+    for channel in channels {
+      channel.send(element)
     }
   }
 
   /// Finishes the subject with either a normal ending or an error.
   /// - Parameter termination: The termination to finish the subject
   public func send(_ termination: Termination<Failure>) {
-    self.state.withCriticalRegion { state in
+    let channels = self.state.withCriticalRegion { state -> [AsyncThrowingBufferedChannel<Element, Error>] in
       state.terminalState = termination
       let channels = Array(state.channels.values)
       state.channels.removeAll()
       state.buffer.removeAll()
       state.bufferSize = 0
-      for channel in channels {
-        switch termination {
-          case .finished:
-            channel.finish()
-          case .failure(let error):
-            channel.fail(error)
-        }
+      return channels
+    }
+
+    for channel in channels {
+      switch termination {
+        case .finished:
+          channel.finish()
+        case .failure(let error):
+          channel.fail(error)
       }
     }
   }
@@ -130,10 +134,10 @@ public final class AsyncThrowingReplaySubject<Element, Failure: Error>: AsyncSub
     }
 
     public mutating func next() async throws -> Element? {
-      try await withTaskCancellationHandler {
-        try await self.iterator.next()
-      } onCancel: { [unregister] in
+      try await withTaskCancellationHandler { [unregister] in
         unregister()
+      } operation: {
+        try await self.iterator.next()
       }
     }
   }