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

Complexity

Conditions	1
Paths	1

Size

Total Lines	15
Code Lines	9

Duplication

Lines	0
Ratio	0 %

Importance

Changes

0

<?php declare(strict_types=1);
    public function subscribe(ObserverInterface $observer, SchedulerInterface $scheduler = null): DisposableInterface
    {
        $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);
        });

        return new EmptyDisposable();
    }
}