5
5
use MongoDB \BSON \Binary ;
6
6
use MongoDB \BSON \ObjectId ;
7
7
use MongoDB \BSON \UTCDateTime ;
8
+ use MongoDB \Driver \Exception \RuntimeException as DriverRuntimeException ;
8
9
use MongoDB \Exception \InvalidArgumentException ;
10
+ use MongoDB \Exception \RuntimeException ;
9
11
10
12
/**
11
13
* WritableStream abstracts the process of writing a GridFS file.
@@ -16,8 +18,7 @@ class WritableStream
16
18
{
17
19
private static $ defaultChunkSizeBytes = 261120 ;
18
20
19
- private $ buffer ;
20
- private $ bufferLength = 0 ;
21
+ private $ buffer = '' ;
21
22
private $ chunkOffset = 0 ;
22
23
private $ chunkSize ;
23
24
private $ collectionWrapper ;
@@ -66,6 +67,10 @@ public function __construct(CollectionWrapper $collectionWrapper, $filename, arr
66
67
throw InvalidArgumentException::invalidType ('"chunkSizeBytes" option ' , $ options ['chunkSizeBytes ' ], 'integer ' );
67
68
}
68
69
70
+ if (isset ($ options ['chunkSizeBytes ' ]) && $ options ['chunkSizeBytes ' ] < 1 ) {
71
+ throw new InvalidArgumentException (sprintf ('Expected "chunkSizeBytes" option to be >= 1, %d given ' , $ options ['chunkSizeBytes ' ]));
72
+ }
73
+
69
74
if (isset ($ options ['contentType ' ]) && ! is_string ($ options ['contentType ' ])) {
70
75
throw InvalidArgumentException::invalidType ('"contentType" option ' , $ options ['contentType ' ], 'string ' );
71
76
}
@@ -76,15 +81,13 @@ public function __construct(CollectionWrapper $collectionWrapper, $filename, arr
76
81
77
82
$ this ->chunkSize = $ options ['chunkSizeBytes ' ];
78
83
$ this ->collectionWrapper = $ collectionWrapper ;
79
- $ this ->buffer = fopen ('php://memory ' , 'w+b ' );
80
84
$ this ->ctx = hash_init ('md5 ' );
81
85
82
86
$ this ->file = [
83
87
'_id ' => $ options ['_id ' ],
84
88
'chunkSize ' => $ this ->chunkSize ,
85
89
'filename ' => (string ) $ filename ,
86
- // TODO: This is necessary until PHPC-536 is implemented
87
- 'uploadDate ' => new UTCDateTime ((int ) floor (microtime (true ) * 1000 )),
90
+ 'uploadDate ' => new UTCDateTime ,
88
91
] + array_intersect_key ($ options , ['aliases ' => 1 , 'contentType ' => 1 , 'metadata ' => 1 ]);
89
92
}
90
93
@@ -113,14 +116,10 @@ public function close()
113
116
return ;
114
117
}
115
118
116
- rewind ($ this ->buffer );
117
- $ cached = stream_get_contents ($ this ->buffer );
118
-
119
- if (strlen ($ cached ) > 0 ) {
120
- $ this ->insertChunk ($ cached );
119
+ if (strlen ($ this ->buffer ) > 0 ) {
120
+ $ this ->insertChunkFromBuffer ();
121
121
}
122
122
123
- fclose ($ this ->buffer );
124
123
$ this ->fileCollectionInsert ();
125
124
$ this ->isClosed = true ;
126
125
}
@@ -151,77 +150,87 @@ public function getSize()
151
150
* Inserts binary data into GridFS via chunks.
152
151
*
153
152
* Data will be buffered internally until chunkSizeBytes are accumulated, at
154
- * which point a chunk's worth of data will be inserted and the buffer
155
- * reset.
153
+ * which point a chunk document will be inserted and the buffer reset.
156
154
*
157
- * @param string $toWrite Binary data to write
155
+ * @param string $data Binary data to write
158
156
* @return integer
159
157
*/
160
- public function insertChunks ( $ toWrite )
158
+ public function writeBytes ( $ data )
161
159
{
162
160
if ($ this ->isClosed ) {
163
161
// TODO: Should this be an error condition? e.g. BadMethodCallException
164
162
return ;
165
163
}
166
164
167
- $ readBytes = 0 ;
165
+ $ bytesRead = 0 ;
168
166
169
- while ($ readBytes != strlen ($ toWrite )) {
170
- $ addToBuffer = substr ($ toWrite , $ readBytes , $ this ->chunkSize - $ this ->bufferLength );
171
- fwrite ($ this ->buffer , $ addToBuffer );
172
- $ readBytes += strlen ($ addToBuffer );
173
- $ this ->bufferLength += strlen ($ addToBuffer );
167
+ while ($ bytesRead != strlen ($ data )) {
168
+ $ initialBufferLength = strlen ($ this ->buffer );
169
+ $ this ->buffer .= substr ($ data , $ bytesRead , $ this ->chunkSize - $ initialBufferLength );
170
+ $ bytesRead += strlen ($ this ->buffer ) - $ initialBufferLength ;
174
171
175
- if ($ this ->bufferLength == $ this ->chunkSize ) {
176
- rewind ($ this ->buffer );
177
- $ this ->insertChunk (stream_get_contents ($ this ->buffer ));
178
- ftruncate ($ this ->buffer , 0 );
179
- $ this ->bufferLength = 0 ;
172
+ if (strlen ($ this ->buffer ) == $ this ->chunkSize ) {
173
+ $ this ->insertChunkFromBuffer ();
180
174
}
181
175
}
182
176
183
- return $ readBytes ;
177
+ return $ bytesRead ;
184
178
}
185
179
186
180
private function abort ()
187
181
{
188
- $ this ->collectionWrapper ->deleteChunksByFilesId ($ this ->file ['_id ' ]);
182
+ try {
183
+ $ this ->collectionWrapper ->deleteChunksByFilesId ($ this ->file ['_id ' ]);
184
+ } catch (DriverRuntimeException $ e ) {
185
+ // We are already handling an error if abort() is called, so suppress this
186
+ }
187
+
189
188
$ this ->isClosed = true ;
190
189
}
191
190
192
191
private function fileCollectionInsert ()
193
192
{
194
- if ($ this ->isClosed ) {
195
- // TODO: Should this be an error condition? e.g. BadMethodCallException
196
- return ;
197
- }
198
-
199
193
$ md5 = hash_final ($ this ->ctx );
200
194
201
195
$ this ->file ['length ' ] = $ this ->length ;
202
196
$ this ->file ['md5 ' ] = $ md5 ;
203
197
204
- $ this ->collectionWrapper ->insertFile ($ this ->file );
198
+ try {
199
+ $ this ->collectionWrapper ->insertFile ($ this ->file );
200
+ } catch (DriverRuntimeException $ e ) {
201
+ $ this ->abort ();
202
+
203
+ throw $ e ;
204
+ }
205
205
206
206
return $ this ->file ['_id ' ];
207
207
}
208
208
209
- private function insertChunk ( $ data )
209
+ private function insertChunkFromBuffer ( )
210
210
{
211
- if ($ this ->isClosed ) {
212
- // TODO: Should this be an error condition? e.g. BadMethodCallException
211
+ if (strlen ($ this ->buffer ) == 0 ) {
213
212
return ;
214
213
}
215
214
216
- $ toUpload = [
215
+ $ data = $ this ->buffer ;
216
+ $ this ->buffer = '' ;
217
+
218
+ $ chunk = [
217
219
'files_id ' => $ this ->file ['_id ' ],
218
220
'n ' => $ this ->chunkOffset ,
219
221
'data ' => new Binary ($ data , Binary::TYPE_GENERIC ),
220
222
];
221
223
222
224
hash_update ($ this ->ctx , $ data );
223
225
224
- $ this ->collectionWrapper ->insertChunk ($ toUpload );
226
+ try {
227
+ $ this ->collectionWrapper ->insertChunk ($ chunk );
228
+ } catch (DriverRuntimeException $ e ) {
229
+ $ this ->abort ();
230
+
231
+ throw $ e ;
232
+ }
233
+
225
234
$ this ->length += strlen ($ data );
226
235
$ this ->chunkOffset ++;
227
236
}
0 commit comments