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

EventParsingIterator.php in Media Cloud Sync 1.4.2, at includes/sdk/s3/Aws/Api/Parser/EventParsingIterator.php

168 lines 5.4 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\Exception\EventStreamDataException;
7	use Dudlewebs\WPMCS\s3\Aws\Api\Parser\Exception\ParserException;
8	use Dudlewebs\WPMCS\s3\Aws\Api\StructureShape;
9	use Dudlewebs\WPMCS\s3\Psr\Http\Message\StreamInterface;
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 EventParsingIterator implements Iterator
15	{
16	/** @var StreamInterface */
17	private $decodingIterator;
18	/** @var StructureShape */
19	private $shape;
20	/** @var AbstractParser */
21	private $parser;
22	public function __construct(StreamInterface $stream, StructureShape $shape, AbstractParser $parser)
23	{
24	$this->decodingIterator = $this->chooseDecodingIterator($stream);
25	$this->shape = $shape;
26	$this->parser = $parser;
27	}
28	/**
29	* This method choose a decoding iterator implementation based on if the stream
30	* is seekable or not.
31	*
32	* @param $stream
33	*
34	* @return Iterator
35	*/
36	private function chooseDecodingIterator($stream)
37	{
38	if ($stream->isSeekable()) {
39	return new DecodingEventStreamIterator($stream);
40	} else {
41	return new NonSeekableStreamDecodingEventStreamIterator($stream);
42	}
43	}
44	/**
45	* @return mixed
46	*/
47	#[\ReturnTypeWillChange]
48	public function current()
49	{
50	return $this->parseEvent($this->decodingIterator->current());
51	}
52	/**
53	* @return mixed
54	*/
55	#[\ReturnTypeWillChange]
56	public function key()
57	{
58	return $this->decodingIterator->key();
59	}
60	/**
61	* @return void
62	*/
63	#[\ReturnTypeWillChange]
64	public function next()
65	{
66	$this->decodingIterator->next();
67	}
68	/**
69	* @return void
70	*/
71	#[\ReturnTypeWillChange]
72	public function rewind()
73	{
74	$this->decodingIterator->rewind();
75	}
76	/**
77	* @return bool
78	*/
79	#[\ReturnTypeWillChange]
80	public function valid()
81	{
82	return $this->decodingIterator->valid();
83	}
84	private function parseEvent(array $event)
85	{
86	if (!empty($event['headers'][':message-type'])) {
87	if ($event['headers'][':message-type'] === 'error') {
88	return $this->parseError($event);
89	}
90	if ($event['headers'][':message-type'] === 'exception') {
91	return $this->parseException($event);
92	}
93	if ($event['headers'][':message-type'] !== 'event') {
94	throw new ParserException('Failed to parse unknown message type.');
95	}
96	}
97	$eventType = $event['headers'][':event-type'] ?? null;
98	if (empty($eventType)) {
99	throw new ParserException('Failed to parse without event type.');
100	}
101	$eventPayload = $event['payload'];
102	if ($eventType === 'initial-response') {
103	return $this->parseInitialResponseEvent($eventPayload);
104	}
105	$eventShape = $this->shape->getMember($eventType);
106	return [$eventType => \array_merge($this->parseEventHeaders($event['headers'], $eventShape), $this->parseEventPayload($eventPayload, $eventShape))];
107	}
108	/**
109	* @param $headers
110	* @param $eventShape
111	*
112	* @return array
113	*/
114	private function parseEventHeaders($headers, $eventShape) : array
115	{
116	$parsedHeaders = [];
117	foreach ($eventShape->getMembers() as $memberName => $memberProps) {
118	if (isset($memberProps['eventheader'])) {
119	$parsedHeaders[$memberName] = $headers[$memberName];
120	}
121	}
122	return $parsedHeaders;
123	}
124	/**
125	* @param $payload
126	* @param $eventShape
127	*
128	* @return array
129	*/
130	private function parseEventPayload($payload, $eventShape) : array
131	{
132	$parsedPayload = [];
133	foreach ($eventShape->getMembers() as $memberName => $memberProps) {
134	$memberShape = $eventShape->getMember($memberName);
135	if (isset($memberProps['eventpayload'])) {
136	if ($memberShape->getType() === 'blob') {
137	$parsedPayload[$memberName] = $payload;
138	} else {
139	$parsedPayload[$memberName] = $this->parser->parseMemberFromStream($payload, $memberShape, null);
140	}
141	break;
142	}
143	}
144	if (empty($parsedPayload) && !empty($payload->getContents())) {
145	/**
146	* If we did not find a member with an eventpayload trait, then we should deserialize the payload
147	* using the event's shape.
148	*/
149	$parsedPayload = $this->parser->parseMemberFromStream($payload, $eventShape, null);
150	}
151	return $parsedPayload;
152	}
153	private function parseError(array $event)
154	{
155	throw new EventStreamDataException($event['headers'][':error-code'], $event['headers'][':error-message']);
156	}
157	private function parseException(array $event)
158	{
159	$payload = $event['payload']?->getContents();
160	$parsedPayload = \json_decode($payload, \true);
161	throw new EventStreamDataException($event['headers'][':exception-type'] ?? 'Unknown', $parsedPayload['message'] ?? $payload);
162	}
163	private function parseInitialResponseEvent($payload) : array
164	{
165	return ['initial-response' => \json_decode($payload, \true)];
166	}
167	}
168