diff --git a/composer.json b/composer.json index 06a9d11..33c4550 100644 --- a/composer.json +++ b/composer.json @@ -41,6 +41,7 @@ "illuminate/broadcasting": "^6.3|^7.0|^8.0", "illuminate/console": "^6.3|^7.0|^8.0", "illuminate/http": "^6.3|^7.0|^8.0", + "illuminate/queue": "^6.3|^7.0|^8.0", "illuminate/routing": "^6.3|^7.0|^8.0", "illuminate/support": "^6.3|^7.0|^8.0", "pusher/pusher-php-server": "^3.0|^4.0", diff --git a/config/websockets.php b/config/websockets.php index 9dcd4f6..9bb34b4 100644 --- a/config/websockets.php +++ b/config/websockets.php @@ -137,7 +137,7 @@ return [ 'redis' => [ - 'connection' => 'default', + 'connection' => env('WEBSOCKETS_REDIS_REPLICATION_CONNECTION', 'default'), /* |-------------------------------------------------------------------------- diff --git a/docs/horizontal-scaling/redis.md b/docs/horizontal-scaling/redis.md index 4f63835..86759db 100644 --- a/docs/horizontal-scaling/redis.md +++ b/docs/horizontal-scaling/redis.md @@ -40,3 +40,29 @@ You can set the connection name to the Redis database under `redis`: ``` The connections can be found in your `config/database.php` file, under the `redis` key. + +## Async Redis Queue + +The default Redis connection also interacts with the queues. Since you might want to dispatch jobs on Redis from the server, you can encounter an anti-pattern of using a blocking I/O connection (like PhpRedis or PRedis) within the WebSockets server. + +To solve this issue, you can configure the built-in queue driver that uses the Async Redis connection when it's possible, like within the WebSockets server. It's highly recommended to switch your queue to it if you are going to use the queues within the server controllers, for example. + +Add the `async-redis` queue driver to your list of connections. The configuration parameters are compatible with the default `redis` driver: + +```php +'connections' => [ + 'async-redis' => [ + 'driver' => 'async-redis', + 'connection' => env('WEBSOCKETS_REDIS_REPLICATION_CONNECTION', 'default'), + 'queue' => env('REDIS_QUEUE', 'default'), + 'retry_after' => 90, + 'block_for' => null, + ], +] +``` + +Also, make sure that the default queue driver is set to `async-redis`: + +``` +QUEUE_CONNECTION=async-redis +``` diff --git a/src/ChannelManagers/RedisChannelManager.php b/src/ChannelManagers/RedisChannelManager.php index c099bbf..51a6d59 100644 --- a/src/ChannelManagers/RedisChannelManager.php +++ b/src/ChannelManagers/RedisChannelManager.php @@ -519,6 +519,16 @@ class RedisChannelManager extends LocalChannelManager return $this->publishClient; } + /** + * Get the Redis client used by other classes. + * + * @return Client + */ + public function getRedisClient() + { + return $this->getPublishClient(); + } + /** * Get the unique identifier for the server. * diff --git a/src/Queue/AsyncRedisConnector.php b/src/Queue/AsyncRedisConnector.php new file mode 100644 index 0000000..ac730c3 --- /dev/null +++ b/src/Queue/AsyncRedisConnector.php @@ -0,0 +1,24 @@ +redis, $config['queue'], + $config['connection'] ?? $this->connection, + $config['retry_after'] ?? 60, + $config['block_for'] ?? null + ); + } +} diff --git a/src/Queue/AsyncRedisQueue.php b/src/Queue/AsyncRedisQueue.php new file mode 100644 index 0000000..9fd35cf --- /dev/null +++ b/src/Queue/AsyncRedisQueue.php @@ -0,0 +1,26 @@ +container->bound(ChannelManager::Class) + ? $this->container->make(ChannelManager::class) + : null; + + return $channelManager && method_exists($channelManager, 'getRedisClient') + ? $channelManager->getRedisClient() + : parent::getConnection(); + } +} diff --git a/src/WebSocketsServiceProvider.php b/src/WebSocketsServiceProvider.php index e498c11..f513caa 100644 --- a/src/WebSocketsServiceProvider.php +++ b/src/WebSocketsServiceProvider.php @@ -11,6 +11,7 @@ use BeyondCode\LaravelWebSockets\Dashboard\Http\Controllers\ShowStatistics; use BeyondCode\LaravelWebSockets\Dashboard\Http\Middleware\Authorize as AuthorizeDashboard; use BeyondCode\LaravelWebSockets\Server\Router; use Illuminate\Support\Facades\Gate; +use Illuminate\Support\Facades\Queue; use Illuminate\Support\Facades\Route; use Illuminate\Support\ServiceProvider; @@ -36,6 +37,12 @@ class WebSocketsServiceProvider extends ServiceProvider __DIR__.'/../database/migrations/0000_00_00_000000_rename_statistics_counters.php' => database_path('migrations/0000_00_00_000000_rename_statistics_counters.php'), ], 'migrations'); + $this->registerAsyncRedisQueueDriver(); + + $this->registerRouter(); + + $this->registerManagers(); + $this->registerStatistics(); $this->registerDashboard(); @@ -50,8 +57,19 @@ class WebSocketsServiceProvider extends ServiceProvider */ public function register() { - $this->registerRouter(); - $this->registerManagers(); + // + } + + /** + * Register the async, non-blocking Redis queue driver. + * + * @return void + */ + protected function registerAsyncRedisQueueDriver() + { + Queue::extend('async-redis', function () { + return new Queue\AsyncRedisConnector($this->app['redis']); + }); } /** diff --git a/tests/LocalQueueTest.php b/tests/LocalQueueTest.php new file mode 100644 index 0000000..dcd8464 --- /dev/null +++ b/tests/LocalQueueTest.php @@ -0,0 +1,273 @@ +runOnlyOnLocalReplication(); + } + + /** + * {@inheritdoc} + */ + protected function tearDown(): void + { + m::close(); + } + + public function testPushProperlyPushesJobOntoRedis() + { + $uuid = Str::uuid(); + + Str::createUuidsUsing(function () use ($uuid) { + return $uuid; + }); + + $queue = $this->getMockBuilder(RedisQueue::class) + ->setMethods(['getRandomId']) + ->setConstructorArgs([$redis = m::mock(Factory::class), 'default']) + ->getMock(); + + $queue->expects($this->once()) + ->method('getRandomId') + ->willReturn('foo'); + + $redis->shouldReceive('connection') + ->once() + ->andReturn($redis); + + $redis->shouldReceive('eval') + ->once() + ->with(LuaScripts::push(), 2, 'queues:default', 'queues:default:notify', json_encode(['uuid' => $uuid, 'displayName' => 'foo', 'job' => 'foo', 'maxTries' => null, 'maxExceptions' => null, 'backoff' => null, 'timeout' => null, 'data' => ['data'], 'id' => 'foo', 'attempts' => 0])); + + $id = $queue->push('foo', ['data']); + + $this->assertSame('foo', $id); + + Str::createUuidsNormally(); + } + + public function testPushProperlyPushesJobOntoRedisWithCustomPayloadHook() + { + $uuid = Str::uuid(); + + Str::createUuidsUsing(function () use ($uuid) { + return $uuid; + }); + + $queue = $this->getMockBuilder(RedisQueue::class) + ->setMethods(['getRandomId']) + ->setConstructorArgs([$redis = m::mock(Factory::class), 'default']) + ->getMock(); + + $queue->expects($this->once()) + ->method('getRandomId') + ->willReturn('foo'); + + $redis->shouldReceive('connection') + ->once() + ->andReturn($redis); + + $redis->shouldReceive('eval') + ->once() + ->with(LuaScripts::push(), 2, 'queues:default', 'queues:default:notify', json_encode(['uuid' => $uuid, 'displayName' => 'foo', 'job' => 'foo', 'maxTries' => null, 'maxExceptions' => null, 'backoff' => null, 'timeout' => null, 'data' => ['data'], 'custom' => 'taylor', 'id' => 'foo', 'attempts' => 0])); + + Queue::createPayloadUsing(function ($connection, $queue, $payload) { + return ['custom' => 'taylor']; + }); + + $id = $queue->push('foo', ['data']); + + $this->assertSame('foo', $id); + + Queue::createPayloadUsing(null); + + Str::createUuidsNormally(); + } + + public function testPushProperlyPushesJobOntoRedisWithTwoCustomPayloadHook() + { + $uuid = Str::uuid(); + + Str::createUuidsUsing(function () use ($uuid) { + return $uuid; + }); + + $queue = $this->getMockBuilder(RedisQueue::class) + ->setMethods(['getRandomId']) + ->setConstructorArgs([$redis = m::mock(Factory::class), 'default']) + ->getMock(); + + $queue->expects($this->once()) + ->method('getRandomId') + ->willReturn('foo'); + + $redis->shouldReceive('connection') + ->once() + ->andReturn($redis); + + $redis->shouldReceive('eval') + ->once() + ->with(LuaScripts::push(), 2, 'queues:default', 'queues:default:notify', json_encode(['uuid' => $uuid, 'displayName' => 'foo', 'job' => 'foo', 'maxTries' => null, 'maxExceptions' => null, 'backoff' => null, 'timeout' => null, 'data' => ['data'], 'custom' => 'taylor', 'bar' => 'foo', 'id' => 'foo', 'attempts' => 0])); + + Queue::createPayloadUsing(function ($connection, $queue, $payload) { + return ['custom' => 'taylor']; + }); + + Queue::createPayloadUsing(function ($connection, $queue, $payload) { + return ['bar' => 'foo']; + }); + + $id = $queue->push('foo', ['data']); + + $this->assertSame('foo', $id); + + Queue::createPayloadUsing(null); + + Str::createUuidsNormally(); + } + + public function testDelayedPushProperlyPushesJobOntoRedis() + { + $uuid = Str::uuid(); + + Str::createUuidsUsing(function () use ($uuid) { + return $uuid; + }); + + $queue = $this->getMockBuilder(RedisQueue::class) + ->setMethods(['availableAt', 'getRandomId']) + ->setConstructorArgs([$redis = m::mock(Factory::class), 'default']) + ->getMock(); + + $queue->expects($this->once()) + ->method('getRandomId') + ->willReturn('foo'); + + $queue->expects($this->once()) + ->method('availableAt') + ->with(1) + ->willReturn(2); + + $redis->shouldReceive('connection') + ->once() + ->andReturn($redis); + + $redis->shouldReceive('zadd') + ->once() + ->with('queues:default:delayed', 2, json_encode(['uuid' => $uuid, 'displayName' => 'foo', 'job' => 'foo', 'maxTries' => null, 'maxExceptions' => null, 'backoff' => null, 'timeout' => null, 'data' => ['data'], 'id' => 'foo', 'attempts' => 0])); + + $id = $queue->later(1, 'foo', ['data']); + + $this->assertSame('foo', $id); + + Str::createUuidsNormally(); + } + + public function testDelayedPushWithDateTimeProperlyPushesJobOntoRedis() + { + $uuid = Str::uuid(); + + Str::createUuidsUsing(function () use ($uuid) { + return $uuid; + }); + + $date = Carbon::now(); + + $queue = $this->getMockBuilder(RedisQueue::class) + ->setMethods(['availableAt', 'getRandomId']) + ->setConstructorArgs([$redis = m::mock(Factory::class), 'default']) + ->getMock(); + + $queue->expects($this->once()) + ->method('getRandomId') + ->willReturn('foo'); + + $queue->expects($this->once()) + ->method('availableAt') + ->with($date) + ->willReturn(2); + + $redis->shouldReceive('connection') + ->once() + ->andReturn($redis); + + $redis->shouldReceive('zadd') + ->once() + ->with('queues:default:delayed', 2, json_encode(['uuid' => $uuid, 'displayName' => 'foo', 'job' => 'foo', 'maxTries' => null, 'maxExceptions' => null, 'backoff' => null, 'timeout' => null, 'data' => ['data'], 'id' => 'foo', 'attempts' => 0])); + + $queue->later($date, 'foo', ['data']); + + Str::createUuidsNormally(); + } + + public function testFireProperlyCallsTheJobHandler() + { + $job = $this->getJob(); + + $job->getContainer() + ->shouldReceive('make') + ->once()->with('foo') + ->andReturn($handler = m::mock(stdClass::class)); + + $handler->shouldReceive('fire') + ->once() + ->with($job, ['data']); + + $job->fire(); + } + + public function testDeleteRemovesTheJobFromRedis() + { + $job = $this->getJob(); + + $job->getRedisQueue() + ->shouldReceive('deleteReserved') + ->once() + ->with('default', $job); + + $job->delete(); + } + + public function testReleaseProperlyReleasesJobOntoRedis() + { + $job = $this->getJob(); + + $job->getRedisQueue() + ->shouldReceive('deleteAndRelease') + ->once() + ->with('default', $job, 1); + + $job->release(1); + } + + protected function getJob() + { + return new RedisJob( + m::mock(Container::class), + m::mock(RedisQueue::class), + json_encode(['job' => 'foo', 'data' => ['data'], 'attempts' => 1]), + json_encode(['job' => 'foo', 'data' => ['data'], 'attempts' => 2]), + 'connection-name', + 'default' + ); + } +} diff --git a/tests/RedisQueueTest.php b/tests/RedisQueueTest.php new file mode 100644 index 0000000..3fbdc63 --- /dev/null +++ b/tests/RedisQueueTest.php @@ -0,0 +1,273 @@ +runOnlyOnRedisReplication(); + } + + /** + * {@inheritdoc} + */ + protected function tearDown(): void + { + m::close(); + } + + public function testPushProperlyPushesJobOntoRedis() + { + $uuid = Str::uuid(); + + Str::createUuidsUsing(function () use ($uuid) { + return $uuid; + }); + + $queue = $this->getMockBuilder(RedisQueue::class) + ->setMethods(['getRandomId']) + ->setConstructorArgs([$redis = m::mock(Factory::class), 'default']) + ->getMock(); + + $queue->expects($this->once()) + ->method('getRandomId') + ->willReturn('foo'); + + $redis->shouldReceive('connection') + ->once() + ->andReturn($redis); + + $redis->shouldReceive('eval') + ->once() + ->with(LuaScripts::push(), 2, 'queues:default', 'queues:default:notify', json_encode(['uuid' => $uuid, 'displayName' => 'foo', 'job' => 'foo', 'maxTries' => null, 'maxExceptions' => null, 'backoff' => null, 'timeout' => null, 'data' => ['data'], 'id' => 'foo', 'attempts' => 0])); + + $id = $queue->push('foo', ['data']); + + $this->assertSame('foo', $id); + + Str::createUuidsNormally(); + } + + public function testPushProperlyPushesJobOntoRedisWithCustomPayloadHook() + { + $uuid = Str::uuid(); + + Str::createUuidsUsing(function () use ($uuid) { + return $uuid; + }); + + $queue = $this->getMockBuilder(RedisQueue::class) + ->setMethods(['getRandomId']) + ->setConstructorArgs([$redis = m::mock(Factory::class), 'default']) + ->getMock(); + + $queue->expects($this->once()) + ->method('getRandomId') + ->willReturn('foo'); + + $redis->shouldReceive('connection') + ->once() + ->andReturn($redis); + + $redis->shouldReceive('eval') + ->once() + ->with(LuaScripts::push(), 2, 'queues:default', 'queues:default:notify', json_encode(['uuid' => $uuid, 'displayName' => 'foo', 'job' => 'foo', 'maxTries' => null, 'maxExceptions' => null, 'backoff' => null, 'timeout' => null, 'data' => ['data'], 'custom' => 'taylor', 'id' => 'foo', 'attempts' => 0])); + + Queue::createPayloadUsing(function ($connection, $queue, $payload) { + return ['custom' => 'taylor']; + }); + + $id = $queue->push('foo', ['data']); + + $this->assertSame('foo', $id); + + Queue::createPayloadUsing(null); + + Str::createUuidsNormally(); + } + + public function testPushProperlyPushesJobOntoRedisWithTwoCustomPayloadHook() + { + $uuid = Str::uuid(); + + Str::createUuidsUsing(function () use ($uuid) { + return $uuid; + }); + + $queue = $this->getMockBuilder(RedisQueue::class) + ->setMethods(['getRandomId']) + ->setConstructorArgs([$redis = m::mock(Factory::class), 'default']) + ->getMock(); + + $queue->expects($this->once()) + ->method('getRandomId') + ->willReturn('foo'); + + $redis->shouldReceive('connection') + ->once() + ->andReturn($redis); + + $redis->shouldReceive('eval') + ->once() + ->with(LuaScripts::push(), 2, 'queues:default', 'queues:default:notify', json_encode(['uuid' => $uuid, 'displayName' => 'foo', 'job' => 'foo', 'maxTries' => null, 'maxExceptions' => null, 'backoff' => null, 'timeout' => null, 'data' => ['data'], 'custom' => 'taylor', 'bar' => 'foo', 'id' => 'foo', 'attempts' => 0])); + + Queue::createPayloadUsing(function ($connection, $queue, $payload) { + return ['custom' => 'taylor']; + }); + + Queue::createPayloadUsing(function ($connection, $queue, $payload) { + return ['bar' => 'foo']; + }); + + $id = $queue->push('foo', ['data']); + + $this->assertSame('foo', $id); + + Queue::createPayloadUsing(null); + + Str::createUuidsNormally(); + } + + public function testDelayedPushProperlyPushesJobOntoRedis() + { + $uuid = Str::uuid(); + + Str::createUuidsUsing(function () use ($uuid) { + return $uuid; + }); + + $queue = $this->getMockBuilder(RedisQueue::class) + ->setMethods(['availableAt', 'getRandomId']) + ->setConstructorArgs([$redis = m::mock(Factory::class), 'default']) + ->getMock(); + + $queue->expects($this->once()) + ->method('getRandomId') + ->willReturn('foo'); + + $queue->expects($this->once()) + ->method('availableAt') + ->with(1) + ->willReturn(2); + + $redis->shouldReceive('connection') + ->once() + ->andReturn($redis); + + $redis->shouldReceive('zadd') + ->once() + ->with('queues:default:delayed', 2, json_encode(['uuid' => $uuid, 'displayName' => 'foo', 'job' => 'foo', 'maxTries' => null, 'maxExceptions' => null, 'backoff' => null, 'timeout' => null, 'data' => ['data'], 'id' => 'foo', 'attempts' => 0])); + + $id = $queue->later(1, 'foo', ['data']); + + $this->assertSame('foo', $id); + + Str::createUuidsNormally(); + } + + public function testDelayedPushWithDateTimeProperlyPushesJobOntoRedis() + { + $uuid = Str::uuid(); + + Str::createUuidsUsing(function () use ($uuid) { + return $uuid; + }); + + $date = Carbon::now(); + + $queue = $this->getMockBuilder(RedisQueue::class) + ->setMethods(['availableAt', 'getRandomId']) + ->setConstructorArgs([$redis = m::mock(Factory::class), 'default']) + ->getMock(); + + $queue->expects($this->once()) + ->method('getRandomId') + ->willReturn('foo'); + + $queue->expects($this->once()) + ->method('availableAt') + ->with($date) + ->willReturn(2); + + $redis->shouldReceive('connection') + ->once() + ->andReturn($redis); + + $redis->shouldReceive('zadd') + ->once() + ->with('queues:default:delayed', 2, json_encode(['uuid' => $uuid, 'displayName' => 'foo', 'job' => 'foo', 'maxTries' => null, 'maxExceptions' => null, 'backoff' => null, 'timeout' => null, 'data' => ['data'], 'id' => 'foo', 'attempts' => 0])); + + $queue->later($date, 'foo', ['data']); + + Str::createUuidsNormally(); + } + + public function testFireProperlyCallsTheJobHandler() + { + $job = $this->getJob(); + + $job->getContainer() + ->shouldReceive('make') + ->once()->with('foo') + ->andReturn($handler = m::mock(stdClass::class)); + + $handler->shouldReceive('fire') + ->once() + ->with($job, ['data']); + + $job->fire(); + } + + public function testDeleteRemovesTheJobFromRedis() + { + $job = $this->getJob(); + + $job->getRedisQueue() + ->shouldReceive('deleteReserved') + ->once() + ->with('default', $job); + + $job->delete(); + } + + public function testReleaseProperlyReleasesJobOntoRedis() + { + $job = $this->getJob(); + + $job->getRedisQueue() + ->shouldReceive('deleteAndRelease') + ->once() + ->with('default', $job, 1); + + $job->release(1); + } + + protected function getJob() + { + return new RedisJob( + m::mock(Container::class), + m::mock(RedisQueue::class), + json_encode(['job' => 'foo', 'data' => ['data'], 'attempts' => 1]), + json_encode(['job' => 'foo', 'data' => ['data'], 'attempts' => 2]), + 'connection-name', + 'default' + ); + } +} diff --git a/tests/TestCase.php b/tests/TestCase.php index 6544731..e331fa5 100644 --- a/tests/TestCase.php +++ b/tests/TestCase.php @@ -124,21 +124,29 @@ abstract class TestCase extends Orchestra 'prefix' => '', ]); - $app['config']->set( - 'broadcasting.connections.websockets', [ - 'driver' => 'pusher', - 'key' => 'TestKey', - 'secret' => 'TestSecret', - 'app_id' => '1234', - 'options' => [ - 'cluster' => 'mt1', - 'encrypted' => true, - 'host' => '127.0.0.1', - 'port' => 6001, - 'scheme' => 'http', - ], - ] - ); + $app['config']->set('broadcasting.connections.websockets', [ + 'driver' => 'pusher', + 'key' => 'TestKey', + 'secret' => 'TestSecret', + 'app_id' => '1234', + 'options' => [ + 'cluster' => 'mt1', + 'encrypted' => true, + 'host' => '127.0.0.1', + 'port' => 6001, + 'scheme' => 'http', + ], + ]); + + $app['config']->set('queue.default', 'async-redis'); + + $app['config']->set('queue.connections.async-redis', [ + 'driver' => 'async-redis', + 'connection' => env('WEBSOCKETS_REDIS_REPLICATION_CONNECTION', 'default'), + 'queue' => env('REDIS_QUEUE', 'default'), + 'retry_after' => 90, + 'block_for' => null, + ]); $app['config']->set('auth.providers.users.model', Models\User::class);