1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
|
//===----------------------------------------------------------------------===//
//
// This source file is part of the Swift Async Algorithms open source project
//
// Copyright (c) 2022 Apple Inc. and the Swift project authors
// Licensed under Apache License v2.0 with Runtime Library Exception
//
// See https://swift.org/LICENSE.txt for license information
//
//===----------------------------------------------------------------------===//
import XCTest
import AsyncAlgorithms
final class TestChannel: XCTestCase {
func test_asyncChannel_delivers_elements_when_several_producers_and_several_consumers() async {
let sents = (1...10)
let expected = Set(sents)
// Given: an AsyncChannel
let sut = AsyncChannel<Int>()
// When: sending elements from tasks in a group
Task {
await withTaskGroup(of: Void.self) { group in
for sent in sents {
group.addTask {
await sut.send(sent)
}
}
}
}
// When: receiving those elements from tasks in a group
let collected = await withTaskGroup(of: Int.self, returning: Set<Int>.self) { group in
for _ in sents {
group.addTask {
var iterator = sut.makeAsyncIterator()
let received = await iterator.next()
return received!
}
}
var collected = Set<Int>()
for await element in group {
collected.update(with: element)
}
return collected
}
// Then: all elements are sent and received
XCTAssertEqual(collected, expected)
}
func test_asyncChannel_resumes_producers_and_discards_additional_elements_when_finish_is_called() async {
// Given: an AsyncChannel
let sut = AsyncChannel<Int>()
// Given: 2 suspended send operations
let task1 = Task {
await sut.send(1)
}
let task2 = Task {
await sut.send(2)
}
// When: finishing the channel
sut.finish()
// Then: the send operations are resumed
_ = await (task1.value, task2.value)
// When: sending an extra value
await sut.send(3)
// Then: the operation and the iteration are immediately resumed
var collected = [Int]()
for await element in sut {
collected.append(element)
}
XCTAssertTrue(collected.isEmpty)
}
func test_asyncChannel_resumes_consumers_when_finish_is_called() async {
// Given: an AsyncChannel
let sut = AsyncChannel<Int>()
// Given: 2 suspended iterations
let task1 = Task<Int?, Never> {
var iterator = sut.makeAsyncIterator()
return await iterator.next()
}
let task2 = Task<Int?, Never> {
var iterator = sut.makeAsyncIterator()
return await iterator.next()
}
// When: finishing the channel
sut.finish()
// Then: the iterations are resumed with nil values
let (collected1, collected2) = await (task1.value, task2.value)
XCTAssertNil(collected1)
XCTAssertNil(collected2)
// When: requesting a next value
var iterator = sut.makeAsyncIterator()
let pastEnd = await iterator.next()
// Then: the past end is nil
XCTAssertNil(pastEnd)
}
func test_asyncChannel_resumes_producer_when_task_is_cancelled() async {
let send1IsResumed = expectation(description: "The first send operation is resumed")
// Given: an AsyncChannel
let sut = AsyncChannel<Int>()
// Given: 2 suspended send operations
let task1 = Task {
await sut.send(1)
send1IsResumed.fulfill()
}
let task2 = Task {
await sut.send(2)
}
// When: cancelling the first task
task1.cancel()
// Then: the first sending operation is resumed
await fulfillment(of: [send1IsResumed], timeout: 1.0)
// When: collecting elements
var iterator = sut.makeAsyncIterator()
let collected = await iterator.next()
// Then: the second operation resumes and the iteration receives the element
_ = await task2.value
XCTAssertEqual(collected, 2)
}
func test_asyncChannel_resumes_consumer_when_task_is_cancelled() async {
// Given: an AsyncChannel
let sut = AsyncChannel<Int>()
// Given: 2 suspended iterations
let task1 = Task<Int?, Never> {
var iterator = sut.makeAsyncIterator()
return await iterator.next()
}
let task2 = Task<Int?, Never> {
var iterator = sut.makeAsyncIterator()
return await iterator.next()
}
// When: cancelling the first task
task1.cancel()
// Then: the iteration is resumed with a nil element
let collected1 = await task1.value
XCTAssertNil(collected1)
// When: sending an element
await sut.send(1)
// Then: the second iteration is resumed with the element
let collected2 = await task2.value
XCTAssertEqual(collected2, 1)
}
}
|
