Skip to content

Commit 35071ab

Browse files
paras-malhotraparas-malhotra
committed
Add ThrottlesExceptionsWithRedis job middleware
1 parent 92a1ce8 commit 35071ab

File tree

3 files changed

+283
-0
lines changed

3 files changed

+283
-0
lines changed

‎src/Illuminate/Queue/Middleware/ThrottlesExceptionsWithRedis.php‎

Lines changed: 62 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,62 @@
1+
<?php
2+
3+
namespace Illuminate\Queue\Middleware;
4+
5+
use Illuminate\Container\Container;
6+
use Illuminate\Contracts\Redis\Factory as Redis;
7+
use Illuminate\Redis\Limiters\DurationLimiter;
8+
use Illuminate\Support\InteractsWithTime;
9+
use Throwable;
10+
11+
class ThrottlesExceptionsWithRedis extends ThrottlesExceptions
12+
{
13+
use InteractsWithTime;
14+
15+
/**
16+
* The Redis factory implementation.
17+
*
18+
* @var \Illuminate\Contracts\Redis\Factory
19+
*/
20+
protected $redis;
21+
22+
/**
23+
* The rate limiter instance.
24+
*
25+
* @var \Illuminate\Redis\Limiters\DurationLimiter
26+
*/
27+
protected $limiter;
28+
29+
/**
30+
* Process the job.
31+
*
32+
* @param mixed $job
33+
* @param callable $next
34+
* @return mixed
35+
*/
36+
public function handle($job, $next)
37+
{
38+
$this->redis = Container::getInstance()->make(Redis::class);
39+
40+
$this->limiter = new DurationLimiter(
41+
$this->redis, $this->getKey($job), $this->maxAttempts, $this->decayMinutes * 60
42+
);
43+
44+
if ($this->limiter->tooManyAttempts()) {
45+
return $job->release($this->limiter->decaysAt - $this->currentTime());
46+
}
47+
48+
try {
49+
$next($job);
50+
51+
$this->limiter->clear();
52+
} catch (Throwable $throwable) {
53+
if ($this->whenCallback && ! call_user_func($this->whenCallback, $throwable)) {
54+
throw $throwable;
55+
}
56+
57+
$this->limiter->acquire();
58+
59+
return $job->release($this->retryAfterMinutes * 60);
60+
}
61+
}
62+
}

‎src/Illuminate/Redis/Limiters/DurationLimiter.php‎

Lines changed: 54 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -111,6 +111,30 @@ public function acquire()
111111
return (bool) $results[0];
112112
}
113113

114+
/**
115+
* Determine if the key has been "accessed" too many times.
116+
*
117+
* @return bool
118+
*/
119+
public function tooManyAttempts()
120+
{
121+
[$this->decaysAt, $this->remaining] = $this->redis->eval(
122+
$this->tooManyAttemptsScript(), 1, $this->name, microtime(true), time(), $this->decay, $this->maxLocks
123+
);
124+
125+
return $this->remaining <= 0;
126+
}
127+
128+
/**
129+
* Clear the limiter.
130+
*
131+
* @return void
132+
*/
133+
public function clear()
134+
{
135+
$this->redis->del($this->name);
136+
}
137+
114138
/**
115139
* Get the Lua script for acquiring a lock.
116140
*
@@ -143,6 +167,36 @@ protected function luaScript()
143167
end
144168
145169
return {reset(), ARGV[2] + ARGV[3], ARGV[4] - 1}
170+
LUA;
171+
}
172+
173+
/**
174+
* Get the Lua script to determine if the key has been "accessed" too many times.
175+
*
176+
* KEYS[1] - The limiter name
177+
* ARGV[1] - Current time in microseconds
178+
* ARGV[2] - Current time in seconds
179+
* ARGV[3] - Duration of the bucket
180+
* ARGV[4] - Allowed number of tasks
181+
*
182+
* @return string
183+
*/
184+
protected function tooManyAttemptsScript()
185+
{
186+
return <<<'LUA'
187+
188+
if redis.call('EXISTS', KEYS[1]) == 0 then
189+
return {0, ARGV[2] + ARGV[3]}
190+
end
191+
192+
if ARGV[1] >= redis.call('HGET', KEYS[1], 'start') and ARGV[1] <= redis.call('HGET', KEYS[1], 'end') then
193+
return {
194+
redis.call('HGET', KEYS[1], 'end'),
195+
ARGV[4] - redis.call('HGET', KEYS[1], 'count')
196+
}
197+
end
198+
199+
return {0, ARGV[2] + ARGV[3]}
146200
LUA;
147201
}
148202
}

‎tests/Integration/Queue/ThrottlesExceptionsWithRedisTest.php‎

Lines changed: 167 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,167 @@
1+
<?php
2+
3+
namespace Illuminate\Tests\Integration\Queue;
4+
5+
use Exception;
6+
use Illuminate\Bus\Dispatcher;
7+
use Illuminate\Bus\Queueable;
8+
use Illuminate\Contracts\Queue\Job;
9+
use Illuminate\Foundation\Testing\Concerns\InteractsWithRedis;
10+
use Illuminate\Queue\CallQueuedHandler;
11+
use Illuminate\Queue\InteractsWithQueue;
12+
use Illuminate\Queue\Middleware\ThrottlesExceptionsWithRedis;
13+
use Illuminate\Support\Str;
14+
use Mockery as m;
15+
use Orchestra\Testbench\TestCase;
16+
17+
/**
18+
* @group integration
19+
*/
20+
class ThrottlesExceptionsWithRedisTest extends TestCase
21+
{
22+
use InteractsWithRedis;
23+
24+
protected function setUp(): void
25+
{
26+
parent::setUp();
27+
28+
$this->setUpRedis();
29+
}
30+
31+
protected function tearDown(): void
32+
{
33+
parent::tearDown();
34+
35+
$this->tearDownRedis();
36+
37+
m::close();
38+
}
39+
40+
public function testCircuitIsOpenedForJobErrors()
41+
{
42+
$this->assertJobWasReleasedImmediately(CircuitBreakerWithRedisTestJob::class, $key = Str::random());
43+
$this->assertJobWasReleasedImmediately(CircuitBreakerWithRedisTestJob::class, $key);
44+
$this->assertJobWasReleasedWithDelay(CircuitBreakerWithRedisTestJob::class, $key);
45+
}
46+
47+
public function testCircuitStaysClosedForSuccessfulJobs()
48+
{
49+
$this->assertJobRanSuccessfully(CircuitBreakerWithRedisSuccessfulJob::class, $key = Str::random());
50+
$this->assertJobRanSuccessfully(CircuitBreakerWithRedisSuccessfulJob::class, $key);
51+
$this->assertJobRanSuccessfully(CircuitBreakerWithRedisSuccessfulJob::class, $key);
52+
}
53+
54+
public function testCircuitResetsAfterSuccess()
55+
{
56+
$this->assertJobWasReleasedImmediately(CircuitBreakerWithRedisTestJob::class, $key = Str::random());
57+
$this->assertJobRanSuccessfully(CircuitBreakerWithRedisSuccessfulJob::class, $key);
58+
$this->assertJobWasReleasedImmediately(CircuitBreakerWithRedisTestJob::class, $key);
59+
$this->assertJobWasReleasedImmediately(CircuitBreakerWithRedisTestJob::class, $key);
60+
$this->assertJobWasReleasedWithDelay(CircuitBreakerWithRedisTestJob::class, $key);
61+
}
62+
63+
protected function assertJobWasReleasedImmediately($class, $key)
64+
{
65+
$class::$handled = false;
66+
$instance = new CallQueuedHandler(new Dispatcher($this->app), $this->app);
67+
68+
$job = m::mock(Job::class);
69+
70+
$job->shouldReceive('hasFailed')->once()->andReturn(false);
71+
$job->shouldReceive('release')->with(0)->once();
72+
$job->shouldReceive('isReleased')->andReturn(true);
73+
$job->shouldReceive('isDeletedOrReleased')->once()->andReturn(true);
74+
75+
$instance->call($job, [
76+
'command' => serialize($command = new $class($key)),
77+
]);
78+
79+
$this->assertTrue($class::$handled);
80+
}
81+
82+
protected function assertJobWasReleasedWithDelay($class, $key)
83+
{
84+
$class::$handled = false;
85+
$instance = new CallQueuedHandler(new Dispatcher($this->app), $this->app);
86+
87+
$job = m::mock(Job::class);
88+
89+
$job->shouldReceive('hasFailed')->once()->andReturn(false);
90+
$job->shouldReceive('release')->withArgs(function ($delay) {
91+
return $delay >= 600;
92+
})->once();
93+
$job->shouldReceive('isReleased')->andReturn(true);
94+
$job->shouldReceive('isDeletedOrReleased')->once()->andReturn(true);
95+
96+
$instance->call($job, [
97+
'command' => serialize($command = new $class($key)),
98+
]);
99+
100+
$this->assertFalse($class::$handled);
101+
}
102+
103+
protected function assertJobRanSuccessfully($class, $key)
104+
{
105+
$class::$handled = false;
106+
$instance = new CallQueuedHandler(new Dispatcher($this->app), $this->app);
107+
108+
$job = m::mock(Job::class);
109+
110+
$job->shouldReceive('hasFailed')->once()->andReturn(false);
111+
$job->shouldReceive('isReleased')->andReturn(false);
112+
$job->shouldReceive('isDeletedOrReleased')->once()->andReturn(false);
113+
$job->shouldReceive('delete')->once();
114+
115+
$instance->call($job, [
116+
'command' => serialize($command = new $class($key)),
117+
]);
118+
119+
$this->assertTrue($class::$handled);
120+
}
121+
}
122+
123+
class CircuitBreakerWithRedisTestJob
124+
{
125+
use InteractsWithQueue, Queueable;
126+
127+
public static $handled = false;
128+
129+
public function __construct($key)
130+
{
131+
$this->key = $key;
132+
}
133+
134+
public function handle()
135+
{
136+
static::$handled = true;
137+
138+
throw new Exception;
139+
}
140+
141+
public function middleware()
142+
{
143+
return [new ThrottlesExceptionsWithRedis(2, 10, 0, $this->key)];
144+
}
145+
}
146+
147+
class CircuitBreakerWithRedisSuccessfulJob
148+
{
149+
use InteractsWithQueue, Queueable;
150+
151+
public static $handled = false;
152+
153+
public function __construct($key)
154+
{
155+
$this->key = $key;
156+
}
157+
158+
public function handle()
159+
{
160+
static::$handled = true;
161+
}
162+
163+
public function middleware()
164+
{
165+
return [new ThrottlesExceptionsWithRedis(2, 10, 0, $this->key)];
166+
}
167+
}

0 commit comments

Comments
 (0)