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