class EventMachine::Synchrony::AMQP::Consumer

def cancel(nowait = false)
  EM::Synchrony::AMQP.sync { |f| self.aconsume(nowait, &EM::Synchrony::AMQP.sync_cb(f)) }
  self
end

def consume(nowait = false)
  ret = EM::Synchrony::AMQP.sync { |f| self.aconsume(nowait, &EM::Synchrony::AMQP.sync_cb(f)) }
  raise Error.new(ret.to_s) unless ret.is_a?(::AMQ::Protocol::Basic::ConsumeOk)
  self
end

def on_delivery(&block)
  Fiber.new do
    aon_delivery(&EM::Synchrony::AMQP.sync_cb(Fiber.current))
    loop { block.call(Fiber.yield) }
  end.resume
  self
end

def resubscribe
  EM::Synchrony::AMQP.sync { |f| self.aconsume(&EM::Synchrony::AMQP.sync_cb(f)) }
  self
end