[FS] Polling Interval

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

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,
         ];
     }
 }

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;
+            }
         }
     }
 

pkg/fs/FsContext.php

@@ -32,19 +32,26 @@ 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);
 
         $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()

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()
         ;
     }
 

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,
             ],
         ];
     }

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();

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());
+    }
 }

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);