Как запустить Net::SSH и AMQP в одном и том же реакторе EventMachine?

Question

Немного предыстории: Геррит предоставляет поток событий через SSH. Это милый трюк, но мне нужно преобразовать эти события в сообщения AMQP. Я пытался сделать это с помощью ruby-amqp и Net::SSH, но, похоже, субкомпонент AMQP вообще не запускается вообще.

Я довольно новичок в EventMachine. Может кто-то указать, что я делаю неправильно? Ответ на вопрос " Несколько серверов в одном реакторе EventMachine) не показался применимым. Программа, также доступная в основном для облегчения доступа:

#!/usr/bin/env ruby                                                                                                                                            

require 'rubygems'
require 'optparse'
require 'net/ssh'
require 'json'
require 'yaml'
require 'amqp'
require 'logger'

trap(:INT) { puts; exit }

options = {
  :logs => 'kili.log',
  :amqp => {
    :host => 'localhost',
    :port => '5672',
  },
  :ssh => {
    :host => 'localhost',
    :port => '22',
    :user => 'nobody',
    :keys => '~/.ssh/id_rsa',
  }
}
optparse = OptionParser.new do|opts|
  opts.banner = "Usage: kili [options]"
  opts.on( '--amqp_host HOST', 'The AMQP host kili will connect to.') do |a|
    options[:amqp][:host] = a
  end
  opts.on( '--amqp_port PORT', 'The port for the AMQP host.') do |ap|
    options[:amqp][:port] = ap
  end
  opts.on( '--ssh_host HOST', 'The SSH host kili will connect to.') do |s|
    options[:ssh][:host] = s
  end
  opts.on( '--ssh_port PORT', 'The SSH port kili will connect on.') do |sp|
    options[:ssh][:port] = sp
  end
  opts.on( '--ssh_keys KEYS', 'Comma delimeted SSH keys for user.') do |sk|
    options[:ssh][:keys] = sk
  end
  opts.on( '--ssh_user USER', 'SSH user for host.') do |su|
    options[:ssh][:user] = su
  end
  opts.on( '-l', '--log LOG', 'The log location of Kili') do |log|
    options[:logs] = log
  end
  opts.on( '-h', '--help', 'Display this screen' ) do
    puts opts
    exit
  end
end


optparse.parse!
log = Logger.new(options[:logs])
log.level = Logger::INFO

amqp = options[:amqp]
sshd = options[:ssh]
queue= EM::Queue.new

EventMachine.run do

  AMQP.connect(:host => amqp[:host], :port => amqp[:port]) do |connection|
    log.info "Connected to AMQP at #{amqp[:host]}:#{amqp[:port]}"
    channel = AMQP::Channel.new(connection)
    exchange = channel.topic("traut", :auto_delete => true)

    queue.pop do |msg|
      log.info("Pulled #{msg} out of queue.")
      exchange.publish(msg[:data], :routing_key => msg[:route]) do
        log.info("On route #{msg[:route]} published:\n#{msg[:data]}")
      end
    end
  end


  Net::SSH.start(sshd[:host], sshd[:user],
    :port => sshd[:port], :keys => sshd[:keys].split(',')) do |ssh|
    log.info "SSH connection to #{sshd[:host]}:#{sshd[:port]} as #{sshd[:user]} made."

    channel = ssh.open_channel do |ch|
      ch.exec "gerrit stream-events" do |ch, success|
        abort "could not stream gerrit events" unless success

        # "on_data" is called when the process writes something to                                                                                             
        # stdout                                                                                                                                               
        ch.on_data do |c, data|
          json = JSON.parse(data)
          if json['type'] == 'change-merged'
            project = json['change']['project']
            route = "com.carepilot.event.code.review.#{project}"
            msg = {:data => data, :route => route}
            queue.push(msg)
            log.info("Pushed #{msg} into queue.")
          else
            log.info("Ignoring event of type #{json['type']}")
          end
        end


    # "on_extended_data" is called when the process writes                                                                                                 
    # something to stderr                                                                                                                                  
    ch.on_extended_data do |c, type, data|
          log.error(data)
    end

    ch.on_close { log.info('Connection closed') }
      end
    end  
  end  

end

Answer 1

Net::SSH не асинхронный, поэтому ваш EventMachine.run() никогда не достигает конца блока, таким образом, никогда не возобновляет поток реактора. Это приводит к тому, что код AMQP никогда не запускается. Я бы предложил запустить ваш SSH-код в другом потоке.

Answer 2

Если вы вернетесь в EventMachine, попробуйте em-ssh https://github.com/simulacre/em-ssh.