actual jsonlio and cborio from idm_mon imported

lib/cborio.rb

@@ -1,54 +1,89 @@
+# vim: set noet sw=2 ts=2 sts=2:
+
 require 'cbor'
+require 'socket'
 
-class CBORIO
+class CBORIO <IO
-  include Enumerable
+	include Enumerable
-  def orig_io() @io end
+	extend Enumerable
-  def unpacker() @un end
+	def orig_io() @io end
-  def close() @io.close end
+	def to_io() self end
-  def write( obj) @io.write CBOR.pack( obj) end
+	def to_i() @io.to_i end
-  def sync() @io.sync end
+	def fileno() @io.fileno end
-  def sync( v) @io.sync = v end
+	def unpacker() @un end
-  def self.pipe() IO.pipe.map &self.method(:new) end
+	def close() @io.close end
-  def self.popen &exe
+	def close_write() @io.close_write end
-    rd, wr = self.pipe
+	def close_read() @io.close_read end
-    pid =
+	def write( obj) @io.write CBOR.pack( obj) end
-      fork do
+	def sync() @io.sync end
-        rd.close
+	def sync=( v) @io.sync = v end
-        yield wr
+	def inspect() "#<#{self.class.name} fd:#{@io.to_i} #{@mode}>" end
-        exit 0
-      end
-    wr.close
-    [pid, rd]
-  end
 
-  def initialize io
+	class <<self
-    @io, @un = io, CBOR::Unpacker.new
+		def stream() Socket.pair( :UNIX, :STREAM, 0).map {|s| self.new s, :rw } end
-    @io.sync = true
-  end
 
-  def self.open io, &exe
+		def pipe
-    if block_given?
+			a, b = IO.pipe
-      s = self.new io
+			[self.new( a, :r), self.new( b, :w)]
-      begin yield s
+		end
-      ensure s.close
-      end
-    else self.new io
-    end
-  end
 
-  def read
+		def popen mode = nil, &exe
-    return @un.read  rescue EOFError
+			s1, s2 =
-    loop do
+				case mode
-      @un.feed @io.readpartial( 1024)
+				when :rw, 'rw', nil then self.stream
-      return @un.read  rescue EOFError
+				when :r, 'r' then self.pipe
-    end
+				when :w, 'w' then self.pipe.reverse
-  end
+				end
+			pid =
+				fork do
+					s1.close
+					yield s2
+					exit 0
+				end
+			s2.close
+			[pid, s1]
+		end
 
-  def each &exe
+		def open io, mode = :r, &exe
-    return to_enum __method__  unless block_given?
+			s = self.new io, mode
-    loop { yield read }
+			if block_given?
-    self
+				begin yield s
-  rescue EOFError
+				ensure s.close
-    self
+				end
-  end
+			else s
+			end
+		end
+
+		def each fn, &exe
+			open fn, 'r' do |f|
+				f.each &exe
+			end
+		end
+	end
+
+	def initialize io, mode = :rw
+		case io
+		when IO then @io = io
+		when String then @io = File.open io, mode
+		when Pathname then @io = io.open mode
+		else raise ArgumentError, "This seems not to be a IO/String/Pathname: #{io.inspect}"
+		end
+		@mode, @un, @io.sync = mode, CBOR::Unpacker.new, true
+	end
+
+	def read
+		return @un.read  rescue EOFError
+		loop do
+			@un.feed @io.readpartial( 1024)
+			return @un.read  rescue EOFError
+		end
+	end
+
+	def each &exe
+		return to_enum __method__  unless block_given?
+		loop { yield read }
+		self
+	rescue EOFError
+		self
+	end
 end

lib/jsonlio.rb

@@ -0,0 +1,90 @@
+# vim: set noet sw=2 ts=2 sts=2:
+
+require 'json'
+require 'socket'
+
+class JSONLIO <IO
+	include Enumerable
+	extend Enumerable
+	def orig_io() @io end
+	def to_io() self end
+	def to_i() @io.to_i end
+	def fileno() @io.fileno end
+	def unpacker() @un end
+	def close() @io.close end
+	def close_write() @io.close_write end
+	def close_read() @io.close_read end
+	def write( obj) @io.write obj.to_json+"\n" end
+	def sync() @io.sync end
+	def sync=( v) @io.sync = v end
+	def inspect() "#<#{self.class.name} fd:#{@io.to_i} #{@mode}>" end
+	def initialize( io) @io = io end
+
+	class <<self
+		def stream() Socket.pair( :UNIX, :STREAM, 0).map {|s| self.new s, :rw } end
+
+		def pipe
+			a, b = IO.pipe
+			[self.new( a, :r), self.new( b, :w)]
+		end
+
+		def popen mode = nil, &exe
+			s1, s2 =
+				case mode
+				when :rw, 'rw', nil then self.stream
+				when :r, 'r' then self.pipe
+				when :w, 'w' then self.pipe.reverse
+				end
+			pid =
+				fork do
+					s1.close
+					yield s2
+					exit 0
+				end
+			s2.close
+			[pid, s1]
+		end
+
+		def open io, mode = :r, &exe
+			s = self.new io, mode
+			if block_given?
+				begin yield s
+				ensure s.close
+				end
+			else s
+			end
+		end
+
+		def each fn, &exe
+			open fn, 'r' do |f|
+				f.each &exe
+			end
+		end
+	end
+
+	attr_reader :lineno
+
+	def initialize io, mode = :rw
+		case io
+		when IO then @io = io
+		when String then @io = File.open io, mode
+		when Pathname then @io = io.open mode
+		else raise ArgumentError, "This seems not to be a IO/String/Pathname: #{io.inspect}"
+		end
+		@lineno, @mode, @io.sync = 0, mode, true
+	end
+
+	def read
+		JSON.parse( @io.readline.chomp).tap { @lineno += 1 }
+	rescue Exception
+		raise $!.class, $!.message + " (JSONL-line: #{@lineno})", $!.backtrace
+	end
+
+	def each &exe
+		return to_enum __method__  unless block_given?
+		loop { yield read }
+		self
+	rescue EOFError
+		self
+	end
+end