You signed in with another tab or window. Reload to refresh your session.You signed out in another tab or window. Reload to refresh your session.You switched accounts on another tab or window. Reload to refresh your session.Dismiss alert
Copy file name to clipboardExpand all lines: lib/protocol/http/body/streamable.rb
+47-80
Original file line number
Diff line number
Diff line change
@@ -17,14 +17,6 @@ module Body
17
17
#
18
18
# When invoking `call(stream)`, the stream can be read from and written to, and closed. However, the stream is only guaranteed to be open for the duration of the `call(stream)` call. Once the method returns, the stream **should** be closed by the server.
19
19
moduleStreamable
20
-
# Raised when an operation is attempted on a closed stream.
21
-
classClosedError < StandardError
22
-
end
23
-
24
-
# Raised when a streaming body is consumed more than once.
25
-
classConsumedError < StandardError
26
-
end
27
-
28
20
defself.new(*arguments)
29
21
ifarguments.size == 1
30
22
DeferredBody.new(*arguments)
@@ -33,100 +25,64 @@ def self.new(*arguments)
33
25
end
34
26
end
35
27
36
-
# Represents an output wrapper around a stream, that can invoke a fiber when `#read`` is called.
37
-
#
38
-
# This behaves a little bit like a generator or lazy enumerator, in that it can be used to generate chunks of data on demand.
39
-
#
40
-
# When closing the the output, the block is invoked one last time with `nil` to indicate the end of the stream.
28
+
defself.request(&block)
29
+
DeferredBody.new(block)
30
+
end
31
+
32
+
defself.response(request, &block)
33
+
Body.new(block,request.body)
34
+
end
35
+
41
36
classOutput
37
+
defself.schedule(input,block)
38
+
self.new(input,block).tap(&:schedule)
39
+
end
40
+
42
41
definitialize(input,block)
43
-
stream=Stream.new(input,self)
44
-
45
-
@from=nil
46
-
47
-
@fiber=Fiber.newdo |from|
48
-
@from=from
49
-
block.call(stream)
50
-
rescue=>error
51
-
# Ignore.
52
-
ensure
53
-
@fiber=nil
54
-
self.close(error)
55
-
end
42
+
@output=Writable.new
43
+
@stream=Stream.new(input,@output)
44
+
@block=block
56
45
end
57
46
58
-
# Can be invoked by the block to write to the stream.
59
-
defwrite(chunk)
60
-
iffrom=@from
61
-
@from=nil
62
-
@from=from.transfer(chunk)
63
-
else
64
-
raiseClosedError,"Stream is not being read!"
47
+
defschedule
48
+
@fiber ||= Fiber.scheduledo
49
+
@block.call(@stream)
65
50
end
66
51
end
67
52
68
-
# Indicates that no further output will be generated.
69
-
defclose_write(error=nil)
70
-
# We might want to specialize the implementation later...
71
-
close(error)
53
+
defread
54
+
@output.read
72
55
end
73
56
74
-
# Can be invoked by the block to close the stream. Closing the output means that no more chunks will be generated.
75
57
defclose(error=nil)
76
-
iffrom=@from
77
-
# We are closing from within the output fiber, so we need to transfer back to `@from`:
78
-
@from=nil
79
-
iferror
80
-
from.raise(error)
81
-
else
82
-
from.transfer(nil)
83
-
end
84
-
elsif@fiber
85
-
# We are closing from outside the output fiber, so we need to resume the fiber appropriately:
86
-
@from=Fiber.current
87
-
88
-
iferror
89
-
# The fiber will be resumed from where it last called write, and we will raise the error there:
90
-
@fiber.raise(error)
91
-
else
92
-
begin
93
-
# If we get here, it means we are closing the fiber from the outside, so we need to transfer control back to the fiber:
# If the fiber then tries to write to the stream, it will raise a ClosedError, and we will end up here. We can ignore it, as we are already closing the stream and don't care about further writes.
97
-
end
98
-
end
99
-
end
100
-
end
101
-
102
-
defread
103
-
raiseRuntimeError,"Stream is already being read!"if@from
104
-
105
-
@fiber&.transfer(Fiber.current)
58
+
@output.close_write(error)
106
59
end
107
60
end
108
61
62
+
# Raised when a streaming body is consumed more than once.
63
+
classConsumedError < StandardError
64
+
end
65
+
109
66
classBody < Readable
110
67
definitialize(block,input=nil)
111
68
@block=block
112
69
@input=input
113
70
@output=nil
114
71
end
115
72
116
-
attr:block
117
-
118
73
defstream?
119
74
true
120
75
end
121
76
122
77
# Invokes the block in a fiber which yields chunks when they are available.
123
78
defread
79
+
# We are reading chunk by chunk, allocate an output stream and execute the block to generate the chunks:
124
80
if@output.nil?
125
81
if@block.nil?
126
82
raiseConsumedError,"Streaming body has already been consumed!"
127
83
end
128
84
129
-
@output=Output.new(@input,@block)
85
+
@output=Output.schedule(@input,@block)
130
86
@block=nil
131
87
end
132
88
@@ -155,15 +111,16 @@ def call(stream)
155
111
156
112
# Closing a stream indicates we are no longer interested in reading from it.
157
113
defclose(error=nil)
158
-
ifinput=@input
159
-
@input=nil
160
-
input.close(error)
161
-
end
162
-
163
114
ifoutput=@output
164
115
@output=nil
116
+
# Closing the output here may take some time, as it may need to finish handling the stream:
165
117
output.close(error)
166
118
end
119
+
120
+
ifinput=@input
121
+
@input=nil
122
+
input.close(error)
123
+
end
167
124
end
168
125
end
169
126
@@ -173,12 +130,22 @@ def initialize(block)
173
130
super(block,Writable.new)
174
131
end
175
132
133
+
# Closing a stream indicates we are no longer interested in reading from it, but in this case that does not mean that the output block is finished generating data.
134
+
defclose(error=nil)
135
+
iferror
136
+
super
137
+
end
138
+
end
139
+
176
140
# Stream the response body into the block's input.
0 commit comments