rxjs.rb

gistfile1.txt

class Rx
  class Subject
    def initialize
      @pipes = []
    end

    def next(object)
      threads = @pipes.map { |pipe| Thread.new { pipe.push object } }
      threads.each(&:join)
    end

    def pipe(*operators)
      pipe = Pipe.new *operators
      @pipes << pipe
      pipe
    end
  end

  class Pipe
    def initialize(*operators)
      @operators = operators
    end

    def subscribe(&block)
      @subscription = Subscriber.new block
    end

    def push(object)
      return unless @subscription
      latest = @operators.reduce(object) { |subject, operator| subject = operator.(subject) }
      @subscription.push latest
    end
  end

  class Subscriber
    def initialize(block)
      @procedure = block || Proc.new {}
    end

    def push(object)
      @procedure.(object)
    end
  end

  class Operator
    def self.tap
      ->(input) do
        yield input
        input
      end
    end

    def self.map
      ->(input) { yield input }
    end
  end
end

require 'bcrypt'

subject = Rx::Subject.new

def generate_password(secret)
  BCrypt::Password.create(secret, :cost => (10..15).to_a.sample)
end

p1 = Proc.new do |secret|
  new_secret = generate_password secret
  puts "p1 -> #{new_secret}"
  new_secret
end

p2 = Proc.new do |secret|
  new_secret = generate_password secret
  puts "p2 -> #{new_secret}"
  new_secret
end

subject.pipe(
  Rx::Operator.tap { |secret| puts "p1 init: #{secret}" },
  Rx::Operator.map(&p1),
  Rx::Operator.map(&p1),
  Rx::Operator.map(&p1),
  Rx::Operator.tap { |secret| puts "tap p1: #{secret}" },
  Rx::Operator.map(&p1),
  Rx::Operator.map(&p1),
  Rx::Operator.map(&p1),
).subscribe { |result| puts "p1 result: #{result}" }

subject.pipe(
  Rx::Operator.tap { |secret| puts "p2 init: #{secret}" },
  Rx::Operator.map(&p2),
  Rx::Operator.map(&p2),
  Rx::Operator.map(&p2),
  Rx::Operator.tap { |secret| puts "tap p2: #{secret}" },
  Rx::Operator.map(&p2),
  Rx::Operator.map(&p2),
  Rx::Operator.map(&p2),
).subscribe { |result| puts "p2 result: #{result}" }