class Net::SSH::Connection::Session

def [](key)
  @properties[key]
end

def []=(key, value)
  @properties[key] = value
end

def busy?(include_invisible = false)
  if include_invisible
    channels.any?
  else
    channels.any? { |id, ch| !ch[:invisible] }
  end
end

def channel_close(packet)
  info { "channel_close: #{packet[:local_id]}" }
  channel = channels[packet[:local_id]]
  channel_closed(channel)
end

def channel_closed(channel)
  channel.remote_closed!
  channel.close
  cleanup_channel(channel)
  channel.do_close
end

def channel_data(packet)
  info { "channel_data: #{packet[:local_id]} #{packet[:data].length}b" }
  channels[packet[:local_id]].do_data(packet[:data])
end

def channel_eof(packet)
  info { "channel_eof: #{packet[:local_id]}" }
  channels[packet[:local_id]].do_eof
end

def channel_extended_data(packet)
  info { "channel_extended_data: #{packet[:local_id]} #{packet[:data_type]} #{packet[:data].length}b" }
  channels[packet[:local_id]].do_extended_data(packet[:data_type], packet[:data])
end

def channel_failure(packet)
  info { "channel_failure: #{packet[:local_id]}" }
  channels[packet[:local_id]].do_failure
end

def channel_open(packet)
  info { "channel open #{packet[:channel_type]}" }
  local_id = get_next_channel_id
  channel = Channel.new(self, packet[:channel_type], local_id, @max_pkt_size, @max_win_size)
  channel.do_open_confirmation(packet[:remote_id], packet[:window_size], packet[:packet_size])
  callback = channel_open_handlers[packet[:channel_type]]
  if callback
    begin
      callback[self, channel, packet]
    rescue ChannelOpenFailed => err
      failure = [err.code, err.reason]
    else
      channels[local_id] = channel
      msg = Buffer.from(:byte, CHANNEL_OPEN_CONFIRMATION, :long, channel.remote_id, :long,
                        channel.local_id, :long, channel.local_maximum_window_size, :long,
                        channel.local_maximum_packet_size)
    end
  else
    failure = [3, "unknown channel type #{channel.type}"]
  end
  if failure
    error { failure.inspect }
    msg = Buffer.from(:byte, CHANNEL_OPEN_FAILURE, :long, channel.remote_id, :long, failure[0], :string, failure[1], :string, "")
  end
  send_message(msg)
end

def channel_open_confirmation(packet)
  info { "channel_open_confirmation: #{packet[:local_id]} #{packet[:remote_id]} #{packet[:window_size]} #{packet[:packet_size]}" }
  channel = channels[packet[:local_id]]
  channel.do_open_confirmation(packet[:remote_id], packet[:window_size], packet[:packet_size])
end

def channel_open_failure(packet)
  error { "channel_open_failed: #{packet[:local_id]} #{packet[:reason_code]} #{packet[:description]}" }
  channel = channels.delete(packet[:local_id])
  channel.do_open_failed(packet[:reason_code], packet[:description])
end

def channel_request(packet)
  info { "channel_request: #{packet[:local_id]} #{packet[:request]} #{packet[:want_reply]}" }
  channels[packet[:local_id]].do_request(packet[:request], packet[:want_reply], packet[:request_data])
end

def channel_success(packet)
  info { "channel_success: #{packet[:local_id]}" }
  channels[packet[:local_id]].do_success
end

def channel_window_adjust(packet)
  info { "channel_window_adjust: #{packet[:local_id]} +#{packet[:extra_bytes]}" }
  channels[packet[:local_id]].do_window_adjust(packet[:extra_bytes])
end

def cleanup_channel(channel)
  if channel.local_closed? and channel.remote_closed?
    info { "#{host} delete channel #{channel.local_id} which closed locally and remotely" }
    channels.delete(channel.local_id)
  end
end

def close
  info { "closing remaining channels (#{channels.length} open)" }
  channels.each { |id, channel| channel.close }
  begin
    loop(0.1) { channels.any? }
  rescue Net::SSH::Disconnect
    raise unless channels.empty?
  end
  transport.close
end

def closed?
  transport.closed?
end

def dispatch_incoming_packets(raise_disconnect_errors: true)
  while packet = transport.poll_message
    raise Net::SSH::Exception, "unexpected response #{packet.type} (#{packet.inspect})" unless MAP.key?(packet.type)
    send(MAP[packet.type], packet)
  end
rescue StandardError
  force_channel_cleanup_on_close if closed?
  raise if raise_disconnect_errors || !$!.is_a?(Net::SSH::Disconnect)
end

def each_channel(&block)
  channels.dup.each(&block)
end

def ev_do_calculate_rw_wait(wait)
  r = listeners.keys
  w = r.select { |w2| w2.respond_to?(:pending_write?) && w2.pending_write? }
  [r, w, io_select_wait(wait)]
end

def ev_do_handle_events(readers, writers)
  Array(readers).each do |reader|
    if listeners[reader]
      listeners[reader].call(reader)
    else
      if reader.fill.zero?
        reader.close
        stop_listening_to(reader)
      end
    end
  end
  Array(writers).each do |writer|
    writer.send_pending
  end
end

def ev_do_postprocess(was_events)
  @keepalive.send_as_needed(was_events)
  transport.rekey_as_needed
  true
end

def ev_preprocess(&block)
  dispatch_incoming_packets(raise_disconnect_errors: false)
  each_channel { |id, channel| channel.process unless channel.local_closed? }
end

def exec(command, status: nil, &block)
  open_channel do |channel|
    channel.exec(command) do |ch, success|
      raise "could not execute command: #{command.inspect}" unless success
      if status
        channel.on_request("exit-status") do |ch2, data|
          status[:exit_code] = data.read_long
        end
        channel.on_request("exit-signal") do |ch2, data|
          status[:exit_signal] = data.read_long
        end
      end
      channel.on_data do |ch2, data|
        if block
          block.call(ch2, :stdout, data)
        else
          $stdout.print(data)
        end
      end
      channel.on_extended_data do |ch2, type, data|
        if block
          block.call(ch2, :stderr, data)
        else
          $stderr.print(data)
        end
      end
    end
  end
end

def exec!(command, status: nil, &block)
  block_or_concat = block || Proc.new do |ch, type, data|
    ch[:result] ||= String.new
    ch[:result] << data
  end
  status ||= {}
  channel = exec(command, status: status, &block_or_concat)
  channel.wait
  channel[:result] ||= String.new unless block
  channel[:result] &&= channel[:result].force_encoding("UTF-8") unless block
  StringWithExitstatus.new(channel[:result], status[:exit_code]) if channel[:result]
end

def force_channel_cleanup_on_close
  channels.each do |id, channel|
    channel_closed(channel)
  end
end

def forward
  @forward ||= Service::Forward.new(self)
end

def get_next_channel_id
  @channel_id_counter += 1
end

def global_request(packet)
  info { "global request received: #{packet[:request_type]} #{packet[:want_reply]}" }
  callback = @on_global_request[packet[:request_type]]
  result = callback ? callback.call(packet[:request_data], packet[:want_reply]) : false
  if result != :sent && result != true && result != false
    raise "expected global request handler for `#{packet[:request_type]}' to return true, false, or :sent, but got #{result.inspect}"
  end
  if packet[:want_reply] && result != :sent
    msg = Buffer.from(:byte, result ? REQUEST_SUCCESS : REQUEST_FAILURE)
    send_message(msg)
  end
end

def host
  transport.host
end

def initialize(transport, options = {})
  self.logger = transport.logger
  @transport = transport
  @options = options
  @channel_id_counter = -1
  @channels = Hash.new(NilChannel.new(self))
  @listeners = { transport.socket => nil }
  @pending_requests = []
  @channel_open_handlers = {}
  @on_global_request = {}
  @properties = (options[:properties] || {}).dup
  @max_pkt_size = (options.key?(:max_pkt_size) ? options[:max_pkt_size] : 0x8000)
  @max_win_size = (options.key?(:max_win_size) ? options[:max_win_size] : 0x20000)
  @keepalive = Keepalive.new(self)
  @event_loop = options[:event_loop] || SingleSessionEventLoop.new
  @event_loop.register(self)
end

def io_select_wait(wait)
  [wait, max_select_wait_time].compact.min
end

def listen_to(io, &callback)
  listeners[io] = callback
end

def loop(wait = nil, &block)
  running = block || Proc.new { busy? }
  loop_forever { break unless process(wait, &running) }
  begin
    process(0)
  rescue IOError => e
    if e.message =~ /closed/
      debug { "stream was closed after loop => shallowing exception so it will be re-raised in next loop" }
    else
      raise
    end
  end
end

def max_select_wait_time
  @keepalive.interval if @keepalive.enabled?
end

def on_global_request(type, &block)
  old, @on_global_request[type] = @on_global_request[type], block
  old
end

def on_open_channel(type, &block)
  channel_open_handlers[type] = block
end

def open_channel(type = "session", *extra, &on_confirm)
  local_id = get_next_channel_id
  channel = Channel.new(self, type, local_id, @max_pkt_size, @max_win_size, &on_confirm)
  msg = Buffer.from(:byte, CHANNEL_OPEN, :string, type, :long, local_id,
                    :long, channel.local_maximum_window_size,
                    :long, channel.local_maximum_packet_size, *extra)
  send_message(msg)
  channels[local_id] = channel
end

def postprocess(readers, writers)
  ev_do_handle_events(readers, writers)
end

def preprocess(&block)
  return false if block_given? && !yield(self)
  ev_preprocess(&block)
  return false if block_given? && !yield(self)
  return true
end

def process(wait = nil, &block)
  @event_loop.process(wait, &block)
rescue StandardError
  force_channel_cleanup_on_close if closed?
  raise
end

def request_failure(packet)
  info { "global request failure" }
  callback = pending_requests.shift
  callback.call(false, packet) if callback
end

def request_success(packet)
  info { "global request success" }
  callback = pending_requests.shift
  callback.call(true, packet) if callback
end

def send_global_request(type, *extra, &callback)
  info { "sending global request #{type}" }
  msg = Buffer.from(:byte, GLOBAL_REQUEST, :string, type.to_s, :bool, !callback.nil?, *extra)
  send_message(msg)
  pending_requests << callback if callback
  self
end

def send_message(message)
  transport.enqueue_message(message)
end

def shutdown!
  transport.shutdown!
end

def stop_listening_to(io)
  listeners.delete(io)
end