Merge pull request #75 from jamesmoey/feature/file_provider

k-k · k-k · commit bf1a6359cca2 · 2015-10-21T16:06:09.000-07:00
File Provider
diff --git a/docs/file-provider.rst b/docs/file-provider.rst
@@ -0,0 +1,23 @@
+File Provider
+-------------
+
+The file provider uses the filesystem to dispatches and resolves queued messages.
+
+Configuration
+^^^^^^^^^^^^^
+
+To designate a queue as file, set the ``driver`` of its provider to ``file``. You will
+need to a read-able and write-able path to store the messages.
+
+.. code-block:: yaml
+
+    #app/config_dev.yml
+
+    uecode_qpush:
+        providers:
+            file_based:
+                driver: file
+                path: [Path to store messages]
+        queues:
+            my_queue_name:
+                provider: file_based
diff --git a/src/Provider/FileProvider.php b/src/Provider/FileProvider.php
@@ -12,9 +12,12 @@
 class FileProvider extends AbstractProvider
 {
     protected $filePointerList = [];
+    protected $queuePath;
 
     public function __construct($name, array $options, $client, Cache $cache, Logger $logger) {
         $this->name     = $name;
+        /** @var md5 only contain numeric and A to F, so it is file system safe */
+        $this->queuePath = $options['path'].DIRECTORY_SEPARATOR.str_replace('-', '', hash('md5', $name));
         $this->options  = $options;
         $this->cache    = $cache;
         $this->logger   = $logger;
@@ -28,8 +31,9 @@ public function getProvider()
     public function create()
     {
         $fs = new Filesystem();
-        if (!$fs->exists($this->options['path'])) {
-            return $fs->mkdir($this->options['path']);
+        if (!$fs->exists($this->queuePath)) {
+            $fs->mkdir($this->queuePath);
+            return $fs->exists($this->queuePath);
         }
         return true;
     }
@@ -39,12 +43,12 @@ public function publish(array $message, array $options = [])
         $fileName = microtime(false);
         $fileName = str_replace(' ', '', $fileName);
         $path = substr(hash('md5', $fileName), 0, 3);
-        if (!is_dir($this->options['path'].DIRECTORY_SEPARATOR.$path)) {
-            mkdir($this->options['path'].DIRECTORY_SEPARATOR.$path);
+        if (!is_dir($this->queuePath.DIRECTORY_SEPARATOR.$path)) {
+            mkdir($this->queuePath.DIRECTORY_SEPARATOR.$path);
         }
         $fs = new Filesystem();
         $fs->dumpFile(
-            $this->options['path'].DIRECTORY_SEPARATOR.$path.DIRECTORY_SEPARATOR.$fileName.'.json',
+            $this->queuePath.DIRECTORY_SEPARATOR.$path.DIRECTORY_SEPARATOR.$fileName.'.json',
             json_encode($message)
         );
         return $fileName;
@@ -63,7 +67,7 @@ public function receive(array $options = [])
             ->ignoreUnreadableDirs(true)
             ->ignoreVCS(true)
             ->name('*.json')
-            ->in($this->options['path'])
+            ->in($this->queuePath)
         ;
         if ($this->options['message_delay'] > 0) {
             $finder->date(
@@ -82,7 +86,7 @@ public function receive(array $options = [])
             $id = substr($file->getFilename(), 0, -5);
             if (!isset($this->filePointerList[$id]) && flock($filePointer, LOCK_EX | LOCK_NB)) {
                 $this->filePointerList[$id] = $filePointer;
-                 $messages[] = new Message($id, json_decode($file->getContents(), true), []);
+                $messages[] = new Message($id, json_decode($file->getContents(), true), []);
             } else {
                 fclose($filePointer);
             }
@@ -101,7 +105,7 @@ public function delete($id)
             $path = substr(hash('md5', (string)$fileName), 0, 3);
             $fs = new Filesystem();
             $fs->remove(
-                $this->options['path'] . DIRECTORY_SEPARATOR . $path . DIRECTORY_SEPARATOR . $fileName . '.json'
+                $this->queuePath . DIRECTORY_SEPARATOR . $path . DIRECTORY_SEPARATOR . $fileName . '.json'
             );
             fclose($this->filePointerList[$id]);
             unset($this->filePointerList[$id]);
@@ -118,7 +122,7 @@ public function cleanUp()
         $finder = new Finder();
         $finder
             ->files()
-            ->in($this->options['path'])
+            ->in($this->queuePath)
             ->ignoreDotFiles(true)
             ->ignoreUnreadableDirs(true)
             ->ignoreVCS(true)
@@ -137,8 +141,9 @@ public function cleanUp()
     public function destroy()
     {
         $fs = new Filesystem();
-        $fs->remove($this->options['path']);
-        return !is_dir($this->options['path']);
+        $fs->remove($this->queuePath);
+        $this->filePointerList = [];
+        return !is_dir($this->queuePath);
     }
 
     /**
diff --git a/tests/Provider/FileProviderTest.php b/tests/Provider/FileProviderTest.php
@@ -12,20 +12,21 @@ class FileProviderTest extends \PHPUnit_Framework_TestCase
 {
     /** @var FileProvider */
     protected $provider;
-    protected $path;
+    protected $basePath;
+    protected $queueHash;
     protected $umask;
 
     public function setUp()
     {
         $this->umask = umask(0);
-        $this->path = rtrim(sys_get_temp_dir(), DIRECTORY_SEPARATOR).DIRECTORY_SEPARATOR.time().rand(0, 1000);
-        mkdir($this->path);
+        $this->basePath = rtrim(sys_get_temp_dir(), DIRECTORY_SEPARATOR).DIRECTORY_SEPARATOR.time().rand(0, 1000);
+        mkdir($this->basePath);
         $this->provider = $this->getFileProvider();
     }
 
     public function tearDown()
     {
-        $this->clean($this->path);
+        $this->clean($this->basePath);
         umask($this->umask);
     }
 
@@ -50,7 +51,7 @@ private function getFileProvider(array $options = [])
     {
         $options = array_merge(
             [
-                'path'                  => $this->path,
+                'path'                  => $this->basePath,
                 'logging_enabled'       => false,
                 'message_delay'         => 0,
                 'message_timeout'       => 30,
@@ -70,6 +71,8 @@ private function getFileProvider(array $options = [])
             'Symfony\Bridge\Monolog\Logger', [], ['qpush.test']
         );
 
+        $this->queueHash = str_replace('-', '', md5('test'));
+
         return new FileProvider('test', $options, null, $cache, $logger);
     }
 
@@ -83,18 +86,19 @@ public function testGetProvider()
     public function testCreate()
     {
         $this->assertTrue($this->provider->create());
-        $this->assertTrue(is_readable($this->path));
-        $this->assertTrue(is_writable($this->path));
+        $this->assertTrue(is_readable($this->basePath.DIRECTORY_SEPARATOR.$this->queueHash));
+        $this->assertTrue(is_writable($this->basePath.DIRECTORY_SEPARATOR.$this->queueHash));
     }
 
     public function testDestroy()
     {
         $this->provider->destroy();
-        $this->assertFalse(is_dir($this->path));
+        $this->assertFalse(is_dir($this->basePath.DIRECTORY_SEPARATOR.$this->queueHash));
     }
 
     public function testReceive()
     {
+        $this->provider->create();
         $this->assertTrue(is_array($this->provider->receive()));
     }
 
@@ -103,8 +107,8 @@ public function testDelete()
         $this->provider->create();
 
         $path = substr(hash('md5', '123'), 0, 3);
-        mkdir($this->path.DIRECTORY_SEPARATOR.$path);
-        touch($this->path.DIRECTORY_SEPARATOR.$path.DIRECTORY_SEPARATOR.'123.json');
+        mkdir($this->basePath.DIRECTORY_SEPARATOR.$this->queueHash.DIRECTORY_SEPARATOR.$path);
+        touch($this->basePath.DIRECTORY_SEPARATOR.$this->queueHash.DIRECTORY_SEPARATOR.$path.DIRECTORY_SEPARATOR.'123.json');
 
         $messages = $this->provider->receive();
         $this->assertNotEmpty($messages);
@@ -113,6 +117,7 @@ public function testDelete()
 
     public function testPublish()
     {
+        $this->provider->create();
         $content = [
             ['testing'],
             ['testing 123']
@@ -129,6 +134,7 @@ public function testPublish()
     }
 
     public function testPublishDelay() {
+        $this->provider->create();
         $provider = $this->getFileProvider([
             'message_delay' => 2,
         ]);
@@ -142,16 +148,17 @@ public function testOnMessageReceived()
         $this->provider->create();
         $id = $this->provider->publish(['foo' => 'bar']);
         $path = substr(hash('md5', $id), 0, 3);
-        $this->assertTrue(is_file($this->path.DIRECTORY_SEPARATOR.$path.DIRECTORY_SEPARATOR.$id.'.json'));
+        $this->assertTrue(is_file($this->basePath.DIRECTORY_SEPARATOR.$this->queueHash.DIRECTORY_SEPARATOR.$path.DIRECTORY_SEPARATOR.$id.'.json'));
         $this->provider->onMessageReceived(new MessageEvent(
             'test',
             $this->provider->receive()[0]
         ));
-        $this->assertFalse(is_file($this->path.DIRECTORY_SEPARATOR.$path.DIRECTORY_SEPARATOR.$id.'.json'));
+        $this->assertFalse(is_file($this->basePath.DIRECTORY_SEPARATOR.$this->queueHash.DIRECTORY_SEPARATOR.$path.DIRECTORY_SEPARATOR.$id.'.json'));
     }
 
     public function testCleanUp()
     {
+        $this->provider->create();
         $provider = $this->getFileProvider([
             'message_expiration' => 1,
         ]);

Original file line number	Diff line number	Diff line change
`@@ -12,20 +12,21 @@ class FileProviderTest extends \PHPUnit_Framework_TestCase`
`12`	`12`	`{`
`13`	`13`	`/** @var FileProvider */`
`14`	`14`	`protected $provider;`
`15`		`- protected $path;`
	`15`	`+ protected $basePath;`
	`16`	`+ protected $queueHash;`
`16`	`17`	`protected $umask;`
`17`	`18`
`18`	`19`	`public function setUp()`
`19`	`20`	`{`
`20`	`21`	`$this->umask = umask(0);`
`21`		`- $this->path = rtrim(sys_get_temp_dir(), DIRECTORY_SEPARATOR).DIRECTORY_SEPARATOR.time().rand(0, 1000);`
`22`		`- mkdir($this->path);`
	`22`	`+ $this->basePath = rtrim(sys_get_temp_dir(), DIRECTORY_SEPARATOR).DIRECTORY_SEPARATOR.time().rand(0, 1000);`
	`23`	`+ mkdir($this->basePath);`
`23`	`24`	`$this->provider = $this->getFileProvider();`
`24`	`25`	`}`
`25`	`26`
`26`	`27`	`public function tearDown()`
`27`	`28`	`{`
`28`		`- $this->clean($this->path);`
	`29`	`+ $this->clean($this->basePath);`
`29`	`30`	`umask($this->umask);`
`30`	`31`	`}`
`31`	`32`
`@@ -50,7 +51,7 @@ private function getFileProvider(array $options = [])`
`50`	`51`	`{`
`51`	`52`	`$options = array_merge(`
`52`	`53`	`[`
`53`		`- 'path' => $this->path,`
	`54`	`+ 'path' => $this->basePath,`
`54`	`55`	`'logging_enabled' => false,`
`55`	`56`	`'message_delay' => 0,`
`56`	`57`	`'message_timeout' => 30,`
`@@ -70,6 +71,8 @@ private function getFileProvider(array $options = [])`
`70`	`71`	`'Symfony\Bridge\Monolog\Logger', [], ['qpush.test']`
`71`	`72`	`);`
`72`	`73`
	`74`	`+ $this->queueHash = str_replace('-', '', md5('test'));`
	`75`	`+`
`73`	`76`	`return new FileProvider('test', $options, null, $cache, $logger);`
`74`	`77`	`}`
`75`	`78`
`@@ -83,18 +86,19 @@ public function testGetProvider()`
`83`	`86`	`public function testCreate()`
`84`	`87`	`{`
`85`	`88`	`$this->assertTrue($this->provider->create());`
`86`		`- $this->assertTrue(is_readable($this->path));`
`87`		`- $this->assertTrue(is_writable($this->path));`
	`89`	`+ $this->assertTrue(is_readable($this->basePath.DIRECTORY_SEPARATOR.$this->queueHash));`
	`90`	`+ $this->assertTrue(is_writable($this->basePath.DIRECTORY_SEPARATOR.$this->queueHash));`
`88`	`91`	`}`
`89`	`92`
`90`	`93`	`public function testDestroy()`
`91`	`94`	`{`
`92`	`95`	`$this->provider->destroy();`
`93`		`- $this->assertFalse(is_dir($this->path));`
	`96`	`+ $this->assertFalse(is_dir($this->basePath.DIRECTORY_SEPARATOR.$this->queueHash));`
`94`	`97`	`}`
`95`	`98`
`96`	`99`	`public function testReceive()`
`97`	`100`	`{`
	`101`	`+ $this->provider->create();`
`98`	`102`	`$this->assertTrue(is_array($this->provider->receive()));`
`99`	`103`	`}`
`100`	`104`
`@@ -103,8 +107,8 @@ public function testDelete()`
`103`	`107`	`$this->provider->create();`
`104`	`108`
`105`	`109`	`$path = substr(hash('md5', '123'), 0, 3);`
`106`		`- mkdir($this->path.DIRECTORY_SEPARATOR.$path);`
`107`		`- touch($this->path.DIRECTORY_SEPARATOR.$path.DIRECTORY_SEPARATOR.'123.json');`
	`110`	`+ mkdir($this->basePath.DIRECTORY_SEPARATOR.$this->queueHash.DIRECTORY_SEPARATOR.$path);`
	`111`	`+ touch($this->basePath.DIRECTORY_SEPARATOR.$this->queueHash.DIRECTORY_SEPARATOR.$path.DIRECTORY_SEPARATOR.'123.json');`
`108`	`112`
`109`	`113`	`$messages = $this->provider->receive();`
`110`	`114`	`$this->assertNotEmpty($messages);`
`@@ -113,6 +117,7 @@ public function testDelete()`
`113`	`117`
`114`	`118`	`public function testPublish()`
`115`	`119`	`{`
	`120`	`+ $this->provider->create();`
`116`	`121`	`$content = [`
`117`	`122`	`['testing'],`
`118`	`123`	`['testing 123']`
`@@ -129,6 +134,7 @@ public function testPublish()`
`129`	`134`	`}`
`130`	`135`
`131`	`136`	`public function testPublishDelay() {`
	`137`	`+ $this->provider->create();`
`132`	`138`	`$provider = $this->getFileProvider([`
`133`	`139`	`'message_delay' => 2,`
`134`	`140`	`]);`
`@@ -142,16 +148,17 @@ public function testOnMessageReceived()`
`142`	`148`	`$this->provider->create();`
`143`	`149`	`$id = $this->provider->publish(['foo' => 'bar']);`
`144`	`150`	`$path = substr(hash('md5', $id), 0, 3);`
`145`		`- $this->assertTrue(is_file($this->path.DIRECTORY_SEPARATOR.$path.DIRECTORY_SEPARATOR.$id.'.json'));`
	`151`	`+ $this->assertTrue(is_file($this->basePath.DIRECTORY_SEPARATOR.$this->queueHash.DIRECTORY_SEPARATOR.$path.DIRECTORY_SEPARATOR.$id.'.json'));`
`146`	`152`	`$this->provider->onMessageReceived(new MessageEvent(`
`147`	`153`	`'test',`
`148`	`154`	`$this->provider->receive()[0]`
`149`	`155`	`));`
`150`		`- $this->assertFalse(is_file($this->path.DIRECTORY_SEPARATOR.$path.DIRECTORY_SEPARATOR.$id.'.json'));`
	`156`	`+ $this->assertFalse(is_file($this->basePath.DIRECTORY_SEPARATOR.$this->queueHash.DIRECTORY_SEPARATOR.$path.DIRECTORY_SEPARATOR.$id.'.json'));`
`151`	`157`	`}`
`152`	`158`
`153`	`159`	`public function testCleanUp()`
`154`	`160`	`{`
	`161`	`+ $this->provider->create();`
`155`	`162`	`$provider = $this->getFileProvider([`
`156`	`163`	`'message_expiration' => 1,`
`157`	`164`	`]);`