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,28 @@ def ==(other)
 
       private
 
+      def rewind(io)
+        io.rewind if io.respond_to? :rewind
+      rescue Errno::ESPIPE, Errno::EPIPE
+        # Pipe IOs respond to `:rewind` but fail when you call it.
+        #
+        # Calling `IO#rewind` on a pipe, fails with *ESPIPE* on MRI,
+        # but *EPIPE* on jRuby.
+        #
+        # - **ESPIPE** -- "Illegal seek."
+        #   Invalid seek operation (such as on a pipe).
+        #
+        # - **EPIPE** -- "Broken pipe."
+        #   There is no process reading from the other end of a pipe. Every
+        #   library function that returns this error code also generates
+        #   a SIGPIPE signal; this signal terminates the program if not handled
+        #   or blocked. Thus, your program will never actually see EPIPE unless
+        #   it has handled or blocked SIGPIPE.
+        #
+        # See: https://www.gnu.org/software/libc/manual/html_node/Error-Codes.html
+        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 }