Skip to content

Commit 6155400

Browse files
authored
Lock async stream inits (#11)
* Lock async stream inits Fixes #10. * wip
1 parent eadb1cb commit 6155400

File tree

3 files changed

+25
-4
lines changed

3 files changed

+25
-4
lines changed

Sources/ConcurrencyExtras/AsyncStream.swift

+7-2
Original file line numberDiff line numberDiff line change
@@ -1,3 +1,5 @@
1+
import Foundation
2+
13
extension AsyncStream {
24
/// Produces an `AsyncStream` from an `AsyncSequence` by consuming the sequence till it
35
/// terminates, ignoring any failure.
@@ -51,10 +53,13 @@ extension AsyncStream {
5153
///
5254
/// - Parameter sequence: An async sequence.
5355
public init<S: AsyncSequence>(_ sequence: S) where S.Element == Element {
56+
let lock = NSLock()
5457
var iterator: S.AsyncIterator?
5558
self.init {
56-
if iterator == nil {
57-
iterator = sequence.makeAsyncIterator()
59+
lock.withLock {
60+
if iterator == nil {
61+
iterator = sequence.makeAsyncIterator()
62+
}
5863
}
5964
return try? await iterator?.next()
6065
}

Sources/ConcurrencyExtras/AsyncThrowingStream.swift

+7-2
Original file line numberDiff line numberDiff line change
@@ -1,13 +1,18 @@
1+
import Foundation
2+
13
extension AsyncThrowingStream where Failure == Error {
24
/// Produces an `AsyncThrowingStream` from an `AsyncSequence` by consuming the sequence till it
35
/// terminates, rethrowing any failure.
46
///
57
/// - Parameter sequence: An async sequence.
68
public init<S: AsyncSequence>(_ sequence: S) where S.Element == Element {
9+
let lock = NSLock()
710
var iterator: S.AsyncIterator?
811
self.init {
9-
if iterator == nil {
10-
iterator = sequence.makeAsyncIterator()
12+
lock.withLock {
13+
if iterator == nil {
14+
iterator = sequence.makeAsyncIterator()
15+
}
1116
}
1217
return try await iterator?.next()
1318
}
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,11 @@
1+
import Foundation
2+
3+
#if !(os(iOS) || os(macOS) || os(tvOS) || os(watchOS))
4+
extension NSLock {
5+
func withLock<R>(_ body: () throws -> R) rethrows -> R {
6+
self.lock()
7+
defer { self.unlock() }
8+
return try body()
9+
}
10+
}
11+
#endif

0 commit comments

Comments
 (0)