named_pipes2.rb

#!/usr/bin/env ruby

require 'parallel'
require 'pp'

class FIFO
  def self.pipe(path)
    unless File.exists? path
      system("mkfifo #{path}")

      raise "mkfifo #{path} failed" unless $?.success?
    end

    reader = File.open(path, 'r+')
    writer = File.open(path, 'w+')

    [reader, writer]
  end
end

class Pipe
  def initialize
    @commands = []
  end

  def add(command, options = {})
    @commands << Command.new(command, options)

    self
  end

  def run
    count = 0

    @commands.each_cons(2) do |c_in, c_out|
      path = "pipe_#{count}"

      reader, writer = FIFO.pipe(path)

      c_out.input = reader
      c_in.output = writer

      count += 1
    end

    Parallel.each(@commands, in_processes: @commands.size) do |command|
      command.run
#      command.input.close
#      command.output.close
    end

    self
  end

  class Command
    attr_accessor :input, :output

    def initialize(command, options)
      @command = command
      @options = options
      @input   = nil
      @output  = nil
    end

    def run
      send @command
    end

    def cat
      @input.each { |record| break if record.chomp == "\0"; @output.puts record } if @input

      File.open(@options[:input]) do |ios|
        ios.each { |record| @output.puts record }
        @output.puts "\0"
      end
    end

    def count
      count = 0

      @input.each { |record| break if record.chomp == "\0"; count += 1 }
      puts count

      if @output
        @output.puts count
        @otuput.puts "\0"
      end
    end

    def dump
      @input.each do |record|
        break if record.chomp == "\0"
        puts record

        @output.write record if @output
      end

      @output.puts "\0" if @output
    end
  end
end

p = Pipe.new
#p.add(:cat, input: "foo.tab").add(:dump).add(:cat, input: "table.txt").add(:dump)
#p.add(:cat, input: "table.txt").add(:cat, input: "foo.tab").add(:dump)
p.add(:cat, input: "big.tab").add(:cat, input: "table.txt").add(:dump)
#p.add(:cat, input: "table.txt").add(:dump)
p.run