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

DecodingEventStreamIterator.php in Media Cloud Sync 1.2.13, at includes/sdk/s3/Aws/Api/Parser/DecodingEventStreamIterator.php

252 lines 8.5 KB

Raw Download Zip

1	<?php
2
3	namespace Dudlewebs\WPMCS\s3\Aws\Api\Parser;
4
5	use Iterator;
6	use Dudlewebs\WPMCS\s3\Aws\Api\DateTimeResult;
7	use Dudlewebs\WPMCS\s3\GuzzleHttp\Psr7;
8	use Dudlewebs\WPMCS\s3\Psr\Http\Message\StreamInterface;
9	use Dudlewebs\WPMCS\s3\Aws\Api\Parser\Exception\ParserException;
10	/**
11	* @internal Implements a decoder for a binary encoded event stream that will
12	* decode, validate, and provide individual events from the stream.
13	*/
14	class DecodingEventStreamIterator implements Iterator
15	{
16	const HEADERS = 'headers';
17	const PAYLOAD = 'payload';
18	const LENGTH_TOTAL = 'total_length';
19	const LENGTH_HEADERS = 'headers_length';
20	const CRC_PRELUDE = 'prelude_crc';
21	const BYTES_PRELUDE = 12;
22	const BYTES_TRAILING = 4;
23	private static $preludeFormat = [self::LENGTH_TOTAL => 'decodeUint32', self::LENGTH_HEADERS => 'decodeUint32', self::CRC_PRELUDE => 'decodeUint32'];
24	private static $lengthFormatMap = [1 => 'decodeUint8', 2 => 'decodeUint16', 4 => 'decodeUint32', 8 => 'decodeUint64'];
25	private static $headerTypeMap = [0 => 'decodeBooleanTrue', 1 => 'decodeBooleanFalse', 2 => 'decodeInt8', 3 => 'decodeInt16', 4 => 'decodeInt32', 5 => 'decodeInt64', 6 => 'decodeBytes', 7 => 'decodeString', 8 => 'decodeTimestamp', 9 => 'decodeUuid'];
26	/** @var StreamInterface Stream of eventstream shape to parse. */
27	protected $stream;
28	/** @var array Currently parsed event. */
29	protected $currentEvent;
30	/** @var int Current in-order event key. */
31	protected $key;
32	/** @var resource\|\HashContext CRC32 hash context for event validation */
33	protected $hashContext;
34	/** @var int $currentPosition */
35	protected $currentPosition;
36	/**
37	* DecodingEventStreamIterator constructor.
38	*
39	* @param StreamInterface $stream
40	*/
41	public function __construct(StreamInterface $stream)
42	{
43	$this->stream = $stream;
44	$this->rewind();
45	}
46	protected function parseHeaders($headerBytes)
47	{
48	$headers = [];
49	$bytesRead = 0;
50	while ($bytesRead < $headerBytes) {
51	list($key, $numBytes) = $this->decodeString(1);
52	$bytesRead += $numBytes;
53	list($type, $numBytes) = $this->decodeUint8();
54	$bytesRead += $numBytes;
55	$f = self::$headerTypeMap[$type];
56	list($value, $numBytes) = $this->{$f}();
57	$bytesRead += $numBytes;
58	if (isset($headers[$key])) {
59	throw new ParserException('Duplicate key in event headers.');
60	}
61	$headers[$key] = $value;
62	}
63	return [$headers, $bytesRead];
64	}
65	protected function parsePrelude()
66	{
67	$prelude = [];
68	$bytesRead = 0;
69	$calculatedCrc = null;
70	foreach (self::$preludeFormat as $key => $decodeFunction) {
71	if ($key === self::CRC_PRELUDE) {
72	$hashCopy = \hash_copy($this->hashContext);
73	$calculatedCrc = \hash_final($this->hashContext, \true);
74	$this->hashContext = $hashCopy;
75	}
76	list($value, $numBytes) = $this->{$decodeFunction}();
77	$bytesRead += $numBytes;
78	$prelude[$key] = $value;
79	}
80	if (\unpack('N', $calculatedCrc)[1] !== $prelude[self::CRC_PRELUDE]) {
81	throw new ParserException('Prelude checksum mismatch.');
82	}
83	return [$prelude, $bytesRead];
84	}
85	/**
86	* This method decodes an event from the stream.
87	*
88	* @return array
89	*/
90	protected function parseEvent()
91	{
92	$event = [];
93	if ($this->stream->tell() < $this->stream->getSize()) {
94	$this->hashContext = \hash_init('crc32b');
95	$bytesLeft = $this->stream->getSize() - $this->stream->tell();
96	list($prelude, $numBytes) = $this->parsePrelude();
97	if ($prelude[self::LENGTH_TOTAL] > $bytesLeft) {
98	throw new ParserException('Message length too long.');
99	}
100	$bytesLeft -= $numBytes;
101	if ($prelude[self::LENGTH_HEADERS] > $bytesLeft) {
102	throw new ParserException('Headers length too long.');
103	}
104	list($event[self::HEADERS], $numBytes) = $this->parseHeaders($prelude[self::LENGTH_HEADERS]);
105	$event[self::PAYLOAD] = Psr7\Utils::streamFor($this->readAndHashBytes($prelude[self::LENGTH_TOTAL] - self::BYTES_PRELUDE - $numBytes - self::BYTES_TRAILING));
106	$calculatedCrc = \hash_final($this->hashContext, \true);
107	$messageCrc = $this->stream->read(4);
108	if ($calculatedCrc !== $messageCrc) {
109	throw new ParserException('Message checksum mismatch.');
110	}
111	}
112	return $event;
113	}
114	// Iterator Functionality
115	/**
116	* @return array
117	*/
118	#[\ReturnTypeWillChange]
119	public function current()
120	{
121	return $this->currentEvent;
122	}
123	/**
124	* @return int
125	*/
126	#[\ReturnTypeWillChange]
127	public function key()
128	{
129	return $this->key;
130	}
131	#[\ReturnTypeWillChange]
132	public function next()
133	{
134	$this->currentPosition = $this->stream->tell();
135	if ($this->valid()) {
136	$this->key++;
137	$this->currentEvent = $this->parseEvent();
138	}
139	}
140	#[\ReturnTypeWillChange]
141	public function rewind()
142	{
143	$this->stream->rewind();
144	$this->key = 0;
145	$this->currentPosition = 0;
146	$this->currentEvent = $this->parseEvent();
147	}
148	/**
149	* @return bool
150	*/
151	#[\ReturnTypeWillChange]
152	public function valid()
153	{
154	return $this->currentPosition < $this->stream->getSize();
155	}
156	// Decoding Utilities
157	protected function readAndHashBytes($num)
158	{
159	$bytes = $this->stream->read($num);
160	\hash_update($this->hashContext, $bytes);
161	return $bytes;
162	}
163	private function decodeBooleanTrue()
164	{
165	return [\true, 0];
166	}
167	private function decodeBooleanFalse()
168	{
169	return [\false, 0];
170	}
171	private function uintToInt($val, $size)
172	{
173	$signedCap = \pow(2, $size - 1);
174	if ($val > $signedCap) {
175	$val -= 2 * $signedCap;
176	}
177	return $val;
178	}
179	private function decodeInt8()
180	{
181	$val = (int) \unpack('C', $this->readAndHashBytes(1))[1];
182	return [$this->uintToInt($val, 8), 1];
183	}
184	private function decodeUint8()
185	{
186	return [\unpack('C', $this->readAndHashBytes(1))[1], 1];
187	}
188	private function decodeInt16()
189	{
190	$val = (int) \unpack('n', $this->readAndHashBytes(2))[1];
191	return [$this->uintToInt($val, 16), 2];
192	}
193	private function decodeUint16()
194	{
195	return [\unpack('n', $this->readAndHashBytes(2))[1], 2];
196	}
197	private function decodeInt32()
198	{
199	$val = (int) \unpack('N', $this->readAndHashBytes(4))[1];
200	return [$this->uintToInt($val, 32), 4];
201	}
202	private function decodeUint32()
203	{
204	return [\unpack('N', $this->readAndHashBytes(4))[1], 4];
205	}
206	private function decodeInt64()
207	{
208	$val = $this->unpackInt64($this->readAndHashBytes(8))[1];
209	return [$this->uintToInt($val, 64), 8];
210	}
211	private function decodeUint64()
212	{
213	return [$this->unpackInt64($this->readAndHashBytes(8))[1], 8];
214	}
215	private function unpackInt64($bytes)
216	{
217	if (\version_compare(\PHP_VERSION, '5.6.3', '<')) {
218	$d = \unpack('N2', $bytes);
219	return [1 => $d[1] << 32 \| $d[2]];
220	}
221	return \unpack('J', $bytes);
222	}
223	private function decodeBytes($lengthBytes = 2)
224	{
225	if (!isset(self::$lengthFormatMap[$lengthBytes])) {
226	throw new ParserException('Undefined variable length format.');
227	}
228	$f = self::$lengthFormatMap[$lengthBytes];
229	list($len, $bytes) = $this->{$f}();
230	return [$this->readAndHashBytes($len), $len + $bytes];
231	}
232	private function decodeString($lengthBytes = 2)
233	{
234	if (!isset(self::$lengthFormatMap[$lengthBytes])) {
235	throw new ParserException('Undefined variable length format.');
236	}
237	$f = self::$lengthFormatMap[$lengthBytes];
238	list($len, $bytes) = $this->{$f}();
239	return [$this->readAndHashBytes($len), $len + $bytes];
240	}
241	private function decodeTimestamp()
242	{
243	list($val, $bytes) = $this->decodeInt64();
244	return [DateTimeResult::createFromFormat('U.u', $val / 1000), $bytes];
245	}
246	private function decodeUuid()
247	{
248	$val = \unpack('H32', $this->readAndHashBytes(16))[1];
249	return [\substr($val, 0, 8) . '-' . \substr($val, 8, 4) . '-' . \substr($val, 12, 4) . '-' . \substr($val, 16, 4) . '-' . \substr($val, 20, 12), 16];
250	}
251	}
252