Class: Protocol::HTTP::Executor::Body::StreamInput

Inherits:
Body::Readable
  • Object
show all
Defined in:
lib/protocol/http/executor/body/input.rb

Overview

An upgraded stream input that follows the initial request body phase.

Instance Method Summary collapse

Constructor Details

#initialize(channel, headers) ⇒ StreamInput

Initialize a stream input.



97
98
99
100
101
 
# File 'lib/protocol/http/executor/body/input.rb', line 97

def initialize(channel, headers)
	@channel = channel
	@headers = headers
	@closed = false
end

Instance Method Details

#close(error = nil) ⇒ Object

Close the stream input channel.



135
136
137
138
139
140
 
# File 'lib/protocol/http/executor/body/input.rb', line 135

def close(error = nil)
	return if @closed
	
	@closed = true
	@channel.close_read
end

#read ⇒ Object

Read the next request or upgraded-stream chunk.



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
 
# File 'lib/protocol/http/executor/body/input.rb', line 104

def read
	return nil if @closed
	
	loop do
		message = @channel.read
		unless message
			@closed = true
			return nil
		end
		
		type, payload = message
		
		case type
		when :chunk, :stream_chunk
			return payload
		when :trailers
			@headers.trailer!
			payload.each{|key, value| @headers.add(key, value, trailer: true)}
		when :end
			# The request body ended; an upgraded stream may follow:
			next
		when :error, :stream_error
			@closed = true
			raise RemoteError.new(payload)
		else
			raise ClosedError, "Unexpected stream input event: #{type.inspect}!"
		end
	end
end