Prototype reactive API for moon

reactive.rb

module Reactive
  module Observable
    def subscribe(listener)
      listeners.push listener
      listener
    end

    def unsubscribe(listener)
      listeners.delete listener
    end

    def notify(object)
      listeners.each do |listener|
        listener.call object
      end
    end
  end

  module Reactable
    include Observable

    def filter(&block)
      subscribe Filter.new(&block)
    end

    def select(&block)
      subscribe Selector.new(&block)
    end

    def reject(&block)
      subscribe Rejector.new(&block)
    end

    def map(&block)
      subscribe Mapper.new(&block)
    end

    def buffer(length)
      subscribe Buffer.new(length)
    end

    def accumulate(length)
      subscribe Accumulator.new(length)
    end

    def case(clause)
      select { |value| clause === value }
    end

    def eq(other)
      select { |value| other == value }
    end

    def not_eq(other)
      reject { |value| other == value }
    end
  end

  class Reactor
    include Reactable

    # @api
    attr_accessor :listeners

    def initialize
      @listeners = []
    end

    def call(value)
      @last_value = value
      invoke value
    end
  end

  class Accumulator < Reactor
    attr_reader :length

    def initialize(length)
      super()
      @length = length
      @b = []
    end

    def flush
      notify @b
    end

    def invoke(value)
      @b << value
      if @b.size >= @length
        flush
        @b.clear
      end
    end
  end

  class Buffer < Accumulator
    def flush
      @b.each do |e|
        notify e
      end
    end
  end

  class Filter < Reactor
    def initialize(&block)
      super()
      @func = block
    end

    def invoke(value)
      @func.call value
      notify value
    end
  end

  class Selector < Filter
    def invoke(value)
      notify value if @func.call value
    end
  end

  class Rejector < Filter
    def invoke(value)
      notify value unless @func.call value
    end
  end

  class Mapper < Reactor
    def invoke(value)
      notify @func.call(value)
    end
  end
end