Change ObservableInterface return types to Observable

examples/accounts-async.php

@@ -1,21 +1,26 @@
 <?php declare(strict_types=1);
+
 use ApiClients\Client\Travis\AsyncClient;
 use ApiClients\Client\Travis\Resource\AccountInterface;
 use React\EventLoop\Factory;
 use function ApiClients\Foundation\resource_pretty_print;
 
 require dirname(__DIR__) . DIRECTORY_SEPARATOR . 'vendor/autoload.php';
 
-$loop = Factory::create();
+$loop   = Factory::create();
 $client = AsyncClient::create($loop, require 'resolve_key.php');
 
-$client->accounts()->subscribeCallback(function (AccountInterface $account) {
-    resource_pretty_print($account);
-    $account->refresh()->then(function ($account) {
-        resource_pretty_print($account);
-    });
-}, function ($e) {
-    echo (string)$e;
-});
+$client->accounts()
+    ->subscribe(
+        function (AccountInterface $account) {
+            resource_pretty_print($account);
+            $account->refresh()->then(function ($account) {
+                resource_pretty_print($account);
+            });
+        },
+        function ($e) {
+            echo (string)$e;
+        }
+    );
 
 $loop->run();

examples/repositories-async.php

@@ -8,7 +8,6 @@
 use ApiClients\Middleware\Pool\PoolMiddleware;
 use React\EventLoop\Factory;
 use ResourcePool\Pool;
-use Rx\Observer\CallbackObserver;
 use function ApiClients\Foundation\resource_pretty_print;
 
 require dirname(__DIR__) . DIRECTORY_SEPARATOR . 'vendor/autoload.php';
@@ -35,18 +34,18 @@
     ]
 );
 
-$client->repositories()->subscribe(new CallbackObserver(
+$client->repositories()->subscribe(
     function (RepositoryInterface $repository) {
         resource_pretty_print($repository);
     },
     function ($e) {
         echo (string)$e,
-    PHP_EOL;
+        PHP_EOL;
     },
     function () {
         echo 'Done!',
-    PHP_EOL;
+        PHP_EOL;
     }
-));
+);
 
 $loop->run();

examples/specific-build-async.php

@@ -8,13 +8,13 @@
 $loop = \React\EventLoop\Factory::create();
 $client = AsyncClient::create($loop);
 
-$client->repository($argv[1] ?? 'WyriHaximus/php-travis-client')->subscribeCallback(function (Repository $repository) use ($argv) {
+$client->repository($argv[1] ?? 'WyriHaximus/php-travis-client')->then(function (Repository $repository) use ($argv) {
     echo 'Repository: ', PHP_EOL;
     echo 'id: ' . $repository->id(), PHP_EOL;
     echo 'slug: ' . $repository->slug(), PHP_EOL;
     echo 'description: ' . $repository->description(), PHP_EOL;
     echo 'Builds:', PHP_EOL;
-    $repository->build($argv[2] ?? 126863927)->subscribeCallback(function (BuildInterface $build) {
+    $repository->build($argv[2] ?? 126863927)->then(function (BuildInterface $build) {
         echo "\t", 'Build', PHP_EOL;
         echo "\t\t" . 'id: ' . $build->id(), PHP_EOL;
         echo "\t\t" . 'commit id: ' . $build->commitId(), PHP_EOL;

examples/specific-job-async.php

@@ -1,39 +1,44 @@
 <?php declare(strict_types=1);
+
 use ApiClients\Client\Travis\AsyncClient;
 use ApiClients\Client\Travis\Resource\Async\Build;
 use ApiClients\Client\Travis\Resource\Async\Job;
 use ApiClients\Client\Travis\Resource\Async\Repository;
 use React\EventLoop\Factory;
+use Rx\Observable;
 
 require dirname(__DIR__) . DIRECTORY_SEPARATOR . 'vendor/autoload.php';
 
 $loop   = Factory::create();
 $client = AsyncClient::create($loop);
 
-$client->repository($argv[1] ?? 'WyriHaximus/php-travis-client')->flatMap(function (Repository $repository) {
-    echo 'Repository: ', PHP_EOL;
-    echo 'id: ' . $repository->id(), PHP_EOL;
-    echo 'slug: ' . $repository->slug(), PHP_EOL;
-    echo 'description: ' . $repository->description(), PHP_EOL;
-    echo 'Builds:', PHP_EOL;
-    //@todo the takeLast operator is isn't implemented yet in RxPHP, so we need to fake it for now
-    return $repository->builds()
-        ->toArray()
-        ->map(function ($builds) {
-            return array_pop($builds);
-        });
-})->flatMap(function (Build $build) use ($argv) {
-    echo "\t", 'Build', PHP_EOL;
-    echo "\t\t" . 'id: ' . $build->id(), PHP_EOL;
-    echo "\t\t" . 'commit id: ' . $build->commitId(), PHP_EOL;
-    echo "\t\t" . 'duration: ' . $build->duration(), PHP_EOL;
+Observable::fromPromise($client->repository($argv[1] ?? 'WyriHaximus/php-travis-client'))
+    ->flatMap(function (Repository $repository) {
+        echo 'Repository: ', PHP_EOL;
+        echo 'id: ' . $repository->id(), PHP_EOL;
+        echo 'slug: ' . $repository->slug(), PHP_EOL;
+        echo 'description: ' . $repository->description(), PHP_EOL;
+        echo 'Builds:', PHP_EOL;
+        //@todo the takeLast operator is isn't implemented yet in RxPHP, so we need to fake it for now
+        return $repository->builds()
+            ->toArray()
+            ->map(function ($builds) {
+                return array_pop($builds);
+            });
+    })
+    ->flatMap(function (Build $build) use ($argv) {
+        echo "\t", 'Build', PHP_EOL;
+        echo "\t\t" . 'id: ' . $build->id(), PHP_EOL;
+        echo "\t\t" . 'commit id: ' . $build->commitId(), PHP_EOL;
+        echo "\t\t" . 'duration: ' . $build->duration(), PHP_EOL;
 
-    return $build->job($argv[2] ?? 128670080);
-})->subscribeCallback(function (Job $job) {
-    echo "\t\t" . 'Job', PHP_EOL;
-    echo "\t\t\t" . 'id: ' . $job->id(), PHP_EOL;
-    echo "\t\t\t" . 'number: ' . $job->number(), PHP_EOL;
-    echo "\t\t\t" . 'state: ' . $job->state(), PHP_EOL;
-});
+        return $build->job($argv[2] ?? 128670080);
+    })
+    ->subscribe(function (Job $job) {
+        echo "\t\t" . 'Job', PHP_EOL;
+        echo "\t\t\t" . 'id: ' . $job->id(), PHP_EOL;
+        echo "\t\t\t" . 'number: ' . $job->number(), PHP_EOL;
+        echo "\t\t\t" . 'state: ' . $job->state(), PHP_EOL;
+    });
 
 $loop->run();

src/AsyncClient.php

@@ -10,7 +10,7 @@
 use React\EventLoop\LoopInterface;
 use React\Promise\CancellablePromiseInterface;
 use React\Promise\PromiseInterface;
-use Rx\ObservableInterface;
+use Rx\Observable;
 use Rx\React\Promise;
 use function ApiClients\Tools\Rx\setAsyncScheduler;
 use function ApiClients\Tools\Rx\unwrapObservableFromPromise;
@@ -90,7 +90,7 @@ public function repository(string $repository): CancellablePromiseInterface
      *
      * {@inheritdoc}
      */
-    public function repositories(callable $filter = null): ObservableInterface
+    public function repositories(callable $filter = null): Observable
     {
         if ($filter === null) {
             $filter = function () {
@@ -124,9 +124,9 @@ public function sshKey(int $id): PromiseInterface
     }
 
     /**
-     * @return ObservableInterface
+     * @return Observable
      */
-    public function hooks(): ObservableInterface
+    public function hooks(): Observable
     {
         return unwrapObservableFromPromise($this->client->handle(
             new Command\HooksCommand()
@@ -136,7 +136,7 @@ public function hooks(): ObservableInterface
     /**
      * {@inheritdoc}
      */
-    public function accounts(): ObservableInterface
+    public function accounts(): Observable
     {
         return unwrapObservableFromPromise($this->client->handle(
             new Command\AccountsCommand()
@@ -146,7 +146,7 @@ public function accounts(): ObservableInterface
     /**
      * {@inheritdoc}
      */
-    public function broadcasts(): ObservableInterface
+    public function broadcasts(): Observable
     {
         return unwrapObservableFromPromise($this->client->handle(
             new Command\BroadcastsCommand()

src/AsyncClientInterface.php

@@ -5,7 +5,7 @@
 use ApiClients\Foundation\Resource\ResourceInterface;
 use React\Promise\CancellablePromiseInterface;
 use React\Promise\PromiseInterface;
-use Rx\ObservableInterface;
+use Rx\Observable;
 
 interface AsyncClientInterface
 {
@@ -43,10 +43,10 @@ public function repository(string $repository): CancellablePromiseInterface;
      * on the hooks observable before it fetches the repository information.
      * It could be used to only fetch new repositories.
      *
-     * @param  callable|null       $filter
-     * @return ObservableInterface
+     * @param  callable|null $filter
+     * @return Observable
      */
-    public function repositories(callable $filter = null): ObservableInterface;
+    public function repositories(callable $filter = null): Observable;
 
     /**
      * Fetch information about the current authenticated user.
@@ -75,25 +75,25 @@ public function sshKey(int $id): PromiseInterface;
      * This will return a stream of Resource\Async\Hook objects.
      * (Requires auth token to be passed to the client!).
      *
-     * @return ObservableInterface
+     * @return Observable
      */
-    public function hooks(): ObservableInterface;
+    public function hooks(): Observable;
 
     /**
      * Fetch a stream of which users and orgs the currently authenticated user has access to.
      * This will return a stream of Resource\Async\Account objects.
      * (Requires auth token to be passed to the client!).
      *
-     * @return ObservableInterface
+     * @return Observable
      */
-    public function accounts(): ObservableInterface;
+    public function accounts(): Observable;
 
     /**
      * Fetch a list of currently active broadcast. Used by the Travis team to spread news on the site.
      * This will return a stream of Resource\Async\Broadcast objects.
      * (Requires auth token to be passed to the client!).
      *
-     * @return ObservableInterface
+     * @return Observable
      */
-    public function broadcasts(): ObservableInterface;
+    public function broadcasts(): Observable;
 }

src/Resource/Async/Repository.php

@@ -11,7 +11,6 @@
 use React\Promise\CancellablePromiseInterface;
 use React\Promise\PromiseInterface;
 use Rx\Observable;
-use Rx\ObservableInterface;
 use Rx\React\Promise;
 use function ApiClients\Tools\Rx\unwrapObservableFromPromise;
 use function React\Promise\resolve;
@@ -42,9 +41,9 @@ public function build(int $id): CancellablePromiseInterface
     }
 
     /**
-     * @return ObservableInterface
+     * @return Observable
      */
-    public function commits(): ObservableInterface
+    public function commits(): Observable
     {
         return unwrapObservableFromPromise($this->handleCommand(
             new Command\CommitsCommand($this->slug())
@@ -102,29 +101,29 @@ public function disable(): PromiseInterface
     }
 
     /**
-     * @return ObservableInterface
+     * @return Observable
      */
-    public function branches(): ObservableInterface
+    public function branches(): Observable
     {
         return unwrapObservableFromPromise($this->handleCommand(
             new Command\BranchesCommand($this->id())
         ));
     }
 
     /**
-     * @return ObservableInterface
+     * @return Observable
      */
-    public function vars(): ObservableInterface
+    public function vars(): Observable
     {
         return unwrapObservableFromPromise($this->handleCommand(
             new Command\VarsCommand($this->id())
         ));
     }
 
     /**
-     * @return ObservableInterface
+     * @return Observable
      */
-    public function caches(): ObservableInterface
+    public function caches(): Observable
     {
         return unwrapObservableFromPromise($this->handleCommand(
             new Command\CachesCommand($this->id())