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
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
|
class TestPool < Test::Unit::TestCase
def pool
@pool ||= EM::Pool.new
end
def go
EM.run { yield }
end
def stop
EM.stop
end
def deferrable
@deferrable ||= EM::DefaultDeferrable.new
end
def test_supports_more_work_than_resources
ran = false
go do
pool.perform do
ran = true
deferrable
end
stop
end
assert_equal false, ran
go do
pool.add :resource
stop
end
assert_equal true, ran
end
def test_reques_resources_on_error
pooled_res, pooled_res2 = nil
pool.add :res
go do
pool.perform do |res|
pooled_res = res
deferrable
end
stop
end
deferrable.fail
go do
pool.perform do |res|
pooled_res2 = res
deferrable
end
stop
end
assert_equal :res, pooled_res
assert_equal pooled_res, pooled_res2
end
def test_supports_custom_error_handler
eres = nil
pool.on_error do |res|
eres = res
end
performs = []
pool.add :res
go do
pool.perform do |res|
performs << res
deferrable
end
pool.perform do |res|
performs << res
deferrable
end
deferrable.fail
stop
end
assert_equal :res, eres
# manual requeues required when error handler is installed:
assert_equal 1, performs.size
assert_equal :res, performs.first
end
def test_catches_successful_deferrables
performs = []
pool.add :res
go do
pool.perform { |res| performs << res; deferrable }
pool.perform { |res| performs << res; deferrable }
stop
end
assert_equal [:res], performs
deferrable.succeed
go { stop }
assert_equal [:res, :res], performs
end
def test_prunes_locked_and_removed_resources
performs = []
pool.add :res
deferrable.succeed
go do
pool.perform { |res| performs << res; pool.remove res; deferrable }
pool.perform { |res| performs << res; pool.remove res; deferrable }
stop
end
assert_equal [:res], performs
end
# Contents is only to be used for inspection of the pool!
def test_contents
pool.add :res
assert_equal [:res], pool.contents
# Assert that modifying the contents list does not affect the pools
# contents.
pool.contents.delete(:res)
assert_equal [:res], pool.contents
end
def test_contents_when_perform_errors_and_on_error_is_not_set
pool.add :res
assert_equal [:res], pool.contents
pool.perform do |r|
d = EM::DefaultDeferrable.new
d.fail
d
end
EM.run { EM.next_tick { EM.stop } }
assert_equal [:res], pool.contents
end
def test_contents_when_perform_errors_and_on_error_is_set
pool.add :res
res = nil
pool.on_error do |r|
res = r
end
assert_equal [:res], pool.contents
pool.perform do |r|
d = EM::DefaultDeferrable.new
d.fail 'foo'
d
end
EM.run { EM.next_tick { EM.stop } }
assert_equal :res, res
assert_equal [], pool.contents
end
def test_num_waiting
pool.add :res
assert_equal 0, pool.num_waiting
pool.perform { |r| EM::DefaultDeferrable.new }
assert_equal 0, pool.num_waiting
10.times { pool.perform { |r| EM::DefaultDeferrable.new } }
EM.run { EM.next_tick { EM.stop } }
assert_equal 10, pool.num_waiting
end
def test_exceptions_in_the_work_block_bubble_up_raise_and_fail_the_resource
pool.add :res
res = nil
pool.on_error { |r| res = r }
pool.perform { raise 'boom' }
assert_raises(RuntimeError) do
EM.run { EM.next_tick { EM.stop } }
end
assert_equal [], pool.contents
assert_equal :res, res
end
def test_removed_list_does_not_leak_on_errors
pool.add :res
pool.on_error do |r|
# This is actually the wrong thing to do, and not required, but some users
# might do it. When they do, they would find that @removed would cause a
# slow leak.
pool.remove r
end
pool.perform { d = EM::DefaultDeferrable.new; d.fail; d }
EM.run { EM.next_tick { EM.stop } }
assert_equal [], pool.instance_variable_get(:@removed)
end
end
|
