Skip to content

Commit 1286bef

Browse files
twittembtwittemb
authored
Merge pull request #32 from sideeffect-io/fix/multicast-upstream-cancellation
multicast: dont cancel upstream sequence when client is cancelled
2 parents f5187fa + 74ed024 commit 1286bef

10 files changed

+118
-188
lines changed

‎CHANGELOG.md‎

Lines changed: 6 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -1,3 +1,9 @@
1+
**v0.5.2 - Oxygen:**
2+
3+
This version is a bug fix version.
4+
5+
- Multicast: don't cancel the upstream sequence when a client is cancelled
6+
17
**v0.5.1 - Nitrogen:**
28

39
This version removes compilation unsafe flags

‎Sources/AsyncSubjects/AsyncCurrentValueSubject.swift‎

Lines changed: 11 additions & 15 deletions
Original file line numberDiff line numberDiff line change
@@ -67,28 +67,24 @@ public final class AsyncCurrentValueSubject<Element>: AsyncSubject where Element
6767
/// Sends a value to all consumers
6868
/// - Parameter element: the value to send
6969
public func send(_ element: Element) {
70-
let channels = self.state.withCriticalRegion { state -> [AsyncBufferedChannel<Element>] in
70+
self.state.withCriticalRegion { state in
7171
state.current = element
72-
return Array(state.channels.values)
73-
}
74-
75-
for channel in channels {
76-
channel.send(element)
72+
for channel in state.channels.values {
73+
channel.send(element)
74+
}
7775
}
7876
}
7977

8078
/// Finishes the async sequences with a normal ending.
8179
/// - Parameter termination: The termination to finish the subject.
8280
public func send(_ termination: Termination<Failure>) {
83-
let channels = self.state.withCriticalRegion { state -> [AsyncBufferedChannel<Element>] in
81+
self.state.withCriticalRegion { state in
8482
state.terminalState = termination
8583
let channels = Array(state.channels.values)
8684
state.channels.removeAll()
87-
return channels
88-
}
89-
90-
for channel in channels {
91-
channel.finish()
85+
for channel in channels {
86+
channel.finish()
87+
}
9288
}
9389
}
9490

@@ -138,10 +134,10 @@ public final class AsyncCurrentValueSubject<Element>: AsyncSubject where Element
138134
}
139135

140136
public mutating func next() async -> Element? {
141-
await withTaskCancellationHandler { [unregister] in
142-
unregister()
143-
} operation: {
137+
await withTaskCancellationHandler {
144138
await self.iterator.next()
139+
} onCancel: { [unregister] in
140+
unregister()
145141
}
146142
}
147143
}

‎Sources/AsyncSubjects/AsyncPassthroughSubject.swift‎

Lines changed: 11 additions & 15 deletions
Original file line numberDiff line numberDiff line change
@@ -52,27 +52,23 @@ public final class AsyncPassthroughSubject<Element>: AsyncSubject {
5252
/// Sends a value to all consumers
5353
/// - Parameter element: the value to send
5454
public func send(_ element: Element) {
55-
let channels = self.state.withCriticalRegion { state in
56-
state.channels.values
57-
}
58-
59-
for channel in channels {
60-
channel.send(element)
55+
self.state.withCriticalRegion { state in
56+
for channel in state.channels.values {
57+
channel.send(element)
58+
}
6159
}
6260
}
6361

6462
/// Finishes the subject with a normal ending.
6563
/// - Parameter termination: The termination to finish the subject
6664
public func send(_ termination: Termination<Failure>) {
67-
let channels = self.state.withCriticalRegion { state -> [AsyncBufferedChannel<Element>] in
65+
self.state.withCriticalRegion { state in
6866
state.terminalState = termination
6967
let channels = Array(state.channels.values)
7068
state.channels.removeAll()
71-
return channels
72-
}
73-
74-
for channel in channels {
75-
channel.finish()
69+
for channel in channels {
70+
channel.finish()
71+
}
7672
}
7773
}
7874

@@ -120,10 +116,10 @@ public final class AsyncPassthroughSubject<Element>: AsyncSubject {
120116
}
121117

122118
public mutating func next() async -> Element? {
123-
await withTaskCancellationHandler { [unregister] in
124-
unregister()
125-
} operation: {
119+
await withTaskCancellationHandler {
126120
await self.iterator.next()
121+
} onCancel: { [unregister] in
122+
unregister()
127123
}
128124
}
129125
}

‎Sources/AsyncSubjects/AsyncReplaySubject.swift‎

Lines changed: 11 additions & 15 deletions
Original file line numberDiff line numberDiff line change
@@ -46,33 +46,29 @@ public final class AsyncReplaySubject<Element>: AsyncSubject where Element: Send
4646
/// Sends a value to all consumers
4747
/// - Parameter element: the value to send
4848
public func send(_ element: Element) {
49-
let channels = self.state.withCriticalRegion { state -> [AsyncBufferedChannel<Element>] in
49+
self.state.withCriticalRegion { state in
5050
if state.buffer.count >= state.bufferSize && !state.buffer.isEmpty {
5151
state.buffer.removeFirst()
5252
}
5353
state.buffer.append(element)
54-
return Array(state.channels.values)
55-
}
56-
57-
for channel in channels {
58-
channel.send(element)
54+
for channel in state.channels.values {
55+
channel.send(element)
56+
}
5957
}
6058
}
6159

6260
/// Finishes the subject with a normal ending.
6361
/// - Parameter termination: The termination to finish the subject.
6462
public func send(_ termination: Termination<Failure>) {
65-
let channels = self.state.withCriticalRegion { state -> [AsyncBufferedChannel<Element>] in
63+
self.state.withCriticalRegion { state in
6664
state.terminalState = termination
6765
let channels = Array(state.channels.values)
6866
state.channels.removeAll()
6967
state.buffer.removeAll()
7068
state.bufferSize = 0
71-
return channels
72-
}
73-
74-
for channel in channels {
75-
channel.finish()
69+
for channel in channels {
70+
channel.finish()
71+
}
7672
}
7773
}
7874

@@ -124,10 +120,10 @@ public final class AsyncReplaySubject<Element>: AsyncSubject where Element: Send
124120
}
125121

126122
public mutating func next() async -> Element? {
127-
await withTaskCancellationHandler { [unregister] in
128-
unregister()
129-
} operation: {
123+
await withTaskCancellationHandler {
130124
await self.iterator.next()
125+
} onCancel: { [unregister] in
126+
unregister()
131127
}
132128
}
133129
}

‎Sources/AsyncSubjects/AsyncThrowingCurrentValueSubject.swift‎

Lines changed: 15 additions & 19 deletions
Original file line numberDiff line numberDiff line change
@@ -67,32 +67,28 @@ public final class AsyncThrowingCurrentValueSubject<Element, Failure: Error>: As
6767
/// Sends a value to all consumers
6868
/// - Parameter element: the value to send
6969
public func send(_ element: Element) {
70-
let channels = self.state.withCriticalRegion { state -> [AsyncThrowingBufferedChannel<Element, Error>] in
70+
self.state.withCriticalRegion { state in
7171
state.current = element
72-
return Array(state.channels.values)
73-
}
74-
75-
for channel in channels {
76-
channel.send(element)
72+
for channel in state.channels.values {
73+
channel.send(element)
74+
}
7775
}
7876
}
7977

8078
/// Finishes the subject with either a normal ending or an error.
8179
/// - Parameter termination: The termination to finish the subject.
8280
public func send(_ termination: Termination<Failure>) {
83-
let channels = self.state.withCriticalRegion { state -> [AsyncThrowingBufferedChannel<Element, Error>] in
81+
self.state.withCriticalRegion { state in
8482
state.terminalState = termination
8583
let channels = Array(state.channels.values)
8684
state.channels.removeAll()
87-
return channels
88-
}
89-
90-
for channel in channels {
91-
switch termination {
92-
case .finished:
93-
channel.finish()
94-
case .failure(let error):
95-
channel.fail(error)
85+
for channel in channels {
86+
switch termination {
87+
case .finished:
88+
channel.finish()
89+
case .failure(let error):
90+
channel.fail(error)
91+
}
9692
}
9793
}
9894
}
@@ -149,10 +145,10 @@ public final class AsyncThrowingCurrentValueSubject<Element, Failure: Error>: As
149145
}
150146

151147
public mutating func next() async throws -> Element? {
152-
try await withTaskCancellationHandler { [unregister] in
153-
unregister()
154-
} operation: {
148+
try await withTaskCancellationHandler {
155149
try await self.iterator.next()
150+
} onCancel: { [unregister] in
151+
unregister()
156152
}
157153
}
158154
}

‎Sources/AsyncSubjects/AsyncThrowingPassthroughSubject.swift‎

Lines changed: 15 additions & 18 deletions
Original file line numberDiff line numberDiff line change
@@ -53,31 +53,28 @@ public final class AsyncThrowingPassthroughSubject<Element, Failure: Error>: Asy
5353
/// Sends a value to all consumers
5454
/// - Parameter element: the value to send
5555
public func send(_ element: Element) {
56-
let channels = self.state.withCriticalRegion { state in
57-
state.channels.values
58-
}
59-
60-
for channel in channels {
61-
channel.send(element)
56+
self.state.withCriticalRegion { state in
57+
for channel in state.channels.values {
58+
channel.send(element)
59+
}
6260
}
6361
}
6462

6563
/// Finishes the subject with either a normal ending or an error.
6664
/// - Parameter termination: The termination to finish the subject
6765
public func send(_ termination: Termination<Failure>) {
68-
let channels = self.state.withCriticalRegion { state -> [AsyncThrowingBufferedChannel<Element, Error>] in
66+
self.state.withCriticalRegion { state in
6967
state.terminalState = termination
7068
let channels = Array(state.channels.values)
7169
state.channels.removeAll()
72-
return channels
73-
}
7470

75-
for channel in channels {
76-
switch termination {
77-
case .finished:
78-
channel.finish()
79-
case .failure(let error):
80-
channel.fail(error)
71+
for channel in channels {
72+
switch termination {
73+
case .finished:
74+
channel.finish()
75+
case .failure(let error):
76+
channel.fail(error)
77+
}
8178
}
8279
}
8380
}
@@ -132,10 +129,10 @@ public final class AsyncThrowingPassthroughSubject<Element, Failure: Error>: Asy
132129
}
133130

134131
public mutating func next() async throws -> Element? {
135-
try await withTaskCancellationHandler { [unregister] in
136-
unregister()
137-
} operation: {
132+
try await withTaskCancellationHandler {
138133
try await self.iterator.next()
134+
} onCancel: { [unregister] in
135+
unregister()
139136
}
140137
}
141138
}

‎Sources/AsyncSubjects/AsyncThrowingReplaySubject.swift‎

Lines changed: 15 additions & 19 deletions
Original file line numberDiff line numberDiff line change
@@ -45,37 +45,33 @@ public final class AsyncThrowingReplaySubject<Element, Failure: Error>: AsyncSub
4545
/// Sends a value to all consumers
4646
/// - Parameter element: the value to send
4747
public func send(_ element: Element) {
48-
let channels = self.state.withCriticalRegion { state -> [AsyncThrowingBufferedChannel<Element, Error>] in
48+
self.state.withCriticalRegion { state in
4949
if state.buffer.count >= state.bufferSize && !state.buffer.isEmpty {
5050
state.buffer.removeFirst()
5151
}
5252
state.buffer.append(element)
53-
return Array(state.channels.values)
54-
}
55-
56-
for channel in channels {
57-
channel.send(element)
53+
for channel in state.channels.values {
54+
channel.send(element)
55+
}
5856
}
5957
}
6058

6159
/// Finishes the subject with either a normal ending or an error.
6260
/// - Parameter termination: The termination to finish the subject
6361
public func send(_ termination: Termination<Failure>) {
64-
let channels = self.state.withCriticalRegion { state -> [AsyncThrowingBufferedChannel<Element, Error>] in
62+
self.state.withCriticalRegion { state in
6563
state.terminalState = termination
6664
let channels = Array(state.channels.values)
6765
state.channels.removeAll()
6866
state.buffer.removeAll()
6967
state.bufferSize = 0
70-
return channels
71-
}
72-
73-
for channel in channels {
74-
switch termination {
75-
case .finished:
76-
channel.finish()
77-
case .failure(let error):
78-
channel.fail(error)
68+
for channel in channels {
69+
switch termination {
70+
case .finished:
71+
channel.finish()
72+
case .failure(let error):
73+
channel.fail(error)
74+
}
7975
}
8076
}
8177
}
@@ -134,10 +130,10 @@ public final class AsyncThrowingReplaySubject<Element, Failure: Error>: AsyncSub
134130
}
135131

136132
public mutating func next() async throws -> Element? {
137-
try await withTaskCancellationHandler { [unregister] in
138-
unregister()
139-
} operation: {
133+
try await withTaskCancellationHandler {
140134
try await self.iterator.next()
135+
} onCancel: { [unregister] in
136+
unregister()
141137
}
142138
}
143139
}

0 commit comments

Comments
 (0)