<?php
/**
 * ------------------------ 
 *  版权所有  www.tecmz.com
 *  商业版本请购买正版授权使用
 * ------------------------
*/ namespace Illuminate\Queue; use Illuminate\Support\Arr; use Illuminate\Support\Str; use Illuminate\Redis\Database; use Illuminate\Queue\Jobs\RedisJob; use Illuminate\Contracts\Queue\Queue as QueueContract; class RedisQueue extends Queue implements QueueContract { protected $redis; protected $connection; protected $default; protected $expire = 60; public function __construct(Database $WNTtE, $Rethz = 'default', $GU5eI = null) { goto a300d; ytCz3: $this->connection = $GU5eI; goto nc0yq; rmJlw: $this->default = $Rethz; goto ytCz3; a300d: $this->redis = $WNTtE; goto rmJlw; nc0yq: } public function push($qy3NR, $dTnlu = '', $mgtaV = null) { return $this->pushRaw($this->createPayload($qy3NR, $dTnlu), $mgtaV); } public function pushRaw($CP9TO, $mgtaV = null, array $Ho5Xg = array()) { $this->getConnection()->rpush($this->getQueue($mgtaV), $CP9TO); return Arr::get(json_decode($CP9TO, true), 'id'); } public function later($VSmEG, $qy3NR, $dTnlu = '', $mgtaV = null) { goto R1oGC; POgp4: $this->getConnection()->zadd($this->getQueue($mgtaV) . ':delayed', $this->getTime() + $VSmEG, $CP9TO); goto KRdlH; KRdlH: return Arr::get(json_decode($CP9TO, true), 'id'); goto HcG06; R1oGC: $CP9TO = $this->createPayload($qy3NR, $dTnlu); goto veVDV; veVDV: $VSmEG = $this->getSeconds($VSmEG); goto POgp4; HcG06: } public function release($mgtaV, $CP9TO, $VSmEG, $h7GTE) { $CP9TO = $this->setMeta($CP9TO, 'attempts', $h7GTE); $this->getConnection()->zadd($this->getQueue($mgtaV) . ':delayed', $this->getTime() + $VSmEG, $CP9TO); } public function pop($mgtaV = null) { goto bf5iU; F55Rr: if (!is_null($this->expire)) { $this->migrateAllExpiredJobs($mgtaV); } goto AUu1e; AUu1e: $qy3NR = $this->getConnection()->lpop($mgtaV); goto I10Y4; QDw2O: $mgtaV = $this->getQueue($mgtaV); goto F55Rr; I10Y4: if (!is_null($qy3NR)) { $this->getConnection()->zadd($mgtaV . ':reserved', $this->getTime() + $this->expire, $qy3NR); return new RedisJob($this->container, $this, $qy3NR, $B9SP6); } goto LrCbK; bf5iU: $B9SP6 = $mgtaV ?: $this->default; goto QDw2O; LrCbK: } public function deleteReserved($mgtaV, $qy3NR) { $this->getConnection()->zrem($this->getQueue($mgtaV) . ':reserved', $qy3NR); } protected function migrateAllExpiredJobs($mgtaV) { $this->migrateExpiredJobs($mgtaV . ':delayed', $mgtaV); $this->migrateExpiredJobs($mgtaV . ':reserved', $mgtaV); } public function migrateExpiredJobs($E_44J, $cJXa5) { $oL7Ir = '
local val = redis.call(\'zrangebyscore\', KEYS[1], \'-inf\', ARGV[1], \'limit\', 0, ARGV[2])
if(next(val) ~= nil) then
    redis.call(\'zremrangebyrank\', KEYS[1], 0, #val - 1)
    for i = 1, #val, 100 do
        redis.call(\'rpush\', KEYS[2], unpack(val, i, math.min(i+99, #val)))
    end
end
return val
'; $this->getConnection()->eval($oL7Ir, 2, $E_44J, $cJXa5, time(), 100); } protected function getExpiredJobs($W2PAv, $E_44J, $EjkFa) { return $W2PAv->zrangebyscore($E_44J, '-inf', $EjkFa); } protected function removeExpiredJobs($W2PAv, $E_44J, $EjkFa) { $W2PAv->multi(); $W2PAv->zremrangebyscore($E_44J, '-inf', $EjkFa); } protected function pushExpiredJobsOntoNewQueue($W2PAv, $cJXa5, $MKyCH) { call_user_func_array(array($W2PAv, 'rpush'), array_merge(array($cJXa5), $MKyCH)); } protected function createPayload($qy3NR, $dTnlu = '', $mgtaV = null) { goto mP4m1; mP4m1: $CP9TO = parent::createPayload($qy3NR, $dTnlu); goto egN18; egN18: $CP9TO = $this->setMeta($CP9TO, 'id', $this->getRandomId()); goto XLzow; XLzow: return $this->setMeta($CP9TO, 'attempts', 1); goto oL6il; oL6il: } protected function getRandomId() { return Str::random(32); } protected function getQueue($mgtaV) { return 'queues:' . ($mgtaV ?: $this->default); } protected function getConnection() { return $this->redis->connection($this->connection); } public function getRedis() { return $this->redis; } public function getExpire() { return $this->expire; } public function setExpire($oWvwl) { $this->expire = $oWvwl; } }