Add each_line enumerator to IO class

lib/piperator.rb

@@ -51,4 +51,32 @@ def self.pipe(enumerable)
   def self.wrap(value)
     Pipeline.wrap(value)
   end
+
+  # Coerce any enumerator to be an IO (via pipe).
+  #
+  # Pro: infinite length without using infinite memory.  Con: unseekable (as is IO::Pipe).
+  #
+  # @param enumerator [Enumerator] source of data; infinite sources are OK
+  # @yieldparam io_r [IO] readable IO
+  def self.infinite_io(enumerator)
+    stop = false
+    ::IO.pipe do |io_r, io_w|  # not the IO from this library
+
+      # a thread writes all the data to the pipe.  the pipe automatically buffers everything for us
+      thr = Thread.new do
+        enumerator.each do |chunk|
+          break if stop
+          io_w.write(chunk)
+        end
+      ensure
+        io_w.close # otherwise a read will hang
+      end
+
+      yield io_r
+    ensure
+      stop = true
+      thr.join
+    end
+  end
+
 end

spec/piperator/infinite_io_spec.rb

@@ -0,0 +1,69 @@
+require 'spec_helper'
+
+
+
+RSpec.describe "Piperator.infinite_io" do
+  repeated_string = "foobar\n"
+  infinite_foobar = Enumerator.new { |y| loop { y << repeated_string } }
+
+  describe '#read' do
+    subject { proc { |&b| Piperator.infinite_io(infinite_foobar) { |x| b.(x) } } }
+
+    it 'reads specific number of bytes' do
+      subject.call do |io|
+        expect(io.read(4)).to eq('foob')
+      end
+    end
+
+    it 'buffers rest and returns on next read' do
+      subject.call do |io|
+        expect(io.read(2)).to eq('fo')
+        expect(io.read(2)).to eq('ob')
+        expect(io.read(2)).to eq('ar')
+      end
+    end
+
+    it 'does not try to reach the end before working' do
+      n = 100000
+      subject.call do |io|
+        expect(io.read(n * repeated_string.length)).to eq(repeated_string * n)
+      end
+    end
+  end
+
+  describe '#gets' do
+    it 'returns characters until the separator' do
+      Piperator.infinite_io(infinite_foobar) do |io|
+        expect(io.gets).to eq(repeated_string)
+      end
+    end
+
+    it 'responds to gets with nil when enumerable is exhausted' do
+      n = 2
+      Piperator.infinite_io(([repeated_string] * n).each) do |io|
+        n.times { expect(io.gets).to eq(repeated_string) }
+        expect(io.gets).to be_nil
+      end
+    end
+  end
+
+  describe '#eof?' do
+    it 'returns eof when enumerable is exhausted' do
+      n = 2
+      Piperator.infinite_io(([repeated_string] * n).each) do |io|
+        expect(io.eof?).to be_falsey
+        n.times { expect(io.gets).to eq(repeated_string) }
+        expect(io.eof?).to be_truthy
+      end
+    end
+  end
+
+  describe "#each_line" do
+    it 'reevaluates line breaks' do
+      Piperator.infinite_io(["foo\n", "bar\n", "baz\nbmp"].lazy.each) do |io|
+        expect(io.each_line.map(&:strip)).to eq(["foo", "bar", "baz", "bmp"])
+      end
+    end
+  end
+
+end