WIP for GridFS Spec

Author: Will Banfield

src/GridFS/Bucket.php

@@ -0,0 +1,258 @@
+<?php
+namespace MongoDB\GridFS;
+
+use MongoDB\Collection;
+use MongoDB\Database;
+use MongoDB\BSON\ObjectId;
+use MongoDB\Driver\ReadPreference;
+use MongoDB\Driver\WriteConcern;
+use MongoDB\Driver\Manager;
+use MongoDB\Exception\InvalidArgumentException;
+use MongoDB\Exception\InvalidArgumentTypeException;
+use MongoDB\Exception\RuntimeException;
+use MongoDB\Exception\UnexpectedValueException;
+/**
+ * Bucket abstracts the GridFS files and chunks collections.
+ *
+ * @api
+ */
+class Bucket
+{
+    private $manager;
+    private $databaseName;
+    private $options;
+    private $filesCollection;
+    private $chunksCollection;
+    private $ensuredIndexes = false;
+    /**
+     * Constructs a GridFS bucket.
+     *
+     * Supported options:
+     *
+     *  * bucketName (string): The bucket name, which will be used as a prefix
+     *    for the files and chunks collections. Defaults to "fs".
+     *
+     *  * chunkSizeBytes (integer): The chunk size in bytes. Defaults to
+     *    261120 (i.e. 255 KiB).
+     *
+     *  * readPreference (MongoDB\Driver\ReadPreference): Read preference.
+     *
+     *  * writeConcern (MongoDB\Driver\WriteConcern): Write concern.
+     *
+     * @param Manager  $manager      Manager instance from the driver
+     * @param string   $databaseName Database name
+     * @param array    $options      Bucket options
+     * @throws InvalidArgumentException
+     */
+    public function __construct(Manager $manager, $databaseName, array $options = [])
+    {
+    	$collectionOpts = array();
+        $options += [
+            'bucketName' => 'fs',
+            'chunkSizeBytes' => 261120,
+        ];
+        if (isset($options['bucketName']) && ! is_string($options['bucketName'])) {
+            throw new InvalidArgumentTypeException('"bucketName" option', $options['bucketName'], 'string');
+        }
+        if (isset($options['chunkSizeBytes']) && ! is_integer($options['chunkSizeBytes'])) {
+            throw new InvalidArgumentTypeException('"chunkSizeBytes" option', $options['chunkSizeBytes'], 'integer');
+        }
+        if (isset($options['readPreference'])) {
+        	if (! $options['readPreference'] instanceof ReadPreference) {
+            	throw new InvalidArgumentTypeException('"readPreference" option', $options['readPreference'], 'MongoDB\Driver\ReadPreference');
+        	} else {
+        		$collectionOpts['readPreference'] = $options['readPreference'];
+        	}
+        }
+        if (isset($options['writeConcern'])) {
+        	if (! $options['writeConcern'] instanceof WriteConcern) {
+            	throw new InvalidArgumentTypeException('"writeConcern" option', $options['writeConcern'], 'MongoDB\Driver\WriteConcern');
+        	} else {
+        		$collectionOpts['writeConcern'] = $options['writeConcern'];
+        	}
+        }
+        $this->manager = $manager;
+        $this->databaseName = (string) $databaseName;
+        $this->options = $options;
+
+        $this->filesCollection = new Collection(
+            $manager,
+            sprintf('%s.%s.files', $this->databaseName, $options['bucketName']),
+            $collectionOpts
+        );
+        $this->chunksCollection = new Collection(
+            $manager,
+            sprintf('%s.%s.chunks', $this->databaseName, $options['bucketName']),
+            $collectionOpts
+        );
+    }
+    /**
+     * Opens a Stream for reading the contents of a file specified by ID.
+     *
+     * @param ObjectId $id
+     * @return Stream
+     */
+    public function openDownloadStream(ObjectId $id)
+    {
+    	fopen("gridfs://$this->databaseName/$id", "r");
+    }
+ 	  /**
+ 	  * Downloads the contents of the stored file specified by id and writes
+ 	  * the contents to the destination Stream.
+ 	  * @param ObjectId $id
+ 	  * @param Stream destination
+ 	  */
+  	public function downloadToStream(ObjectId $id, $destination)
+  	{
+  		$result = $this->filesCollection->findOne(["_id" => $id]);
+  		if($result == null) {
+  			return;
+  		}
+  		if($result->length == 0){
+  			return;
+  		}
+
+  		$n=0;
+  		$results = $this->chunksCollection->find(["files_id" => $result->_id]);
+  		foreach($results as $chunk) {
+  			if($chunk->n != $n){
+  				return;
+  			}
+  			fwrite($destination, $chunk->data);
+  			$n++;
+  		}
+  	}
+
+    /**
+     * Return the chunkSizeBytes option for this Bucket.
+     *
+     * @return integer
+     */
+    public function getChunkSizeBytes()
+    {
+        return $this->options['chunkSizeBytes'];
+    }
+    /**
+     * Opens a Stream for writing the contents of a file.
+     *
+     * @param string   $filename     file to up
+     * @param array	   $options      options for the stream
+     * @return Stream  uploadStream  what stream to upload from
+     */
+    public function openUploadStream($filename, array $options = [])
+    {
+    	return fopen("gridfs://$this->databaseName/$filename", "w");
+    }
+     /**
+     * Upload a file to this bucket by specifying the source stream file
+     *
+     * @param String fileName
+     * @param Stream source
+     * @param array Options
+     * @return ObjectId
+     */
+
+    public function uploadFromStream($filename, $source, array $options = []) {
+        $id = \MongoDB\BSON\ObjectId();
+        $ctx = hash_init('md5');
+
+        $file_options = [];
+    	if (isset($options['contentType'])) {
+    		if(is_string($options['contentType'])) {
+    			$file_options['contentType'] = $options['contentType'];
+    		} else {
+    			throw new InvalidArgumentTypeException('"contentType" option', $options['contentType'], 'string');
+    		}
+        }
+       	if (isset($options['aliases'])) {
+    		if(isStringArray($options['aliases'])) {
+    			$file_options['aliases'] = $options['aliases'];
+    		} else {
+    			throw new InvalidArgumentTypeException('"aliases" option', $options['aliases'], 'array of strings');
+    		}
+        }
+
+        if (isset($options['metadata'])) {
+    		if(is_array($options['metadata']) || is_object($options['metadata'])) {
+    			$file_options['metadata'] = $options['metadata'];
+    		} else {
+    			throw new InvalidArgumentTypeException('"metadata" option', $options['metadata'], 'object or array');
+    		}
+        }
+
+        $this->ensureIndexes();
+
+    	$n=0;
+    	while ($data = fread($source, $this->getChunkSizeBytes())) {
+    		$toUpload = ["files_id" => $id, "n" => $n, "data" => $data];
+    		$this->chunksCollection->insertOne($toUpload);
+    		hash_update($ctx, $data);
+    		$n++;
+    	}
+    	$uploadDate = time();
+    	$md5 = hash_final($ctx, true);
+    	$file = ["chunkSize" => $this->getChunkSizeBytes(), "filename" => $filename, "uploadDate" => $uploadDate,
+    			"md5" => $md5, "length" => $this->n];
+    	$file = array_merge($file, $file_options);
+    	$doc = $this->filesCollection->insertOne($file);
+    	return $id;
+    }
+
+    private function isStringArray($input) {
+    	if (!is_array($input)){
+    		return false;
+    	}
+    	foreach($input as $item) {
+    		if (!is_string($item)) {
+    			return false;
+    		}
+    	}
+    	return true;
+    }
+
+    private function ensureIndexes()
+    {
+        // Indexes should only be ensured once before the first write operation
+        if ($this->ensuredIndexes) {
+            return;
+        }
+        if ( ! $this->isFilesCollectionEmpty()) {
+            return;
+        }
+        $this->ensureFilesIndex();
+        $this->ensureChunksIndex();
+        $this->ensuredIndexes = true;
+    }
+    private function ensureChunksIndex()
+    {
+        foreach ($this->chunksCollection->listIndexes() as $index) {
+            if ($index->isUnique() && $index->getKey() === ['files_id' => 1, 'n' => 1]) {
+                return;
+            }
+        }
+        $this->chunksCollection->createIndex(['files_id' => 1, 'n' => 1], ['unique' => true]);
+    }
+    private function ensureFilesIndex()
+    {
+        foreach ($this->filesCollection->listIndexes() as $index) {
+            if ($index->getKey() === ['filename' => 1, 'uploadDate' => 1]) {
+                return;
+            }
+        }
+        $this->filesCollection->createIndex(['filename' => 1, 'uploadDate' => 1]);
+    }
+    private function isFilesCollectionEmpty()
+    {
+        return null === $this->filesCollection->findOne([], [
+            'readPreference' => new ReadPreference(ReadPreference::RP_PRIMARY),
+            'projection' => ['_id' => 1],
+        ]);
+    }
+    public function delete(ObjectId $id)
+    {
+    	$options = array("writeConcern" => $this->writeConcern);
+    	$this->chunksCollection->deleteMany(["file_id" => $id], $options);
+
+    	$this->filesCollection->deleteOne(["_id" => $id], $options);
+    }
+}