feat: add straming support on JSON RPC parser

Add event streaming capabilities on the JSON RPC parser, so that operations such as StartLiveTail from CloudwatchLogs are properly handled/parsed. This change also includes an extension of the DecodingEventStreamIterator for handling streamable responses where the stream as response is not seekable, and a handling for when the `event-type` is `initial-response`.

.changes/nextrelease/json-rpc-streaming-support.json

@@ -0,0 +1,7 @@
+[
+  {
+    "type": "feature",
+    "category": "Parser",
+    "description": "Adds support for event streaming on JSON-RPC protocol parser"
+  }
+]

src/Api/Parser/DecodingEventStreamIterator.php

@@ -52,19 +52,19 @@ class DecodingEventStreamIterator implements Iterator
     ];
 
     /** @var StreamInterface Stream of eventstream shape to parse. */
-    private $stream;
+    protected $stream;
 
     /** @var array Currently parsed event. */
-    private $currentEvent;
+    protected $currentEvent;
 
     /** @var int Current in-order event key. */
-    private $key;
+    protected $key;
 
     /** @var resource|\HashContext CRC32 hash context for event validation */
-    private $hashContext;
+    protected $hashContext;
 
     /** @var int $currentPosition */
-    private $currentPosition;
+    protected $currentPosition;
 
     /**
      * DecodingEventStreamIterator constructor.
@@ -77,7 +77,7 @@ public function __construct(StreamInterface $stream)
         $this->rewind();
     }
 
-    private function parseHeaders($headerBytes)
+    protected function parseHeaders($headerBytes)
     {
         $headers = [];
         $bytesRead = 0;
@@ -102,7 +102,7 @@ private function parseHeaders($headerBytes)
         return [$headers, $bytesRead];
     }
 
-    private function parsePrelude()
+    protected function parsePrelude()
     {
         $prelude = [];
         $bytesRead = 0;
@@ -127,7 +127,12 @@ private function parsePrelude()
         return [$prelude, $bytesRead];
     }
 
-    private function parseEvent()
+    /**
+     * This method decodes an event from the stream.
+     *
+     * @return array
+     */
+    protected function parseEvent()
     {
         $event = [];
 
@@ -217,7 +222,7 @@ public function valid()
 
     // Decoding Utilities
 
-    private function readAndHashBytes($num)
+    protected function readAndHashBytes($num)
     {
         $bytes = $this->stream->read($num);
         hash_update($this->hashContext, $bytes);

src/Api/Parser/EventParsingIterator.php

@@ -28,11 +28,28 @@ public function __construct(
         StructureShape $shape,
         AbstractParser $parser
     ) {
-        $this->decodingIterator = new DecodingEventStreamIterator($stream);
+        $this->decodingIterator = $this->chooseDecodingIterator($stream);
         $this->shape = $shape;
         $this->parser = $parser;
     }
 
+    /**
+     * This method choose a decoding iterator implementation based on if the stream
+     * is seekable or not.
+     *
+     * @param $stream
+     *
+     * @return Iterator
+     */
+    private function chooseDecodingIterator($stream)
+    {
+        if ($stream->isSeekable()) {
+            return new DecodingEventStreamIterator($stream);
+        } else {
+            return new NonSeekableStreamDecodingEventStreamIterator($stream);
+        }
+    }
+
     #[\ReturnTypeWillChange]
     public function current()
     {
@@ -80,8 +97,12 @@ private function parseEvent(array $event)
             throw new ParserException('Failed to parse without event type.');
         }
 
-        $eventShape = $this->shape->getMember($eventType);
         $eventPayload = $event['payload'];
+        if ($eventType === 'initial-response') {
+            return $this->parseInitialResponseEvent($eventPayload);
+        }
+
+        $eventShape = $this->shape->getMember($eventType);
 
         return [
             $eventType => array_merge(
@@ -153,4 +174,9 @@ private function parseError(array $event)
             $event['headers'][':error-message']
         );
     }
+
+    private function parseInitialResponseEvent($payload): array
+    {
+        return ['initial-response' => json_decode($payload, true)];
+    }
 }

src/Api/Parser/JsonRpcParser.php

@@ -1,6 +1,7 @@
 <?php
 namespace Aws\Api\Parser;
 
+use Aws\Api\Operation;
 use Aws\Api\StructureShape;
 use Aws\Api\Service;
 use Aws\Result;
@@ -30,15 +31,45 @@ public function __invoke(
         ResponseInterface $response
     ) {
         $operation = $this->api->getOperation($command->getName());
-        $result = null === $operation['output']
-            ? null
-            : $this->parseMemberFromStream(
+
+        return $this->parseResponse($response, $operation);
+    }
+
+    /**
+     * This method parses a response based on JSON RPC protocol.
+     *
+     * @param ResponseInterface $response the response to parse.
+     * @param Operation $operation the operation which holds information for
+     *        parsing the response.
+     *
+     * @return Result
+     */
+    private function parseResponse(ResponseInterface $response, Operation $operation)
+    {
+        if (null === $operation['output']) {
+            return new Result([]);
+        }
+
+        $outputShape = $operation->getOutput();
+        foreach ($outputShape->getMembers() as $memberName => $memberProps) {
+            if (!empty($memberProps['eventstream'])) {
+                return new Result([
+                    $memberName => new EventParsingIterator(
+                        $response->getBody(),
+                        $outputShape->getMember($memberName),
+                        $this
+                    )
+                ]);
+            }
+        }
+
+        $result = $this->parseMemberFromStream(
                 $response->getBody(),
                 $operation->getOutput(),
                 $response
             );
 
-        return new Result($result ?: []);
+        return new Result(is_null($result) ? [] : $result);
     }
 
     public function parseMemberFromStream(

src/Api/Parser/NonSeekableStreamDecodingEventStreamIterator.php

@@ -0,0 +1,101 @@
+<?php
+
+namespace Aws\Api\Parser;
+
+use GuzzleHttp\Psr7;
+use Psr\Http\Message\StreamInterface;
+use Aws\Api\Parser\Exception\ParserException;
+
+/**
+ * @inheritDoc
+ */
+class NonSeekableStreamDecodingEventStreamIterator extends DecodingEventStreamIterator
+{
+    /** @var array $tempBuffer */
+    private $tempBuffer;
+
+    /**
+     * NonSeekableStreamDecodingEventStreamIterator constructor.
+     *
+     * @param StreamInterface $stream
+     */
+    public function __construct(StreamInterface $stream)
+    {
+        $this->stream = $stream;
+        if ($this->stream->isSeekable()) {
+            throw new \InvalidArgumentException('The stream provided must be not seekable.');
+        }
+
+        $this->tempBuffer = [];
+    }
+
+    /**
+     * @inheritDoc
+     *
+     * @return array
+     */
+    protected function parseEvent(): array
+    {
+        $event = [];
+        $this->hashContext = hash_init('crc32b');
+        $prelude = $this->parsePrelude()[0];
+        list(
+            $event[self::HEADERS],
+            $numBytes
+        ) = $this->parseHeaders($prelude[self::LENGTH_HEADERS]);
+        $event[self::PAYLOAD] = Psr7\Utils::streamFor(
+            $this->readAndHashBytes(
+                $prelude[self::LENGTH_TOTAL] - self::BYTES_PRELUDE
+                - $numBytes - self::BYTES_TRAILING
+            )
+        );
+        $calculatedCrc = hash_final($this->hashContext, true);
+        $messageCrc = $this->stream->read(4);
+        if ($calculatedCrc !== $messageCrc) {
+            throw new ParserException('Message checksum mismatch.');
+        }
+
+        return $event;
+    }
+
+    protected function readAndHashBytes($num): string
+    {
+        $bytes = '';
+        while (!empty($this->tempBuffer) && $num > 0) {
+            $byte = array_shift($this->tempBuffer);
+            $bytes .= $byte;
+            $num = $num - 1;
+        }
+
+        $bytes = $bytes . $this->stream->read($num);
+        hash_update($this->hashContext, $bytes);
+
+        return $bytes;
+    }
+
+    // Iterator Functionality
+
+    #[\ReturnTypeWillChange]
+    public function rewind()
+    {
+        $this->currentEvent = $this->parseEvent();
+    }
+
+    public function next()
+    {
+        $this->tempBuffer[] = $this->stream->read(1);
+        if ($this->valid()) {
+            $this->key++;
+            $this->currentEvent = $this->parseEvent();
+        }
+    }
+
+    /**
+     * @return bool
+     */
+    #[\ReturnTypeWillChange]
+    public function valid()
+    {
+        return !$this->stream->eof();
+    }
+}

src/CloudWatchLogs/CloudWatchLogsClient.php

@@ -2,6 +2,8 @@
 namespace Aws\CloudWatchLogs;
 
 use Aws\AwsClient;
+use Aws\CommandInterface;
+use Generator;
 
 /**
  * This client is used to interact with the **Amazon CloudWatch Logs** service.
@@ -155,4 +157,64 @@
  * @method \Aws\Result updateLogAnomalyDetector(array $args = [])
  * @method \GuzzleHttp\Promise\Promise updateLogAnomalyDetectorAsync(array $args = [])
  */
-class CloudWatchLogsClient extends AwsClient {}
+class CloudWatchLogsClient extends AwsClient {
+    static $streamingCommands = [
+        'StartLiveTail' => true
+    ];
+
+    public function __construct(array $args)
+    {
+        parent::__construct($args);
+        $this->addStreamingFlagMiddleware();
+    }
+
+    private function addStreamingFlagMiddleware()
+    {
+        $this->getHandlerList()
+            -> appendInit(
+                $this->getStreamingFlagMiddleware(),
+                'streaming-flag-middleware'
+            );
+    }
+
+    private function getStreamingFlagMiddleware(): callable
+    {
+        return function (callable $handler) {
+            return function (CommandInterface $command, $request = null) use ($handler) {
+                if (!empty(self::$streamingCommands[$command->getName()])) {
+                    $command['@http']['stream'] = true;
+                }
+
+                return $handler($command, $request);
+            };
+        };
+    }
+
+    /**
+     * Helper method for 'startLiveTail' operation that checks for results.
+     *
+     * Initiates 'startLiveTail' operation with given arguments, and continuously
+     * checks response stream for session updates or results, yielding each
+     * stream chunk when results are not empty. This method abstracts from users
+     * the need of checking if there are logs entry available to be watched, which means
+     * that users will always get a next item to be iterated when more log entries are
+     * available.
+     *
+     * @param array $args Command arguments.
+     *
+     * @return Generator Yields session update or result stream chunks.
+     */
+    public function startLiveTailCheckingForResults(array $args): Generator
+    {
+        $response = $this->startLiveTail($args);
+        foreach ($response['responseStream'] as $streamChunk) {
+            if (isset($streamChunk['sessionUpdate'])) {
+                if (!empty($streamChunk['sessionUpdate']['sessionResults'])) {
+                    yield $streamChunk;
+                }
+            } else {
+                yield $streamChunk;
+            }
+        }
+    }
+}