public function consume($queue, $prefetchCount = null, $prefetchSize = null, $consumerId = null, $opts = [])
    {
        return $this->channel()
            ->doOnNext(function (Channel $channel) use ($prefetchCount, $prefetchSize) {
                $channel->qos($prefetchSize, $prefetchCount);
            })
            ->flatMap(
                function (Channel $channel) use ($queue, $consumerId, $opts) {
                    return $this->queue($queue, $opts, $channel)
                        ->consume($consumerId);
                }
            );
    }