Improve streaming interface to provide response info

.github/workflows/ci.yml

@@ -40,7 +40,13 @@ jobs:
     strategy:
       fail-fast: false
       matrix:
-        ruby: [ '2.6', '2.7', '3.0', '3.1', truffleruby-head ]
+        ruby: [ '2.6', '2.7', '3.0', '3.1' ]
+        experimental: [false]
+        include:
+          - ruby: head
+            experimental: true
+          - ruby: truffleruby-head
+            experimental: true
 
     steps:
       - uses: actions/checkout@v3
@@ -49,7 +55,13 @@ jobs:
           ruby-version: ${{ matrix.ruby }}
           bundler-cache: true
 
-      - name: Test
+      - name: RSpec
         continue-on-error: ${{ matrix.experimental }}
         run: bundle exec rake
 
+      - name: Test External Adapters
+        if: ${{ matrix.ruby != '2.6' }}
+        continue-on-error: ${{ matrix.experimental }}
+        run: bundle exec bake test:external
+
+

.gitignore

@@ -14,6 +14,7 @@ tmp
 .bundle
 Gemfile.lock
 vendor/bundle
+external
 
 ## PROJECT::SPECIFIC
 .rbx

.rubocop.yml

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

Gemfile

@@ -7,6 +7,7 @@ source 'https://rubygems.org'
 gem 'jruby-openssl', '~> 0.11.0', platforms: :jruby
 
 group :development, :test do
+  gem 'bake-test-external'
   gem 'coveralls_reborn', require: false
   gem 'pry'
   gem 'rack', '~> 2.2'
@@ -19,7 +20,7 @@ end
 
 group :development, :lint do
   gem 'rubocop'
-  gem 'rubocop-packaging', '~> 0.5'
+  gem 'rubocop-packaging', github: 'utkarsh2102/rubocop-packaging' # '~> 0.5'
   gem 'rubocop-performance', '~> 1.0'
   gem 'yard-junk'
 end

config/external.yaml

@@ -0,0 +1,3 @@
+faraday-net_http:
+  url: https://github.com/lostisland/faraday-net_http.git
+  command: bundle exec rspec

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,39 @@
       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]])
+          # TODO: enable this after updating all existing adapters to the new streaming API
+          # 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)
+          # TODO: enable this after updating all existing adapters to the new streaming API
+          # expect(env).to be_a(Faraday::Env)
+          # expect(env.status).to eq(200)
         end
       end
     end