media-cloud-sync / includes / sdk / s3 / Aws / Api / Parser / NonSeekableStreamDecodingEventStreamIterator.php

NonSeekableStreamDecodingEventStreamIterator.php in Media Cloud Sync 1.4.1, at includes/sdk/s3/Aws/Api/Parser/NonSeekableStreamDecodingEventStreamIterator.php

82 lines 2.4 KB

Raw Download Zip

1	<?php
2
3	namespace Dudlewebs\WPMCS\s3\Aws\Api\Parser;
4
5	use Dudlewebs\WPMCS\s3\GuzzleHttp\Psr7;
6	use Dudlewebs\WPMCS\s3\Psr\Http\Message\StreamInterface;
7	use Dudlewebs\WPMCS\s3\Aws\Api\Parser\Exception\ParserException;
8	/**
9	* @inheritDoc
10	*/
11	class NonSeekableStreamDecodingEventStreamIterator extends DecodingEventStreamIterator
12	{
13	/** @var array $tempBuffer */
14	private $tempBuffer;
15	/**
16	* NonSeekableStreamDecodingEventStreamIterator constructor.
17	*
18	* @param StreamInterface $stream
19	*/
20	public function __construct(StreamInterface $stream)
21	{
22	$this->stream = $stream;
23	if ($this->stream->isSeekable()) {
24	throw new \InvalidArgumentException('The stream provided must be not seekable.');
25	}
26	$this->tempBuffer = [];
27	}
28	/**
29	* @inheritDoc
30	*
31	* @return array
32	*/
33	protected function parseEvent() : array
34	{
35	$event = [];
36	$this->hashContext = \hash_init('crc32b');
37	$prelude = $this->parsePrelude()[0];
38	list($event[self::HEADERS], $numBytes) = $this->parseHeaders($prelude[self::LENGTH_HEADERS]);
39	$event[self::PAYLOAD] = Psr7\Utils::streamFor($this->readAndHashBytes($prelude[self::LENGTH_TOTAL] - self::BYTES_PRELUDE - $numBytes - self::BYTES_TRAILING));
40	$calculatedCrc = \hash_final($this->hashContext, \true);
41	$messageCrc = $this->stream->read(4);
42	if ($calculatedCrc !== $messageCrc) {
43	throw new ParserException('Message checksum mismatch.');
44	}
45	return $event;
46	}
47	protected function readAndHashBytes($num) : string
48	{
49	$bytes = '';
50	while (!empty($this->tempBuffer) && $num > 0) {
51	$byte = \array_shift($this->tempBuffer);
52	$bytes .= $byte;
53	$num = $num - 1;
54	}
55	$bytes = $bytes . $this->stream->read($num);
56	\hash_update($this->hashContext, $bytes);
57	return $bytes;
58	}
59	// Iterator Functionality
60	#[\ReturnTypeWillChange]
61	public function rewind()
62	{
63	$this->currentEvent = $this->parseEvent();
64	}
65	public function next()
66	{
67	$this->tempBuffer[] = $this->stream->read(1);
68	if ($this->valid()) {
69	$this->key++;
70	$this->currentEvent = $this->parseEvent();
71	}
72	}
73	/**
74	* @return bool
75	*/
76	#[\ReturnTypeWillChange]
77	public function valid()
78	{
79	return !$this->stream->eof();
80	}
81	}
82