title: Is Ruby pipe streaming asynchronous?
body: <p>I have a <a href="http://en.wikipedia.org/wiki/MapReduce" rel="nofollow">map/reduce</a> streaming <a href="http://en.wikipedia.org/wiki/Pipeline_%28Unix%29" rel="nofollow">pipeline</a> written in Ruby which is behaving strangely. The pipeline looks like this:</p>
  
  <p><code>mapper | sort | reducer | expander | sort | splitter | uploader</code></p>
  
  <p>The mapper writes to STDOUT (via <a href="http://ruby-doc.org/core-1.9.3/Kernel.html#method-i-puts" rel="nofollow">puts</a>), the reducer reads from STDIN (via <a href="http://ruby-doc.org/core-1.9.3/ARGF.html#method-i-each" rel="nofollow">ARGF.each</a>) and writes to STDOUT (via puts) etc. etc.</p>
  
  <p>It seems like when the uploader is executed the files the splitter should've have created are not created yet. So the uploader doesn't upload anything.</p>
  
  <p>Here's my Pipeline class:</p>
  
  <pre><code>class Pipeline
  
    def run(context)
      raise ArgumentError, 'context is nil' unless context
      raise ArgumentError, 'context[:logger] is nil' unless context[:logger]
  
      current_path = File.dirname(__FILE__)
      mapper       = File.join(current_path, 'mapper.rb')
      reducer      = File.join(current_path, 'reducer.rb')
      expander     = File.join(current_path, 'expander.rb')
      splitter     = File.join(current_path, 'splitter.rb')
      uploader     = File.join(current_path, 'uploader.rb')
  
      mapper_args = context[:order_id] == nil ? nil : " #{context[:order_id]}"
  
      command_line = "ruby #{mapper}#{mapper_args} | sort | ruby #{reducer} | ruby #{expander} | sort | ruby #{splitter} | ruby #{uploader}"
  
      context[:logger].debug command_line
  
      %x{#{command_line}}
    end
  
  end
  </code></pre>
  
  <p>If piped streaming is asynchronous in Ruby, I'm wondering if doing what RubyMine does would fix this.  For example, before they run a Ruby script they prepend the command line like this:  <code>ruby -e $stdout.sync=true;$stderr.sync=true;load($0=ARGV.shift)</code>.</p>
  
  <p>I've updated my code with this technique, however, I want to know if this is correctly.  Or, if there is a better way?</p>
  
  <pre><code>class Pipeline
  
    def run(context)
      raise ArgumentError, 'context is nil' unless context
      raise ArgumentError, 'context[:logger] is nil' unless context[:logger]
  
      current_path = File.dirname(__FILE__)
      ruby         = 'ruby -e $stdout.sync=true;$stderr.sync=true;load($0=ARGV.shift)'
      mapper       = File.join(current_path, 'mapper.rb')
      reducer      = File.join(current_path, 'reducer.rb')
      expander     = File.join(current_path, 'expander.rb')
      splitter     = File.join(current_path, 'splitter.rb')
      uploader     = File.join(current_path, 'uploader.rb')
  
      mapper_args = context[:order_id] == nil ? nil : " #{context[:order_id]}"
  
      create_reports_command_line = "#{ruby} #{mapper}#{mapper_args} | sort | #{ruby} #{reducer} | #{ruby} #{expander} | sort | #{ruby} #{splitter}"
  
      context[:logger].debug create_reports_command_line
  
      %x{#{create_reports_command_line}}
  
      sleep 60 # Sleep for 1 min, just in case...
  
      upload_reports_command_line = "#{ruby} #{uploader}"
  
      context[:logger].debug upload_reports_command_line
  
      %x{#{upload_reports_command_line}}
    end
  
  end
  </code></pre>