Ytake\KsqlClient\Result\StreamResult::result()

Complexity

Conditions	4
Paths	2

Size

Total Lines	18
Code Lines	12

Duplication

Lines	0
Ratio	0 %

Importance

Changes

0

<?php
    public function result(): EntityInterface
    {
        $stream = $this->response->getBody();
        $streamed = [];
        if ($stream instanceof Stream) {
            while (!$stream->eof()) {
                $line = trim(\GuzzleHttp\Psr7\readline($stream));
                if (!empty($line)) {
                    $decode = \GuzzleHttp\json_decode($line, true);
                    $row = new StreamedRow($decode['row']);
                    call_user_func_array($this->callback, [$row]);
                    $streamed[] = $row;
                }
            }
        }

        return new StreamedRows($streamed);
    }
}