ApiClients\Foundation\Transport\StreamingResponse::subscribe()

Complexity

Conditions	1
Paths	1

Size

Total Lines	18
Code Lines	11

Duplication

Lines	0
Ratio	0 %

Importance

Changes

0

<?php declare(strict_types=1);
    public function subscribe(): Observable
    {
        return Observable::create(function (
            ObserverInterface $observer,
            SchedulerInterface $scheduler

The parameter $scheduler is not used and could be removed.

        ) {
            $body = $this->response->getBody();
            $body->on('data', function (string $data) use ($observer) {
                $observer->onNext($data);
            });
            $body->on('end', function () use ($observer) {
                $observer->onCompleted();
            });
            $body->on('error', function ($error) use ($observer) {
                $observer->onError($error);
            });
        });
    }
}