Refactor GridFS internals

The Bucket's public API remains unchanged. More logic has been moved into the CollectionWrapper and the stream classes have also been renamed.

The StreamWrapper's registration has been changed to allow a configurable protocol (Bucket defaults to "gridfs"). While file paths are generated for readable and writable streams (now in a more robust manner), necessary arguments are passed through context options. We may decide to simplify the path for readable streams down the line, unless the current path is beneficial for debugging.

Closes #167

Author: jmikola

src/GridFS/Bucket.php

@@ -10,6 +10,7 @@
 use MongoDB\Exception\InvalidArgumentException;
 use MongoDB\GridFS\Exception\FileNotFoundException;
 use MongoDB\Operation\Find;
+use stdClass;
 
 /**
  * Bucket provides a public API for interacting with the GridFS files and chunks
@@ -19,8 +20,8 @@
  */
 class Bucket
 {
-    private static $streamWrapper;
     private static $defaultChunkSizeBytes = 261120;
+    private static $streamWrapperProtocol = 'gridfs';
 
     private $collectionWrapper;
     private $databaseName;
@@ -75,7 +76,7 @@ public function __construct(Manager $manager, $databaseName, array $options = []
         $collectionOptions = array_intersect_key($options, ['readPreference' => 1, 'writeConcern' => 1]);
 
         $this->collectionWrapper = new CollectionWrapper($manager, $databaseName, $options['bucketName'], $collectionOptions);
-        $this->registerStreamWrapper($manager);
+        $this->registerStreamWrapper();
     }
 
     /**
@@ -89,14 +90,12 @@ public function __construct(Manager $manager, $databaseName, array $options = []
      */
     public function delete(ObjectId $id)
     {
-        $file = $this->collectionWrapper->getFilesCollection()->findOne(['_id' => $id]);
-        $this->collectionWrapper->getFilesCollection()->deleteOne(['_id' => $id]);
-        $this->collectionWrapper->getChunksCollection()->deleteMany(['files_id' => $id]);
+        $file = $this->collectionWrapper->findFileById($id);
+        $this->collectionWrapper->deleteFileAndChunksById($id);
 
         if ($file === null) {
-            throw FileNotFoundException::byId($id, $this->collectionWrapper->getFilesCollection()->getNameSpace());
+            throw FileNotFoundException::byId($id, $this->getFilesNamespace());
         }
-
     }
 
     /**
@@ -108,17 +107,14 @@ public function delete(ObjectId $id)
      */
     public function downloadToStream(ObjectId $id, $destination)
     {
-        $file = $this->collectionWrapper->getFilesCollection()->findOne(
-            ['_id' => $id],
-            ['typeMap' => ['root' => 'stdClass']]
-        );
+        $file = $this->collectionWrapper->findFileById($id);
 
         if ($file === null) {
-            throw FileNotFoundException::byId($id, $this->collectionWrapper->getFilesCollection()->getNameSpace());
+            throw FileNotFoundException::byId($id, $this->getFilesNamespace());
         }
 
-        $gridFsStream = new GridFSDownload($this->collectionWrapper, $file);
-        $gridFsStream->downloadToStream($destination);
+        $stream = new ReadableStream($this->collectionWrapper, $file);
+        $stream ->downloadToStream($destination);
     }
 
     /**
@@ -148,23 +144,29 @@ public function downloadToStream(ObjectId $id, $destination)
     public function downloadToStreamByName($filename, $destination, array $options = [])
     {
         $options += ['revision' => -1];
-        $file = $this->findFileRevision($filename, $options['revision']);
-        $gridFsStream = new GridFSDownload($this->collectionWrapper, $file);
-        $gridFsStream->downloadToStream($destination);
+
+        $file = $this->collectionWrapper->findFileByFilenameAndRevision($filename, $options['revision']);
+
+        if ($file === null) {
+            throw FileNotFoundException::byFilenameAndRevision($filename, $options['revision'], $this->getFilesNamespace());
+        }
+
+        $stream = new ReadableStream($this->collectionWrapper, $file);
+        $stream->downloadToStream($destination);
     }
 
     /**
-    * Drops the files and chunks collection associated with GridFS this bucket
-    *
-    */
-
+     * Drops the files and chunks collections associated with this GridFS
+     * bucket.
+     */
     public function drop()
     {
         $this->collectionWrapper->dropCollections();
     }
 
     /**
-     * Find files from the GridFS bucket's files collection.
+     * Finds documents from the GridFS bucket's files collection matching the
+     * query.
      *
      * @see Find::__construct() for supported options
      * @param array|object $filter  Query by which to filter documents
@@ -173,10 +175,10 @@ public function drop()
      */
     public function find($filter, array $options = [])
     {
-        return $this->collectionWrapper->getFilesCollection()->find($filter, $options);
+        return $this->collectionWrapper->findFiles($filter, $options);
     }
 
-    public function getCollectionsWrapper()
+    public function getCollectionWrapper()
     {
         return $this->collectionWrapper;
     }
@@ -200,7 +202,7 @@ public function getIdFromStream($stream)
             return $metadata['wrapper_data']->getId();
         }
 
-        return;
+        // TODO: Throw if we cannot access the ID
     }
 
     /**
@@ -212,13 +214,10 @@ public function getIdFromStream($stream)
      */
     public function openDownloadStream(ObjectId $id)
     {
-        $file = $this->collectionWrapper->getFilesCollection()->findOne(
-            ['_id' => $id],
-            ['typeMap' => ['root' => 'stdClass']]
-        );
+        $file = $this->collectionWrapper->findFileById($id);
 
         if ($file === null) {
-            throw FileNotFoundException::byId($id, $this->collectionWrapper->getFilesCollection()->getNameSpace());
+            throw FileNotFoundException::byId($id, $this->getFilesNamespace());
         }
 
         return $this->openDownloadStreamByFile($file);
@@ -251,7 +250,12 @@ public function openDownloadStream(ObjectId $id)
     public function openDownloadStreamByName($filename, array $options = [])
     {
         $options += ['revision' => -1];
-        $file = $this->findFileRevision($filename, $options['revision']);
+
+        $file = $this->collectionWrapper->findFileByFilenameAndRevision($filename, $options['revision']);
+
+        if ($file === null) {
+            throw FileNotFoundException::byFilenameAndRevision($filename, $options['revision'], $this->getFilesNamespace());
+        }
 
         return $this->openDownloadStreamByFile($file);
     }
@@ -265,36 +269,51 @@ public function openDownloadStreamByName($filename, array $options = [])
      *    bucket's chunk size.
      *
      * @param string $filename File name
-     * @param array  $options  Stream options
+     * @param array  $options  Upload options
      * @return resource
      */
     public function openUploadStream($filename, array $options = [])
     {
         $options += ['chunkSizeBytes' => $this->options['chunkSizeBytes']];
 
-        $streamOptions = [
-            'collectionWrapper' => $this->collectionWrapper,
-            'uploadOptions' => $options,
-        ];
-
-        $context = stream_context_create(['gridfs' => $streamOptions]);
+        $path = $this->createPathForUpload();
+        $context = stream_context_create([
+            self::$streamWrapperProtocol => [
+                'collectionWrapper' => $this->collectionWrapper,
+                'filename' => $filename,
+                'options' => $options,
+            ],
+        ]);
 
-        return fopen(sprintf('gridfs://%s/%s', $this->databaseName, $filename), 'w', false, $context);
+        return fopen($path, 'w', false, $context);
     }
 
     /**
      * Renames the GridFS file with the specified ID.
      *
      * @param ObjectId $id          ID of the file to rename
      * @param string   $newFilename New file name
-     * @throws GridFSFileNotFoundException
+     * @throws FileNotFoundException
      */
     public function rename(ObjectId $id, $newFilename)
     {
-        $filesCollection = $this->collectionWrapper->getFilesCollection();
-        $result = $filesCollection->updateOne(['_id' => $id], ['$set' => ['filename' => $newFilename]]);
-        if($result->getModifiedCount() == 0) {
-            throw FileNotFoundException::byId($id, $this->collectionWrapper->getFilesCollection()->getNameSpace());
+        $updateResult = $this->collectionWrapper->updateFilenameForId($id, $newFilename);
+
+        if ($updateResult->getModifiedCount() === 1) {
+            return;
+        }
+
+        /* If the update resulted in no modification, it's possible that the
+         * file did not exist, in which case we must raise an error. Checking
+         * the write result's matched count will be most efficient, but fall
+         * back to a findOne operation if necessary (i.e. legacy writes).
+         */
+        $found = $updateResult->getMatchedCount() !== null
+            ? $updateResult->getMatchedCount() === 1
+            : $this->collectionWrapper->findFileById($id) !== null;
+
+        if ( ! $found) {
+            throw FileNotFoundException::byId($id, $this->getFilesNamespace());
         }
     }
 
@@ -310,61 +329,93 @@ public function rename(ObjectId $id, $newFilename)
      * @param resource $source   Readable stream
      * @param array    $options  Stream options
      * @return ObjectId
+     * @throws InvalidArgumentException
      */
     public function uploadFromStream($filename, $source, array $options = [])
     {
         $options += ['chunkSizeBytes' => $this->options['chunkSizeBytes']];
-        $gridFsStream = new GridFSUpload($this->collectionWrapper, $filename, $options);
 
-        return $gridFsStream->uploadFromStream($source);
+        $stream = new WritableStream($this->collectionWrapper, $filename, $options);
+
+        return $stream->uploadFromStream($source);
     }
 
-    private function findFileRevision($filename, $revision)
+    /**
+     * Creates a path for an existing GridFS file.
+     *
+     * @param stdClass $file GridFS file document
+     * @return string
+     */
+    private function createPathForFile(stdClass $file)
     {
-        if ($revision < 0) {
-            $skip = abs($revision) - 1;
-            $sortOrder = -1;
+        if ( ! is_object($file->_id) || method_exists($file->_id, '__toString')) {
+            $id = (string) $file->_id;
         } else {
-            $skip = $revision;
-            $sortOrder = 1;
+            $id = \MongoDB\BSON\toJSON(\MongoDB\BSON\fromPHP(['_id' => $file->_id]));
         }
 
-        $filesCollection = $this->collectionWrapper->getFilesCollection();
-        $file = $filesCollection->findOne(
-            ['filename' => $filename],
-            [
-                'skip' => $skip,
-                'sort' => ['uploadDate' => $sortOrder],
-                'typeMap' => ['root' => 'stdClass'],
-            ]
+        return sprintf(
+            '%s://%s/%s.files/%s',
+            self::$streamWrapperProtocol,
+            urlencode($this->databaseName),
+            urlencode($this->options['bucketName']),
+            urlencode($id)
         );
-
-        if ($file === null) {
-            throw FileNotFoundException::byFilenameAndRevision($filename, $revision, $filesCollection->getNameSpace());
-        }
-
-        return $file;
     }
 
-    private function openDownloadStreamByFile($file)
+    /**
+     * Creates a path for a new GridFS file, which does not yet have an ID.
+     *
+     * @return string
+     */
+    private function createPathForUpload()
     {
-        $options = [
-            'collectionWrapper' => $this->collectionWrapper,
-            'file' => $file,
-        ];
+        return sprintf(
+            '%s://%s/%s.files',
+            self::$streamWrapperProtocol,
+            urlencode($this->databaseName),
+            urlencode($this->options['bucketName'])
+        );
+    }
 
-        $context = stream_context_create(['gridfs' => $options]);
+    /**
+     * Returns the names of the files collection.
+     *
+     * @return string
+     */
+    private function getFilesNamespace()
+    {
+        return sprintf('%s.%s.files', $this->databaseName, $this->options['bucketName']);
+    }
 
-        return fopen(sprintf('gridfs://%s/%s', $this->databaseName, $file->filename), 'r', false, $context);
+    /**
+     * Opens a readable stream for the GridFS file.
+     *
+     * @param stdClass $file GridFS file document
+     * @return resource
+     */
+    private function openDownloadStreamByFile(stdClass $file)
+    {
+        $path = $this->createPathForFile($file);
+        $context = stream_context_create([
+            self::$streamWrapperProtocol => [
+                'collectionWrapper' => $this->collectionWrapper,
+                'file' => $file,
+            ],
+        ]);
+
+        return fopen($path, 'r', false, $context);
     }
 
-    private function registerStreamWrapper(Manager $manager)
+    /**
+     * Registers the GridFS stream wrapper if it is not already registered.
+     */
+    private function registerStreamWrapper()
     {
-        if (isset(self::$streamWrapper)) {
+        if (in_array(self::$streamWrapperProtocol, stream_get_wrappers())) {
             return;
         }
 
-        self::$streamWrapper = new StreamWrapper();
-        self::$streamWrapper->register($manager);
+        StreamWrapper::register(self::$streamWrapperProtocol);
     }
 }