public function listen($exchange, $routing_key, $callback, $type = self::TYPE_TOPIC)
    {
        list ($queueName) = $this->channel->queue_declare();
        if ($type == Amqp::TYPE_DIRECT) {
            $this->channel->exchange_declare($exchange, $type, false, true, false);
        }
        $this->channel->queue_bind($queueName, $exchange, $routing_key);
        $this->channel->basic_consume($queueName, '', false, true, false, false, $callback);

        while (count($this->channel->callbacks)) {
            $this->channel->wait();
        }

        $this->channel->close();
        $this->connection->close();
    }