class Async::HTTP::Protocol::HTTP11
Implements basic HTTP/1.1 request/response.
def call(request)
def call(request) @count += 1 request.version ||= self.version Async.logger.debug(self) {"#{request.method} #{request.path} #{request.headers.inspect}"} write_request(request.authority, request.method, request.path, request.version, request.headers, request.body) return Response.new(*read_response) rescue EOFError Async.logger.debug(self) {"Connection failed with EOFError after #{@count} requests."} return nil end
def initialize(stream)
def initialize(stream) super(stream, CRLF) @persistent = true @count = 0 end
def multiplex
def multiplex 1 end
def persistent?(headers)
def persistent?(headers) headers['connection'] != CLOSE end
def read_body(headers)
def read_body(headers) if headers['transfer-encoding'] == 'chunked' return Body::Chunked.new(self) elsif content_length = headers['content-length'] return Body::Fixed.new(@stream, Integer(content_length)) end end
def read_headers
def read_headers fields = [] each_line do |line| if line =~ /^([a-zA-Z\-]+):\s*(.+?)\s*$/ fields << [$1, $2] else break end end return Headers.new(fields) end
def read_request
def read_request method, path, version = read_line.split(/\s+/, 3) headers = read_headers body = read_body(headers) return headers.delete('host'), method, path, version, headers, body end
def read_response
def read_response version, status, reason = read_line.split(/\s+/, 3) headers = read_headers body = read_body(headers) @persistent = persistent?(headers) return version, Integer(status), reason, headers, body end
def receive_requests(task: Task.current)
def receive_requests(task: Task.current) while true request = Request.new(*read_request) @count += 1 response = yield request response.version ||= request.version write_response(response.version, response.status, response.headers, response.body) request.finish unless persistent?(request.headers) and persistent?(response.headers) @persistent = false break end # This ensures we yield at least once every iteration of the loop and allow other fibers to execute. task.yield end end
def reusable?
def reusable? @persistent end
def version
def version VERSION end
def write_body(body, chunked = true)
def write_body(body, chunked = true) if body.nil? or body.empty? @stream.write("Content-Length: 0\r\n\r\n") body.read if body elsif chunked @stream.write("Transfer-Encoding: chunked\r\n\r\n") body.each do |chunk| next if chunk.size == 0 @stream.write("#{chunk.bytesize.to_s(16).upcase}\r\n") @stream.write(chunk) @stream.write(CRLF) @stream.flush end @stream.write("0\r\n\r\n") else body = Body::Buffered.for(body) @stream.write("Content-Length: #{body.bytesize}\r\n\r\n") body.each do |chunk| @stream.write(chunk) end end end
def write_headers(headers)
def write_headers(headers) headers.each do |name, value| @stream.write("#{name}: #{value}\r\n") end end
def write_request(authority, method, path, version, headers, body)
def write_request(authority, method, path, version, headers, body) @stream.write("#{method} #{path} #{version}\r\n") @stream.write("Host: #{authority}\r\n") write_headers(headers) write_body(body) @stream.flush return true end
def write_response(version, status, headers, body)
def write_response(version, status, headers, body) @stream.write("#{version} #{status}\r\n") write_headers(headers) write_body(body) @stream.flush return true end