improve EM::Stdout by using EM.attach

TODO.textile

@@ -2,7 +2,6 @@ h1. Todo
 
 h2. Features
 
-* Repository specific build configuration (.travis/config.yml), allow custom build scripts
 * Ability to log in w/ Github OAuth and subscribe to own/watched repositories, see https://github.com/blog/656-github-oauth2-support
 * Build history
 * Display queue

lib/em/stdout.rb

@@ -1,7 +1,23 @@
 require 'eventmachine'
 
-module EM
-  class Stdout
+module EventMachine
+  class << self
+    def split_stdout
+      stdout = nil
+      EM.next_tick do
+        stdout, read, write = STDOUT.clone, *IO.pipe
+        EM.attach(read, Stdout) do |connection|
+          connection.stdout = stdout
+          yield connection if block_given?
+        end
+        STDOUT.reopen(write)
+      end
+      sleep(0.01) until stdout
+      stdout
+    end
+  end
+
+  class Stdout < EventMachine::Connection
     class << self
       def output
         defined?(@@output) ? @@output : @@output = true
@@ -12,47 +28,19 @@ def output=(output)
       end
     end
 
-    attr_reader :read, :write, :stdout, :callback
+    attr_accessor :stdout
 
-    def initialize(&callback)
-      @read, @write = IO.pipe
-      $_stdout = @stdout = STDOUT.dup
-      @callback = callback
-      STDOUT.reopen(write)
-      EM.next_tick { pipe! }
+    def callback(&block)
+      @callback = block
     end
 
-    def split
-      yield
-      sleep(1) # TODO get rid of this
-      close
-    rescue Exception => e
-      close
-      stdout.puts e.message
-      e.backtrace.each { |line| stdout.puts line }
-    # ensure
-    #   sleep(0.1) until read.eof?
+    def receive_data(data)
+      stdout.print(data) if self.class.output
+      @callback.call(data) if @callback
     end
 
-    def close
+    def unbind
       STDOUT.reopen(stdout)
-      write.close if write && !write.closed?
     end
-
-    protected
-
-      def pipe!
-        unless read.eof?
-          data = read.readpartial(1024)
-          stdout << data if self.class.output
-          callback.call(data)
-          sleep(0.5)
-        end
-      rescue Exception => e
-        stdout.puts e.message
-        e.backtrace.each { |line| stdout.puts line }
-      ensure
-        EM.next_tick { pipe! }
-      end
   end
 end

lib/travis/builder.rb

@@ -27,8 +27,9 @@ def perform(meta_id, payload)
           sleep(0.01) until EM.reactor_running?
           EM.defer do
             new(meta_id, payload).work!
-            sleep(1) # TODO get rid of this
-            EM.stop
+            EM.add_timer(1) do # TODO get rid of this
+              EM.stop
+            end
           end
         end
       end

lib/travis/builder/base.rb

@@ -11,7 +11,8 @@ def initialize(meta_id, build)
       def work!
         on_start
         build['status'] = buildable.build!
-        sleep(1)
+        puts "\nDone. Build script exited with: #{build['status']}"
+        # sleep(1)
         on_finish
       end
 

lib/travis/builder/pusher.rb

@@ -23,8 +23,9 @@ def on_finish
 
         def push(event, data)
           channel = :"repository_#{repository_id}"
-          $_stdout.puts "Pusher: notifying channel #{channel} about #{event}: #{data.inspect}"
+          # stdout.puts "Pusher: notifying channel #{channel} about #{event}: #{data.inspect}"
           pusher(channel).trigger(event, data)
+          sleep(0.1) # TODO how to better synchronize websocket messages
         end
 
         def pusher(channel)

lib/travis/builder/stdout.rb

@@ -3,13 +3,13 @@
 module Travis
   class Builder
     module Stdout
+      attr_reader :stdout
+
       def work!
-        stdout = EM::Stdout.new do |chars|
-          on_log(chars)
-        end
-        stdout.split do
-          super
+        @stdout = EM.split_stdout do |c|
+          c.callback { |data| on_log(data) }
         end
+        super
       end
     end
   end

play/em_stdout.rb

@@ -0,0 +1,52 @@
+require 'rubygems'
+require 'em/stdout'
+
+EventMachine.module_eval do
+  def self.split_stdout(&block)
+    stdout = nil
+    EM.next_tick do
+      read, write = IO.pipe
+      stdout = STDOUT.clone
+      EM.attach(read, Splitter) { |c|
+        c.stdout = stdout
+        c.callback = block
+      }
+      STDOUT.reopen(write)
+    end
+    sleep(0.01) until stdout
+    stdout
+  end
+
+  class Splitter < EventMachine::Connection
+    attr_accessor :stdout, :callback
+
+    def receive_data(data)
+      callback.call(data)
+    end
+
+    def unbind
+      STDOUT.reopen(stdout)
+      stdout.puts "unbound"
+    end
+  end
+end
+
+
+STDOUT.sync = true
+
+EM.run do
+  EM.defer do
+    stdout = EM.split_stdout do |data|
+      stdout.puts '--: ' + data.inspect
+      sleep(1)
+    end
+
+    stdout.puts 'starting'
+
+    10.times { puts 'output ... '; sleep(0.2) }
+
+    stdout.puts 'stopping'
+    EM.stop
+  end
+end
+puts 'fin'

test/em/stdout_test.rb

@@ -8,10 +8,11 @@ class EMStdoutTest < Test::Unit::TestCase
     EM::Stdout.output = false
 
     EM.run do
-      stream = EM::Stdout.new { |data| out << data }
       EM.defer do
+        EM.split_stdout do |c|
+          c.callback { |data| out << data }
+        end
         print '.'
-        stream.close
         EM.stop
       end
     end

test/travis/builder/stdout_test.rb

@@ -1,7 +1,6 @@
 require 'test_helper'
 require 'travis/builder'
 require 'travis/builder/stdout'
-require 'em/stdout'
 
 STDOUT.sync = true