wip: interactive mode

lib/foreman/buffer.rb

@@ -0,0 +1,38 @@
+ANSI_TOKEN = /\e\[(?:\??\d{1,4}(?:;\d{0,4})*)?[A-Za-z]|\e=|\e>/
+NEWLINE_TOKEN = /\n/
+TOKENIZER = Regexp.new("(#{ANSI_TOKEN}|#{NEWLINE_TOKEN})")
+
+class Buffer
+  @buffer = ''
+
+  def initialize(initial = '')
+    @buffer = initial.dup
+    @fd = File.open("/tmp/buffer.#{initial}.log", "w+")
+    @fd.sync = true
+  end
+
+  def each_token
+    remainder = ''
+    @fd.puts @buffer.split(TOKENIZER).inspect
+    @buffer.split(TOKENIZER).each do |token|
+      if token.include?("\e") && !token.match(ANSI_TOKEN)
+        remainder << token
+      else
+        yield token unless token.empty?
+      end
+    end
+    @buffer = remainder
+  end
+
+  def gets
+    return nil unless @buffer.include?("\n")
+
+    line, @buffer = @buffer.split("\n", 2)
+    line
+  end
+
+  def write(data)
+    @buffer << data
+    @fd.puts "write: #{data.inspect}"
+  end
+end

lib/foreman/cli.rb

@@ -19,12 +19,13 @@ class Foreman::CLI < Foreman::Thor
 
   desc "start [PROCESS]", "Start the application (or a specific PROCESS)"
 
-  method_option :color,     :type => :boolean, :aliases => "-c", :desc => "Force color to be enabled"
-  method_option :env,       :type => :string,  :aliases => "-e", :desc => "Specify an environment file to load, defaults to .env"
-  method_option :formation, :type => :string,  :aliases => "-m", :banner => '"alpha=5,bar=3"', :desc => 'Specify what processes will run and how many. Default: "all=1"'
-  method_option :port,      :type => :numeric, :aliases => "-p"
-  method_option :timeout,   :type => :numeric, :aliases => "-t", :desc => "Specify the amount of time (in seconds) processes have to shutdown gracefully before receiving a SIGKILL, defaults to 5."
-  method_option :timestamp, :type => :boolean, :default => true, :desc => "Include timestamp in output"
+  method_option :color,       :type => :boolean, :aliases => "-c",  :desc => "Force color to be enabled"
+  method_option :env,         :type => :string,  :aliases => "-e",  :desc => "Specify an environment file to load, defaults to .env"
+  method_option :formation,   :type => :string,  :aliases => "-m",  :banner => '"alpha=5,bar=3"', :desc => 'Specify what processes will run and how many. Default: "all=1"'
+  method_option :interactive, :type => :string,  :aliases => "-i",  :desc => "Run a process interactively"
+  method_option :port,        :type => :numeric, :aliases => "-p"
+  method_option :timeout,     :type => :numeric, :aliases => "-t",  :desc => "Specify the amount of time (in seconds) processes have to shutdown gracefully before receiving a SIGKILL, defaults to 5."
+  method_option :timestamp,   :type => :boolean, :default => false, :desc => "Include timestamp in output"
 
   class << self
     # Hackery. Take the run method away from Thor so that we can redefine it.

lib/foreman/engine.rb

@@ -1,7 +1,9 @@
 require "foreman"
+require "foreman/buffer"
 require "foreman/env"
 require "foreman/process"
 require "foreman/procfile"
+require "pty"
 require "tempfile"
 require "fileutils"
 require "thread"
@@ -30,16 +32,18 @@ def initialize(options={})
     @options[:formation] ||= "all=1"
     @options[:timeout] ||= 5
 
+    @buffers   = {}
     @env       = {}
     @mutex     = Mutex.new
     @names     = {}
+    @prefixed  = {}
     @processes = []
     @running   = {}
     @readers   = {}
     @shutdown  = false
 
     # Self-pipe for deferred signal-handling (ala djb: http://cr.yp.to/docs/selfpipe.html)
-    reader, writer       = create_pipe
+    reader, writer       = self.class.create_pipe
     reader.close_on_exec = true if reader.respond_to?(:close_on_exec)
     writer.close_on_exec = true if writer.respond_to?(:close_on_exec)
     @selfpipe            = { :reader => reader, :writer => writer }
@@ -148,6 +152,7 @@ def handle_signal_forward(signal)
   def register(name, command, options={})
     options[:env] ||= env
     options[:cwd] ||= File.dirname(command.split(" ").first)
+    options[:interactive] ||= @options[:interactive] == name
     process = Foreman::Process.new(command, options)
     @names[process] = name
     @processes << process
@@ -307,7 +312,7 @@ def shutdown
 
 ## Helpers ##########################################################
 
-  def create_pipe
+  def self.create_pipe
     IO.method(:pipe).arity.zero? ? IO.pipe : IO.pipe("BINARY")
   end
 
@@ -320,6 +325,10 @@ def name_for_index(process, index)
     [ @names[process], index.to_s ].compact.join(".")
   end
 
+  def process_for(reader)
+    @running[@readers.invert[reader]].first
+  end
+
   def parse_formation(formation)
     pairs = formation.to_s.gsub(/\s/, "").split(",")
 
@@ -350,30 +359,26 @@ def termination_message_for(status)
     end
   end
 
-  def flush_reader(reader)
-    until reader.eof?
-      data = reader.gets
-      output_with_mutex name_for(@readers.key(reader)), data
-    end
-  end
-
 ## Engine ###########################################################
 
   def spawn_processes
     @processes.each do |process|
       1.upto(formation[@names[process]]) do |n|
-        reader, writer = create_pipe
         begin
-          pid = process.run(:output => writer, :env => {
-            "PORT" => port_for(process, n).to_s,
-            "PS" => name_for_index(process, n)
-          })
-          writer.puts "started with pid #{pid}"
+          pid = process.run(
+            env: {
+              'PORT' => port_for(process, n).to_s,
+              'PS' => name_for_index(process, n)
+            }
+          )
+          # writer.puts "started with pid #{pid}"
         rescue Errno::ENOENT
-          writer.puts "unknown command: #{process.command}"
+          # writer.puts "unknown command: #{process.command}"
         end
+        @buffers[process.reader] = Buffer.new(@names[process])
+        @prefixed[process.reader] = false
         @running[pid] = [process, n]
-        @readers[pid] = reader
+        @readers[pid] = process.reader
       end
     end
   end
@@ -395,14 +400,64 @@ def handle_io(readers)
       next if reader == @selfpipe[:reader]
 
       if reader.eof?
-        @readers.delete_if { |key, value| value == reader }
+        @buffers.delete(reader)
+        @readers.delete_if { |_key, value| value == reader }
+      elsif process_for(reader).interactive?
+        handle_io_interactive reader
       else
-        data = reader.gets
-        output_with_mutex name_for(@readers.invert[reader]), data
+        handle_io_noninteractive reader
       end
     end
   end
 
+  def handle_io_interactive(reader)
+    done = false
+    name = name_for(@readers.invert[reader])
+    indent = prefix(name).gsub(ANSI_TOKEN, "").length
+
+    loop do
+      @buffers[reader].write(reader.read_nonblock(10))
+
+      @buffers[reader].each_token do |token|
+        case token
+        when /^\e\[(\d+)G$/
+          output_partial "\e[#{Regexp.last_match(1).to_i + indent}G"
+        when ANSI_TOKEN
+          output_partial token
+        when "\r"
+          output_partial "\e[#{indent+1}G"
+        when "\n"
+          output_partial "\r\n"
+          @prefixed[reader] = false
+        else
+          unless @prefixed[reader]
+            output_partial "\e[1G"
+            output_partial prefix(name)
+            @prefixed[reader] = true
+          end
+          output_partial token
+        end
+        done = (token == "\n")
+      end
+    rescue IO::WaitReadable
+      retry if IO.select([reader], [], [], 1)
+      return if done
+    rescue EOFError
+    end
+  end
+
+  def handle_io_noninteractive(reader)
+    @buffers[reader].write(reader.read_nonblock(4096))
+    while line = @buffers[reader].gets
+      output_with_mutex name_for(@readers.invert[reader]), line
+    end
+  end
+
+  def output_prefix(reader)
+    output_partial prefix(name_for(@readers.invert[reader]))
+    @prefixed[reader] = true
+  end
+
   def watch_for_output
     Thread.new do
       begin

lib/foreman/engine/cli.rb

@@ -1,4 +1,5 @@
 require "foreman/engine"
+require "io/console"
 
 class Foreman::Engine::CLI < Foreman::Engine
 
@@ -55,20 +56,30 @@ def startup
 
   def output(name, data)
     data.to_s.lines.map(&:chomp).each do |message|
-      output  = ""
-      output += $stdout.color(@colors[name.split(".").first].to_sym)
-      output += "#{Time.now.strftime("%H:%M:%S")} " if options[:timestamp]
-      output += "#{pad_process_name(name)} | "
-      output += $stdout.color(:reset)
-      output += message
-      $stdout.puts output
+      $stdout.write prefix(name)
+      $stdout.puts message + "\r"
       $stdout.flush
     end
   rescue Errno::EPIPE
     terminate_gracefully
   end
 
+  def output_partial(data)
+    $stdout.write data
+    $stdout.flush
+  end
+
+  def prefix(name)
+    output  = ''
+    output += $stdout.color(@colors[name.split('.').first].to_sym)
+    output += "#{Time.now.strftime('%H:%M:%S')} " if options[:timestamp]
+    output += "#{pad_process_name(name)} | "
+    output += $stdout.color(:reset)
+    output
+  end
+
   def shutdown
+    $stdin.cooked!
   end
 
 private

lib/foreman/process.rb

@@ -1,10 +1,18 @@
 require "foreman"
+require "io/console"
 require "shellwords"
 
 class Foreman::Process
 
+  @noninteractive_stdin = $stdin
+
+  class << self
+    attr_accessor :noninteractive_stdin
+  end
+
   attr_reader :command
   attr_reader :env
+  attr_reader :reader
 
   # Create a Process
   #
@@ -19,6 +27,8 @@ def initialize(command, options={})
     @options = options.dup
 
     @options[:env] ||= {}
+
+    self.class.noninteractive_stdin = :close if options[:interactive]
   end
 
   # Get environment-expanded command for a +Process+
@@ -40,18 +50,31 @@ def expanded_command(custom_env={})
   #
   # @param [Hash] options
   #
-  # @option options :env    ({})       Environment variables to set for this execution
-  # @option options :output ($stdout)  The output stream
+  # @option options :env ({}) Environment variables to set for this execution
   #
   # @returns [Fixnum] pid  The +pid+ of the process
   #
   def run(options={})
     env    = @options[:env].merge(options[:env] || {})
-    output = options[:output] || $stdout
     runner = "#{Foreman.runner}".shellescape
-
-    Dir.chdir(cwd) do
-      Process.spawn env, expanded_command(env), :out => output, :err => output
+
+    if interactive?
+      $stdin.raw!
+      @reader, tty = PTY.open
+      Thread.new do
+        loop do
+          data = $stdin.readpartial(4096)
+          if data.include?("\03")
+            Process.kill("INT", Process.pid)
+            data.gsub!("\03", "")
+          end
+          @reader.write(data)
+        end
+      end
+      Process.spawn env, expanded_command(env), chdir: cwd, in: tty, out: tty, err: tty
+    else
+      @reader, writer = Foreman::Engine::create_pipe
+      Process.spawn env, expanded_command(env), chdir: cwd, in: self.class.noninteractive_stdin, out: writer, err: writer
     end
   end
 
@@ -77,4 +100,7 @@ def cwd
     File.expand_path(@options[:cwd] || ".")
   end
 
+  def interactive?
+    @options[:interactive]
+  end
 end