1
0

AssemblyStream.php 6.9 KB

123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245246247248249250251252253254255256257258259260261262263264265266267268269270271272273274275276277278279280281282283284285286287288289290291292293294295296297298299300301302303304
  1. <?php
  2. /**
  3. * @copyright Copyright (c) 2016, ownCloud, Inc.
  4. *
  5. * @author Christoph Wurst <christoph@winzerhof-wurst.at>
  6. * @author J0WI <J0WI@users.noreply.github.com>
  7. * @author Lukas Reschke <lukas@statuscode.ch>
  8. * @author Markus Goetz <markus@woboq.com>
  9. * @author Robin Appelman <robin@icewind.nl>
  10. * @author Roeland Jago Douma <roeland@famdouma.nl>
  11. * @author Thomas Müller <thomas.mueller@tmit.eu>
  12. * @author Vincent Petry <vincent@nextcloud.com>
  13. *
  14. * @license AGPL-3.0
  15. *
  16. * This code is free software: you can redistribute it and/or modify
  17. * it under the terms of the GNU Affero General Public License, version 3,
  18. * as published by the Free Software Foundation.
  19. *
  20. * This program is distributed in the hope that it will be useful,
  21. * but WITHOUT ANY WARRANTY; without even the implied warranty of
  22. * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
  23. * GNU Affero General Public License for more details.
  24. *
  25. * You should have received a copy of the GNU Affero General Public License, version 3,
  26. * along with this program. If not, see <http://www.gnu.org/licenses/>
  27. *
  28. */
  29. namespace OCA\DAV\Upload;
  30. use Sabre\DAV\IFile;
  31. /**
  32. * Class AssemblyStream
  33. *
  34. * The assembly stream is a virtual stream that wraps multiple chunks.
  35. * Reading from the stream transparently accessed the underlying chunks and
  36. * give a representation as if they were already merged together.
  37. *
  38. * @package OCA\DAV\Upload
  39. */
  40. class AssemblyStream implements \Icewind\Streams\File {
  41. /** @var resource */
  42. private $context;
  43. /** @var IFile[] */
  44. private $nodes;
  45. /** @var int */
  46. private $pos = 0;
  47. /** @var int */
  48. private $size = 0;
  49. /** @var resource */
  50. private $currentStream = null;
  51. /** @var int */
  52. private $currentNode = 0;
  53. /** @var int */
  54. private $currentNodeRead = 0;
  55. /**
  56. * @param string $path
  57. * @param string $mode
  58. * @param int $options
  59. * @param string &$opened_path
  60. * @return bool
  61. */
  62. public function stream_open($path, $mode, $options, &$opened_path) {
  63. $this->loadContext('assembly');
  64. $nodes = $this->nodes;
  65. usort($nodes, function (IFile $a, IFile $b) {
  66. return strnatcmp($a->getName(), $b->getName());
  67. });
  68. $this->nodes = array_values($nodes);
  69. $this->size = array_reduce($this->nodes, function ($size, IFile $file) {
  70. return $size + $file->getSize();
  71. }, 0);
  72. return true;
  73. }
  74. /**
  75. * @param int $offset
  76. * @param int $whence
  77. * @return bool
  78. */
  79. public function stream_seek($offset, $whence = SEEK_SET) {
  80. if ($whence === SEEK_CUR) {
  81. $offset = $this->stream_tell() + $offset;
  82. } elseif ($whence === SEEK_END) {
  83. $offset = $this->size + $offset;
  84. }
  85. if ($offset > $this->size) {
  86. return false;
  87. }
  88. $nodeIndex = 0;
  89. $nodeStart = 0;
  90. while (true) {
  91. if (!isset($this->nodes[$nodeIndex + 1])) {
  92. break;
  93. }
  94. $node = $this->nodes[$nodeIndex];
  95. if ($nodeStart + $node->getSize() > $offset) {
  96. break;
  97. }
  98. $nodeIndex++;
  99. $nodeStart += $node->getSize();
  100. }
  101. $stream = $this->getStream($this->nodes[$nodeIndex]);
  102. $nodeOffset = $offset - $nodeStart;
  103. if (fseek($stream, $nodeOffset) === -1) {
  104. return false;
  105. }
  106. $this->currentNode = $nodeIndex;
  107. $this->currentNodeRead = $nodeOffset;
  108. $this->currentStream = $stream;
  109. $this->pos = $offset;
  110. return true;
  111. }
  112. /**
  113. * @return int
  114. */
  115. public function stream_tell() {
  116. return $this->pos;
  117. }
  118. /**
  119. * @param int $count
  120. * @return string
  121. */
  122. public function stream_read($count) {
  123. if (is_null($this->currentStream)) {
  124. if ($this->currentNode < count($this->nodes)) {
  125. $this->currentStream = $this->getStream($this->nodes[$this->currentNode]);
  126. } else {
  127. return '';
  128. }
  129. }
  130. do {
  131. $data = fread($this->currentStream, $count);
  132. $read = strlen($data);
  133. $this->currentNodeRead += $read;
  134. if (feof($this->currentStream)) {
  135. fclose($this->currentStream);
  136. $currentNodeSize = $this->nodes[$this->currentNode]->getSize();
  137. if ($this->currentNodeRead < $currentNodeSize) {
  138. throw new \Exception('Stream from assembly node shorter than expected, got ' . $this->currentNodeRead . ' bytes, expected ' . $currentNodeSize);
  139. }
  140. $this->currentNode++;
  141. $this->currentNodeRead = 0;
  142. if ($this->currentNode < count($this->nodes)) {
  143. $this->currentStream = $this->getStream($this->nodes[$this->currentNode]);
  144. } else {
  145. $this->currentStream = null;
  146. }
  147. }
  148. // if no data read, try again with the next node because
  149. // returning empty data can make the caller think there is no more
  150. // data left to read
  151. } while ($read === 0 && !is_null($this->currentStream));
  152. // update position
  153. $this->pos += $read;
  154. return $data;
  155. }
  156. /**
  157. * @param string $data
  158. * @return int
  159. */
  160. public function stream_write($data) {
  161. return false;
  162. }
  163. /**
  164. * @param int $option
  165. * @param int $arg1
  166. * @param int $arg2
  167. * @return bool
  168. */
  169. public function stream_set_option($option, $arg1, $arg2) {
  170. return false;
  171. }
  172. /**
  173. * @param int $size
  174. * @return bool
  175. */
  176. public function stream_truncate($size) {
  177. return false;
  178. }
  179. /**
  180. * @return array
  181. */
  182. public function stream_stat() {
  183. return [
  184. 'size' => $this->size,
  185. ];
  186. }
  187. /**
  188. * @param int $operation
  189. * @return bool
  190. */
  191. public function stream_lock($operation) {
  192. return false;
  193. }
  194. /**
  195. * @return bool
  196. */
  197. public function stream_flush() {
  198. return false;
  199. }
  200. /**
  201. * @return bool
  202. */
  203. public function stream_eof() {
  204. return $this->pos >= $this->size || ($this->currentNode >= count($this->nodes) && $this->currentNode === null);
  205. }
  206. /**
  207. * @return bool
  208. */
  209. public function stream_close() {
  210. return true;
  211. }
  212. /**
  213. * Load the source from the stream context and return the context options
  214. *
  215. * @param string $name
  216. * @return array
  217. * @throws \BadMethodCallException
  218. */
  219. protected function loadContext($name) {
  220. $context = stream_context_get_options($this->context);
  221. if (isset($context[$name])) {
  222. $context = $context[$name];
  223. } else {
  224. throw new \BadMethodCallException('Invalid context, "' . $name . '" options not set');
  225. }
  226. if (isset($context['nodes']) and is_array($context['nodes'])) {
  227. $this->nodes = $context['nodes'];
  228. } else {
  229. throw new \BadMethodCallException('Invalid context, nodes not set');
  230. }
  231. return $context;
  232. }
  233. /**
  234. * @param IFile[] $nodes
  235. * @return resource
  236. *
  237. * @throws \BadMethodCallException
  238. */
  239. public static function wrap(array $nodes) {
  240. $context = stream_context_create([
  241. 'assembly' => [
  242. 'nodes' => $nodes
  243. ]
  244. ]);
  245. stream_wrapper_register('assembly', self::class);
  246. try {
  247. $wrapped = fopen('assembly://', 'r', false, $context);
  248. } catch (\BadMethodCallException $e) {
  249. stream_wrapper_unregister('assembly');
  250. throw $e;
  251. }
  252. stream_wrapper_unregister('assembly');
  253. return $wrapped;
  254. }
  255. /**
  256. * @param IFile $node
  257. * @return resource
  258. */
  259. private function getStream(IFile $node) {
  260. $data = $node->get();
  261. if (is_resource($data)) {
  262. return $data;
  263. } else {
  264. $tmp = fopen('php://temp', 'w+');
  265. fwrite($tmp, $data);
  266. rewind($tmp);
  267. return $tmp;
  268. }
  269. }
  270. }