reserve

@param string[] $queues
@param \QueueJitsu\Queue\Adapter\AdapterInterface $adapter

@return null|\QueueJitsu\Job\Job
970505
public function reserve(array $queues, AdapterInterface $adapter): ?Job
    {
        if (in_array('*', $queues, true)) {
            $queues = $adapter->getAllQueueNames();
        }

        foreach ($queues as $queue) {
            $this->log->debug(sprintf('Checking %s', $queue));

            $job = $adapter->reserve($queue);

            if (!is_null($job)) {
                $this->log->debug(sprintf('Found job on %s', $queue));

                return $job;
            }
        }

        return null;
    }
105171
public function setAdapter($adapter)
    {
        if (\is_callable($adapter)) {
            $adapter = \call_user_func($adapter, $this->container);
        }

        $this->adapter = $adapter;
    }
1116002
public function waitFor(array $stack, int $options=0): Scheduler
    {
        $jobs = array_map(function($job) {
            if(!($job instanceof Process)) {
                throw new InvalidArgumentException('waitFor() requires a stack of Process[]');
            }

            return $job->getId();
        }, $stack);

        $cursor = $this->events->getCursor([
            'job' => ['$in' => $jobs],
            'status' => ['$gte' => JobInterface::STATUS_DONE],
        ]);

        $expected = count($stack);
        $done = 0;

        while (true) {
            if (null === $cursor->current()) {
                if ($cursor->getInnerIterator()->isDead()) {
                    $this->events->create();

                    return $this->waitFor($stack, $options);
                }

                $this->events->next($cursor, function () use($stack, $options) {
                    $this->waitFor($stack, $options);
                });

                continue;
            }

            $event = $cursor->current();
            $this->events->next($cursor, function () use($stack, $options) {
                $this->waitFor($stack, $options);
            });

            if (JobInterface::STATUS_FAILED === $event['status'] && isset($event['exception']) && $options & self::OPTION_THROW_EXCEPTION) {
                throw new $event['exception']['class'](
                    $event['exception']['message'],
                    $event['exception']['code']
                );
            }

            $done++;

            if($done >= $expected) {
                return $this;
            }
        }
    }
371393
private function getRightAdapter(array $adapters)
    {
        $adapter = null;
        if (1 < count($adapters) && in_array($this->defaultAdapter, $adapters)) {
            $adapter = $this->defaultAdapter;
        } else {
            $adapter = reset($adapters);
        }

        return $adapter;
    }
384391
public function reserve()
    {
        $queues = $this->queues();
        if (!is_array($queues)) {
            return;
        }
        foreach ($queues as $queue) {
            $this->log(array('message' => 'Checking ' . $queue, 'data' => array('type' => 'check', 'queue' => $queue)), self::LOG_TYPE_DEBUG);
            $job = Resque_Job::reserve($queue);
            if ($job) {
                $this->log(array('message' => 'Found job on ' . $queue, 'data' => array('type' => 'found', 'queue' => $queue)), self::LOG_TYPE_DEBUG);
                return $job;
            }
        }

        return false;
    }
645121
public function ack($queueMessage)
    {
        $queueName = $queueMessage->getQueueName();
        $queueData = $queueMessage->getQueueData();

        return $this->__handler->zrem($queueName.':reserved', $queueData);
    }
26105
def reserve(cls, queues, res, worker=None, timeout=10):
        
        if isinstance(queues, string_types):
            queues = [queues]
        queue, payload = res.pop(queues, timeout=timeout)
        if payload:
            return cls(queue, payload, res, worker)
628377
public function reserve($timeout = null)
    {
        if (isset($timeout)) {
            $this->_write(sprintf('reserve-with-timeout %d', $timeout));
        } else {
            $this->_write('reserve');
        }
        $status = strtok($this->_read(), ' ');

        switch ($status) {
            case 'RESERVED':
                return [
                    'id' => (integer) strtok(' '),
                    'body' => $this->_read((integer) strtok(' ')),
                ];
            case 'TIMED_OUT':
                return false;
            case 'DEADLINE_SOON':
                throw new DeadlineSoonException('Reserve failed: '.$status);
            default:
                throw new ServerException('Reserve failed: '.$status);
        }
    }
635408
public function factory( $adapter, $options = null )
    {
        $adapter = parent::factory( $adapter, $options );
        $adapter->setModel( $this->getModel() );

        if ( $adapter instanceof ServiceLocatorAwareInterface )
        {
            $adapter->setServiceLocator( $this->getServiceLocator() );
        }

        return $adapter;
    }
1092604
protected function retrieveNextJob($queue)
    {
        return $this->getConnection()->eval(
            LuaScripts::pop(), 2, $queue, $queue.':reserved',
            $this->availableAt($this->retryAfter)
        );
    }
961641
protected function getNextAvailableJob($queue)
	{
		$this->database->beginTransaction();

		$job = $this->database->table($this->table)
					->lockForUpdate()
					->where('queue', $this->getQueue($queue))
					->where('reserved', 0)
					->where('available_at', '<=', $this->getTime())
					->orderBy('id', 'asc')
					->first();

		return $job ? (object) $job : null;
	}
7040
public function queueDeclare(array $params)
    {
        list(
            $queue,
            $passive,
            $durable,
            $exclusive,
            $auto_delete,
            $nowait,
            $arguments,
            $ticket
            ) = $params;

        return parent::queue_declare(
            $queue,
            $passive,
            $durable,
            $exclusive,
            $auto_delete,
            $nowait,
            $arguments,
            $ticket
        );
    }
302666
public function get(array $adapters, CacheableInterface $item)
    {
        foreach ($adapters as $adapter) {
            try {
                return $adapter->get($item);
            } catch (\Exception $e) {
                continue;
            }
        }

        throw new UnhandledCacheException('No cache system is available');
    }
1102945
private function getAdapter(string $adapter = null, array $params, SymfonyStyle $inputOutput): Adapter
    {
        $adapterName = $adapter ?? $inputOutput->askQuestion(new AdapterChoice());
        $adapterName = strtolower($adapterName);

        if (empty($params)) {
            $questionGenerator = new AdapterQuestionProvider();
            foreach ($questionGenerator->getQuestionsFor($adapterName) as $question) {
                $params[] = $inputOutput->askQuestion($question);
            }
        }

        return $this->adapterFactory->build($adapterName, $params);
    }
786112
public function listenToQueue(array $handlers , $routingKey = null, $options =[] )
    {
        /* Look for handlers */
        $handlersMap = array();
        foreach ($handlers as $handlerClassPath) {
            if (!class_exists($handlerClassPath)) {
                $handlerClassPath = "Kontoulis\\RabbitMQLaravel\\Handlers\\DefaultHandler";
                if (!class_exists($handlerClassPath)) {
                    throw new BrokerException(
                        "Class $handlerClassPath was not found!"
                    );
                }
            }
            $handlerOb = new $handlerClassPath();
            $classPathParts = explode("\\", $handlerClassPath);
            $handlersMap[$classPathParts[count($classPathParts) - 1]] = $handlerOb;
        }

        $this->queueDeclareBind($routingKey);
        /* Start consuming */
        $this->basic_qos(
            (isset($options["prefetch_size"]) ? $options["prefetch_size"] : null),
            (isset($options["prefetch_count"]) ? $options["prefetch_count"] : 1),
            (isset($options["a_global"]) ? $options["a_global"] : null)
        );
        $this->basic_consume(
            $routingKey,
            (isset($options["consumer_tag"]) ? $options["consumer_tag"] : ''),
            (isset($options["no_local"]) ? (bool)$options["no_local"] : false),
            (isset($options["no_ack"]) ? (bool)$options["no_ack"] : false),
            (isset($options["exclusive"]) ? (bool)$options["exclusive"] : false),
            (isset($options["no_wait"]) ? (bool)$options["no_wait"] : false),
            function (AMQPMessage $amqpMsg) use($handlersMap) {
                $msg = Message::fromAMQPMessage($amqpMsg);
                $this->handleMessage($msg, $handlersMap);
            }
        );
        return $this->waitConsume($options);
    }
508494
public function adapter($adapter = null)
    {
        if ($adapter === null) {
            return $this->_adapter;
        }

        $this->_adapter = $adapter;
    }
717110
def reserve
      queues.each do |queue|
        log_with_severity :debug, "Checking #{queue}"
        if job = Resque.reserve(queue)
          log_with_severity :debug, "Found job on #{queue}"
          return job
        end
      end

      nil
    rescue Exception => e
      log_with_severity :error, "Error reserving job: #{e.inspect}"
      log_with_severity :error, e.backtrace.join("\n")
      raise e
    end