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
|
# frozen_string_literal: true
require 'spec_helper'
describe Ethon::Multi do
describe ".new" do
it "inits curl" do
expect(Ethon::Curl).to receive(:init)
Ethon::Multi.new
end
context "with default options" do
it "allows running #perform with the default execution_mode" do
Ethon::Multi.new.perform
end
it "refuses to run #socket_action" do
expect { Ethon::Multi.new.socket_action }.to raise_error(ArgumentError)
end
end
context "when options not empty" do
context "when pipelining is set" do
let(:options) { { :pipelining => true } }
it "sets pipelining" do
expect_any_instance_of(Ethon::Multi).to receive(:pipelining=).with(true)
Ethon::Multi.new(options)
end
end
context "when execution_mode option is :socket_action" do
let(:options) { { :execution_mode => :socket_action } }
let(:multi) { Ethon::Multi.new(options) }
it "refuses to run #perform" do
expect { multi.perform }.to raise_error(ArgumentError)
end
it "allows running #socket_action" do
multi.socket_action
end
end
end
end
describe "#socket_action" do
let(:options) { { :execution_mode => :socket_action } }
let(:select_state) { { :readers => [], :writers => [], :timeout => 0 } }
let(:multi) {
multi = Ethon::Multi.new(options)
multi.timerfunction = proc do |handle, timeout_ms, userp|
timeout_ms = nil if timeout_ms == -1
select_state[:timeout] = timeout_ms
:ok
end
multi.socketfunction = proc do |handle, sock, what, userp, socketp|
case what
when :remove
select_state[:readers].delete(sock)
select_state[:writers].delete(sock)
when :in
select_state[:readers].push(sock) unless select_state[:readers].include? sock
select_state[:writers].delete(sock)
when :out
select_state[:readers].delete(sock)
select_state[:writers].push(sock) unless select_state[:writers].include? sock
when :inout
select_state[:readers].push(sock) unless select_state[:readers].include? sock
select_state[:writers].push(sock) unless select_state[:writers].include? sock
else
raise ArgumentError, "invalid value for 'what' in socketfunction callback"
end
:ok
end
multi
}
def fds_to_ios(fds)
fds.map do |fd|
IO.for_fd(fd).tap { |io| io.autoclose = false }
end
end
def perform_socket_action_until_complete
multi.socket_action # start things off
while multi.ongoing?
readers, writers, _ = IO.select(
fds_to_ios(select_state[:readers]),
fds_to_ios(select_state[:writers]),
[],
select_state[:timeout]
)
to_notify = Hash.new { |hash, key| hash[key] = [] }
unless readers.nil?
readers.each do |reader|
to_notify[reader] << :in
end
end
unless writers.nil?
writers.each do |writer|
to_notify[writer] << :out
end
end
to_notify.each do |io, readiness|
multi.socket_action(io, readiness)
end
# if we didn't have anything to notify, then we timed out
multi.socket_action if to_notify.empty?
end
ensure
multi.easy_handles.dup.each do |h|
multi.delete(h)
end
end
it "supports an end-to-end request" do
easy = Ethon::Easy.new
easy.url = "http://localhost:3001/"
multi.add(easy)
perform_socket_action_until_complete
expect(multi.ongoing?).to eq(false)
end
it "supports multiple concurrent requests" do
handles = []
10.times do
easy = Ethon::Easy.new
easy.url = "http://localhost:3001/?delay=1"
multi.add(easy)
handles << easy
end
start = Time.now
perform_socket_action_until_complete
duration = Time.now - start
# these should have happened concurrently
expect(duration).to be < 2
expect(multi.ongoing?).to eq(false)
handles.each do |handle|
expect(handle.response_code).to eq(200)
end
end
end
end
|
