require 'open3' require 'io/nonblock' require 'io/wait' require 'tmpdir' module Session #{{{ VERSION = '2.1.7' class << self #{{{ attr :use_spawn, true attr :debug, true def new(*args, &block) #{{{ Sh::new(*args, &block) #}}} end alias [] new #}}} end @use_spawn = ENV['SESSION_USE_SPAWN'] @debug = ENV['SESSION_DEBUG'] class PipeError < StandardError; end class ExecutionError < StandardError; end class History #{{{ def initialize; @a = []; end def method_missing(m, *a, &b); @a.send(m, *a, &b); end def to_s #{{{ @a.inject(''){|s,cmd| s << format("%d: %s\n", cmd.cmdno, cmd)} #}}} end alias to_str to_s def inspect #{{{ @a.inject(''){|s,cmd| s << cmd.inspect} #}}} end #}}} end class Command #{{{ class << self #{{{ @@cmdno = 0 def cmdno; @@cmdno; end def cmdno= n; @@cmdno = n; end #}}} end # attributes #{{{ attr :cmd attr :cmdno attr :out,true attr :err,true attr :cid attr :begin_out attr :end_out attr :begin_out_pat attr :end_out_pat attr :begin_err attr :end_err attr :begin_err_pat attr :end_err_pat #}}} def initialize(command) #{{{ @cmd = command.to_s @cmdno = self.class.cmdno self.class.cmdno += 1 @err = '' @out = '' @cid = "%d_%d_%d" % [$$, cmdno, rand(Time.now.usec)] @begin_out = "__CMD_OUT_%s_BEGIN__" % cid @end_out = "__CMD_OUT_%s_END__" % cid @begin_out_pat = %r/#{ Regexp.escape(@begin_out) }/ @end_out_pat = %r/#{ Regexp.escape(@end_out) }/ @begin_err = "__CMD_ERR_%s_BEGIN__" % cid @end_err = "__CMD_ERR_%s_END__" % cid @begin_err_pat = %r/#{ Regexp.escape(@begin_err) }/ @end_err_pat = %r/#{ Regexp.escape(@end_err) }/ #}}} end def to_s; cmd end def to_str; cmd end def inspect #{{{ format("cmd: <%s>\n\tcmdno: <%s>\n\tout: <%s>\n\terr: <%s>\n\tcid: <%s>\n", cmd, cmdno, out.inspect, err.inspect, cid) #}}} end #}}} end class LineBuffer #{{{ def initialize #{{{ @idx = 0 @buf = [] #}}} end def push(buffer) #{{{ buffer.map do |line| if last and not complete? last last << line else @buf << line end line end #}}} end def pop #{{{ @buf.shift #}}} end def top #{{{ @buf[0] #}}} end def last #{{{ @buf[-1] #}}} end def complete? line #{{{ line[-1,1] == "\n" #}}} end def size #{{{ @buf.size #}}} end def empty? #{{{ @buf.empty? #}}} end def clear #{{{ @buf.clear #}}} end def each #{{{ while top and complete? top yield pop end #}}} end def inspect #{{{ @buf.inspect #}}} end def to_s #{{{ @buf.inspect #}}} end #}}} end class AbstractSession #{{{ # class methods class << self #{{{ def default_prog #{{{ return @default_prog if defined? @default_prog and @default_prog if defined? self::DEFAULT_PROG return @default_prog = self::DEFAULT_PROG else @default_prog = ENV["SESSION_#{ self }_PROG"] end nil #}}} end def default_prog= prog #{{{ @default_prog = prog #}}} end attr :use_spawn, true attr :debug, true def init #{{{ @use_spawn = nil @debug = nil #}}} end alias [] new #}}} end # class init init # attributes #{{{ attr :opts attr :prog attr :stdin alias i stdin attr :stdout alias o stdout attr :stderr alias e stderr attr :history attr :track_history, true attr :outproc, true attr :errproc, true attr :use_spawn attr :debug, true alias debug? debug #}}} # instance methods def initialize(*args) #{{{ @opts = hashify(*args) @prog = opts[:prog] || self.class.default_prog raise(ArgumentError, "no program specified") unless @prog @track_history = nil @track_history = opts[:history] if opts.has_key? :history @track_history = opts[:track_history] if opts.has_key? :track_history @history = History.new if @track_history @debug = Session.debug unless Session.debug.nil? @debug = self.class.debug unless self.class.debug.nil? @debug = opts[:debug] if opts.has_key? :debug @debug = false if @debug.nil? @use_spawn = nil @use_spawn = Session.use_spawn unless Session.use_spawn.nil? @use_spawn = self.class.use_spawn unless self.class.use_spawn.nil? @use_spawn = opts[:use_spawn] if opts.has_key? :use_spawn @outproc = nil @errproc = nil @stdin, @stdout, @stderr = if @use_spawn Spawn::spawn @prog else Open3::popen3 @prog end clear if block_given? ret = nil begin ret = yield self ensure self.close! end return ret end return self #}}} end # abstract methods def clear #{{{ raise NotImplementedError #}}} end alias flush clear def path #{{{ raise NotImplementedError #}}} end def path= #{{{ raise NotImplementedError #}}} end def send_command cmd #{{{ raise NotImplementedError #}}} end # concrete methods def ready? #{{{ (stdin and stdout and stderr) and (IO === stdin and IO === stdout and IO === stderr) and (not (stdin.closed? or stdout.closed? or stderr.closed?))# and #(not (stdout.eof? or stderr.eof?)) #}}} end def close! #{{{ [stdin, stdout, stderr].each{|pipe| pipe.close} stdin, stdout, stderr = nil, nil, nil true #}}} end alias close close! def hashify(*a) #{{{ a.inject({}){|o,h| o.update(h)} #}}} end private :hashify def execute(command, redirects = {}) #{{{ $command = command if @debug raise(PipeError, command) unless ready? # clear buffers clear # setup redirects rerr = redirects[:e] || redirects[:err] || redirects[2] || redirects[:stderr] || redirects['stderr'] rout = redirects[:o] || redirects[:out] || redirects[1] || redirects[:stdout] || redirects['stdout'] # create cmd object and add to history cmd = Command.new(command.to_s) # store cmd if tracking history if track_history history << cmd end # send command in the background so we can begin processing output # immediately - thanks to tanaka akira for this suggestion Thread.new { send_command cmd } # process stdout/stderr err = { :io => stderr, :cmd => cmd.err, :name => 'stderr', #:buf => [], #:lines => [], :begin => false, :end => false, :begin_pat => cmd.begin_err_pat, :end_pat => cmd.end_err_pat, :redirect => rerr, :proc => errproc, :yield => lambda{|b| yield(nil, b)}, :buffer => LineBuffer.new, } out = { :io => stdout, :cmd => cmd.out, :name => 'stdout', #:buf => [], #:lines => [], :begin => false, :end => false, :begin_pat => cmd.begin_out_pat, :end_pat => cmd.end_out_pat, :redirect => rout, :proc => outproc, :yield => lambda{|b| yield(b, nil)}, :buffer => LineBuffer.new, } iodat = nil buf = '' # process until end token is found in stream loop do raise(PipeError, command) unless ready? raise ExecutionError, err.inspect if err[:end] and not err[:begin] raise ExecutionError, out.inspect if out[:end] and not out[:begin] break if err[:end] and out[:end] # wait for data selected = [] selected << out[:io] unless out[:end] selected << err[:io] unless err[:end] $selecting = true if @debug rfds, wfds, efds = select selected, nil, selected $selecting = false if @debug # blow up on io err unless efds.empty? msg = efds.map{|io| "err on <#{ io.inspect }>"}.join(',') raise PipeError, msg end # handle any io that's ready rfds.each do |io| next if io.eof? # chose the io iodat = case io when err[:io] err when out[:io] out else raise ExecutionError, "unknown IO #{ io.inspect }" end $iodat = iodat if @debug # blow up if we get data after we should be thru raise ExecutionError, iodat.inspect if iodat[:end] # # read anything that's there - even incomplete lines # read is protected so multi-threaded apps don't hang # buf = '' # this is the desired call - but threads block on it! #iodat[:io].nonblock{ buf = iodat[:io].read } # 2.1.5 #sleep(0.042) and Thread.pass until iodat[:io].ready? #buf << iodat[:io].getc while iodat[:io].ready? # 2.1.6 # decided to go ahead and read lines since # - loop only terminates when cmd_end_pat has been found and this # must be newline terminated # - although this will block mid-line in the case of parital lines we # know that, eventually, a newline will appear and this delay is # acceptable so we can make larger (line buffered) reads instead of # character ones. eventually i'd like to use io.nonblock{} to read # 'as much as is there' but this does not play well with threads at the # moment buf << iodat[:io].gets while iodat[:io].ready? iodat[:buffer].push buf # handle __only__ complete lines iodat[:buffer].each do |line| raise ExecutionError, iodat.inspect unless line[-1,1] == "\n" case line when iodat[:end_pat] # handle the special case of non-newline terminated output if((m = %r/(.+)__CMD/o.match(line)) and (pre = m[1])) iodat[:cmd] << pre iodat[:redirect] << pre if iodat[:redirect] iodat[:proc].call pre if iodat[:proc] iodat[:yield].call pre if block_given? end iodat[:end] = true when iodat[:begin_pat] iodat[:begin] = true else next unless iodat[:begin] and not iodat[:end] # ignore chaff iodat[:cmd] << line iodat[:redirect] << line if iodat[:redirect] iodat[:proc].call line if iodat[:proc] iodat[:yield].call line if block_given? end end end # each end # loop # get the exit status get_status if self.respond_to? :get_status out = err = iodat = nil return [cmd.out, cmd.err] #}}} end #}}} end # class AbstractSession # IDL => interactive data language - see http://www.rsinc.com/ class IDL < AbstractSession class LicenseManagerError < StandardError; end #{{{ DEFAULT_PROG = 'idl' MAX_TRIES = 32 def initialize(*args) #{{{ tries = 0 ret = nil begin ret = super rescue LicenseManagerError => e tries += 1 if tries < MAX_TRIES sleep 1 retry else raise LicenseManagerError, "<#{ MAX_TRIES }> attempts <#{ e.message }>" end end ret #}}} end def clear #{{{ stdin.puts "retall" stdin.puts "printf, -2, '__clear__'" stdin.puts "printf, -1, '__clear__'" stdin.flush while((line = stderr.gets) and line !~ %r/__clear__/o) raise LicenseManagerError, line if line =~ %r/license\s*manager/io end while((line = stdout.gets) and line !~ %r/__clear__/o) raise LicenseManagerError, line if line =~ %r/license\s*manager/io end self #}}} end def send_command cmd #{{{ stdin.printf "printf, -2, '%s'\n", cmd.begin_err stdin.printf "printf, -1, '%s'\n", cmd.begin_out stdin.printf "%s\n", cmd.cmd stdin.printf "retall\n" stdin.printf "printf, -2, '%s'\n", cmd.end_err stdin.printf "printf, -1, '%s'\n", cmd.end_out stdin.flush #}}} end def path #{{{ stdout, stderr = execute "print, !path" stdout.strip.split %r/:/o #}}} end def path= arg #{{{ case arg when Array arg = arg.join ':' else arg = arg.to_s.strip end stdout, stderr = execute "!path='#{ arg }'" self.path #}}} end #}}} end # class IDL class Sh < AbstractSession #{{{ DEFAULT_PROG = 'sh' ECHO = 'echo' attr :status alias exit_status status alias exitstatus status def clear #{{{ stdin.puts "#{ ECHO } __clear__ 1>&2" stdin.puts "#{ ECHO } __clear__" stdin.flush while((line = stderr.gets) and line !~ %r/__clear__/o); end while((line = stdout.gets) and line !~ %r/__clear__/o); end self #}}} end def send_command cmd #{{{ stdin.printf "%s '%s' 1>&2\n", ECHO, cmd.begin_err stdin.printf "%s '%s' \n", ECHO, cmd.begin_out stdin.printf "%s\n", cmd.cmd stdin.printf "export __exit_status__=$?\n" stdin.printf "%s '%s' 1>&2\n", ECHO, cmd.end_err stdin.printf "%s '%s' \n", ECHO, cmd.end_out stdin.flush #}}} end def get_status #{{{ @status = get_var '__exit_status__' unless @status =~ /^\s*\d+\s*$/o raise ExecutionError, "could not determine exit status from <#{ @status.inspect }>" end @status = Integer @status #}}} end def set_var name, value #{{{ stdin.puts "export #{ name }=#{ value }" stdin.flush #}}} end def get_var name #{{{ stdin.puts "#{ ECHO } \"#{ name }=${#{ name }}\"" stdin.flush var = nil while((line = stdout.gets)) m = %r/#{ name }\s*=\s*(.*)/.match line if m var = m[1] raise ExecutionError, "could not determine <#{ name }> from <#{ line.inspect }>" unless var break end end var #}}} end def path #{{{ var = get_var 'PATH' var.strip.split %r/:/o #}}} end def path= arg #{{{ case arg when Array arg = arg.join ':' else arg = arg.to_s.strip end set_var 'PATH', "'#{ arg }'" self.path #}}} end #}}} end # class Sh class Bash < Sh; DEFAULT_PROG = 'bash'; end class Shell < Bash; end #}}} end module Spawn #{{{ class << self def spawn command #{{{ ipath = tmpfifo opath = tmpfifo epath = tmpfifo cmd = "#{ command } < #{ ipath } 1> #{ opath } 2> #{ epath } &" system cmd i = open ipath, 'w' o = open opath, 'r' e = open epath, 'r' [i,o,e] #}}} end def tmpfifo #{{{ path = nil 42.times do |i| tpath = File.join(Dir.tmpdir, "#{ $$ }.#{ rand }.#{ i }") system "mkfifo #{ tpath }" next unless $? == 0 path = tpath at_exit{ File.unlink(path) rescue STDERR.puts("rm <#{ path }> failed") } break end raise "could not generate tmpfifo" unless path path #}}} end end #}}} end