Added compatibility with the Net::HTTP response streaming API

Rakefile

@@ -24,19 +24,8 @@ rescue LoadError
   puts "Jeweler (or a dependency) not available. Install it with: gem install jeweler"
 end
 
-require 'spec/rake/spectask'
-Spec::Rake::SpecTask.new(:spec) do |spec|
-  spec.libs << 'lib' << 'spec'
-  spec.spec_files = FileList['spec/**/*_spec.rb']
-end
-
-Spec::Rake::SpecTask.new(:rcov) do |spec|
-  spec.libs << 'lib' << 'spec'
-  spec.pattern = 'spec/**/*_spec.rb'
-  spec.rcov = true
-end
-
-task :spec => :check_dependencies
+require 'rspec/core/rake_task'
+RSpec::Core::RakeTask.new(:spec)
 
 task :default => :spec
 

lib/em-net-http.rb

@@ -10,12 +10,16 @@ class Response
       attr_reader :code, :body, :header, :message, :http_version
       alias_method :msg, :message
 
-      def initialize(res)
-        @code = res.response_header.http_status
-        @message = res.response_header.http_reason
-        @http_version = res.response_header.http_version
-        @header = res.response_header
-        @body = res.response
+      def initialize(response_header)
+        @code = response_header.http_status
+        @message = response_header.http_reason
+        @http_version = response_header.http_version
+        @header = response_header
+      end
+
+      def set_body body
+        @already_buffered = true
+        @body = body
       end
 
       def content_type
@@ -70,8 +74,28 @@ class << self
     alias_method :orig_net_http_read_body, :read_body
 
     def read_body(dest=nil, &block)
+      return @body if @already_buffered
       return orig_net_http_read_body(dest, &block) unless ::EM.reactor_running?
-      @body
+      if block_given?
+        f = Fiber.current
+        @httpreq.callback { |res| f.resume }
+        @httpreq.stream &block
+        Fiber.yield
+      else
+        unless @body || @already_buffered
+          if self.class.body_permitted?
+            f = Fiber.current
+            io = StringIO.new '', 'wb'
+            io.set_encoding 'ASCII-8BIT'
+            @httpreq.callback { |res| f.resume io.string }
+            @httpreq.errback { |err| f.resume err }
+            @httpreq.stream { |chunk| io.write chunk }
+            @body = Fiber.yield
+          end
+          @already_buffered = true
+        end
+        @body
+      end
     end
   end
 
@@ -116,7 +140,8 @@ def request(req, body = nil, &block)
       f=Fiber.current
 
       convert_em_http_response = lambda do |res|
-        emres = EM::NetHTTP::Response.new(res)
+        emres = EM::NetHTTP::Response.new(res.response_header)
+        emres.set_body res.response
         nhresclass = Net::HTTPResponse.response_class(emres.code)
         nhres = nhresclass.new(emres.http_version, emres.code, emres.message)
         emres.to_hash.each do |k, v|
@@ -127,16 +152,36 @@ def request(req, body = nil, &block)
         f.resume nhres
       end
 
-      httpreq.callback &convert_em_http_response
-      httpreq.errback {|err|f.resume(:error)}
-      res = Fiber.yield
-      if res == :error
-        raise 'EM::HttpRequest error - request timed out' if Time.now - self.read_timeout > t0
-        raise 'EM::HttpRequest error - unknown error'
-      end
 
-      yield res if block_given?
-      res
+      if block_given?
+        httpreq.headers { |headers|
+
+          emres = EM::NetHTTP::Response.new(headers)
+          nhresclass = Net::HTTPResponse.response_class(emres.code)
+          nhres = nhresclass.new(emres.http_version, emres.code, emres.message)
+          emres.to_hash.each do |k, v|
+            nhres.add_field(k, v)
+          end
+          f.resume nhres
+        }
+
+        nhres = Fiber.yield
+        nhres.instance_variable_set :@httpreq, httpreq
+
+        yield nhres
+        nhres
+      else
+        httpreq.callback &convert_em_http_response
+        httpreq.errback {|err|f.resume(:error)}
+        res = Fiber.yield
+
+        if res == :error
+          raise 'EM::HttpRequest error - request timed out' if Time.now - self.read_timeout > t0
+          raise 'EM::HttpRequest error - unknown error'
+        end
+
+        res
+      end
     end
 
   end

spec/em-net-http_spec.rb

@@ -1,9 +1,39 @@
 require File.expand_path(File.dirname(__FILE__) + '/spec_helper')
 
 describe "em-net-http" do
+
+  it 'should support streaming the response' do
+    assert_identical(:streamed => true) {
+      body = StringIO.new '', 'wb'
+
+      Net::HTTP.start('localhost', Mimic::MIMIC_DEFAULT_PORT) do |http|
+        http.request_get "/image" do |resp|
+          resp.should be_a_kind_of(Net::HTTPOK)
+          resp.read_body { |chunk| body.write chunk }
+          resp
+        end
+      end.tap do |resp|
+        resp.instance_variable_set :@streamed_body, body.string
+      end
+    }
+  end
+
+  it 'should support buffering the response' do
+    assert_identical {
+      Net::HTTP.start('localhost', Mimic::MIMIC_DEFAULT_PORT) do |http|
+        respone = http.request_get "/image" do |resp|
+          resp.should be_a_kind_of(Net::HTTPOK)
+          resp.read_body # force reading the body before the test tears down the EM loop
+          resp
+        end
+        respone.tap { respone.should be_a_kind_of(Net::HTTPOK) }
+      end
+    }
+  end
+
   describe 'should be compatible' do
     it 'for Net::HTTP.get()' do
-      run_requests {Net::HTTP.get(URI.parse('http://localhost/hello'))}
+      run_requests {Net::HTTP.get(URI.parse("http://localhost:#{Mimic::MIMIC_DEFAULT_PORT}/hello"))}
       @expected_res.should == @actual_res
     end
 
@@ -17,7 +47,10 @@
       it "for Net::HTTP.start(host, port, &block) with response code #{code}" do
         assert_identical {
           Net::HTTP.start('localhost', Mimic::MIMIC_DEFAULT_PORT) do |http|
-            http.get("/code/#{code}")
+            http.get("/code/#{code}").tap { |resp|
+              # Force the response to be buffered while we are still in the EM loop, since we shut it down EM before the verifications
+              resp.body
+            }
           end
         }
       end
@@ -68,18 +101,20 @@ def run_requests(&block)
     end
   end
 
-  def assert_identical(&block)
+  def assert_identical(streamed=false, &block)
     run_requests(&block)
-    @actual_res.should match_response(@expected_res)
+    @actual_res.should be_a_kind_of(Net::HTTPResponse)
+    @actual_res.should match_response(@expected_res, :streamed => streamed)
   end
 
-  def match_response(expected)
-    ResponseMatcher.new(expected)
+  def match_response(expected, streamed=false)
+    ResponseMatcher.new(expected, streamed)
   end
 
   class ResponseMatcher
-    def initialize(expected)
+    def initialize(expected, streamed=false)
       @expected = expected
+      @streamed = streamed
     end
 
     def matches?(actual)
@@ -88,8 +123,16 @@ def matches?(actual)
       actual_date = Time.parse(actual.delete('date').join)
       actual_date.should >= expected_date
       actual_date.should <= expected_date + 2
-      [:class, :code, :to_hash, :body].each do |i|
-        actual.send(i).should == @expected.send(i)
+
+      actual.class.should == @expected.class
+      actual.code.should == @expected.code
+      actual.to_hash.should == @expected.to_hash.merge({"connection" => ['close']})
+
+      if @streamed
+        actual.instance_variable_get(:@streamed_body).should ==
+          @expected.instance_variable_get(:@streamed_body)
+      else
+        actual.body.should == @expected.body
       end
       true
     end

spec/spec_helper.rb

@@ -2,12 +2,11 @@
 $LOAD_PATH.unshift(File.join(File.dirname(__FILE__), '..', 'lib'))
 require 'em-net-http'
 require 'time'
-require 'spec'
-require 'spec/autorun'
+require 'rspec'
 
 require 'mimic'
 
-Spec::Runner.configure do |config|
+RSpec.configure do |config|
   config.before(:all) do
     Mimic.mimic do
       Net::HTTPResponse::CODE_TO_OBJ.each do |code, klass|
@@ -16,6 +15,18 @@
 
       get('/hello').returning('Hello World!', 200, {'Content-Type'=>'text/plain'})
 
+      class BigImageResponse
+        def each
+          ::File.open('spec/image.jpg', "rb") { |file|
+            while part = file.read(8192)
+              yield part
+            end
+          }
+        end
+      end
+      resp = BigImageResponse.new
+      get('/image').returning(resp, 200, {"Content-Type" => 'image/jpeg'})
+
       post('/testpost') do
         "You said #{request.body.read}."
       end