public function consume($msgAmount, $timeout=0)
    {
        if ($timeout > 0) {
            // save initial time
            $loopBegin = time();
            $remaining = $timeout;

            // reimplement parent::consume() to inject the timeout
            $this->target = $msgAmount;
            $this->setupConsumer();
            while (count($this->getChannel()->callbacks)) {
                // avoid waiting more than timeout seconds for message reception
                $this->setIdleTimeout($remaining);
                $this->maybeStopConsumer();
                try {
                    $this->getChannel()->wait(null, true, $this->getIdleTimeout());
                } catch (AMQPTimeoutException $e) {
                    return;
                }
                $remaining = $loopBegin + $timeout - time();
                if ($remaining <= 0) {
                    $this->forceStopConsumer();
                }
            }
        } else {
            $this->loopBegin = null;
            parent::consume($msgAmount);
        }
    }