allow aborting early while streaming

lib/posix/spawn.rb

@@ -337,6 +337,10 @@ class MaximumOutputExceeded < StandardError
     class TimeoutExceeded < StandardError
     end
 
+    # Exception raised when output streaming is aborted early.
+    class Aborted < StandardError
+    end
+
     private
 
     # Turns the various varargs incantations supported by Process::spawn into a

lib/posix/spawn/child.rb

@@ -260,21 +260,26 @@ def read_and_write(input, stdin, stdout, stderr, timeout=nil, max=nil)
               fd.close
             end
 
+            abort = false
             if chunk
               if fd == stdout
                 if @streaming && @stdout_block
-                  @stdout_block.call(chunk)
+                  abort = !!@stdout_block.call(chunk)
                 else
                   @out << chunk
                 end
               else
                 if @streaming && @stderr_block
-                  @stderr_block.call(chunk)
+                  abort = !!@stderr_block.call(chunk)
                 else
                   @err << chunk
                 end
               end
             end
+
+            if @streaming && abort
+              raise Aborted
+            end
           end
 
           # keep tabs on the total amount of time we've spent here

test/test_child.rb

@@ -228,6 +228,8 @@ def test_streaming_stdout
     stdout_buf = ""
     stdout_stream = Proc.new do |chunk|
       stdout_buf << chunk
+
+      false
     end
 
     input = "hello!"
@@ -243,17 +245,44 @@ def test_streaming_stderr
     stderr_buf = ""
     stderr_stream = Proc.new do |chunk|
       stderr_buf << chunk
+
+      false
     end
 
-    input = "hello!"
-    p = Child.new('ls', '-?', :input => input, :streams => {
+    p = Child.new('ls', '-?', :streams => {
       :stderr => stderr_stream
     })
 
     refute p.success?
     refute stderr_buf.empty?
   end
 
+  def test_streaming_stdout_aborted
+    stdout_stream = Proc.new do |chunk|
+      true
+    end
+
+    input = "hello!"
+    assert_raises POSIX::Spawn::Aborted do
+      p = Child.new('cat', :input => input, :streams => {
+        :stdout => stdout_stream
+      })
+    end
+  end
+
+  def test_streaming_stderr_aborted
+    stderr_stream = Proc.new do |chunk|
+      true
+    end
+
+    input = "hello!"
+    assert_raises POSIX::Spawn::Aborted do
+      p = Child.new('ls', '-?', :streams => {
+        :stderr => stderr_stream
+      })
+    end
+  end
+
   ##
   # Assertion Helpers