Fix pipes support in bodies.

lib/http/request/body.rb

@@ -35,10 +35,12 @@ def each(&block)
           yield @source
         elsif @source.respond_to?(:read)
           IO.copy_stream(@source, ProcIO.new(block))
-          @source.rewind if @source.respond_to?(:rewind)
+          rewind(@source)
         elsif @source.is_a?(Enumerable)
           @source.each(&block)
         end
+
+        self
       end
 
       # Request bodies are equivalent when they have the same source.
@@ -48,6 +50,13 @@ def ==(other)
 
       private
 
+      def rewind(io)
+        io.rewind if io.respond_to? :rewind
+      rescue Errno::ESPIPE
+        # Pipe IOs respond to `:rewind` but fail when you call it.
+        nil
+      end
+
       def validate_source_type!
         return if @source.is_a?(String)
         return if @source.respond_to?(:read)

spec/lib/http/request/body_spec.rb

@@ -125,6 +125,22 @@
       end
     end
 
+    context "when body is a pipe" do
+      let(:ios)  { IO.pipe }
+      let(:body) { ios[0] }
+
+      before do
+        Thread.new(ios[1]) do |io|
+          16_384.times { io << "abcdef" }
+          io.close
+        end
+      end
+
+      it "yields chunks of content" do
+        expect(chunks.inject("", :+)).to eq("abcdef" * 16_384)
+      end
+    end
+
     context "when body is an Enumerable IO" do
       let(:data) { "a" * 16 * 1024 + "b" * 10 * 1024 }
       let(:body) { StringIO.new data }