mirror of
https://github.com/HoneyryderChuck/httpx.git
synced 2025-08-12 00:01:58 -04:00
refactored the http2 stream callbacks using curry methods, which allowed also to shave LOCs in the push-promise plugin implementation
This commit is contained in:
parent
87d908ba9b
commit
53303b65bc
@ -92,37 +92,13 @@ module HTTPX
|
||||
end
|
||||
|
||||
def handle_stream(stream, request)
|
||||
stream.on(:close) do |error|
|
||||
if request.expects?
|
||||
return handle(request, stream)
|
||||
end
|
||||
response = request.response || ErrorResponse.new(error, @retries)
|
||||
emit(:response, request, response)
|
||||
log(2, "#{stream.id}: ") { "closing stream" }
|
||||
|
||||
|
||||
@streams.delete(request)
|
||||
send(@pending.shift) unless @pending.empty?
|
||||
end
|
||||
stream.on(:close, &method(:on_stream_close).curry[stream, request])
|
||||
stream.on(:half_close) do
|
||||
log(2, "#{stream.id}: ") { "waiting for response..." }
|
||||
end
|
||||
# stream.on(:altsvc)
|
||||
stream.on(:headers) do |h|
|
||||
log(stream.id) do
|
||||
h.map { |k, v| "<- HEADER: #{k}: #{v}" }.join("\n")
|
||||
end
|
||||
_, status = h.shift
|
||||
headers = @options.headers_class.new(h)
|
||||
response = @options.response_class.new(request, status, "2.0", headers, @options)
|
||||
request.response = response
|
||||
@streams[request] = stream
|
||||
end
|
||||
stream.on(:data) do |data|
|
||||
log(1, "#{stream.id}: ") { "<- DATA: #{data.bytesize} bytes..." }
|
||||
log(2, "#{stream.id}: ") { "<- #{data.inspect}" }
|
||||
request.response << data
|
||||
end
|
||||
stream.on(:headers, &method(:on_stream_headers).curry[stream, request])
|
||||
stream.on(:data, &method(:on_stream_data).curry[stream, request])
|
||||
end
|
||||
|
||||
def join_headers(stream, request)
|
||||
@ -158,6 +134,36 @@ module HTTPX
|
||||
# HTTP/2 Callbacks
|
||||
######
|
||||
|
||||
def on_stream_headers(stream, request, h)
|
||||
log(stream.id) do
|
||||
h.map { |k, v| "<- HEADER: #{k}: #{v}" }.join("\n")
|
||||
end
|
||||
_, status = h.shift
|
||||
headers = @options.headers_class.new(h)
|
||||
response = @options.response_class.new(request, status, "2.0", headers, @options)
|
||||
request.response = response
|
||||
@streams[request] = stream
|
||||
end
|
||||
|
||||
def on_stream_data(stream, request, data)
|
||||
log(1, "#{stream.id}: ") { "<- DATA: #{data.bytesize} bytes..." }
|
||||
log(2, "#{stream.id}: ") { "<- #{data.inspect}" }
|
||||
request.response << data
|
||||
end
|
||||
|
||||
def on_stream_close(stream, request, error)
|
||||
if request.expects?
|
||||
return handle(request, stream)
|
||||
end
|
||||
response = request.response || ErrorResponse.new(error, @retries)
|
||||
emit(:response, request, response)
|
||||
log(2, "#{stream.id}: ") { "closing stream" }
|
||||
|
||||
|
||||
@streams.delete(request)
|
||||
send(@pending.shift) unless @pending.empty?
|
||||
end
|
||||
|
||||
def on_frame(bytes)
|
||||
@buffer << bytes
|
||||
end
|
||||
|
@ -1,5 +1,23 @@
|
||||
# frozen_string_literal: true
|
||||
|
||||
unless Method.method_defined?(:curry)
|
||||
|
||||
# Backport
|
||||
#
|
||||
# Ruby 2.1 and lower implement curry only for Procs.
|
||||
#
|
||||
# Why not using Refinements? Because they don't work for Method (tested with ruby 2.1.9).
|
||||
#
|
||||
module CurryMethods # :nodoc:
|
||||
# Backport for the Method#curry method, which is part of ruby core since 2.2 .
|
||||
#
|
||||
def curry(*args)
|
||||
to_proc.curry(*args)
|
||||
end
|
||||
end
|
||||
Method.__send__(:include, CurryMethods)
|
||||
end
|
||||
|
||||
unless String.method_defined?(:+@)
|
||||
# Backport for +"", to initialize unfrozen strings from the string literal.
|
||||
#
|
||||
@ -21,6 +39,7 @@ unless Numeric.method_defined?(:positive?)
|
||||
end
|
||||
Numeric.__send__(:include, PosMethods)
|
||||
end
|
||||
|
||||
unless Numeric.method_defined?(:negative?)
|
||||
# Ruby 2.3 Backport (Numeric#negative?)
|
||||
#
|
||||
|
@ -48,34 +48,12 @@ module HTTPX
|
||||
end
|
||||
|
||||
def __on_promise_response(parser, stream, h)
|
||||
log(1, "#{stream.id}(promise): ") do
|
||||
h.map { |k, v| "<- HEADER: #{k}: #{v}" }.join("\n")
|
||||
end
|
||||
request = @promise_headers.delete(stream)
|
||||
return unless request
|
||||
_, status = h.shift
|
||||
headers = @options.headers_class.new(h)
|
||||
response = @options.response_class.new(request, status, "2.0", headers, @options)
|
||||
request.response = response
|
||||
parser.__send__(:on_stream_headers, stream, request, h)
|
||||
request.transition(:done)
|
||||
parser.streams[request] = stream
|
||||
stream.on(:data) do |data|
|
||||
log(1, "#{stream.id}(promise): ") { "<- DATA: #{data.bytesize} bytes..." }
|
||||
log(2, "#{stream.id}(promise): ") { "<- #{data.inspect}" }
|
||||
request.response << data
|
||||
end
|
||||
stream.on(:close) do |error|
|
||||
|
||||
if request.expects?
|
||||
return handle(request, stream)
|
||||
end
|
||||
response = request.response || ErrorResponse.new(error, retries)
|
||||
on_response(request, response)
|
||||
log(2, "#{stream.id}(promise): ") { "closing stream" }
|
||||
|
||||
|
||||
parser.streams.delete(request)
|
||||
end
|
||||
stream.on(:data, &parser.method(:on_stream_data).curry[stream, request])
|
||||
stream.on(:close, &parser.method(:on_stream_close).curry[stream, request])
|
||||
end
|
||||
end
|
||||
end
|
||||
|
Loading…
x
Reference in New Issue
Block a user