FirstChunk example

if (null === $client) {
                    $client = $r->client;
                } elseif ($r->client !== $client) {
                    throw new TransportException('Cannot stream AsyncResponse objects with many clients.');
                }

                $asyncMap[$r->response] = $r;
                $wrappedResponses[] = $r->response;

                if ($r->stream) {
                    yield from self::passthruStream($response = $r->response, $r, new FirstChunk(), $asyncMap);

                    if (!isset($asyncMap[$response])) {
                        array_pop($wrappedResponses);
                    }

                    if ($r->response !== $response && !isset($asyncMap[$r->response])) {
                        $asyncMap[$r->response] = $r;
                        $wrappedResponses[] = $r->response;
                    }
                }
            }

            

try {
            foreach ($mockHttpClient->stream($mockHttpClient->request('GET', 'https://symfony.com', [])) as $chunk) {
                $chunks[] = $chunk;
            }
            $this->fail();
        } catch (TransportException $e) {
            $this->assertEquals(new \RuntimeException('bar ccc'), $e->getPrevious());
            $this->assertSame('bar ccc', $e->getMessage());
        }

        $this->assertCount(3, $chunks);
        $this->assertEquals(new FirstChunk(0, ''), $chunks[0]);
        $this->assertEquals(new DataChunk(0, 'bar'), $chunks[1]);
        $this->assertInstanceOf(ErrorChunk::class, $chunks[2]);
        $this->assertSame(3, $chunks[2]->getOffset());
        $this->assertSame('bar ccc', $chunks[2]->getError());
    }

    public function testMergeDefaultOptions()
    {
        $mockHttpClient = new MockHttpClient(null, 'https://example.com');

        $this->expectException(InvalidArgumentException::class);
        

try {
            /* @var Response $response */
            if (null === $response = yield from self::getPushedResponse($request, $multi, $info, $headers, $options, $logger)) {
                $logger?->info(sprintf('Request: "%s %s"', $info['http_method'], $info['url']));

                $response = yield from self::followRedirects($request, $multi, $info, $headers, $canceller, $options, $onProgress, $handle, $logger, $pause);
            }

            $options = null;

            $multi->handlesActivity[$id][] = new FirstChunk();

            if ('HEAD' === $response->getRequest()->getMethod() || \in_array($info['http_code'], [204, 304], true)) {
                $multi->handlesActivity[$id][] = null;
                $multi->handlesActivity[$id][] = null;
                self::stopLoop();

                return;
            }

            if ($response->hasHeader('content-length')) {
                $info['download_content_length'] = (float) $response->getHeader('content-length');
            }

TXT;

        $chunk = new DataChunk(0, $data);
        $response = new MockResponse('', ['canceled' => false, 'http_method' => 'GET', 'url' => 'http://localhost:8080/events', 'response_headers' => ['content-type: text/event-stream']]);
        $responseStream = new ResponseStream((function D) use ($response, $chunk) {
            yield $response => new FirstChunk();
            yield $response => $chunk;
            yield $response => new ErrorChunk(0, 'timeout');
        })());

        $hasCorrectHeaders = function D$options) {
            $this->assertSame(['Accept: text/event-stream', 'Cache-Control: no-cache'], $options['headers']);

            return true;
        };

        $httpClient = $this->createMock(HttpClientInterface::class);
        

// Create dechunk buffers         if (isset($this->headers['content-length'])) {
            $this->remaining = (int) $this->headers['content-length'][0];
        } elseif ('chunked' === ($this->headers['transfer-encoding'][0] ?? null)) {
            stream_filter_append($this->buffer, 'dechunk', \STREAM_FILTER_WRITE);
            $this->remaining = -1;
        } else {
            $this->remaining = -2;
        }

        $this->multi->handlesActivity[$this->id] = [new FirstChunk()];

        if ('HEAD' === $context['http']['method'] || \in_array($this->info['http_code'], [204, 304], true)) {
            $this->multi->handlesActivity[$this->id][] = null;
            $this->multi->handlesActivity[$this->id][] = null;

            return;
        }

        $host = parse_url($this->info['redirect_url'] ?? $this->url, \PHP_URL_HOST);
        $this->multi->lastTimeout = null;
        $this->multi->openHandles[$this->id] = [&$this->pauseExpiry, $h, $this->buffer, $this->onProgress, &$this->remaining, &$this->info, $host];
        

curl_multi_remove_handle($multi->handle, $ch);
                    $waitFor[1] = (string) ((int) $waitFor[1] - 1); // decrement the retry counter                     curl_setopt($ch, \CURLOPT_PRIVATE, $waitFor);
                    curl_setopt($ch, \CURLOPT_FORBID_REUSE, true);

                    if (0 === curl_multi_add_handle($multi->handle, $ch)) {
                        continue;
                    }
                }

                if (\CURLE_RECV_ERROR === $result && 'H' === $waitFor[0] && 400 <= ($responses[(int) $ch]->info['http_code'] ?? 0)) {
                    $multi->handlesActivity[$id][] = new FirstChunk();
                }

                $multi->handlesActivity[$id][] = null;
                $multi->handlesActivity[$id][] = \in_array($result, [\CURLE_OK, \CURLE_TOO_MANY_REDIRECTS], true) || '_0' === $waitFor || curl_getinfo($ch, \CURLINFO_SIZE_DOWNLOAD) === curl_getinfo($ch, \CURLINFO_CONTENT_LENGTH_DOWNLOAD) ? null : new TransportException(ucfirst(curl_error($ch) ?: curl_strerror($result)).sprintf(' for "%s".', curl_getinfo($ch, \CURLINFO_EFFECTIVE_URL)));
            }
        } finally {
            $multi->performing = false;
        }
    }

    /** * @param CurlClientState $multi */

 elseif (null === $chunk = array_shift($response->body)) {
                // Last chunk                 $multi->handlesActivity[$id][] = null;
                $multi->handlesActivity[$id][] = array_shift($response->body);
            } elseif (\is_array($chunk)) {
                // First chunk                 try {
                    $offset = 0;
                    $chunk[1]->getStatusCode();
                    $chunk[1]->getHeaders(false);
                    self::readResponse($response, $chunk[0], $chunk[1], $offset);
                    $multi->handlesActivity[$id][] = new FirstChunk();
                } catch (\Throwable $e) {
                    $multi->handlesActivity[$id][] = null;
                    $multi->handlesActivity[$id][] = $e;
                }
            } elseif ($chunk instanceof \Throwable) {
                $multi->handlesActivity[$id][] = null;
                $multi->handlesActivity[$id][] = $chunk;
            } else {
                // Data or timeout chunk                 $multi->handlesActivity[$id][] = $chunk;
            }
        }