-
Notifications
You must be signed in to change notification settings - Fork 21
/
TestFtBasicUsage.scala
140 lines (111 loc) · 2.99 KB
/
TestFtBasicUsage.scala
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
package cps.stream
import org.junit.{Test,Ignore}
import org.junit.Assert._
import scala.concurrent.*
import scala.concurrent.ExecutionContext.Implicits.global
import scala.util.*
import cps.*
import cps.stream.*
import cps.monads.{*, given}
import cps.testconfig.given
import cps.util.FutureCompleter
class TestFtBasicUsage:
val N = 10000
@Test def simpleLoopAsyncListFt() =
val stream = asyncStream[AsyncList[Future, Int]]{ out =>
for(i <- 1 to N)
out.emit(i)
}
val read = async[Future]{
var done = false
var current = stream
var i = 1
while(!done) {
await(current.next) match
case Some((h,next)) =>
current = next
assert(h == i)
i = i + 1
case None =>
done = true
}
}
FutureCompleter(read)
@Test def simpleLoopExceptionAsyncListFt() =
val stream = asyncStream[AsyncList[Future,Int]] { out =>
for(i <- 1 to 10) {
if (i == 5) then
throw new RuntimeException("bye")
out.emit(i)
}
}
val listSum = stream.fold(0)(_ + _)
val res = listSum.failed.map(ex =>
assert(ex.getMessage()=="bye")
)
FutureCompleter(res)
@Test def fewSmallLoopsInAsyncList() =
val M = 1000
val N = 100
val folds: Seq[Future[Int]] = for(i <- 1 to M) yield {
val list = asyncStream[AsyncList[Future,Int]] { out =>
for(i <- 1 to N) {
out.emit(i)
//println("emitted: "+i)
}
}
list.fold(0)(_ + _)
}
val expected = (1 to N).sum
val retval = folds.foldLeft(Future.successful(())){ (s,e) =>
s.flatMap{ r =>
e.map{ x =>
assert(x == expected)
} }
}
FutureCompleter(retval)
@Test def testFib() =
val stream = asyncStream[AsyncList[Future,Int]] { out =>
var n = 1
var m = 1
for(i <- 1 to 10) {
out.emit(n)
val tmp = n
n = m
m = m + tmp
}
}
import scala.collection.mutable.ArrayBuffer
val readAll = stream.fold(ArrayBuffer[Int]())((s,e)=>s.addOne(e))
val res = readAll.map{ l =>
assert(l(0)==1)
assert(l(1)==1)
assert(l(2)==2)
assert(l(3)==3)
assert(l(4)==5)
assert(l(5)==8)
assert(l(6)==13)
assert(l(7)==21)
}
FutureCompleter(res)
@Test def testIterator() =
val stream = asyncStream[AsyncList[Future,Int]] { out =>
var s = 0
for(i <- 1 to 10) {
s+=i
out.emit(s)
}
}
val check = async[Future] {
val it = stream.iterator
var last = 0
while{
await(it.next) match
case Some(e) => last = e
true
case None => false
} do ()
val sampleSum = (1 to 10).sum
assert(last == sampleSum)
}
FutureCompleter(check)