redeliver unacknowledged messages

@param bool $reQueue
181032
public function basicRecover($reQueue = false)
    {
        $args = $this->frameBuilder->basicRecover($reQueue);
        $this->sendMethodFrame(array(60, 100), $args);
    }
137020
public function seeNumberOfMessagesInQueue($queue, $expected)
    {
        $messageCount = $this->_countMessage($queue);
        $this->assertEquals($expected, $messageCount);
    }
873727
public function requeue($delay, $backoff = false)
    {
        $this->isResponded = true;
        $this->delegate->onRequeue($this, $delay, $backoff);
    }
512374
public function changeMessageVisibility($receiptHandle, $visibilityTimeout)
    {
        $this->deleteMessage($receiptHandle);
        if ($visibilityTimeout > 0) {
            $this->client->zadd($this->queueName . ':delayed', [$receiptHandle => time() + $visibilityTimeout]);
        } else {
            $this->client->rpush($this->queueName, [$receiptHandle]);
        }
    }
640985
private function requeue(string $queueName, AMQPMessage $msg, NotProcessableException $exception): void
    {
        $tries = 0;
        $headers = $msg->get_properties()['application_headers'];
        if ($headers instanceof AMQPTable) {
            $tries = $headers->getNativeData()['x-number-of-tries'] ?? 0;
        }

        if ($exception->getMaxTries() !== null && $tries >= $exception->getMaxTries()) {
            $this->logger->debug('No tries left for message. Marking it as an error', ['message' => $msg, 'exception' => $exception]);
            $this->handleError($queueName, $msg, $exception);
            return;
        }

        // Init delay exchange
        $channel = $this->amqp->channel();
        $delayExchange = "$queueName.delayed";
        $channel->exchange_declare($delayExchange, 'x-delayed-message', false, true, true, false, false, new AMQPTable([
            'x-delayed-type' => 'direct',
        ]));
        $channel->queue_bind($queueName, $delayExchange);

        // Send message
        $delayDuration = 1000 * $exception->getSecondsBeforeRetry() * (int)pow($exception->getProgressionMultiplier(), $tries);
        $delayMessage = new AMQPMessage($msg->getBody(), array_merge($msg->get_properties(), [
            'application_headers' => new AMQPTable([
                'x-delay' => $delayDuration,
                'x-number-of-tries' => $tries + 1,
            ]),
        ]));
        $channel->basic_publish($delayMessage, $delayExchange, '');
        $this->logger->debug('Delayed message for ' . $delayDuration, ['message' => $msg, 'exception' => $exception]);
    }
635691
public function nack($requeue = true)
    {
        $promise = $this->channel->nack($this->message, false, $requeue);
        return \Rxnet\fromPromise($promise);
    }
211188
public function reject(Message $message, $requeue = true)
    {
        return $this->rejectImpl($message->deliveryTag, $requeue);
    }
451761
public function pop($queue = null)
    {
        if (static::supportDelayed()) {
            $this->requeueDelayedItems();
        }

        $queue = $queue ?: $this->queue;

        $job = Resque::pop($queue);

        if (!$job) {
            return null;
        }

        $message = new QueueMessage();

        $attempts = $job['attempts'];
        $attempts++;

        $message->setId($job['id']);
        $message->setBody($job);
        $message->setRawBody(json_encode($job));
        $message->setQueueName($queue ?: $this->queue);
        $message->setAttempts($attempts);
        $message->set('attempts', $attempts);

        return $message;
    }
1268909
public static function redisKey($queue = null, $suffix = null)
    {
        if (is_null($queue)) {
            return 'queues';
        }

        return (strpos($queue, 'queue:') !== 0 ? 'queue:' : '').$queue.($suffix ? ':'.$suffix : '');
    }
999711
public function setRetry($retry)
    {
        Argument::i()
            ->test(1, 'int');

        $this->retry = $retry;
        return $this->setData($this->message);
    }
1267096
@Override
	public final void setJMSRedelivered(boolean redelivered)
    {
        this.redelivered = redelivered;
        
        // Update raw cache accordingly
        if (rawMessage != null)
        {
        	byte flags = rawMessage.readByte(1);
        	if (redelivered)
        		flags = (byte)(flags | (1 << 4));
        	else
        		flags = (byte)(flags & ~(1 << 4));
        	
        	rawMessage.writeByte(flags, 1);
        }
    }
413351
protected function ackOrNackBlock()
    {
        if (! $this->lastDeliveryTag) {
            return;
        }

        $result = $this->flushDeferred();

        switch ($result) {
            case FlushDeferredResult::MSG_ACK():
                $this->ack();
                break;
            case FlushDeferredResult::MSG_REJECT():
                $this->nackAll(false);
                break;
            case FlushDeferredResult::MSG_REJECT_REQUEUE():
                $this->nackAll(true);
                break;
        }
    }
280880
public function queueDelete($queue = '', $ifUnused = false, $ifEmpty = false, $nowait = false)
    {
        return $this->getClient()->queueDelete($this->getChannelId(), $queue, $ifUnused, $ifEmpty, $nowait);
    }
138742
protected function reindex(array $events, int &$reindexed, SymfonyStyle $io, array $context, int $batch, bool $dryRun = false, bool $skipErrors = false): void
    {
        if ($dryRun) {
            $io->note(sprintf('DRY RUN - Would reindex event batch %d here.', $batch));
        } else {
            try {
                $this->getPbjx()->getEventSearch()->indexEvents($events, $context);
            } catch (\Throwable $e) {
                $io->error($e->getMessage());
                $io->note(sprintf('Failed to index batch %d.', $batch));
                $io->newLine(2);

                if (!$skipErrors) {
                    throw $e;
                }
            }
        }

        $reindexed += count($events);
    }
458900
static public function reenable( $reactivate=false )
    {
        self::$has_run = false;
        if ( $reactivate )
        {
            self::$activated = true;
        }
    }
1182527
function unqueue(done) {
    var storage = this;

    this.redis().lpop(this.queueKey(), function(err, key) {
        if (err) return done(err);

        storage.redis().get(key, function(err, task) {
            if (err) return done(err);
            if (!task) return done();

            storage.redis().del(key);
            done(null, JSON.parse(task), key);
        });
    });
}
79754
def re_enqueue(self, item):
        """"""
        if 'retries' in item:
            retries = item['retries']
            if retries >= self.MAX_RETRIES:
                log.warn("Failed to execute {} after {} retries, give it "
                         " up.".format(item['method'], retries))
            else:
                retries += 1
                item['retries'] = retries
                self._q.put_nowait(item)
        else:
            item['retries'] = 1
            self._q.put_nowait(item)