Introduce new streaming API (#1439)

* Backwards-compatible
* Allow adapters to provide response info to on_call block
* Provide `stream_response` helper method

.rubocop.yml

@@ -14,6 +14,9 @@ Metrics/BlockLength:
     - spec/**/*.rb
     - examples/**/*.rb
 
+Metrics/ParameterLists:
+  Max: 6
+
 Layout/EmptyLinesAroundAttributeAccessor: # (0.83)
   Enabled: true
 

docs/usage/streaming.md

@@ -22,9 +22,10 @@ This example implements such a callback:
 streamed = []
 
 conn.get('/stream/10') do |req|
-  # Set a callback which will receive tuples of chunk Strings
-  # and the sum of characters received so far
-  req.options.on_data = Proc.new do |chunk, overall_received_bytes|
+  # Set a callback which will receive tuples of chunk Strings,
+  # the sum of characters received so far, and the response environment.
+  # The latter will allow access to the response status, headers and reason, as well as the request info.
+  req.options.on_data = Proc.new do |chunk, overall_received_bytes, env|
     puts "Received #{overall_received_bytes} characters"
     streamed << chunk
   end
@@ -36,6 +37,8 @@ streamed.join
 
 The `on_data` streaming is currently only supported by some adapters.
 To see which ones, please refer to [Awesome Faraday][awesome] comparative table or check the adapter documentation.
+Moreover, the `env` parameter was only recently added, which means some adapters may only have partial support
+(i.e. only `chunk` and `overall_received_bytes` will be passed to your block).
 
 [awesome]:      https://github.com/lostisland/awesome-faraday/#adapters
 

lib/faraday/adapter.rb

@@ -59,7 +59,7 @@ def call(env)
 
     private
 
-    def save_response(env, status, body, headers = nil, reason_phrase = nil)
+    def save_response(env, status, body, headers = nil, reason_phrase = nil, finished: true)
       env.status = status
       env.body = body
       env.reason_phrase = reason_phrase&.to_s&.strip
@@ -68,7 +68,7 @@ def save_response(env, status, body, headers = nil, reason_phrase = nil)
         yield(response_headers) if block_given?
       end
 
-      env.response.finish(env) unless env.parallel?
+      env.response.finish(env) unless env.parallel? || !finished
       env.response
     end
 

lib/faraday/options/env.rb

@@ -157,6 +157,24 @@ def inspect
       %(#<#{self.class}#{attrs.join(' ')}>)
     end
 
+    def stream_response?
+      request.stream_response?
+    end
+
+    def stream_response(&block)
+      size = 0
+      yielded = false
+      block_result = block.call do |chunk| # rubocop:disable Performance/RedundantBlockCall
+        if chunk.bytesize.positive? || size.positive?
+          yielded = true
+          size += chunk.bytesize
+          request.on_data.call(chunk, size, self)
+        end
+      end
+      request.on_data.call(+'', 0, self) unless yielded
+      block_result
+    end
+
     # @private
     def custom_members
       @custom_members ||= {}

spec/support/shared_examples/request_method.rb

@@ -153,25 +153,37 @@
       let(:streamed) { [] }
 
       context 'when response is empty' do
-        it do
+        it 'handles streaming' do
+          env = nil
           conn.public_send(http_method, '/') do |req|
-            req.options.on_data = proc { |*args| streamed << args }
+            req.options.on_data = proc do |chunk, size, block_env|
+              streamed << [chunk, size]
+              env ||= block_env
+            end
           end
 
           expect(streamed).to eq([['', 0]])
+          expect(env).to be_a(Faraday::Env)
+          expect(env.status).to eq(200)
         end
       end
 
       context 'when response contains big data' do
         before { request_stub.to_return(body: big_string) }
 
         it 'handles streaming' do
+          env = nil
           response = conn.public_send(http_method, '/') do |req|
-            req.options.on_data = proc { |*args| streamed << args }
+            req.options.on_data = proc do |chunk, size, block_env|
+              streamed << [chunk, size]
+              env ||= block_env
+            end
           end
 
           expect(response.body).to eq('')
           check_streaming_response(streamed, chunk_size: 16 * 1024)
+          expect(env).to be_a(Faraday::Env)
+          expect(env.status).to eq(200)
         end
       end
     end