Skip to content

Commit 623a98b

Browse files
author
Will Banfield
committed
PHPLIB-147: Implement GridFS download
1 parent 56023b0 commit 623a98b

File tree

7 files changed

+280
-91
lines changed

7 files changed

+280
-91
lines changed

‎src/Exception/GridFSCorruptFileException.php

Lines changed: 7 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,7 @@
1+
<?php
2+
3+
namespace MongoDB\Exception;
4+
5+
class GridFSCorruptFileException extends \MongoDB\Driver\Exception\RuntimeException implements Exception
6+
{
7+
}

‎src/Exception/GridFSFileNotFoundException.php

Lines changed: 10 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,10 @@
1+
<?php
2+
3+
namespace MongoDB\Exception;
4+
5+
class GridFSFileNotFoundException extends \MongoDB\Driver\Exception\RuntimeException implements Exception
6+
{
7+
public function __construct($fname, $bucketName, $databaseName){
8+
parent::__construct(sprintf('Unable to find file by: %s in %s.%s', $fname,$databaseName, $bucketName));
9+
}
10+
}

‎src/GridFS/Bucket.php

Lines changed: 7 additions & 41 deletions
Original file line numberDiff line numberDiff line change
@@ -22,7 +22,6 @@ class Bucket
2222
private $options;
2323
private $filesCollection;
2424
private $chunksCollection;
25-
private $indexChecker;
2625
private $ensuredIndexes = false;
2726
/**
2827
* Constructs a GridFS bucket.
@@ -85,42 +84,6 @@ public function __construct(Manager $manager, $databaseName, array $options = []
8584
$collectionOptions
8685
);
8786
}
88-
/**
89-
* Opens a Stream for reading the contents of a file specified by ID.
90-
*
91-
* @param ObjectId $id
92-
* @return Stream
93-
*/
94-
public function openDownloadStream(ObjectId $id)
95-
{
96-
fopen('gridfs://$this->databaseName/$id', 'r');
97-
}
98-
/**
99-
* Downloads the contents of the stored file specified by id and writes
100-
* the contents to the destination Stream.
101-
* @param ObjectId $id GridFS File Id
102-
* @param Stream $destination Destination Stream
103-
*/
104-
public function downloadToStream(ObjectId $id, $destination)
105-
{
106-
$result = $this->filesCollection->findOne(['_id' => $id]);
107-
if ($result == null) {
108-
return;
109-
}
110-
if ($result->length == 0){
111-
return;
112-
}
113-
114-
$n=0;
115-
$results = $this->chunksCollection->find(['files_id' => $result->_id]);
116-
foreach ($results as $chunk) {
117-
if ($chunk->n != $n) {
118-
return;
119-
}
120-
fwrite($destination, $chunk->data);
121-
$n++;
122-
}
123-
}
12487
/**
12588
* Return the chunkSizeBytes option for this Bucket.
12689
*
@@ -130,6 +93,7 @@ public function getChunkSizeBytes()
13093
{
13194
return $this->options['chunkSizeBytes'];
13295
}
96+
13397
public function getDatabaseName()
13498
{
13599
return $this->databaseName;
@@ -138,10 +102,15 @@ public function getFilesCollection()
138102
{
139103
return $this->filesCollection;
140104
}
105+
141106
public function getChunksCollection()
142107
{
143108
return $this->chunksCollection;
144109
}
110+
public function getBucketName()
111+
{
112+
return $this->options['bucketName'];
113+
}
145114
public function find($filter, array $options =[])
146115
{
147116
//add proper validation for the filter and for the options
@@ -168,7 +137,6 @@ public function ensureIndexes()
168137
$this->ensureChunksIndex();
169138
$this->ensuredIndexes = true;
170139
}
171-
172140
private function ensureChunksIndex()
173141
{
174142
foreach ($this->chunksCollection->listIndexes() as $index) {
@@ -178,7 +146,6 @@ private function ensureChunksIndex()
178146
}
179147
$this->chunksCollection->createIndex(['files_id' => 1, 'n' => 1], ['unique' => true]);
180148
}
181-
182149
private function ensureFilesIndex()
183150
{
184151
foreach ($this->filesCollection->listIndexes() as $index) {
@@ -188,19 +155,18 @@ private function ensureFilesIndex()
188155
}
189156
$this->filesCollection->createIndex(['filename' => 1, 'uploadDate' => 1]);
190157
}
191-
192158
private function isFilesCollectionEmpty()
193159
{
194160
return null === $this->filesCollection->findOne([], [
195161
'readPreference' => new ReadPreference(ReadPreference::RP_PRIMARY),
196162
'projection' => ['_id' => 1],
197163
]);
198164
}
199-
200165
public function delete(ObjectId $id)
201166
{
202167
$options = ['writeConcern' => $this->writeConcern];
203168
$this->chunksCollection->deleteMany(['file_id' => $id], $options);
169+
204170
$this->filesCollection->deleteOne(['_id' => $id], $options);
205171
}
206172
}

‎src/GridFS/BucketReadWriter.php

Lines changed: 26 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -38,4 +38,30 @@ public function uploadFromStream($filename, $source, array $options = [])
3838
$gridFsStream = new GridFsUpload($this->bucket, $filename, $options);
3939
return $gridFsStream->uploadFromStream($source);
4040
}
41+
/**
42+
* Opens a Stream for reading the contents of a file specified by ID.
43+
*
44+
* @param ObjectId $id
45+
* @return Stream
46+
*/
47+
public function openDownloadStream(\MongoDB\BSON\ObjectId $id)
48+
{
49+
$options = [
50+
'bucket' => $this->bucket
51+
];
52+
$context = stream_context_create(['gridfs' => $options]);
53+
return fopen(sprintf('gridfs://%s/%s', $this->bucket->getDatabaseName(), $id), 'r', false, $context);
54+
}
55+
/**
56+
* Downloads the contents of the stored file specified by id and writes
57+
* the contents to the destination Stream.
58+
* @param ObjectId $id GridFS File Id
59+
* @param Stream $destination Destination Stream
60+
*/
61+
public function downloadToStream(\MongoDB\BSON\ObjectId $id, $destination)
62+
{
63+
$gridFsStream = new GridFsDownload($this->bucket, $id);
64+
$gridFsStream->downloadToStream($destination);
65+
}
66+
4167
}

‎src/GridFS/GridFsDownload.php

Lines changed: 144 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,144 @@
1+
<?php
2+
namespace MongoDB\GridFS;
3+
4+
use MongoDB\Collection;
5+
use MongoDB\Exception\RuntimeException;
6+
use MongoDB\BSON\ObjectId;
7+
/**
8+
* GridFsupload abstracts the processes of inserting into a GridFSBucket
9+
*
10+
* @api
11+
*/
12+
class GridFsDownload extends GridFsStream
13+
{
14+
private $chunksIterator;
15+
private $bytesSeen=0;
16+
private $numChunks;
17+
private $iteratorEmpty=false;
18+
private $firstCheck=true;
19+
private $bufferFresh=true;
20+
private $bufferEmpty=true;
21+
/**
22+
* Constructs a GridFS upload stream
23+
*
24+
* Supported options:
25+
*
26+
* * contentType (string): DEPRECATED content type to be stored with the file.
27+
* This information should now be added to the metadata
28+
*
29+
* * aliases (array of strings): DEPRECATED An array of aliases.
30+
* Applications wishing to store aliases should add an aliases field to the
31+
* metadata document instead.
32+
*
33+
* * metadata (array or object): User data for the 'metadata' field of the files
34+
* collection document.
35+
*
36+
* * writeConcern (MongoDB\Driver\WriteConcern): Write concern.
37+
*
38+
* @param array $options File options
39+
* @throws FileNotFoundException
40+
*/
41+
public function __construct(
42+
Bucket $bucket,
43+
ObjectId $objectId
44+
)
45+
{
46+
$this->file = $bucket->getFilesCollection()->findOne(['_id' => $objectId]);
47+
if (is_null($this->file)) {
48+
//MUST RAISE AN ERROR ! (WHICH ONE I DON'T)
49+
throw new \MongoDB\Exception\GridFSFileNotFoundException($objectId, $bucket->getBucketName(), $bucket->getDatabaseName());
50+
}
51+
if ($this->file->length > 0) {
52+
$cursor = $bucket->getChunksCollection()->find(['files_id' => $this->file->_id], ['sort' => ['n' => 1]]);
53+
$this->chunksIterator = new \IteratorIterator($cursor);
54+
$this->numChunks = ceil($this->file->length / $this->file->chunkSize);
55+
}
56+
parent::__construct($bucket);
57+
}
58+
/**
59+
* Reads data from a stream into GridFS
60+
*
61+
* @param Stream $source Source Stream
62+
* @return ObjectId
63+
*/
64+
public function downloadToStream($destination)
65+
{
66+
while($this->advanceChunks()) {
67+
fwrite($destination, $this->chunksIterator->current()->data->getData());
68+
}
69+
}
70+
71+
public function downloadNumBytes($numToRead) {
72+
$output = "";
73+
if ($this->bufferFresh) {
74+
rewind($this->buffer);
75+
$this->bufferFresh=false;
76+
}
77+
78+
$output = fread($this->buffer, $numToRead);
79+
if (strlen($output) == $numToRead) {
80+
return $output;
81+
}
82+
fclose($this->buffer);
83+
$this->buffer = fopen("php://temp", "w+");
84+
85+
$this->bufferFresh=true;
86+
$this->bufferEmpty=true;
87+
88+
$bytesLeft = $numToRead - strlen($output);
89+
90+
while(strlen($output) < $numToRead && $this->advanceChunks()) {
91+
$bytesLeft = $numToRead - strlen($output);
92+
$output .= substr($this->chunksIterator->current()->data, 0, $bytesLeft);
93+
}
94+
if ($bytesLeft < strlen($this->chunksIterator->current()->data)) {
95+
fwrite($this->buffer, substr($this->chunksIterator->current()->data, $bytesLeft));
96+
$this->bufferEmpty=false;
97+
}
98+
return $output;
99+
}
100+
101+
private function advanceChunks()
102+
{
103+
if($this->n >= $this->numChunks) {
104+
$this->iteratorEmpty=true;
105+
return false;
106+
}
107+
if($this->firstCheck) {
108+
$this->chunksIterator->rewind();
109+
$this->firstCheck=false;
110+
} else {
111+
$this->chunksIterator->next();
112+
}
113+
if (!$this->chunksIterator->valid()) {
114+
throw new \MongoDB\Exception\GridFSCorruptFileException();
115+
}
116+
if ($this->chunksIterator->current()->n != $this->n) {
117+
throw new \MongoDB\Exception\GridFSCorruptFileException();
118+
}
119+
$chunkSizeIs = strlen($this->chunksIterator->current()->data->getData());
120+
if ($this->n == $this->numChunks - 1) {
121+
$chunkSizeShouldBe = $this->file->length - $this->bytesSeen;
122+
if($chunkSizeShouldBe != $chunkSizeIs) {
123+
throw new \MongoDB\Exception\GridFSCorruptFileException();
124+
}
125+
} else if ($this->n < $this->numChunks - 1) {
126+
if($chunkSizeIs != $this->file->chunkSize) {
127+
throw new \MongoDB\Exception\GridFSCorruptFileException();
128+
}
129+
}
130+
$this->bytesSeen+= $chunkSizeIs;
131+
$this->n++;
132+
return true;
133+
}
134+
public function close()
135+
{
136+
fclose($this->buffer);
137+
}
138+
139+
public function isEOF()
140+
{
141+
$eof = $this->iteratorEmpty && $this->bufferEmpty;
142+
return $eof;
143+
}
144+
}

‎src/GridFS/StreamWrapper.php

Lines changed: 3 additions & 38 deletions
Original file line numberDiff line numberDiff line change
@@ -33,60 +33,27 @@ public static function register()
3333
}
3434
stream_wrapper_register('gridfs', get_called_class(), STREAM_IS_URL);
3535
}
36-
3736
private function initProtocol($path)
3837
{
3938
$parsed_path = parse_url($path);
4039
$this->databaseName = $parsed_path["host"];
4140
$this->identifier = substr($parsed_path["path"], 1);
4241
}
43-
4442
public function stream_write($data)
4543
{
4644
$this->gridFsStream->insertChunks($data);
4745
return strlen($data);
4846
}
49-
5047
public function stream_read($count) {
51-
$out ="";
52-
if ($this->dirtyCache) {
53-
$out = fread($this->buffer, $count);
54-
if (strlen($out) == $count) {
55-
return $out;
56-
} else {
57-
fclose($out);
58-
$this->dirtyCache = false;
59-
}
60-
$this->n++;
61-
}
62-
63-
if ($this->file->length <= $this->n) {
64-
return false;
65-
}
66-
67-
while(strlen($out) < $count && $this ->n <$this->file->length) {
68-
$bytes_left = $count - strlen($out);
69-
$next = $this->chunksCollection->findOne(['files_id' => $this->file->_id, "n" => $this->n]);
70-
$out .= substr($next->data, 0, $bytes_left);
71-
$this->n++;
72-
}
73-
if ($bytes_left < strlen($next->data)) {
74-
$this->buffer = tmpfile();
75-
fwrite($this->buffer, substr($next->data, $bytes_left));
76-
$this->dirtyCache =true;
77-
}
78-
return $out;
48+
return $this->gridFsStream->downloadNumBytes($count);
7949
}
80-
8150
public function stream_eof() {
82-
return $this->n >= $this->file->length;
51+
return $this->gridFsStream->isEOF();
8352
}
84-
8553
public function stream_close() {
8654
$this->gridFsStream->close();
8755

8856
}
89-
9057
public function stream_open($path, $mode, $options, &$openedPath)
9158
{
9259
$this->initProtocol($path);
@@ -99,17 +66,15 @@ public function stream_open($path, $mode, $options, &$openedPath)
9966
default: return false;
10067
}
10168
}
102-
10369
public function openWriteStream() {
10470
$context = stream_context_get_options($this->context);
10571
$options =$context['gridfs']['uploadOptions'];
10672
$this->gridFsStream = new GridFsUpload($this->bucket, $this->identifier, $options);
10773
return true;
10874
}
109-
11075
public function openReadStream() {
11176
$objectId = new \MongoDB\BSON\ObjectId($this->identifier);
112-
$this->file = $this->filesCollection->findOne(['_id' => $objectId]);
77+
$this->gridFsStream = new GridFsDownload($this->bucket, $objectId);
11378
return true;
11479
}
11580
}

0 commit comments

Comments
 (0)