initial commit mongof flow

Author: daniel-sc

src/Flow/File.php

@@ -7,7 +7,7 @@ class File
     /**
      * @var RequestInterface
      */
-    private $request;
+    protected $request;
 
     /**
      * @var ConfigInterface

src/Flow/Mongo/MongoConfig.php

@@ -0,0 +1,28 @@
+<?php
+
+namespace Flow\Mongo;
+
+use Flow\Config;
+
+class MongoConfig extends Config implements MongoConfigInterface
+{
+    private $gridFs;
+
+    /**
+     * @param \MongoGridFS $gridFS storage of the upload (and chunks)
+     */
+    function __construct(\MongoGridFS $gridFS)
+    {
+        parent::__construct();
+        $this->gridFs = $gridFS;
+    }
+
+
+    /**
+     * @return \MongoGridFS
+     */
+    public function getGridFs()
+    {
+        return $this->gridFs;
+    }
+}

src/Flow/Mongo/MongoConfigInterface.php

@@ -0,0 +1,18 @@
+<?php
+
+namespace Flow\Mongo;
+
+use Flow\ConfigInterface;
+
+/**
+ * @package Flow
+ */
+interface MongoConfigInterface extends ConfigInterface
+{
+
+    /**
+     * @return \MongoGridFS
+     */
+    public function getGridFs();
+
+}

src/Flow/Mongo/MongoFile.php

@@ -0,0 +1,163 @@
+<?php
+
+namespace Flow\Mongo;
+
+use Flow\File;
+use Flow\Request;
+use Flow\RequestInterface;
+
+
+/**
+ * Notes:
+ *
+ * - One should ensure indices on the gridfs collection on the property 'flowIdentifier'.
+ * - Chunk preprocessor not supported (must not modify chunks size)!
+ * - Must use 'forceChunkSize=true' on client side.
+ *
+ * @package Flow
+ */
+class MongoFile extends File
+{
+    private $uploadGridFsFile;
+
+    /**
+     * @var MongoConfigInterface
+     */
+    private $config;
+
+    function __construct(MongoConfigInterface $config, RequestInterface $request = null)
+    {
+        if ($request === null) {
+            $request = new Request();
+        }
+        parent::__construct($config, $request);
+        $this->config = $config;
+    }
+
+    /**
+     * return array
+     */
+    protected function getGridFsFile()
+    {
+        if (!$this->uploadGridFsFile) {
+            $gridFsFileQuery = $this->getGridFsFileQuery();
+            $changed = $gridFsFileQuery;
+            $changed['flowUpdated'] = new \MongoDate();
+            $this->uploadGridFsFile = $this->config->getGridFs()->findAndModify($gridFsFileQuery, $changed, null,
+                ['upsert' => true]);
+        }
+
+        return $this->uploadGridFsFile;
+    }
+
+    /**
+     * @param $index int|string 1-based
+     * @return bool
+     */
+    public function chunkExists($index)
+    {
+        return $this->config->getGridFs()->chunks->find([
+            'files_id' => $this->getGridFsFile()['_id'],
+            'n' => (intval($index) - 1)
+        ])->limit(1)->hasNext();
+    }
+
+    public function checkChunk()
+    {
+        return $this->chunkExists($this->request->getCurrentChunkNumber());
+    }
+
+    /**
+     * Save chunk
+     *
+     * @return bool
+     */
+    public function saveChunk()
+    {
+        try {
+            $file = $this->request->getFile();
+
+            $chunkQuery = [
+                'files_id' => $this->getGridFsFile()['_id'],
+                'n' => intval($this->request->getCurrentChunkNumber()) - 1,
+            ];
+            $chunk = $chunkQuery;
+            $data = file_get_contents($file['tmp_name']);
+            $actualChunkSize = strlen($data);
+            if ($actualChunkSize > $this->request->getDefaultChunkSize() ||
+                ($actualChunkSize < $this->request->getDefaultChunkSize() &&
+                    $this->request->getCurrentChunkNumber() != $this->request->getTotalChunks())
+            ) {
+                throw new \Exception("Invalid upload! (size: {$actualChunkSize})");
+            }
+            $chunk['data'] = new \MongoBinData($data);
+            $this->config->getGridFs()->chunks->findAndModify($chunkQuery, $chunk, [], ['upsert' => true]);
+            unlink($file['tmp_name']);
+
+            return true;
+        } catch (\Exception $e) {
+            return false;
+        }
+    }
+
+    /**
+     * @return bool
+     */
+    public function validateFile()
+    {
+        $totalChunks = $this->request->getTotalChunks();
+
+        for ($i = 1; $i <= $totalChunks; $i++) {
+            if (!$this->chunkExists($i)) {
+                return false;
+            }
+        }
+
+        return true;
+    }
+
+
+    /**
+     * Merge all chunks to single file
+     * @param $metadata array additional metadata for final file
+     * @return \MongoId|bool of saved file or false if file was already saved
+     * @throws \Exception
+     */
+    public function saveToGridFs($metadata = null)
+    {
+        $file = $this->getGridFsFile();
+        $file['flowStatus'] = 'finished';
+        $file['metadata'] = $metadata;
+        $result = $this->config->getGridFs()->findAndModify($this->getGridFsFileQuery(), $file);
+        // on second invocation no more file can be found, as the flowStatus changed:
+        if (is_null($result)) {
+            return false;
+        } else {
+            return $file['_id'];
+        }
+    }
+
+    public function save($destination)
+    {
+        throw new \Exception("Must not use 'save' on MongoFile - use 'saveToGridFs'!");
+    }
+
+    public function deleteChunks()
+    {
+        // nothing to do, as chunks are directly part of the final file
+    }
+
+    /**
+     * @return array
+     */
+    protected function getGridFsFileQuery()
+    {
+        return [
+            'flowIdentifier' => $this->request->getIdentifier(),
+            'flowStatus' => 'uploading',
+            'filename' => $this->request->getFileName(),
+            'chunkSize' => intval($this->request->getDefaultChunkSize()),
+            'length' => intval($this->request->getTotalSize())
+        ];
+    }
+}

src/Flow/Mongo/MongoUploader.php

@@ -0,0 +1,28 @@
+<?php
+
+namespace Flow\Mongo;
+
+use Flow\FileOpenException;
+
+class MongoUploader
+{
+    /**
+     * Delete chunks older than expiration time.
+     *
+     * @param \MongoGridFS $gridFs
+     * @param int $expirationTime seconds
+     *
+     * @throws FileOpenException
+     */
+    public static function pruneChunks($gridFs, $expirationTime = 172800)
+    {
+        $result = $gridFs->remove([
+            'flowUpdated' => ['$lt' => new \MongoDate(time() - $expirationTime)],
+            'flowStatus' => 'uploading'
+        ]);
+
+        if (!$result) {
+            throw new FileOpenException("Could not remove chunks!");
+        }
+    }
+}

src/Flow/Mongo/README.md

@@ -0,0 +1,46 @@
+Usage
+--------------
+
+The usage is analogous to the 'normal' flow-php:
+
+```php
+$config = new \Flow\Mongo\MongoConfig($yourGridFs);
+$file = new \Flow\Mongo\MongoFile($config);
+
+if ($_SERVER['REQUEST_METHOD'] === 'GET') {
+    if ($file->checkChunk()) {
+        header("HTTP/1.1 200 Ok");
+    } else {
+        header("HTTP/1.1 204 No Content");
+        return ;
+    }
+} else {
+  if ($file->validateChunk()) {
+      $file->saveChunk();
+  } else {
+      // error, invalid chunk upload request, retry
+      header("HTTP/1.1 400 Bad Request");
+      return ;
+  }
+}
+if ($file->validateFile()) {
+    // File upload was completed
+    $id = $file->saveToGridFs(['your metadata'=>'value']);
+    if($id) {
+      //do custom post processing here, $id is the MongoId of the gridfs file 
+    }
+} else {
+    // This is not a final chunk, continue to upload
+}
+```
+
+Delete unfinished files
+-----------------------
+
+For this you should setup cron, which would check each chunk upload time.
+If chunk is uploaded long time ago, then chunk should be deleted.
+
+Helper method for checking this:
+```php
+\Flow\Mongo\MongoUploader::pruneChunks($yourGridFs);
+```