propagating channel errors to parsers, allowing to reset

lib/httpx/channel.rb

@@ -66,6 +66,7 @@ module HTTPX
       @write_buffer = Buffer.new(BUFFER_SIZE)
       @pending = []
       @state = :idle
+      on(:error) { |ex| on_error(ex) }
     end
 
     def match?(uri)
@@ -239,6 +240,11 @@ module HTTPX
       emit(:close)
     end
 
+    def on_error(ex)
+      handle_error(ex)
+      reset
+    end
+
     def handle_error(e)
       parser.handle_error(e)
       response = ErrorResponse.new(e, 0, @options)

lib/httpx/channel/http1.rb

@@ -147,6 +147,14 @@ module HTTPX
       emit(:reset)
     end
 
+    def handle_error(ex)
+      @requests.each do |request|
+        emit(:error, request, ex)
+      end
+    end
+
+    private
+
     def disable_concurrency
       return if @requests.empty?
       @requests.each { |r| r.transition(:idle) }
@@ -156,14 +164,6 @@ module HTTPX
       @max_concurrent_requests = 1
     end
 
-    def handle_error(ex)
-      @requests.each do |request|
-        emit(:error, request, ex)
-      end
-    end
-
-    private
-
     def set_request_headers(request)
       request.headers["host"] ||= request.authority
       request.headers["connection"] ||= "keep-alive"