Merge branch 'feature-error-handling-middlware'

Author: WyriHaximus

composer.json

@@ -13,7 +13,7 @@
     "php": "^7.0",
     "api-clients/middleware": "^2.0",
     "api-clients/resource": "^3.0",
-    "api-clients/rx": "dev-master",
+    "api-clients/rx": "^2.0",
     "api-clients/service": "dev-master",
     "clue/block-react": "^1.1",
     "clue/buzz-react": "^1.1",

composer.lock

@@ -34,21 +34,17 @@ public function getResponse(): ResponseInterface
 
     /**
      * @param ObserverInterface $observer
-     * @param SchedulerInterface $scheduler
      * @return DisposableInterface
+     * @ignoreCodeCoverage
      */
-    public function subscribe(ObserverInterface $observer, $scheduler = null): DisposableInterface
+    // @codingStandardsIgnoreStart
+    protected function _subscribe(ObserverInterface $observer): DisposableInterface
     {
+        // @codingStandardsIgnoreEnd
         $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);
-        });
+        $body->on('data', [$observer, 'onNext']);
+        $body->on('error', [$observer, 'onError']);
+        $body->on('end', [$observer, 'onCompleted']);
 
         return new EmptyDisposable();
     }

src/StreamingResponse.php

@@ -27,7 +27,7 @@ public function testResponse()
         $psr7Response = new Response(200, [], new ReadableBodyStream($stream));
         $response = new StreamingResponse($psr7Response);
         self::assertSame($psr7Response, $response->getResponse());
-        $result = await(Promise::fromObservable(Promise::toObservable(resolve($response))->switchLatest()), $loop);
+        $result = await(Promise::fromObservable(Promise::toObservable(resolve($response))->switch()), $loop);
         self::assertSame($string, $result);
     }
 
@@ -44,6 +44,6 @@ public function testResponseError()
         });
         $psr7Response = new Response(200, [], new ReadableBodyStream($stream));
         $response = new StreamingResponse($psr7Response);
-        await(Promise::fromObservable(Promise::toObservable(resolve($response))->switchLatest()), $loop);
+        await(Promise::fromObservable(Promise::toObservable(resolve($response))->switch()), $loop);
     }
 }