The Buckblog

def initialize(servers, command, callback, options, actor)
  @servers = servers
  @command = command.strip.gsub(/\r?\n/, "\\\n")
  @callback = callback
  @options = options
  @actor = actor
  @channels = open_channels
end

def open_channels
  @servers.map do |server|
    @actor.sessions[server].open_channel do |channel|
      ...
    end
  end
end

channel[:host] = server
channel[:actor] = @actor
channel.request_pty :want_reply => true

# The on_success handler is called when the server
# responds to our request_pty message, but only if
# a pty was allocated. We use this opportunity to send
# the actual command to the server, along with any data
# that should be piped to it on stdin.
channel.on_success do |ch|
  ch.exec command
  ch.send_data options[:data] if options[:data]
end

# Just as on_success is called when the server was able
# to allocate a pty, on_failure is called when it can't.
# In that case, we log a message and move on.
channel.on_failure do |ch|
  logger.important "could not open channel", ch[:host]
  ch.close
end

# Any time the remote command emits data on its stdout,
# Net::SSH will call the channel's on_data callback. We
# delegate to the callback hook given when the Command
# was instantiated.
channel.on_data do |ch, data|
  @callback[ch, :out, data] if @callback
end

# Stderr (and any other, non-stdout data) gets sent to
# the on_extended_data hook. We treat it all as stderr
# and delegate it to the primary callback.
channel.on_extended_data do |ch, type, data|
  @callback[ch, :err, data] if @callback
end

# The on_request hook is used for most other kinds of
# response from the server. All we care about is the
# 'exit-status' reply, which we use to determine whether
# or not the command completed successfully.
channel.on_request do |ch, request, reply, data|
  ch[:status] = data.read_long if request == "exit-status"
end

# When the command finishes, the on_close hook is called.
# We set a flag here that let's us easily query whether
# the channel is still active or not.
channel.on_close do |ch|
  ch[:closed] = true
end

def process!
  # First, we mark the current time. This is so that we
  # can ping each connection every second, so that
  # long-running commands don't result in the connection
  # timing out.
  since = Time.now

  # This begins the event loop...
  loop do

    # This indicates how many channels are still active.
    # When there are no more active channels, we can
    # terminate the event loop.
    active = 0

    # For every active channel, have it's associated
    # connection process any pending events. (The 'true'
    # parameter tells the poll not to block, if no
    # events are pending.)
    @channels.each do |ch|
      next if ch[:closed]
      active += 1
      ch.connection.process(true)
    end

    # If there aren't any active channels, break out of
    # the loop
    break if active == 0

    # If it has been at least a second since the last
    # ping, ping every connection. Note that we ping
    # whether the channel is active or not, since the
    # connection itself IS, and we don't want it timing
    # out just because one of the other channels is
    # lagging behind the others.
    if Time.now - since >= 1
      since = Time.now
      @channels.each { |ch| ch.connection.ping! }
    end
    
    # a brief respite, to keep the CPU from going crazy
    sleep 0.01
  end

  # If any command terminated with a non-zero exit
  # status, then we raise an exception. Ultimately,
  # Capistrano::Actor will catch that exception and try
  # to rollback the current task (if a rollback handler
  # is defined for it.)
  if failed = @channels.detect { |ch| ch[:status] != 0 }
    raise "command #{@command.inspect} failed on #{failed[:host]}"
  end

  self
end