Merge pull request #12072 from nextcloud/writestream

extend storage api to allow directly writing a stream to storage

apps/dav/lib/Connector/Sabre/File.php

@@ -164,14 +164,19 @@ class File extends Node implements IFile {
 				$this->changeLock(ILockingProvider::LOCK_EXCLUSIVE);
 			}
 
-			$target = $partStorage->fopen($internalPartPath, 'wb');
-			if ($target === false) {
-				\OC::$server->getLogger()->error('\OC\Files\Filesystem::fopen() failed', ['app' => 'webdav']);
-				// because we have no clue about the cause we can only throw back a 500/Internal Server Error
-				throw new Exception('Could not write file contents');
+			if ($partStorage->instanceOfStorage(Storage\IWriteStreamStorage::class)) {
+				$count = $partStorage->writeStream($internalPartPath, $data);
+				$result = $count > 0;
+			} else {
+				$target = $partStorage->fopen($internalPartPath, 'wb');
+				if ($target === false) {
+					\OC::$server->getLogger()->error('\OC\Files\Filesystem::fopen() failed', ['app' => 'webdav']);
+					// because we have no clue about the cause we can only throw back a 500/Internal Server Error
+					throw new Exception('Could not write file contents');
+				}
+				list($count, $result) = \OC_Helper::streamCopy($data, $target);
+				fclose($target);
 			}
-			list($count, $result) = \OC_Helper::streamCopy($data, $target);
-			fclose($target);
 
 			if ($result === false) {
 				$expected = -1;
@@ -185,7 +190,7 @@ class File extends Node implements IFile {
 			// double check if the file was fully received
 			// compare expected and actual size
 			if (isset($_SERVER['CONTENT_LENGTH']) && $_SERVER['REQUEST_METHOD'] === 'PUT') {
-				$expected = (int) $_SERVER['CONTENT_LENGTH'];
+				$expected = (int)$_SERVER['CONTENT_LENGTH'];
 				if ($count !== $expected) {
 					throw new BadRequest('expected filesize ' . $expected . ' got ' . $count);
 				}
@@ -219,7 +224,7 @@ class File extends Node implements IFile {
 					$renameOkay = $storage->moveFromStorage($partStorage, $internalPartPath, $internalPath);
 					$fileExists = $storage->file_exists($internalPath);
 					if ($renameOkay === false || $fileExists === false) {
-						\OC::$server->getLogger()->error('renaming part file to final file failed ($run: ' . ( $run ? 'true' : 'false' ) . ', $renameOkay: '  . ( $renameOkay ? 'true' : 'false' ) . ', $fileExists: ' . ( $fileExists ? 'true' : 'false' ) . ')', ['app' => 'webdav']);
+						\OC::$server->getLogger()->error('renaming part file to final file failed $renameOkay: ' . ($renameOkay ? 'true' : 'false') . ', $fileExists: ' . ($fileExists ? 'true' : 'false') . ')', ['app' => 'webdav']);
 						throw new Exception('Could not rename part file to final file');
 					}
 				} catch (ForbiddenException $ex) {
@@ -246,7 +251,7 @@ class File extends Node implements IFile {
 					$this->header('X-OC-MTime: accepted');
 				}
 			}
-					
+
 			if ($view) {
 				$this->emitPostHooks($exists);
 			}
@@ -443,7 +448,7 @@ class File extends Node implements IFile {
 		//detect aborted upload
 		if (isset ($_SERVER['REQUEST_METHOD']) && $_SERVER['REQUEST_METHOD'] === 'PUT') {
 			if (isset($_SERVER['CONTENT_LENGTH'])) {
-				$expected = (int) $_SERVER['CONTENT_LENGTH'];
+				$expected = (int)$_SERVER['CONTENT_LENGTH'];
 				if ($bytesWritten !== $expected) {
 					$chunk_handler->remove($info['index']);
 					throw new BadRequest(

apps/dav/tests/unit/Connector/Sabre/FileTest.php

@@ -164,7 +164,7 @@ class FileTest extends \Test\TestCase {
 	public function testSimplePutFails($thrownException, $expectedException, $checkPreviousClass = true) {
 		// setup
 		$storage = $this->getMockBuilder(Local::class)
-			->setMethods(['fopen'])
+			->setMethods(['writeStream'])
 			->setConstructorArgs([['datadir' => \OC::$server->getTempManager()->getTemporaryFolder()]])
 			->getMock();
 		\OC\Files\Filesystem::mount($storage, [], $this->user . '/');
@@ -182,11 +182,11 @@ class FileTest extends \Test\TestCase {
 
 		if ($thrownException !== null) {
 			$storage->expects($this->once())
-				->method('fopen')
+				->method('writeStream')
 				->will($this->throwException($thrownException));
 		} else {
 			$storage->expects($this->once())
-				->method('fopen')
+				->method('writeStream')
 				->will($this->returnValue(false));
 		}
 

lib/composer/composer/autoload_classmap.php

@@ -231,6 +231,7 @@ return array(
     'OCP\\Files\\Storage\\INotifyStorage' => $baseDir . '/lib/public/Files/Storage/INotifyStorage.php',
     'OCP\\Files\\Storage\\IStorage' => $baseDir . '/lib/public/Files/Storage/IStorage.php',
     'OCP\\Files\\Storage\\IStorageFactory' => $baseDir . '/lib/public/Files/Storage/IStorageFactory.php',
+    'OCP\\Files\\Storage\\IWriteStreamStorage' => $baseDir . '/lib/public/Files/Storage/IWriteStreamStorage.php',
     'OCP\\Files\\UnseekableException' => $baseDir . '/lib/public/Files/UnseekableException.php',
     'OCP\\Files_FullTextSearch\\Model\\AFilesDocument' => $baseDir . '/lib/public/Files_FullTextSearch/Model/AFilesDocument.php',
     'OCP\\FullTextSearch\\Exceptions\\FullTextSearchAppNotAvailableException' => $baseDir . '/lib/public/FullTextSearch/Exceptions/FullTextSearchAppNotAvailableException.php',
@@ -821,6 +822,7 @@ return array(
     'OC\\Files\\Storage\\Wrapper\\PermissionsMask' => $baseDir . '/lib/private/Files/Storage/Wrapper/PermissionsMask.php',
     'OC\\Files\\Storage\\Wrapper\\Quota' => $baseDir . '/lib/private/Files/Storage/Wrapper/Quota.php',
     'OC\\Files\\Storage\\Wrapper\\Wrapper' => $baseDir . '/lib/private/Files/Storage/Wrapper/Wrapper.php',
+    'OC\\Files\\Stream\\CountReadStream' => $baseDir . '/lib/private/Files/Stream/CountReadStream.php',
     'OC\\Files\\Stream\\Encryption' => $baseDir . '/lib/private/Files/Stream/Encryption.php',
     'OC\\Files\\Stream\\Quota' => $baseDir . '/lib/private/Files/Stream/Quota.php',
     'OC\\Files\\Type\\Detection' => $baseDir . '/lib/private/Files/Type/Detection.php',

lib/composer/composer/autoload_static.php

@@ -261,6 +261,7 @@ class ComposerStaticInit53792487c5a8370acc0b06b1a864ff4c
         'OCP\\Files\\Storage\\INotifyStorage' => __DIR__ . '/../../..' . '/lib/public/Files/Storage/INotifyStorage.php',
         'OCP\\Files\\Storage\\IStorage' => __DIR__ . '/../../..' . '/lib/public/Files/Storage/IStorage.php',
         'OCP\\Files\\Storage\\IStorageFactory' => __DIR__ . '/../../..' . '/lib/public/Files/Storage/IStorageFactory.php',
+        'OCP\\Files\\Storage\\IWriteStreamStorage' => __DIR__ . '/../../..' . '/lib/public/Files/Storage/IWriteStreamStorage.php',
         'OCP\\Files\\UnseekableException' => __DIR__ . '/../../..' . '/lib/public/Files/UnseekableException.php',
         'OCP\\Files_FullTextSearch\\Model\\AFilesDocument' => __DIR__ . '/../../..' . '/lib/public/Files_FullTextSearch/Model/AFilesDocument.php',
         'OCP\\FullTextSearch\\Exceptions\\FullTextSearchAppNotAvailableException' => __DIR__ . '/../../..' . '/lib/public/FullTextSearch/Exceptions/FullTextSearchAppNotAvailableException.php',
@@ -851,6 +852,7 @@ class ComposerStaticInit53792487c5a8370acc0b06b1a864ff4c
         'OC\\Files\\Storage\\Wrapper\\PermissionsMask' => __DIR__ . '/../../..' . '/lib/private/Files/Storage/Wrapper/PermissionsMask.php',
         'OC\\Files\\Storage\\Wrapper\\Quota' => __DIR__ . '/../../..' . '/lib/private/Files/Storage/Wrapper/Quota.php',
         'OC\\Files\\Storage\\Wrapper\\Wrapper' => __DIR__ . '/../../..' . '/lib/private/Files/Storage/Wrapper/Wrapper.php',
+        'OC\\Files\\Stream\\CountReadStream' => __DIR__ . '/../../..' . '/lib/private/Files/Stream/CountReadStream.php',
         'OC\\Files\\Stream\\Encryption' => __DIR__ . '/../../..' . '/lib/private/Files/Stream/Encryption.php',
         'OC\\Files\\Stream\\Quota' => __DIR__ . '/../../..' . '/lib/private/Files/Stream/Quota.php',
         'OC\\Files\\Type\\Detection' => __DIR__ . '/../../..' . '/lib/private/Files/Type/Detection.php',

lib/private/Files/ObjectStore/ObjectStoreStorage.php

@@ -28,6 +28,7 @@ namespace OC\Files\ObjectStore;
 use Icewind\Streams\CallbackWrapper;
 use Icewind\Streams\IteratorDirectory;
 use OC\Files\Cache\CacheEntry;
+use OC\Files\Stream\CountReadStream;
 use OCP\Files\ObjectStore\IObjectStore;
 
 class ObjectStoreStorage extends \OC\Files\Storage\Common {
@@ -382,25 +383,48 @@ class ObjectStoreStorage extends \OC\Files\Storage\Common {
 	}
 
 	public function writeBack($tmpFile, $path) {
+		$size = filesize($tmpFile);
+		$this->writeStream($path, fopen($tmpFile, 'r'), $size);
+	}
+
+	/**
+	 * external changes are not supported, exclusive access to the object storage is assumed
+	 *
+	 * @param string $path
+	 * @param int $time
+	 * @return false
+	 */
+	public function hasUpdated($path, $time) {
+		return false;
+	}
+
+	public function needsPartFile() {
+		return false;
+	}
+
+	public function file_put_contents($path, $data) {
+		$stream = fopen('php://temp', 'r+');
+		fwrite($stream, $data);
+		rewind($stream);
+		return $this->writeStream($path, $stream, strlen($data)) > 0;
+	}
+
+	public function writeStream(string $path, $stream, int $size = null): int {
 		$stat = $this->stat($path);
 		if (empty($stat)) {
 			// create new file
-			$stat = array(
+			$stat = [
 				'permissions' => \OCP\Constants::PERMISSION_ALL - \OCP\Constants::PERMISSION_CREATE,
-			);
+			];
 		}
 		// update stat with new data
 		$mTime = time();
-		$stat['size'] = filesize($tmpFile);
+		$stat['size'] = (int)$size;
 		$stat['mtime'] = $mTime;
 		$stat['storage_mtime'] = $mTime;
 
-		// run path based detection first, to use file extension because $tmpFile is only a random string
 		$mimetypeDetector = \OC::$server->getMimeTypeDetector();
 		$mimetype = $mimetypeDetector->detectPath($path);
-		if ($mimetype === 'application/octet-stream') {
-			$mimetype = $mimetypeDetector->detect($tmpFile);
-		}
 
 		$stat['mimetype'] = $mimetype;
 		$stat['etag'] = $this->getETag($path);
@@ -408,7 +432,20 @@ class ObjectStoreStorage extends \OC\Files\Storage\Common {
 		$fileId = $this->getCache()->put($path, $stat);
 		try {
 			//upload to object storage
-			$this->objectStore->writeObject($this->getURN($fileId), fopen($tmpFile, 'r'));
+			if ($size === null) {
+				$countStream = CountReadStream::wrap($stream, function ($writtenSize) use ($fileId, &$size) {
+					$this->getCache()->update($fileId, [
+						'size' => $writtenSize
+					]);
+					$size = $writtenSize;
+				});
+				$this->objectStore->writeObject($this->getURN($fileId), $countStream);
+				if (is_resource($countStream)) {
+					fclose($countStream);
+				}
+			} else {
+				$this->objectStore->writeObject($this->getURN($fileId), $stream);
+			}
 		} catch (\Exception $ex) {
 			$this->getCache()->remove($path);
 			$this->logger->logException($ex, [
@@ -417,20 +454,7 @@ class ObjectStoreStorage extends \OC\Files\Storage\Common {
 			]);
 			throw $ex; // make this bubble up
 		}
-	}
 
-	/**
-	 * external changes are not supported, exclusive access to the object storage is assumed
-	 *
-	 * @param string $path
-	 * @param int $time
-	 * @return false
-	 */
-	public function hasUpdated($path, $time) {
-		return false;
-	}
-
-	public function needsPartFile() {
-		return false;
+		return $size;
 	}
 }

lib/private/Files/Storage/Common.php

@@ -54,6 +54,7 @@ use OCP\Files\InvalidPathException;
 use OCP\Files\ReservedWordException;
 use OCP\Files\Storage\ILockingStorage;
 use OCP\Files\Storage\IStorage;
+use OCP\Files\Storage\IWriteStreamStorage;
 use OCP\ILogger;
 use OCP\Lock\ILockingProvider;
 use OCP\Lock\LockedException;
@@ -69,7 +70,7 @@ use OCP\Lock\LockedException;
  * Some \OC\Files\Storage\Common methods call functions which are first defined
  * in classes which extend it, e.g. $this->stat() .
  */
-abstract class Common implements Storage, ILockingStorage {
+abstract class Common implements Storage, ILockingStorage, IWriteStreamStorage {
 
 	use LocalTempFileTrait;
 
@@ -809,4 +810,23 @@ abstract class Common implements Storage, ILockingStorage {
 	public function needsPartFile() {
 		return true;
 	}
+
+	/**
+	 * fallback implementation
+	 *
+	 * @param string $path
+	 * @param resource $stream
+	 * @param int $size
+	 * @return int
+	 */
+	public function writeStream(string $path, $stream, int $size = null): int {
+		$target = $this->fopen($path, 'w');
+		if (!$target) {
+			return 0;
+		}
+		list($count, $result) = \OC_Helper::streamCopy($stream, $target);
+		fclose($stream);
+		fclose($target);
+		return $count;
+	}
 }

lib/private/Files/Storage/Local.php

@@ -462,4 +462,8 @@ class Local extends \OC\Files\Storage\Common {
 			return parent::moveFromStorage($sourceStorage, $sourceInternalPath, $targetInternalPath);
 		}
 	}
+
+	public function writeStream(string $path, $stream, int $size = null): int {
+		return (int)file_put_contents($this->getSourcePath($path), $stream);
+	}
 }

lib/private/Files/Storage/Wrapper/Encryption.php

@@ -1029,4 +1029,13 @@ class Encryption extends Wrapper {
 
 	}
 
+	public function writeStream(string $path, $stream, int $size = null): int {
+		// always fall back to fopen
+		$target = $this->fopen($path, 'w');
+		list($count, $result) = \OC_Helper::streamCopy($stream, $target);
+		fclose($stream);
+		fclose($target);
+		return $count;
+	}
+
 }

lib/private/Files/Storage/Wrapper/Jail.php

@@ -29,6 +29,7 @@ use OC\Files\Cache\Wrapper\CacheJail;
 use OC\Files\Cache\Wrapper\JailPropagator;
 use OC\Files\Filesystem;
 use OCP\Files\Storage\IStorage;
+use OCP\Files\Storage\IWriteStreamStorage;
 use OCP\Lock\ILockingProvider;
 
 /**
@@ -515,4 +516,18 @@ class Jail extends Wrapper {
 		$this->propagator = new JailPropagator($storage, \OC::$server->getDatabaseConnection());
 		return $this->propagator;
 	}
+
+	public function writeStream(string $path, $stream, int $size = null): int {
+		$storage = $this->getWrapperStorage();
+		if ($storage->instanceOfStorage(IWriteStreamStorage::class)) {
+			/** @var IWriteStreamStorage $storage */
+			return $storage->writeStream($this->getUnjailedPath($path), $stream, $size);
+		} else {
+			$target = $this->fopen($path, 'w');
+			list($count, $result) = \OC_Helper::streamCopy($stream, $target);
+			fclose($stream);
+			fclose($target);
+			return $count;
+		}
+	}
 }

lib/private/Files/Storage/Wrapper/Wrapper.php

@@ -32,9 +32,10 @@ namespace OC\Files\Storage\Wrapper;
 use OCP\Files\InvalidPathException;
 use OCP\Files\Storage\ILockingStorage;
 use OCP\Files\Storage\IStorage;
+use OCP\Files\Storage\IWriteStreamStorage;
 use OCP\Lock\ILockingProvider;
 
-class Wrapper implements \OC\Files\Storage\Storage, ILockingStorage {
+class Wrapper implements \OC\Files\Storage\Storage, ILockingStorage, IWriteStreamStorage {
 	/**
 	 * @var \OC\Files\Storage\Storage $storage
 	 */
@@ -621,4 +622,18 @@ class Wrapper implements \OC\Files\Storage\Storage, ILockingStorage {
 	public function needsPartFile() {
 		return $this->getWrapperStorage()->needsPartFile();
 	}
+
+	public function writeStream(string $path, $stream, int $size = null): int {
+		$storage = $this->getWrapperStorage();
+		if ($storage->instanceOfStorage(IWriteStreamStorage::class)) {
+			/** @var IWriteStreamStorage $storage */
+			return $storage->writeStream($path, $stream, $size);
+		} else {
+			$target = $this->fopen($path, 'w');
+			list($count, $result) = \OC_Helper::streamCopy($stream, $target);
+			fclose($stream);
+			fclose($target);
+			return $count;
+		}
+	}
 }

lib/private/Files/Stream/CountReadStream.php

@@ -0,0 +1,65 @@
+<?php declare(strict_types=1);
+/**
+ * @copyright Copyright (c) 2018 Robin Appelman <robin@icewind.nl>
+ *
+ * @license GNU AGPL version 3 or any later version
+ *
+ * This program is free software: you can redistribute it and/or modify
+ * it under the terms of the GNU Affero General Public License as
+ * published by the Free Software Foundation, either version 3 of the
+ * License, or (at your option) any later version.
+ *
+ * This program is distributed in the hope that it will be useful,
+ * but WITHOUT ANY WARRANTY; without even the implied warranty of
+ * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
+ * GNU Affero General Public License for more details.
+ *
+ * You should have received a copy of the GNU Affero General Public License
+ * along with this program.  If not, see <http://www.gnu.org/licenses/>.
+ *
+ */
+
+namespace OC\Files\Stream;
+
+use Icewind\Streams\Wrapper;
+
+class CountReadStream extends Wrapper {
+	/** @var int */
+	private $count;
+
+	/** @var callback */
+	private $callback;
+
+	public static function wrap($source, $callback) {
+		$context = stream_context_create(array(
+			'count' => array(
+				'source' => $source,
+				'callback' => $callback,
+			)
+		));
+		return Wrapper::wrapSource($source, $context, 'count', self::class);
+	}
+
+	public function dir_opendir($path, $options) {
+		return false;
+	}
+
+	public function stream_open($path, $mode, $options, &$opened_path) {
+		$context = $this->loadContext('count');
+
+		$this->callback = $context['callback'];
+		return true;
+	}
+
+	public function stream_read($count) {
+		$result = parent::stream_read($count);
+		$this->count += strlen($result);
+		return $result;
+	}
+
+	public function stream_close() {
+		$result = parent::stream_close();
+		call_user_func($this->callback, $this->count);
+		return $result;
+	}
+}

lib/public/Files/Storage/IWriteStreamStorage.php

@@ -0,0 +1,40 @@
+<?php declare(strict_types=1);
+/**
+ * @copyright Copyright (c) 2018 Robin Appelman <robin@icewind.nl>
+ *
+ * @license GNU AGPL version 3 or any later version
+ *
+ * This program is free software: you can redistribute it and/or modify
+ * it under the terms of the GNU Affero General Public License as
+ * published by the Free Software Foundation, either version 3 of the
+ * License, or (at your option) any later version.
+ *
+ * This program is distributed in the hope that it will be useful,
+ * but WITHOUT ANY WARRANTY; without even the implied warranty of
+ * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
+ * GNU Affero General Public License for more details.
+ *
+ * You should have received a copy of the GNU Affero General Public License
+ * along with this program.  If not, see <http://www.gnu.org/licenses/>.
+ *
+ */
+
+namespace OCP\Files\Storage;
+
+/**
+ * Interface that adds the ability to write a stream directly to file
+ *
+ * @since 15.0.0
+ */
+interface IWriteStreamStorage extends IStorage {
+	/**
+	 * Write the data from a stream to a file
+	 *
+	 * @param string $path
+	 * @param resource $stream
+	 * @param int|null $size the size of the stream if known in advance
+	 * @return int the number of bytes written
+	 * @since 15.0.0
+	 */
+	public function writeStream(string $path, $stream, int $size = null): int;
+}

tests/lib/Files/Storage/Storage.php

@@ -23,6 +23,7 @@
 namespace Test\Files\Storage;
 
 use OC\Files\Cache\Watcher;
+use OCP\Files\Storage\IWriteStreamStorage;
 
 abstract class Storage extends \Test\TestCase {
 	/**
@@ -628,4 +629,20 @@ abstract class Storage extends \Test\TestCase {
 		$this->instance->rename('bar.txt.part', 'bar.txt');
 		$this->assertEquals('bar', $this->instance->file_get_contents('bar.txt'));
 	}
+
+	public function testWriteStream() {
+		$textFile = \OC::$SERVERROOT . '/tests/data/lorem.txt';
+
+		if (!$this->instance->instanceOfStorage(IWriteStreamStorage::class)) {
+			$this->markTestSkipped('Not a WriteSteamStorage');
+		}
+		/** @var IWriteStreamStorage $storage */
+		$storage = $this->instance;
+
+		$source = fopen($textFile, 'r');
+
+		$storage->writeStream('test.txt', $source);
+		$this->assertTrue($storage->file_exists('test.txt'));
+		$this->assertEquals(file_get_contents($textFile), $storage->file_get_contents('test.txt'));
+	}
 }

tests/lib/Files/Stream/CountReadStreamTest.php

@@ -0,0 +1,49 @@
+<?php declare(strict_types=1);
+/**
+ * @copyright Copyright (c) 2018 Robin Appelman <robin@icewind.nl>
+ *
+ * @license GNU AGPL version 3 or any later version
+ *
+ * This program is free software: you can redistribute it and/or modify
+ * it under the terms of the GNU Affero General Public License as
+ * published by the Free Software Foundation, either version 3 of the
+ * License, or (at your option) any later version.
+ *
+ * This program is distributed in the hope that it will be useful,
+ * but WITHOUT ANY WARRANTY; without even the implied warranty of
+ * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
+ * GNU Affero General Public License for more details.
+ *
+ * You should have received a copy of the GNU Affero General Public License
+ * along with this program.  If not, see <http://www.gnu.org/licenses/>.
+ *
+ */
+
+namespace Test\Files\Stream;
+
+use OC\Files\Stream\CountReadStream;
+use Test\TestCase;
+
+class CountReadStreamTest extends TestCase {
+	private function getStream($data) {
+		$handle = fopen('php://temp', 'w+');
+		fwrite($handle, $data);
+		rewind($handle);
+		return $handle;
+	}
+
+	public function testBasicCount() {
+		$source = $this->getStream('foo');
+		$stream = CountReadStream::wrap($source, function ($size) {
+			$this->assertEquals(3, $size);
+		});
+		stream_get_contents($stream);
+	}
+
+	public function testLarger() {
+		$stream = CountReadStream::wrap(fopen(__DIR__ . '/../../../data/testimage.mp4', 'r'), function ($size) {
+			$this->assertEquals(383631, $size);
+		});
+		stream_get_contents($stream);
+	}
+}