Merge branch 'issue-151' into 'master'

New http timeouts

Closes #151

See merge request honeyryderchuck/httpx!212

lib/httpx/connection.rb

@@ -34,6 +34,7 @@ module HTTPX
     include Callbacks
 
     using URIExtensions
+    using NumericExtensions
 
     require "httpx/connection/http2"
     require "httpx/connection/http1"
@@ -233,6 +234,7 @@ module HTTPX
           # when pushing a request into an existing connection, we have to check whether there
           # is the possibility that the connection might have extended the keep alive timeout.
           # for such cases, we want to ping for availability before deciding to shovel requests.
+          log(level: 3) { "keep alive timeout expired, pinging connection..." }
           @pending << request
           parser.ping
           transition(:active) if @state == :inactive
@@ -430,6 +432,8 @@ module HTTPX
       @inflight += 1
       parser.send(request)
 
+      set_request_timeouts(request)
+
       return unless @state == :inactive
 
       transition(:active)
@@ -591,10 +595,45 @@ module HTTPX
     def handle_error(error)
       parser.handle_error(error) if @parser && parser.respond_to?(:handle_error)
       while (request = @pending.shift)
-        response = ErrorResponse.new(request, error, @options)
+        response = ErrorResponse.new(request, error, request.options)
         request.response = response
         request.emit(:response, response)
       end
     end
+
+    def set_request_timeouts(request)
+      write_timeout = request.write_timeout
+      request.once(:headers) do
+        @timers.after(write_timeout) { write_timeout_callback(request, write_timeout) }
+      end unless write_timeout.nil? || write_timeout.infinite?
+
+      read_timeout = request.read_timeout
+      request.once(:done) do
+        @timers.after(read_timeout) { read_timeout_callback(request, read_timeout) }
+      end unless read_timeout.nil? || read_timeout.infinite?
+
+      request_timeout = request.request_timeout
+      request.once(:headers) do
+        @timers.after(request_timeout) { read_timeout_callback(request, request_timeout, RequestTimeoutError) }
+      end unless request_timeout.nil? || request_timeout.infinite?
+    end
+
+    def write_timeout_callback(request, write_timeout)
+      return if request.state == :done
+
+      @write_buffer.clear
+      error = WriteTimeoutError.new(request, nil, write_timeout)
+      on_error(error)
+    end
+
+    def read_timeout_callback(request, read_timeout, error_type = ReadTimeoutError)
+      response = request.response
+
+      return if response && response.finished?
+
+      @write_buffer.clear
+      error = error_type.new(request, request.response, read_timeout)
+      on_error(error)
+    end
   end
 end

lib/httpx/connection/http1.rb

@@ -118,7 +118,7 @@ module HTTPX
       log(color: :yellow) { response.headers.each.map { |f, v| "-> HEADER: #{f}: #{v}" }.join("\n") }
 
       @request.response = response
-      on_complete if response.complete?
+      on_complete if response.finished?
     end
 
     def on_trailers(h)
@@ -158,6 +158,7 @@ module HTTPX
       @request = nil
       @requests.shift
       response = request.response
+      response.finish!
       emit(:response, request, response)
 
       if @parser.upgrade?

lib/httpx/errors.rb

@@ -24,6 +24,24 @@ module HTTPX
 
   class ConnectTimeoutError < TimeoutError; end
 
+  class RequestTimeoutError < TimeoutError
+    attr_reader :request
+
+    def initialize(request, response, timeout)
+      @request = request
+      @response = response
+      super(timeout, "Timed out after #{timeout} seconds")
+    end
+
+    def marshal_dump
+      [message]
+    end
+  end
+
+  class ReadTimeoutError < RequestTimeoutError; end
+
+  class WriteTimeoutError < RequestTimeoutError; end
+
   class SettingsTimeoutError < TimeoutError; end
 
   class ResolveTimeoutError < TimeoutError; end

lib/httpx/extensions.rb

@@ -54,6 +54,14 @@ module HTTPX
     Numeric.__send__(:include, NegMethods)
   end
 
+  module NumericExtensions
+    refine Numeric do
+      def infinite?
+        self == Float::INFINITY
+      end unless Numeric.method_defined?(:infinite?)
+    end
+  end
+
   module StringExtensions
     refine String do
       def delete_suffix!(suffix)

lib/httpx/options.rb

@@ -10,6 +10,7 @@ module HTTPX
     OPERATION_TIMEOUT = 60
     KEEP_ALIVE_TIMEOUT = 20
     SETTINGS_TIMEOUT = 10
+    READ_TIMEOUT = WRITE_TIMEOUT = REQUEST_TIMEOUT = Float::INFINITY
 
     # https://github.com/ruby/resolv/blob/095f1c003f6073730500f02acbdbc55f83d70987/lib/resolv.rb#L408
     ip_address_families = begin
@@ -34,6 +35,9 @@ module HTTPX
         settings_timeout: SETTINGS_TIMEOUT,
         operation_timeout: OPERATION_TIMEOUT,
         keep_alive_timeout: KEEP_ALIVE_TIMEOUT,
+        read_timeout: READ_TIMEOUT,
+        write_timeout: WRITE_TIMEOUT,
+        request_timeout: REQUEST_TIMEOUT,
       },
       :headers => {},
       :window_size => WINDOW_SIZE,

lib/httpx/request.rb

@@ -64,6 +64,18 @@ module HTTPX
       @state = :idle
     end
 
+    def read_timeout
+      @options.timeout[:read_timeout]
+    end
+
+    def write_timeout
+      @options.timeout[:write_timeout]
+    end
+
+    def request_timeout
+      @options.timeout[:request_timeout]
+    end
+
     def trailers?
       defined?(@trailers)
     end
@@ -108,6 +120,7 @@ module HTTPX
 
     def path
       path = uri.path.dup
+      path =  +"" if path.nil?
       path << "/" if path.empty?
       path << "?#{query}" unless query.empty?
       path

lib/httpx/resolver/native.rb

@@ -77,7 +77,8 @@ module HTTPX
       nil
     rescue Errno::EHOSTUNREACH => e
       @ns_index += 1
-      if @ns_index < @nameserver.size
+      nameserver = @nameserver
+      if nameserver && @ns_index < nameserver.size
         log { "resolver: failed resolving on nameserver #{@nameserver[@ns_index - 1]} (#{e.message})" }
         transition(:idle)
       else

lib/httpx/response.rb

@@ -31,6 +31,7 @@ module HTTPX
       @status = Integer(status)
       @headers = @options.headers_class.new(headers)
       @body = @options.response_body_class.new(self, @options)
+      @finished = complete?
     end
 
     def merge_headers(h)
@@ -41,15 +42,24 @@ module HTTPX
       @body.write(data)
     end
 
+    def content_type
+      @content_type ||= ContentType.new(@headers["content-type"])
+    end
+
+    def finished?
+      @finished
+    end
+
+    def finish!
+      @finished = true
+      @headers.freeze
+    end
+
     def bodyless?
       @request.verb == :head ||
         no_data?
     end
 
-    def content_type
-      @content_type ||= ContentType.new(@headers["content-type"])
-    end
-
     def complete?
       bodyless? || (@request.verb == :connect && @status == 200)
     end
@@ -102,7 +112,7 @@ module HTTPX
     end
 
     def no_data?
-      @status < 200 ||
+      @status < 200 || # informational response
         @status == 204 ||
         @status == 205 ||
         @status == 304 || begin
@@ -339,6 +349,10 @@ module HTTPX
       end
     end
 
+    def finished?
+      true
+    end
+
     def raise_for_status
       raise @error
     end

lib/httpx/timers.rb

@@ -37,9 +37,12 @@ module HTTPX
       elapsed_time = Utils.elapsed_time(@next_interval_at)
 
       @intervals.delete_if { |interval| interval.elapse(elapsed_time) <= 0 }
+
+      @next_interval_at = nil if @intervals.empty?
     end
 
     def cancel
+      @next_interval_at = nil
       @intervals.clear
     end
 

sig/connection.rbs

@@ -28,6 +28,7 @@ module HTTPX
     attr_reader options: Options
     attr_writer timers: Timers
 
+    @type: io_type
     @origins: Array[URI::Generic]
     @window_size: Integer
     @read_buffer: Buffer
@@ -40,7 +41,7 @@ module HTTPX
 
     def addresses=: (Array[ipaddr]) -> void
 
-    def match?: (URI::Generic, options) -> bool
+    def match?: (URI::Generic uri, Options options) -> bool
 
     def mergeable?: (Connection) -> bool
 
@@ -54,13 +55,15 @@ module HTTPX
 
     def match_altsvcs?: (URI::Generic uri) -> bool
 
+    def match_altsvc_options?: (URI::Generic uri, Options options) -> bool
+
     def connecting?: () -> bool
 
     def inflight?: () -> boolish
 
     def interests: () -> io_interests?
 
-    def to_io: () -> IO
+    def to_io: () -> ::IO
 
     def call: () -> void
 
@@ -77,7 +80,7 @@ module HTTPX
 
     private
 
-    def initialize: (String, URI::Generic, options) -> untyped
+    def initialize: (io_type, URI::Generic, options) -> untyped
 
     def connect: () -> void
 
@@ -103,5 +106,11 @@ module HTTPX
     def handle_error: (StandardError) -> void
 
     def purge_after_closed: () -> void
+
+    def set_request_timeouts: (Request request) -> void
+
+    def write_timeout_callback: (Request request, Numeric write_timeout) -> void
+
+    def read_timeout_callback: (Request request, Numeric read_timeout, ?singleton(RequestTimeoutError) error_type) -> void
   end
 end

sig/errors.rbs

@@ -23,6 +23,19 @@ module HTTPX
   class ResolveTimeoutError < TimeoutError
   end
 
+  class RequestTimeoutError < TimeoutError
+    attr_reader request: Request
+    attr_reader response: response?
+
+    def initialize: (Request request, response? response, Numeric timeout) -> void
+  end
+
+  class ReadTimeoutError < RequestTimeoutError
+  end
+
+  class WriteTimeoutError < RequestTimeoutError
+  end
+
   class ResolveError < Error
   end
 

sig/io.rbs

@@ -0,0 +1,6 @@
+module HTTPX
+  type io_type = "udp" | "tcp" | "ssl" | "unix"
+
+  module IO
+  end
+end

sig/options.rbs

@@ -10,7 +10,7 @@ module HTTPX
     SETTINGS_TIMEOUT: Integer
     DEFAULT_OPTIONS: Hash[Symbol, untyped]
 
-    type timeout_type = :connect_timeout | :settings_timeout | :operation_timeout | :keep_alive_timeout | :total_timeout
+    type timeout_type = :connect_timeout | :settings_timeout | :operation_timeout | :keep_alive_timeout | :total_timeout | :read_timeout | :write_timeout | :request_timeout
     type timeout = Hash[timeout_type, Numeric]
 
     def self.new: (?options) -> instance
@@ -65,6 +65,9 @@ module HTTPX
     # body
     attr_reader origin: URI::Generic?
 
+    # base_path
+    attr_reader base_path: String?
+
     # ssl
 
     # http2_settings

sig/request.rbs

@@ -50,6 +50,12 @@ module HTTPX
 
     def trailers?: () -> boolish
 
+    def read_timeout: () -> Numeric
+
+    def write_timeout: () -> Numeric
+
+    def request_timeout: () -> Numeric
+
     class Body
       @headers: Headers
       @body: body_encoder?

sig/resolver/native.rbs

@@ -10,7 +10,10 @@ module HTTPX
       @family: ip_family
       @options: Options
       @ns_index: Integer
-      @nameserver: String
+      @nameserver: Array[String]?
+      @ndots: Integer
+      @start_timeout: Numeric?
+      @search: Array[String]
       @_timeouts: Array[Numeric]
       @timeouts: Hash[String, Array[Numeric]]
       @connections: Array[Connection]
@@ -37,7 +40,7 @@ module HTTPX
 
       def consume: () -> void
 
-      def do_retry: (?Numeric loop_time) -> void
+      def do_retry: (?Numeric? loop_time) -> void
 
       def dread: (Integer) -> void
                | () -> void

sig/response.rbs

@@ -1,5 +1,7 @@
 module HTTPX
   interface _Response
+    def finished?: () -> bool
+
     def raise_for_status: () -> self
 
     def error: () -> StandardError?

test/altsvc_test.rb

@@ -6,14 +6,14 @@ class AltSvcTest < Minitest::Test
   include HTTPX
 
   def test_altsvc_cache
-    assert AltSvc.cached_altsvc("http://www.example.com").empty?
-    AltSvc.cached_altsvc_set("http://www.example.com", { "origin" => "http://alt.example.com", "ma" => 2 })
-    entries = AltSvc.cached_altsvc("http://www.example.com")
+    assert AltSvc.cached_altsvc("http://www.example-altsvc-cache.com").empty?
+    AltSvc.cached_altsvc_set("http://www.example-altsvc-cache.com", { "origin" => "http://alt.example-altsvc-cache.com", "ma" => 2 })
+    entries = AltSvc.cached_altsvc("http://www.example-altsvc-cache.com")
     assert !entries.empty?
     entry = entries.first
-    assert entry["origin"] == "http://alt.example.com"
+    assert entry["origin"] == "http://alt.example-altsvc-cache.com"
     sleep 3
-    assert AltSvc.cached_altsvc("http://www.example.com").empty?
+    assert AltSvc.cached_altsvc("http://www.example-altsvc-cache.com").empty?
   end
 
   def test_altsvc_parse_svc
@@ -52,16 +52,16 @@ class AltSvcTest < Minitest::Test
   end
 
   def test_altsvc_clear_cache
-    AltSvc.cached_altsvc_set("http://www.example.com", { "origin" => "http://alt.example.com", "ma" => 2 })
-    entries = AltSvc.cached_altsvc("http://www.example.com")
+    AltSvc.cached_altsvc_set("http://www.example-clear-cache.com", { "origin" => "http://alt.example-clear-cache.com", "ma" => 2 })
+    entries = AltSvc.cached_altsvc("http://www.example-clear-cache.com")
     assert !entries.empty?
 
-    req = Request.new(:get, "http://www.example.com/")
+    req = Request.new(:get, "http://www.example-clear-cache.com/")
     res = Response.new(req, 200, "2.0", { "alt-svc" => "clear" })
 
     AltSvc.emit(req, res)
 
-    entries = AltSvc.cached_altsvc("http://www.example.com")
+    entries = AltSvc.cached_altsvc("http://www.example-clear-cache.com")
     assert entries.empty?
   end
 end

test/options_test.rb

@@ -107,6 +107,9 @@ class OptionsTest < Minitest::Test
         settings_timeout: 10,
         operation_timeout: 60,
         keep_alive_timeout: 20,
+        read_timeout: Float::INFINITY,
+        write_timeout: Float::INFINITY,
+        request_timeout: Float::INFINITY,
       },
       :ssl => { :foo => "bar" },
       :http2_settings => { :settings_enable_push => 0 },

test/session_test.rb

@@ -3,19 +3,18 @@
 require_relative "test_helper"
 
 class SessionTest < Minitest::Test
-  include HTTPX
   include HTTPHelpers
 
   def test_session_block
     yielded = nil
-    Session.new do |cli|
+    HTTPX::Session.new do |cli|
       yielded = cli
     end
-    assert yielded.is_a?(Session), "session should have been yielded"
+    assert yielded.is_a?(HTTPX::Session), "session should have been yielded"
   end
 
   def test_session_plugin
-    klient_class = Class.new(Session)
+    klient_class = Class.new(HTTPX::Session)
     klient_class.plugin(TestPlugin)
     session = klient_class.new
     assert session.respond_to?(:foo), "instance methods weren't added"
@@ -69,6 +68,40 @@ class SessionTest < Minitest::Test
     end
   end
 
+  def test_session_timeouts_read_timeout
+    uri = build_uri("/drip?numbytes=10&duration=4&delay=2&code=200")
+    session = HTTPX.with_timeout(read_timeout: 3, operation_timeout: 10)
+    response = session.get(uri)
+    verify_error_response(response, HTTPX::ReadTimeoutError)
+
+    uri = build_uri("/drip?numbytes=10&duration=1&delay=0&code=200")
+    response1 = session.get(uri)
+    verify_status(response1, 200)
+  end
+
+  def test_session_timeouts_write_timeout
+    start_test_servlet(SlowReader) do |server|
+      uri = URI("#{server.origin}/")
+      session = HTTPX.with(timeout: { write_timeout: 4, operation_timeout: 10 })
+      response = session.post(uri, body: StringIO.new("a" * 65_536 * 3 * 5))
+      verify_error_response(response, HTTPX::WriteTimeoutError)
+
+      response1 = session.post(uri, body: StringIO.new("a" * 65_536 * 2 * 5))
+      verify_status(response1, 200)
+    end
+  end
+
+  def test_session_timeouts_request_timeout
+    uri = build_uri("/drip?numbytes=10&duration=4&delay=2&code=200")
+    session = HTTPX.with_timeout(request_timeout: 3, operation_timeout: 10)
+    response = session.get(uri)
+    verify_error_response(response, HTTPX::RequestTimeoutError)
+
+    uri = build_uri("/drip?numbytes=10&duration=1&delay=0&code=200")
+    response1 = session.get(uri)
+    verify_status(response1, 200)
+  end
+
   # def test_http_timeouts_operation_timeout
   #   uri = build_uri("/delay/2")
   #   session = HTTPX.with_timeout(operation_timeout: 1)

test/support/servlets/slow_reader.rb

@@ -0,0 +1,46 @@
+# frozen_string_literal: true
+
+require "socket"
+
+class SlowReader
+  def initialize
+    @server = TCPServer.new("127.0.0.1", 0)
+    @server.setsockopt(Socket::SOL_SOCKET, Socket::SO_RCVBUF, 5)
+    @can_log = ENV.key?("HTTPX_DEBUG")
+  end
+
+  def origin
+    _, sock, ip, _ = @server.addr
+    "http://#{ip}:#{sock}"
+  end
+
+  def start
+    loop do
+      sock = @server.accept
+
+      begin
+        sock.setsockopt(Socket::SOL_SOCKET, Socket::SO_RCVBUF, 5)
+        request = +""
+        5.times do
+          request << sock.readpartial(2048)
+          warn "buffered request: #{request.size} (closed? #{sock.closed?})" if @can_log
+          sleep(1)
+        end
+        sock.setsockopt(Socket::SOL_SOCKET, Socket::SO_RCVBUF, 65_535)
+        request << sock.readpartial(2048)
+        # warn "request: #{request.size}" if @can_log
+        response = "HTTP/1.1 200\r\nContent-Length: 0\r\nConnection: close\r\n\r\n"
+        sock.puts(response)
+      rescue IOError => e
+        warn e.message
+      ensure
+        sock.close
+      end
+    end
+  rescue IOError
+  end
+
+  def shutdown
+    @server.close
+  end
+end