diff --git a/bin/test b/bin/test index 6641bc743..ea5789a92 100755 --- a/bin/test +++ b/bin/test @@ -23,9 +23,9 @@ function waitForService() waitForService rabbitmq 5672 50 waitForService mysql 3306 50 waitForService redis 6379 50 -waitForService beanstalkd 11300 -waitForService gearmand 4730 -waitForService kafka 9092 +waitForService beanstalkd 11300 50 +waitForService gearmand 4730 50 +waitForService kafka 9092 50 php pkg/job-queue/Tests/Functional/app/console doctrine:database:create php pkg/job-queue/Tests/Functional/app/console doctrine:schema:update --force diff --git a/pkg/fs/FsConnectionFactory.php b/pkg/fs/FsConnectionFactory.php index 8353db497..f155c9df9 100644 --- a/pkg/fs/FsConnectionFactory.php +++ b/pkg/fs/FsConnectionFactory.php @@ -18,6 +18,7 @@ class FsConnectionFactory implements PsrConnectionFactory * 'path' => 'the directory where all queue\topic files remain. For example /home/foo/enqueue', * 'pre_fetch_count' => 'Integer. Defines how many messages to fetch from the file.', * 'chmod' => 'Defines a mode the files are created with', + * 'polling_interval' => 'How often query for new messages, default 100 (milliseconds)', * ] * * or @@ -48,7 +49,12 @@ public function __construct($config = 'file://') */ public function createContext() { - return new FsContext($this->config['path'], $this->config['pre_fetch_count'], $this->config['chmod']); + return new FsContext( + $this->config['path'], + $this->config['pre_fetch_count'], + $this->config['chmod'], + $this->config['polling_interval'] + ); } /** @@ -99,6 +105,7 @@ private function defaultConfig() 'path' => null, 'pre_fetch_count' => 1, 'chmod' => 0600, + 'polling_interval' => 100, ]; } } diff --git a/pkg/fs/FsConsumer.php b/pkg/fs/FsConsumer.php index c9ef71552..1bd36c498 100644 --- a/pkg/fs/FsConsumer.php +++ b/pkg/fs/FsConsumer.php @@ -28,6 +28,11 @@ class FsConsumer implements PsrConsumer */ private $preFetchedMessages; + /** + * @var int microseconds + */ + private $pollingInterval = 100000; + /** * @param FsContext $context * @param FsDestination $destination @@ -42,6 +47,26 @@ public function __construct(FsContext $context, FsDestination $destination, $pre $this->preFetchedMessages = []; } + /** + * Set polling interval in milliseconds. + * + * @param int $msec + */ + public function setPollingInterval($msec) + { + $this->pollingInterval = $msec * 1000; + } + + /** + * Get polling interval in milliseconds. + * + * @return int + */ + public function getPollingInterval() + { + return (int) $this->pollingInterval / 1000; + } + /** * {@inheritdoc} * @@ -59,13 +84,25 @@ public function getQueue() */ public function receive($timeout = 0) { - $end = microtime(true) + ($timeout / 1000); - while (0 === $timeout || microtime(true) < $end) { - if ($message = $this->receiveNoWait()) { + $timeout /= 1000; + $startAt = microtime(true); + + while (true) { + $message = $this->receiveNoWait(); + + if ($message) { return $message; } - usleep(100); + if ($timeout && (microtime(true) - $startAt) >= $timeout) { + return; + } + + usleep($this->pollingInterval); + + if ($timeout && (microtime(true) - $startAt) >= $timeout) { + return; + } } } diff --git a/pkg/fs/FsContext.php b/pkg/fs/FsContext.php index a711e88e7..542975545 100644 --- a/pkg/fs/FsContext.php +++ b/pkg/fs/FsContext.php @@ -32,12 +32,18 @@ class FsContext implements PsrContext */ private $lockHandlers; + /** + * @var null + */ + private $pollingInterval; + /** * @param string $storeDir * @param int $preFetchCount * @param int $chmod + * @param null $pollingInterval */ - public function __construct($storeDir, $preFetchCount, $chmod) + public function __construct($storeDir, $preFetchCount, $chmod, $pollingInterval = null) { $fs = new Filesystem(); $fs->mkdir($storeDir); @@ -45,6 +51,7 @@ public function __construct($storeDir, $preFetchCount, $chmod) $this->storeDir = $storeDir; $this->preFetchCount = $preFetchCount; $this->chmod = $chmod; + $this->pollingInterval = $pollingInterval; $this->lockHandlers = []; } @@ -160,7 +167,13 @@ public function createConsumer(PsrDestination $destination) { InvalidDestinationException::assertDestinationInstanceOf($destination, FsDestination::class); - return new FsConsumer($this, $destination, $this->preFetchCount); + $consumer = new FsConsumer($this, $destination, $this->preFetchCount); + + if ($this->pollingInterval) { + $consumer->setPollingInterval($this->pollingInterval); + } + + return $consumer; } public function close() diff --git a/pkg/fs/Symfony/FsTransportFactory.php b/pkg/fs/Symfony/FsTransportFactory.php index 482cddc87..9908f8914 100644 --- a/pkg/fs/Symfony/FsTransportFactory.php +++ b/pkg/fs/Symfony/FsTransportFactory.php @@ -56,6 +56,11 @@ public function addConfiguration(ArrayNodeDefinition $builder) ->defaultValue(0600) ->info('The queue files are created with this given permissions if not exist.') ->end() + ->integerNode('polling_interval') + ->defaultValue(100) + ->min(50) + ->info('How often query for new messages.') + ->end() ; } diff --git a/pkg/fs/Tests/FsConnectionFactoryConfigTest.php b/pkg/fs/Tests/FsConnectionFactoryConfigTest.php index 71c52c2d0..a42ede8c6 100644 --- a/pkg/fs/Tests/FsConnectionFactoryConfigTest.php +++ b/pkg/fs/Tests/FsConnectionFactoryConfigTest.php @@ -58,6 +58,7 @@ public static function provideConfigs() 'path' => sys_get_temp_dir().'/enqueue', 'pre_fetch_count' => 1, 'chmod' => 0600, + 'polling_interval' => 100, ], ]; @@ -67,6 +68,7 @@ public static function provideConfigs() 'path' => sys_get_temp_dir().'/enqueue', 'pre_fetch_count' => 1, 'chmod' => 0600, + 'polling_interval' => 100, ], ]; @@ -76,6 +78,7 @@ public static function provideConfigs() 'path' => sys_get_temp_dir().'/enqueue', 'pre_fetch_count' => 1, 'chmod' => 0600, + 'polling_interval' => 100, ], ]; @@ -85,6 +88,7 @@ public static function provideConfigs() 'path' => sys_get_temp_dir().'/enqueue', 'pre_fetch_count' => 1, 'chmod' => 0600, + 'polling_interval' => 100, ], ]; @@ -94,6 +98,7 @@ public static function provideConfigs() 'path' => '/foo/bar/baz', 'pre_fetch_count' => 1, 'chmod' => 0600, + 'polling_interval' => 100, ], ]; @@ -103,6 +108,7 @@ public static function provideConfigs() 'path' => '/foo/bar/baz', 'pre_fetch_count' => 1, 'chmod' => 0600, + 'polling_interval' => 100, ], ]; @@ -112,6 +118,7 @@ public static function provideConfigs() 'path' => '/foo/bar/baz', 'pre_fetch_count' => 100, 'chmod' => 0666, + 'polling_interval' => 100, ], ]; } diff --git a/pkg/fs/Tests/FsConsumerTest.php b/pkg/fs/Tests/FsConsumerTest.php index 408cfd2cb..b3f14266c 100644 --- a/pkg/fs/Tests/FsConsumerTest.php +++ b/pkg/fs/Tests/FsConsumerTest.php @@ -64,6 +64,14 @@ public function testShouldDoNothingOnReject() $consumer->reject(new FsMessage()); } + public function testCouldSetAndGetPollingInterval() + { + $consumer = new FsConsumer($this->createContextMock(), new FsDestination(TempFile::generate()), 123); + $consumer->setPollingInterval(123456); + + $this->assertEquals(123456, $consumer->getPollingInterval()); + } + public function testShouldSendSameMessageToDestinationOnReQueue() { $message = new FsMessage(); diff --git a/pkg/fs/Tests/FsContextTest.php b/pkg/fs/Tests/FsContextTest.php index 3c3c53c59..5ce324a49 100644 --- a/pkg/fs/Tests/FsContextTest.php +++ b/pkg/fs/Tests/FsContextTest.php @@ -233,4 +233,18 @@ public function testShouldCreateFileOnFilesystemIfNotExistOnDeclareDestination() unlink($tmpFile); } + + public function testShouldCreateMessageConsumerAndSetPollingInterval() + { + $tmpFile = new TempFile(sys_get_temp_dir().'/foo'); + + $context = new FsContext(sys_get_temp_dir(), 1, 0666, 123456); + + $queue = $context->createQueue($tmpFile->getFilename()); + + $consumer = $context->createConsumer($queue); + + $this->assertInstanceOf(FsConsumer::class, $consumer); + $this->assertEquals(123456, $consumer->getPollingInterval()); + } } diff --git a/pkg/fs/Tests/Symfony/FsTransportFactoryTest.php b/pkg/fs/Tests/Symfony/FsTransportFactoryTest.php index c135a866d..1fccf154e 100644 --- a/pkg/fs/Tests/Symfony/FsTransportFactoryTest.php +++ b/pkg/fs/Tests/Symfony/FsTransportFactoryTest.php @@ -52,6 +52,7 @@ public function testShouldAllowAddConfiguration() 'path' => sys_get_temp_dir(), 'pre_fetch_count' => 1, 'chmod' => 0600, + 'polling_interval' => 100, ], $config); } @@ -69,6 +70,7 @@ public function testShouldAllowAddConfigurationAsString() 'dsn' => 'fileDSN', 'pre_fetch_count' => 1, 'chmod' => 0600, + 'polling_interval' => 100, ], $config); } @@ -82,6 +84,7 @@ public function testShouldCreateConnectionFactory() 'path' => sys_get_temp_dir(), 'pre_fetch_count' => 1, 'chmod' => 0600, + 'polling_interval' => 100, ]); $this->assertTrue($container->hasDefinition($serviceId)); @@ -91,6 +94,7 @@ public function testShouldCreateConnectionFactory() 'path' => sys_get_temp_dir(), 'pre_fetch_count' => 1, 'chmod' => 0600, + 'polling_interval' => 100, ]], $factory->getArguments()); } @@ -120,6 +124,7 @@ public function testShouldCreateContext() 'path' => sys_get_temp_dir(), 'pre_fetch_count' => 1, 'chmod' => 0600, + 'polling_interval' => 100, ]); $this->assertEquals('enqueue.transport.fs.context', $serviceId);