@@ -5,7 +5,7 @@ final class CurrentValueRelay<Output>: Publisher {
55 typealias Failure = Never
66
77 private var currentValue : Output
8- private let lock : os_unfair_lock_t
8+ private let lock : NSRecursiveLock
99 private var subscriptions = ContiguousArray < Subscription > ( )
1010
1111 var value : Output {
@@ -15,13 +15,7 @@ final class CurrentValueRelay<Output>: Publisher {
1515
1616 init ( _ value: Output ) {
1717 self . currentValue = value
18- self . lock = os_unfair_lock_t. allocate ( capacity: 1 )
19- self . lock. initialize ( to: os_unfair_lock ( ) )
20- }
21-
22- deinit {
23- self . lock. deinitialize ( count: 1 )
24- self . lock. deallocate ( )
18+ self . lock = NSRecursiveLock ( )
2519 }
2620
2721 func receive( subscriber: some Subscriber < Output , Never > ) {
@@ -36,7 +30,7 @@ final class CurrentValueRelay<Output>: Publisher {
3630 self . lock. sync {
3731 self . currentValue = value
3832 }
39- for subscription in self . lock. sync ( { self . subscriptions } ) {
33+ for subscription in self . lock. sync ( work : { self . subscriptions } ) {
4034 subscription. receive ( value)
4135 }
4236 }
@@ -52,27 +46,28 @@ final class CurrentValueRelay<Output>: Publisher {
5246
5347extension CurrentValueRelay {
5448 fileprivate final class Subscription : Combine . Subscription , Equatable {
55- private var demand = Subscribers . Demand. none
56- private var downstream : ( any Subscriber < Output , Never > ) ?
57- private let lock : os_unfair_lock_t
49+ private var _demand = Subscribers . Demand. none
50+
51+ private var _downstream : ( any Subscriber < Output , Never > ) ?
52+ var downstream : ( any Subscriber < Output , Never > ) ? {
53+ var downstream : ( any Subscriber < Output , Never > ) ?
54+ self . lock. sync { downstream = _downstream }
55+ return downstream
56+ }
57+
58+ private let lock : NSRecursiveLock
5859 private var receivedLastValue = false
5960 private var upstream : CurrentValueRelay ?
6061
6162 init ( upstream: CurrentValueRelay , downstream: any Subscriber < Output , Never > ) {
6263 self . upstream = upstream
63- self . downstream = downstream
64- self . lock = os_unfair_lock_t. allocate ( capacity: 1 )
65- self . lock. initialize ( to: os_unfair_lock ( ) )
66- }
67-
68- deinit {
69- self . lock. deinitialize ( count: 1 )
70- self . lock. deallocate ( )
64+ self . _downstream = downstream
65+ self . lock = upstream. lock
7166 }
7267
7368 func cancel( ) {
7469 self . lock. sync {
75- self . downstream = nil
70+ self . _downstream = nil
7671 self . upstream? . remove ( self )
7772 self . upstream = nil
7873 }
@@ -81,24 +76,24 @@ extension CurrentValueRelay {
8176 func receive( _ value: Output ) {
8277 guard let downstream else { return }
8378
84- switch self . demand {
79+ self . lock. lock ( )
80+ switch self . _demand {
8581 case . unlimited:
82+ self . lock. unlock ( )
8683 // NB: Adding to unlimited demand has no effect and can be ignored.
8784 _ = downstream. receive ( value)
8885
8986 case . none:
90- self . lock. sync {
91- self . receivedLastValue = false
92- }
87+ self . receivedLastValue = false
88+ self . lock. unlock ( )
9389
9490 default :
95- self . lock. sync {
96- self . receivedLastValue = true
97- self . demand -= 1
98- }
91+ self . receivedLastValue = true
92+ self . _demand -= 1
93+ self . lock. unlock ( )
9994 let moreDemand = downstream. receive ( value)
10095 self . lock. sync {
101- self . demand += moreDemand
96+ self . _demand += moreDemand
10297 }
10398 }
10499 }
@@ -109,7 +104,7 @@ extension CurrentValueRelay {
109104 guard let downstream else { return }
110105
111106 self . lock. lock ( )
112- self . demand += demand
107+ self . _demand += demand
113108
114109 guard
115110 !self . receivedLastValue,
@@ -121,18 +116,18 @@ extension CurrentValueRelay {
121116
122117 self . receivedLastValue = true
123118
124- switch self . demand {
119+ switch self . _demand {
125120 case . unlimited:
126121 self . lock. unlock ( )
127122 // NB: Adding to unlimited demand has no effect and can be ignored.
128123 _ = downstream. receive ( value)
129124
130125 default :
131- self . demand -= 1
126+ self . _demand -= 1
132127 self . lock. unlock ( )
133128 let moreDemand = downstream. receive ( value)
134129 self . lock. lock ( )
135- self . demand += moreDemand
130+ self . _demand += moreDemand
136131 self . lock. unlock ( )
137132 }
138133 }
0 commit comments