Express stream handling via middlewares

(This is an intermediate commit for review-ability purposes).

Streaming downloads to files is core functionality that needs to go in
the base HTTP class, so remove special casing from middlewares for
decompressing responses.

5 files changed, 51 insertions, 5 deletions

diff --git a/lib/chef/http/authenticator.rb b/lib/chef/http/authenticator.rb
index 006b5d890a..489675ad66 100644
--- a/lib/chef/http/authenticator.rb
+++ b/lib/chef/http/authenticator.rb
@@ -48,6 +48,10 @@ class Chef
         [http_response, rest_request, return_value]
       end
 
+      def stream_response_handler(response)
+        nil
+      end
+
       def sign_requests?
         auth_credentials.sign_requests? && @sign_request
       end
diff --git a/lib/chef/http/cookie_manager.rb b/lib/chef/http/cookie_manager.rb
index 5c96b9e115..067dc3ed22 100644
--- a/lib/chef/http/cookie_manager.rb
+++ b/lib/chef/http/cookie_manager.rb
@@ -46,6 +46,11 @@ class Chef
         [http_response, rest_request, return_value]
       end
 
+      def stream_response_handler(response)
+        nil
+      end
+
+
     end
   end
 end
diff --git a/lib/chef/http/decompressor.rb b/lib/chef/http/decompressor.rb
index 846a29db0a..6010ffa698 100644
--- a/lib/chef/http/decompressor.rb
+++ b/lib/chef/http/decompressor.rb
@@ -28,6 +28,21 @@ class Chef
         def inflate(chunk)
           chunk
         end
+        alias :handle_chunk :inflate
+      end
+
+      class GzipInflater < Zlib::Inflate
+        def initialize
+          super(Zlib::MAX_WBITS + 16)
+        end
+        alias :handle_chunk :inflate
+      end
+
+      class DeflateInflater < Zlib::Inflate
+        def initialize
+          super
+        end
+        alias :handle_chunk :inflate
       end
 
       CONTENT_ENCODING  = "content-encoding".freeze
@@ -73,17 +88,17 @@ class Chef
 
       # This isn't used when this class is used as middleware; it returns an
       # object you can use to unzip/inflate a streaming response.
-      def stream_decompressor_for(response)
+      def stream_response_handler(response)
         if gzip_disabled?
           NoopInflater.new
         else
           case response[CONTENT_ENCODING]
           when GZIP
             Chef::Log.debug "decompressing gzip stream"
-            Zlib::Inflate.new(Zlib::MAX_WBITS + 16)
+            GzipInflater.new
           when DEFLATE
             Chef::Log.debug "decompressing inflate stream"
-            Zlib::Inflate.new
+            DeflateInflater.new
           else
             NoopInflater.new
           end
diff --git a/lib/chef/http/json_to_model_inflater.rb b/lib/chef/http/json_to_model_inflater.rb
index bcdc5647aa..3e941ad105 100644
--- a/lib/chef/http/json_to_model_inflater.rb
+++ b/lib/chef/http/json_to_model_inflater.rb
@@ -51,6 +51,10 @@ class Chef
         end
       end
 
+      def stream_response_handler(response)
+        nil
+      end
+
     end
   end
 end
diff --git a/lib/chef/rest.rb b/lib/chef/rest.rb
index f5a270248b..033153f739 100644
--- a/lib/chef/rest.rb
+++ b/lib/chef/rest.rb
@@ -194,6 +194,24 @@ class Chef
       BasicClient.new(create_url(url))
     end
 
+    class StreamHandler
+      def initialize(middlewares, response)
+        middlewares = middlewares.flatten
+        @stream_handlers = []
+        middlewares.each do |middleware|
+          stream_handler = middleware.stream_response_handler(response)
+          @stream_handlers << stream_handler unless stream_handler.nil?
+        end
+      end
+
+      def handle_chunk(next_chunk)
+        @stream_handlers.inject(next_chunk) do |chunk, handler|
+          handler.handle_chunk(chunk)
+        end
+      end
+
+    end
+
     private
 
     def stream_to_tempfile(url, response)
@@ -205,10 +223,10 @@ class Chef
       # Stolen from http://www.ruby-forum.com/topic/166423
       # Kudos to _why!
 
-      inflater = @decompressor.stream_decompressor_for(response)
+      stream_handler = StreamHandler.new(middlewares, response)
 
       response.read_body do |chunk|
-        tf.write(inflater.inflate(chunk))
+        tf.write(stream_handler.handle_chunk(chunk))
       end
       tf.close
       tf