From 30dbf4bebcfe87c3e50f6442ffecae1b65a20665 Mon Sep 17 00:00:00 2001 From: Alexander Kozienko Date: Fri, 8 Sep 2017 11:04:15 +0300 Subject: [PATCH 1/3] fs polling interval --- pkg/fs/FsConnectionFactory.php | 9 +++- pkg/fs/FsConsumer.php | 45 +++++++++++++++++-- pkg/fs/FsContext.php | 17 ++++++- pkg/fs/Symfony/FsTransportFactory.php | 5 +++ .../Tests/FsConnectionFactoryConfigTest.php | 7 +++ pkg/fs/Tests/FsConsumerTest.php | 8 ++++ pkg/fs/Tests/FsContextTest.php | 14 ++++++ .../Tests/Symfony/FsTransportFactoryTest.php | 5 +++ 8 files changed, 103 insertions(+), 7 deletions(-) diff --git a/pkg/fs/FsConnectionFactory.php b/pkg/fs/FsConnectionFactory.php index 8353db497..e078555fe 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 1000 (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' => 1000, ]; } } diff --git a/pkg/fs/FsConsumer.php b/pkg/fs/FsConsumer.php index c9ef71552..a3301600f 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 = 1000000; + /** * @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..4bd4f7216 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(1000) + ->min(100) + ->info('How often query for new messages.') + ->end() ; } diff --git a/pkg/fs/Tests/FsConnectionFactoryConfigTest.php b/pkg/fs/Tests/FsConnectionFactoryConfigTest.php index 71c52c2d0..83c33227b 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' => '1000', ], ]; @@ -67,6 +68,7 @@ public static function provideConfigs() 'path' => sys_get_temp_dir().'/enqueue', 'pre_fetch_count' => 1, 'chmod' => 0600, + 'polling_interval' => '1000', ], ]; @@ -76,6 +78,7 @@ public static function provideConfigs() 'path' => sys_get_temp_dir().'/enqueue', 'pre_fetch_count' => 1, 'chmod' => 0600, + 'polling_interval' => '1000', ], ]; @@ -85,6 +88,7 @@ public static function provideConfigs() 'path' => sys_get_temp_dir().'/enqueue', 'pre_fetch_count' => 1, 'chmod' => 0600, + 'polling_interval' => '1000', ], ]; @@ -94,6 +98,7 @@ public static function provideConfigs() 'path' => '/foo/bar/baz', 'pre_fetch_count' => 1, 'chmod' => 0600, + 'polling_interval' => '1000', ], ]; @@ -103,6 +108,7 @@ public static function provideConfigs() 'path' => '/foo/bar/baz', 'pre_fetch_count' => 1, 'chmod' => 0600, + 'polling_interval' => '1000', ], ]; @@ -112,6 +118,7 @@ public static function provideConfigs() 'path' => '/foo/bar/baz', 'pre_fetch_count' => 100, 'chmod' => 0666, + 'polling_interval' => '1000', ], ]; } 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..7f5ecc038 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' => 1000, ], $config); } @@ -69,6 +70,7 @@ public function testShouldAllowAddConfigurationAsString() 'dsn' => 'fileDSN', 'pre_fetch_count' => 1, 'chmod' => 0600, + 'polling_interval' => 1000, ], $config); } @@ -82,6 +84,7 @@ public function testShouldCreateConnectionFactory() 'path' => sys_get_temp_dir(), 'pre_fetch_count' => 1, 'chmod' => 0600, + 'polling_interval' => 1000, ]); $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' => 1000, ]], $factory->getArguments()); } @@ -120,6 +124,7 @@ public function testShouldCreateContext() 'path' => sys_get_temp_dir(), 'pre_fetch_count' => 1, 'chmod' => 0600, + 'polling_interval' => 1000, ]); $this->assertEquals('enqueue.transport.fs.context', $serviceId); From db7e05d45251644bb3d07c821d5f6d49a991aa25 Mon Sep 17 00:00:00 2001 From: Alexander Kozienko Date: Fri, 8 Sep 2017 11:13:22 +0300 Subject: [PATCH 2/3] fs polling interval --- pkg/fs/FsConnectionFactory.php | 4 ++-- pkg/fs/FsConsumer.php | 2 +- pkg/fs/Symfony/FsTransportFactory.php | 4 ++-- pkg/fs/Tests/FsConnectionFactoryConfigTest.php | 14 +++++++------- pkg/fs/Tests/Symfony/FsTransportFactoryTest.php | 10 +++++----- 5 files changed, 17 insertions(+), 17 deletions(-) diff --git a/pkg/fs/FsConnectionFactory.php b/pkg/fs/FsConnectionFactory.php index e078555fe..f155c9df9 100644 --- a/pkg/fs/FsConnectionFactory.php +++ b/pkg/fs/FsConnectionFactory.php @@ -18,7 +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 1000 (milliseconds)', + * 'polling_interval' => 'How often query for new messages, default 100 (milliseconds)', * ] * * or @@ -105,7 +105,7 @@ private function defaultConfig() 'path' => null, 'pre_fetch_count' => 1, 'chmod' => 0600, - 'polling_interval' => 1000, + 'polling_interval' => 100, ]; } } diff --git a/pkg/fs/FsConsumer.php b/pkg/fs/FsConsumer.php index a3301600f..1bd36c498 100644 --- a/pkg/fs/FsConsumer.php +++ b/pkg/fs/FsConsumer.php @@ -31,7 +31,7 @@ class FsConsumer implements PsrConsumer /** * @var int microseconds */ - private $pollingInterval = 1000000; + private $pollingInterval = 100000; /** * @param FsContext $context diff --git a/pkg/fs/Symfony/FsTransportFactory.php b/pkg/fs/Symfony/FsTransportFactory.php index 4bd4f7216..9908f8914 100644 --- a/pkg/fs/Symfony/FsTransportFactory.php +++ b/pkg/fs/Symfony/FsTransportFactory.php @@ -57,8 +57,8 @@ public function addConfiguration(ArrayNodeDefinition $builder) ->info('The queue files are created with this given permissions if not exist.') ->end() ->integerNode('polling_interval') - ->defaultValue(1000) - ->min(100) + ->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 83c33227b..a42ede8c6 100644 --- a/pkg/fs/Tests/FsConnectionFactoryConfigTest.php +++ b/pkg/fs/Tests/FsConnectionFactoryConfigTest.php @@ -58,7 +58,7 @@ public static function provideConfigs() 'path' => sys_get_temp_dir().'/enqueue', 'pre_fetch_count' => 1, 'chmod' => 0600, - 'polling_interval' => '1000', + 'polling_interval' => 100, ], ]; @@ -68,7 +68,7 @@ public static function provideConfigs() 'path' => sys_get_temp_dir().'/enqueue', 'pre_fetch_count' => 1, 'chmod' => 0600, - 'polling_interval' => '1000', + 'polling_interval' => 100, ], ]; @@ -78,7 +78,7 @@ public static function provideConfigs() 'path' => sys_get_temp_dir().'/enqueue', 'pre_fetch_count' => 1, 'chmod' => 0600, - 'polling_interval' => '1000', + 'polling_interval' => 100, ], ]; @@ -88,7 +88,7 @@ public static function provideConfigs() 'path' => sys_get_temp_dir().'/enqueue', 'pre_fetch_count' => 1, 'chmod' => 0600, - 'polling_interval' => '1000', + 'polling_interval' => 100, ], ]; @@ -98,7 +98,7 @@ public static function provideConfigs() 'path' => '/foo/bar/baz', 'pre_fetch_count' => 1, 'chmod' => 0600, - 'polling_interval' => '1000', + 'polling_interval' => 100, ], ]; @@ -108,7 +108,7 @@ public static function provideConfigs() 'path' => '/foo/bar/baz', 'pre_fetch_count' => 1, 'chmod' => 0600, - 'polling_interval' => '1000', + 'polling_interval' => 100, ], ]; @@ -118,7 +118,7 @@ public static function provideConfigs() 'path' => '/foo/bar/baz', 'pre_fetch_count' => 100, 'chmod' => 0666, - 'polling_interval' => '1000', + 'polling_interval' => 100, ], ]; } diff --git a/pkg/fs/Tests/Symfony/FsTransportFactoryTest.php b/pkg/fs/Tests/Symfony/FsTransportFactoryTest.php index 7f5ecc038..1fccf154e 100644 --- a/pkg/fs/Tests/Symfony/FsTransportFactoryTest.php +++ b/pkg/fs/Tests/Symfony/FsTransportFactoryTest.php @@ -52,7 +52,7 @@ public function testShouldAllowAddConfiguration() 'path' => sys_get_temp_dir(), 'pre_fetch_count' => 1, 'chmod' => 0600, - 'polling_interval' => 1000, + 'polling_interval' => 100, ], $config); } @@ -70,7 +70,7 @@ public function testShouldAllowAddConfigurationAsString() 'dsn' => 'fileDSN', 'pre_fetch_count' => 1, 'chmod' => 0600, - 'polling_interval' => 1000, + 'polling_interval' => 100, ], $config); } @@ -84,7 +84,7 @@ public function testShouldCreateConnectionFactory() 'path' => sys_get_temp_dir(), 'pre_fetch_count' => 1, 'chmod' => 0600, - 'polling_interval' => 1000, + 'polling_interval' => 100, ]); $this->assertTrue($container->hasDefinition($serviceId)); @@ -94,7 +94,7 @@ public function testShouldCreateConnectionFactory() 'path' => sys_get_temp_dir(), 'pre_fetch_count' => 1, 'chmod' => 0600, - 'polling_interval' => 1000, + 'polling_interval' => 100, ]], $factory->getArguments()); } @@ -124,7 +124,7 @@ public function testShouldCreateContext() 'path' => sys_get_temp_dir(), 'pre_fetch_count' => 1, 'chmod' => 0600, - 'polling_interval' => 1000, + 'polling_interval' => 100, ]); $this->assertEquals('enqueue.transport.fs.context', $serviceId); From 337bb0080f16ecbaf8e3a1e6a267e5a78969666e Mon Sep 17 00:00:00 2001 From: Alexander Kozienko Date: Fri, 8 Sep 2017 11:45:26 +0300 Subject: [PATCH 3/3] fix travis --- bin/test | 6 +++--- 1 file changed, 3 insertions(+), 3 deletions(-) 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