*/ private array $decoderOptions; private RequestInterface $originalRequest; private ?JsonStreamDecoder $decoder; private string $decodeType; private ?ResponseInterface $response; private stdClass $status; /** * @param callable $httpHandler * @param string $decodeType * @param array $decoderOptions */ public function __construct(callable $httpHandler, string $decodeType, array $decoderOptions) { $this->httpHandler = $httpHandler; $this->decodeType = $decodeType; $this->decoderOptions = $decoderOptions; } /** * {@inheritdoc} */ public function start($request, array $headers = [], array $callOptions = []) { $this->originalRequest = $this->appendHeaders($request, $headers); try { $handler = $this->httpHandler; $response = $handler($this->originalRequest, $callOptions)->wait(); } catch (\Exception $ex) { if ($ex instanceof RequestException && $ex->hasResponse()) { $ex = ApiException::createFromRequestException( $ex, /* isStream */ \true ); } throw $ex; } // Create an OK Status for a successful request just so that it // has a return value. $this->status = new stdClass(); $this->status->code = Code::OK; $this->status->message = ApiStatus::OK; $this->status->details = []; $this->response = $response; } /** * @param RequestInterface $request * @param array $headers * @return RequestInterface */ private function appendHeaders(RequestInterface $request, array $headers) { foreach ($headers as $key => $value) { $request = $request->hasHeader($key) ? $request->withAddedHeader($key, $value) : $request->withHeader($key, $value); } return $request; } /** * {@inheritdoc} */ public function responses() { if (\is_null($this->response)) { throw new \Exception('Stream has not been started.'); } // Decode the stream and yield responses as they are read. $this->decoder = new JsonStreamDecoder($this->response->getBody(), $this->decodeType, $this->decoderOptions); foreach ($this->decoder->decode() as $message) { (yield $message); } } /** * Return the status of the server stream. If the call has not been started * this will be null. * * @return stdClass The status, with integer $code, string * $details, and array $metadata members */ public function getStatus() { return $this->status; } /** * {@inheritdoc} */ public function getMetadata() { return \is_null($this->response) ? null : $this->response->getHeaders(); } /** * The Rest transport does not support trailing metadata. This is a * passthrough to getMetadata(). */ public function getTrailingMetadata() { return $this->getMetadata(); } /** * {@inheritdoc} */ public function getPeer() { return $this->originalRequest->getUri(); } /** * {@inheritdoc} */ public function cancel() { if (!\is_null($this->decoder)) { $this->decoder->close(); } } /** * For the REST transport this is a no-op. * {@inheritdoc} */ public function setCallCredentials($call_credentials) { // Do nothing. } }