Ruby de Streem

example.rb

require "./streem"
using CoreExt

STDIN | STDOUT

seq(100) | ->x{ 
  if x % 15 == 0
    'FizzBuzz'
  elsif x % 3 == 0
    'Fizz'
  elsif x % 5 == 0
    'Buzz'
  else
    x
  end
} | STDOUT

require "prime"
(1..100) | ->x{ x if x.prime? } | :compact | RUN # => [2, 3, 5, 7, 11, 13, 17, 19, 23, 29, 31, 37, 41, 43, 47, 53, 59, 61, 67, 71, 73, 79, 83, 89, 97]

"hello, world of ruby" | :split | ->w{ w.capitalize } | RUN # => ["Hello,", "World", "Of", "Ruby"]

streem.rb

class Streem
  def initialize(obj)
    @obj = obj
    @procs = []
  end

  def |(proc)
    return run if proc==RUN
    return puts run if proc==STDOUT
    @procs << proc
    self
  end

  def run
    @procs.inject(@obj) do |obj, proc|
      case proc
      when Symbol
        obj.send proc
      else
        case obj
        when Enumerable
          thread_with(obj) { |e| proc.call e }
        else
          proc.call(obj)
        end
      end
    end
  end

  private
  def thread_with(enum)
    mem = []
    enum.map.with_index do |*item, i|
      Thread.new(*item) do |*_item|
        mem << [i, yield(*_item)]
      end
    end.each(&:join)
    mem.sort.map(&:last)
  end
end

module CoreExt
  [Object, Array].each do |klass|
    refine klass do
      def |(other)
        return super unless [Proc, Symbol].any? { |k| other.is_a? k }
        Streem.new(self) | other
      end
    end
  end

  def STDIN.|(other)
    Streem.new(ARGF.readlines) | other
  end

  refine Object do
    def seq(last)
      (1..last)
    end
    
    Object::RUN = nil
  end
end