Skip to content

Commit 4ad0e84

Browse files
hanggahangga
committed
refactor
1 parent 89e7c1f commit 4ad0e84

File tree

1 file changed

+70
-114
lines changed

1 file changed

+70
-114
lines changed

‎core-kotlin-modules/core-kotlin-collections-6/src/test/kotlin/com/baeldung/parallelOperationsCollections/ParallelOperationCollectionsUnitTest.kt

Lines changed: 70 additions & 114 deletions
Original file line numberDiff line numberDiff line change
@@ -8,6 +8,9 @@ import kotlinx.coroutines.flow.*
88
import org.assertj.core.api.Assertions.assertThat
99
import org.junit.jupiter.api.Test
1010
import java.text.SimpleDateFormat
11+
import java.time.Duration
12+
import java.time.Instant
13+
import java.util.*
1114
import java.util.concurrent.Callable
1215
import java.util.concurrent.Executors
1316
import java.util.stream.Collectors
@@ -26,212 +29,165 @@ class ParallelOperationCollectionsUnitTest {
2629
Person("Charlie", 40)
2730
)
2831

32+
private fun List<Person>.assertResultsTrue() {
33+
assertThat(this).containsExactly(
34+
Person("Bob", 16, false), Person("Alice", 30, true), Person("Charlie", 40, true), Person("Ahmad", 42, true)
35+
)
36+
}
37+
2938
private val dateFormat = SimpleDateFormat("yyyy-MM-dd:HH:mm:ss:SSS")
3039

31-
private fun assertResults(filteredPeople: List<Person>) {
32-
assertThat(filteredPeople).containsExactly(
33-
Person("Bob", 16, false), Person("Alice", 30, true), Person("Charlie", 40, true), Person("Ahmad", 42, true)
40+
private fun Person.printFormattedInfo() {
41+
println(
42+
"%-30s %-40s %s".format(
43+
dateFormat.format(Date.from(Instant.now())), Thread.currentThread().name, this
44+
)
3445
)
3546
}
3647

37-
@Test
38-
fun `using coroutines for parallel operations`() = runBlocking {
48+
private fun printHeader() {
3949
println(
4050
"%-30s %-40s %s".format(
4151
"Time", "Thread name", "Operation"
4252
)
4353
)
44-
val startTime = System.currentTimeMillis()
54+
println("-".repeat(115))
55+
}
56+
57+
private fun Instant.printFooter() {
58+
val endTime = Instant.now()
59+
val duration = Duration.between(this, endTime)
60+
println("Total time taken: ${duration.toMillis()} ms")
61+
println()
62+
}
63+
64+
@Test
65+
fun `using coroutines for parallel operations`() = runBlocking {
66+
printHeader()
67+
val startTime = Instant.now()
68+
4569
val filteredPeople = people.map { person ->
4670
async {
4771
launch {
4872
person.isAdult = person.age >= 18
49-
println(
50-
"%-30s %-40s %s".format(
51-
dateFormat.format(System.currentTimeMillis()), Thread.currentThread().name, person
52-
)
53-
)
73+
person.printFormattedInfo()
5474
}
5575
person
5676
}
5777
}.awaitAll().filter { it.age > 15 }.sortedBy { it.age }
5878

59-
val endTime = System.currentTimeMillis()
60-
val duration = endTime - startTime
61-
println("Total time taken: $duration ms")
62-
println()
79+
startTime.printFooter()
6380

64-
assertResults(filteredPeople)
81+
filteredPeople.assertResultsTrue()
6582
}
6683

6784
@OptIn(ExperimentalCoroutinesApi::class)
6885
@Test
6986
fun `using coroutines for parallel operations with Flow`() = runBlocking {
70-
println(
71-
"%-30s %-40s %s".format(
72-
"Time", "Thread name", "Operation"
73-
)
74-
)
75-
val startTime = System.currentTimeMillis()
87+
printHeader()
88+
val startTime = Instant.now()
89+
7690
val filteredPeople = people.asFlow().flatMapMerge { person ->
7791
flow {
7892
emit(async {
7993
person.isAdult = person.age >= 18
80-
81-
println(
82-
"%-30s %-40s %s".format(
83-
dateFormat.format(System.currentTimeMillis()), Thread.currentThread().name, person
84-
)
85-
)
94+
person.printFormattedInfo()
8695
person
8796
}.await())
8897
}
8998
}.filter { it.age > 15 }.toList().sortedBy { it.age }
9099

91-
val endTime = System.currentTimeMillis()
92-
val duration = endTime - startTime
93-
println("Total time taken: $duration ms")
94-
println()
100+
startTime.printFooter()
95101

96-
assertResults(filteredPeople)
102+
filteredPeople.assertResultsTrue()
97103
}
98104

99105
@Test
100106
fun `using RxJava for parallel operations`() { // Observable.class from io.reactivex;
101-
println(
102-
"%-30s %-40s %s".format(
103-
"Time", "Thread name", "Operation"
104-
)
105-
)
106-
val startTime = System.currentTimeMillis()
107+
printHeader()
108+
val startTime = Instant.now()
109+
107110
val observable = Observable.fromIterable(people).flatMap({
108111
Observable.just(it).subscribeOn(Schedulers.computation()).doOnNext { person ->
109112
person.isAdult = person.age >= 18
110-
println(
111-
"%-30s %-40s %s".format(
112-
dateFormat.format(System.currentTimeMillis()), Thread.currentThread().name, person
113-
)
114-
)
113+
person.printFormattedInfo()
115114
}
116115
}, people.size) // Uses maxConcurrency for the number of elements
117116
.filter { it.age > 15 }.toList().map { it.sortedBy { person -> person.age } }.blockingGet()
118117

119-
val endTime = System.currentTimeMillis()
120-
val duration = endTime - startTime
121-
println("Total time taken: $duration ms")
122-
println()
118+
startTime.printFooter()
123119

124-
assertResults(observable)
120+
observable.assertResultsTrue()
125121
}
126122

127123
@Test
128124
fun `using RxKotlin for parallel operations`() { // ObservableKt.kt.class from io.reactivex.rxkotlin
129-
println(
130-
"%-30s %-40s %s".format(
131-
"Time", "Thread name", "Operation"
132-
)
133-
)
134-
val startTime = System.currentTimeMillis()
125+
printHeader()
126+
val startTime = Instant.now()
127+
135128
val observable = people.toObservable().flatMap({
136129
Observable.just(it).subscribeOn(Schedulers.computation()).doOnNext { person ->
137130
person.isAdult = person.age >= 18
138-
println(
139-
"%-30s %-40s %s".format(
140-
dateFormat.format(System.currentTimeMillis()), Thread.currentThread().name, person
141-
)
142-
)
131+
person.printFormattedInfo()
143132
}
144133
}, people.size) // Uses maxConcurrency for the number of elements
145134
.filter { it.age > 15 }.toList().map { it.sortedBy { person -> person.age } }.blockingGet()
146135

147-
val endTime = System.currentTimeMillis()
148-
val duration = endTime - startTime
149-
println("Total time taken: $duration ms")
150-
println()
136+
startTime.printFooter()
151137

152-
assertResults(observable)
138+
observable.assertResultsTrue()
153139
}
154140

155141
@Test
156142
fun `using RxKotlin but still use 1 thread`() { // ObservableKt.kt.class from io.reactivex.rxkotlin
157-
println(
158-
"%-30s %-40s %s".format(
159-
"Time", "Thread name", "Operation"
160-
)
161-
)
162-
val startTime = System.currentTimeMillis()
143+
printHeader()
144+
val startTime = Instant.now()
145+
163146
val observable =
164147
people.toObservable().subscribeOn(Schedulers.io()).flatMap { Observable.just(it) }.doOnNext { person ->
165148
person.isAdult = person.age >= 18
166-
println(
167-
"%-30s %-40s %s".format(
168-
dateFormat.format(System.currentTimeMillis()), Thread.currentThread().name, person
169-
)
170-
)
149+
person.printFormattedInfo()
171150
}.filter { it.age > 15 }.toList().map { it.sortedBy { person -> person.age } }.blockingGet()
172151

173-
val endTime = System.currentTimeMillis()
174-
val duration = endTime - startTime
175-
println("Total time taken: $duration ms")
176-
println()
152+
startTime.printFooter()
177153

178-
assertResults(observable)
154+
observable.assertResultsTrue()
179155
}
180156

181157
@Test
182158
fun `using parallelStream()`() {
183-
println(
184-
"%-30s %-40s %s".format(
185-
"Time", "Thread name", "Operation"
186-
)
187-
)
188-
val startTime = System.currentTimeMillis()
159+
printHeader()
160+
val startTime = Instant.now()
161+
189162
val filteredPeople = people.parallelStream().map { person ->
190163
person.isAdult = person.age >= 18
191-
println(
192-
"%-30s %-40s %s".format(
193-
dateFormat.format(System.currentTimeMillis()), Thread.currentThread().name, person
194-
)
195-
)
164+
person.printFormattedInfo()
196165
person
197166
}.filter { it.age > 15 }.sorted { p1, p2 -> p1.age.compareTo(p2.age) }.collect(Collectors.toList())
198167

199-
val endTime = System.currentTimeMillis()
200-
val duration = endTime - startTime
201-
println("Total time taken: $duration ms")
202-
println()
168+
startTime.printFooter()
203169

204-
assertResults(filteredPeople)
170+
filteredPeople.assertResultsTrue()
205171
}
206172

207173
@Test
208174
fun `using ScheduledExecutorService for parallel operations`() {
209-
println(
210-
"%-30s %-40s %s".format(
211-
"Time", "Thread name", "Operation"
212-
)
213-
)
214-
val startTime = System.currentTimeMillis()
175+
printHeader()
176+
val startTime = Instant.now()
177+
215178
val executor = Executors.newFixedThreadPool(people.size)
216179
val futures = people.map { person ->
217180
executor.submit(Callable {
218181
person.isAdult = person.age >= 18
219-
println(
220-
"%-30s %-40s %s".format(
221-
dateFormat.format(System.currentTimeMillis()), Thread.currentThread().name, person
222-
)
223-
)
182+
person.printFormattedInfo()
224183
person
225184
})
226185
}.map { it.get() }.filter { it.age > 15 }.sortedBy { it.age }
227186

228-
val endTime = System.currentTimeMillis()
229-
val duration = endTime - startTime
230-
println("Total time taken: $duration ms")
231-
println()
187+
executor.shutdown()
232188

233-
assertResults(futures)
189+
startTime.printFooter()
234190

235-
executor.shutdown()
191+
futures.assertResultsTrue()
236192
}
237193
}

0 commit comments

Comments
 (0)