gistfile1.rb

require 'rubygems'
require 'eventmachine'
require 'em-http'
require 'dataflow'

Thread.new {
  EM.run{}
}

module EventMachine
  module DataflowDeferrable 
    include Dataflow

    def succeded?
      Dataflow::barrier(results)
      @deferred_status == :succeeded
    end

    def failed?
      Dataflow::barrier(results)
      @deferred_status == :failed
    end

    def results
      @results ||= Variable.new
    end

    def set_deferred_status(status, *args)
      @deferred_status = status
      unify results, @deferred_args
    end
  end
end

class EventMachine::HttpClient
  include EventMachine::DataflowDeferrable
end

def dataflow_http
  results = %w(http://www.yahoo.com/ http://www.slashdot.org/ http://www.github.com/ http://www.oib.com/).map do |url|
    EventMachine::HttpRequest.new(url).get
  end

  results.each do |http|
    if http.succeded?
      p http.response
    else
      p "bad stuff happened"
    end
  end
end

dataflow_http