Improve pusher integration

composer.json

@@ -19,17 +19,18 @@
     "php": "^7.0",
     "api-clients/client-services": "dev-master",
     "api-clients/foundation": "dev-master",
-    "api-clients/middleware-json": "dev-master",
+    "api-clients/middleware-json": "^1.0",
     "api-clients/middleware-token-authorization": "^2.0",
     "api-clients/middleware-user-agent": "^1.0",
-    "api-clients/pusher": "dev-master",
-    "api-clients/rx": "dev-master"
+    "api-clients/pusher": "dev-improvement-handle-disconnects-and-reconnect-when-appropriate",
+    "api-clients/rx": "^2.0",
+    "api-clients/rx-operators": "^2.0"
   },
   "require-dev": {
-    "api-clients/middleware-delay": "dev-master",
-    "api-clients/middleware-pool": "dev-master",
+    "api-clients/middleware-delay": "^2.0",
+    "api-clients/middleware-pool": "^2.0",
     "api-clients/resource-generator": "dev-master",
-    "api-clients/test-utilities": "^2.0"
+    "api-clients/test-utilities": "^3.0.1"
   },
   "suggest": {
     "api-clients/middleware-delay": "Add a delay between requests to not hammer Travis",

src/AsyncClient.php

@@ -9,11 +9,10 @@
 use React\EventLoop\LoopInterface;
 use React\Promise\CancellablePromiseInterface;
 use React\Promise\PromiseInterface;
-use Rx\Observable;
 use Rx\ObservableInterface;
 use Rx\React\Promise;
+use Rx\Scheduler;
 use function ApiClients\Tools\Rx\unwrapObservableFromPromise;
-use function React\Promise\resolve;
 
 final class AsyncClient implements AsyncClientInterface
 {
@@ -35,6 +34,13 @@ public static function create(
         string $token = '',
         array $options = []
     ): self {
+        try {
+            Scheduler::setAsyncFactory(function () use ($loop) {
+                return new Scheduler\EventLoopScheduler($loop);
+            });
+        } catch (\Throwable $t) {
+        }
+
         $options = ApiSettings::getOptions($token, 'Async', $options);
         $client = Factory::create($loop, $options);
         return new self($client);

src/CommandBus/Command/JobLogCommand.php

@@ -0,0 +1,32 @@
+<?php declare(strict_types=1);
+
+namespace ApiClients\Client\Travis\CommandBus\Command;
+
+use WyriHaximus\Tactician\CommandHandler\Annotations\Handler;
+
+/**
+ * @Handler("ApiClients\Client\Travis\CommandBus\Handler\JobLogHandler")
+ */
+final class JobLogCommand
+{
+    /**
+     * @var int
+     */
+    private $id;
+
+    /**
+     * @param int $id
+     */
+    public function __construct(int $id)
+    {
+        $this->id = $id;
+    }
+
+    /**
+     * @return int
+     */
+    public function getId(): int
+    {
+        return $this->id;
+    }
+}

src/CommandBus/Command/RepositoryEventsCommand.php

@@ -0,0 +1,32 @@
+<?php declare(strict_types=1);
+
+namespace ApiClients\Client\Travis\CommandBus\Command;
+
+use WyriHaximus\Tactician\CommandHandler\Annotations\Handler;
+
+/**
+ * @Handler("ApiClients\Client\Travis\CommandBus\Handler\RepositoryEventsHandler")
+ */
+final class RepositoryEventsCommand
+{
+    /**
+     * @var int
+     */
+    private $id;
+
+    /**
+     * @param int $id
+     */
+    public function __construct(int $id)
+    {
+        $this->id = $id;
+    }
+
+    /**
+     * @return int
+     */
+    public function getId(): int
+    {
+        return $this->id;
+    }
+}

src/CommandBus/Handler/AccountsHandler.php

@@ -6,7 +6,6 @@
 use ApiClients\Client\Travis\Resource\AccountInterface;
 use ApiClients\Tools\Services\Client\FetchAndIterateService;
 use React\Promise\PromiseInterface;
-use Rx\Observable;
 use function React\Promise\resolve;
 use function WyriHaximus\React\futureFunctionPromise;
 
@@ -33,6 +32,6 @@ public function __construct(FetchAndIterateService $fetchAndIterateService)
      */
     public function handle(AccountsCommand $command): PromiseInterface
     {
-        return $this->fetchAndIterateService->handle('accounts', 'accounts', AccountInterface::HYDRATE_CLASS);
+        return resolve($this->fetchAndIterateService->iterate('accounts', 'accounts', AccountInterface::HYDRATE_CLASS));
     }
 }

src/CommandBus/Handler/AnnotationsHandler.php

@@ -33,10 +33,10 @@ public function __construct(FetchAndIterateService $service)
      */
     public function handle(AnnotationsCommand $command): PromiseInterface
     {
-        return $this->service->handle(
+        return resolve($this->service->iterate(
             'jobs/' . (string)$command->getRepositoryId() . '/annotations',
             'annotations',
             AnnotationInterface::HYDRATE_CLASS
-        );
+        ));
     }
 }

src/CommandBus/Handler/BranchesHandler.php

@@ -3,7 +3,6 @@
 namespace ApiClients\Client\Travis\CommandBus\Handler;
 
 use ApiClients\Client\Travis\CommandBus\Command\BranchesCommand;
-use ApiClients\Client\Travis\CommandBus\Command\CachesCommand;
 use ApiClients\Client\Travis\Resource\BranchInterface;
 use ApiClients\Tools\Services\Client\FetchAndHydrateService;
 use ApiClients\Tools\Services\Client\FetchAndIterateService;
@@ -34,10 +33,10 @@ public function __construct(FetchAndIterateService $service)
      */
     public function handle(BranchesCommand $command): PromiseInterface
     {
-        return $this->service->handle(
+        return resolve($this->service->iterate(
             'repos/' . (string)$command->getRepositoryId() . '/branches',
             'branches',
             BranchInterface::HYDRATE_CLASS
-        );
+        ));
     }
 }

src/CommandBus/Handler/BroadcastsHandler.php

@@ -6,7 +6,6 @@
 use ApiClients\Client\Travis\Resource\BroadcastInterface;
 use ApiClients\Tools\Services\Client\FetchAndIterateService;
 use React\Promise\PromiseInterface;
-use Rx\Observable;
 use function React\Promise\resolve;
 use function WyriHaximus\React\futureFunctionPromise;
 
@@ -33,6 +32,8 @@ public function __construct(FetchAndIterateService $fetchAndIterateService)
      */
     public function handle(BroadcastsCommand $command): PromiseInterface
     {
-        return $this->fetchAndIterateService->handle('broadcasts', 'broadcasts', BroadcastInterface::HYDRATE_CLASS);
+        return resolve(
+            $this->fetchAndIterateService->iterate('broadcasts', 'broadcasts', BroadcastInterface::HYDRATE_CLASS)
+        );
     }
 }

src/CommandBus/Handler/BuildHandler.php

@@ -32,7 +32,7 @@ public function __construct(FetchAndHydrateService $service)
      */
     public function handle(BuildCommand $command): PromiseInterface
     {
-        return $this->service->handle(
+        return $this->service->fetch(
             'builds/' . (string)$command->getId(),
             'build',
             BuildInterface::HYDRATE_CLASS

src/CommandBus/Handler/BuildsHandler.php

@@ -33,10 +33,10 @@ public function __construct(FetchAndIterateService $service)
      */
     public function handle(BuildsCommand $command): PromiseInterface
     {
-        return $this->service->handle(
+        return resolve($this->service->iterate(
             'repos/' . $command->getRepository() . '/builds',
             'builds',
             BuildInterface::HYDRATE_CLASS
-        );
+        ));
     }
 }

src/CommandBus/Handler/CachesHandler.php

@@ -3,9 +3,7 @@
 namespace ApiClients\Client\Travis\CommandBus\Handler;
 
 use ApiClients\Client\Travis\CommandBus\Command\CachesCommand;
-use ApiClients\Client\Travis\CommandBus\Command\RepositoryCommand;
 use ApiClients\Client\Travis\Resource\CacheInterface;
-use ApiClients\Client\Travis\Resource\RepositoryInterface;
 use ApiClients\Tools\Services\Client\FetchAndHydrateService;
 use ApiClients\Tools\Services\Client\FetchAndIterateService;
 use React\Promise\PromiseInterface;
@@ -35,10 +33,10 @@ public function __construct(FetchAndIterateService $service)
      */
     public function handle(CachesCommand $command): PromiseInterface
     {
-        return $this->service->handle(
+        return resolve($this->service->iterate(
             'repos/' . (string)$command->getRepositoryId() . '/caches',
             'caches',
             CacheInterface::HYDRATE_CLASS
-        );
+        ));
     }
 }

src/CommandBus/Handler/CommitsHandler.php

@@ -33,10 +33,10 @@ public function __construct(FetchAndIterateService $service)
      */
     public function handle(CommitsCommand $command): PromiseInterface
     {
-        return $this->service->handle(
+        return resolve($this->service->iterate(
             'repos/' . $command->getRepository() . '/builds',
             'commits',
             CommitInterface::HYDRATE_CLASS
-        );
+        ));
     }
 }

src/CommandBus/Handler/HooksHandler.php

@@ -3,11 +3,9 @@
 namespace ApiClients\Client\Travis\CommandBus\Handler;
 
 use ApiClients\Client\Travis\CommandBus\Command\HooksCommand;
-use ApiClients\Client\Travis\Resource\AccountInterface;
 use ApiClients\Client\Travis\Resource\HookInterface;
 use ApiClients\Tools\Services\Client\FetchAndIterateService;
 use React\Promise\PromiseInterface;
-use Rx\Observable;
 use function React\Promise\resolve;
 use function WyriHaximus\React\futureFunctionPromise;
 
@@ -34,6 +32,6 @@ public function __construct(FetchAndIterateService $fetchAndIterateService)
      */
     public function handle(HooksCommand $command): PromiseInterface
     {
-        return $this->fetchAndIterateService->handle('hooks', 'hooks', HookInterface::HYDRATE_CLASS);
+        return resolve($this->fetchAndIterateService->iterate('hooks', 'hooks', HookInterface::HYDRATE_CLASS));
     }
 }

src/CommandBus/Handler/JobHandler.php

@@ -32,7 +32,7 @@ public function __construct(FetchAndHydrateService $service)
      */
     public function handle(JobCommand $command): PromiseInterface
     {
-        return $this->service->handle(
+        return $this->service->fetch(
             'jobs/' . (string)$command->getId(),
             'job',
             JobInterface::HYDRATE_CLASS

src/CommandBus/Handler/JobLogHandler.php

@@ -0,0 +1,84 @@
+<?php declare(strict_types=1);
+
+namespace ApiClients\Client\Travis\CommandBus\Handler;
+
+use ApiClients\Client\Pusher\AsyncClient as PusherAsyncClient;
+use ApiClients\Client\Pusher\Event;
+use ApiClients\Client\Pusher\Service\SharedAppClientService;
+use ApiClients\Client\Travis\ApiSettings;
+use ApiClients\Client\Travis\CommandBus\Command\JobLogCommand;
+use ApiClients\Client\Travis\Resource\LogLineInterface;
+use ApiClients\Foundation\Hydrator\Hydrator;
+use React\Promise\PromiseInterface;
+use Rx\Observable;
+use Rx\ObserverInterface;
+use Rx\SchedulerInterface;
+use function ApiClients\Tools\Rx\unwrapObservableFromPromise;
+use function React\Promise\resolve;
+use function WyriHaximus\React\futureFunctionPromise;
+
+final class JobLogHandler
+{
+    /**
+     * @var SharedAppClientService
+     */
+    private $pusher;
+
+    /**
+     * @var Hydrator
+     */
+    private $hydrator;
+
+    /**
+     * JobLogHandler constructor.
+     * @param SharedAppClientService $pusher
+     * @param Hydrator $hydrator
+     */
+    public function __construct(SharedAppClientService $pusher, Hydrator $hydrator)
+    {
+        $this->pusher = $pusher;
+        $this->hydrator = $hydrator;
+    }
+
+    /**
+     * Fetch the given repository and hydrate it
+     *
+     * @param JobLogCommand $command
+     * @return PromiseInterface
+     */
+    public function handle(JobLogCommand $command): PromiseInterface
+    {
+        return $this->pusher->share(
+            ApiSettings::PUSHER_KEY
+        )->then(function (PusherAsyncClient $pusher) use ($command) {
+            return resolve(Observable::create(function (
+                ObserverInterface $observer,
+                SchedulerInterface $scheduler
+            ) use (
+                $pusher,
+                $command
+            ) {
+                $subscription = $pusher->channel('job-' . (string)$command->getId())->filter(function (Event $event) {
+                    return $event->getEvent() === 'job:log';
+                })->map(function (Event $event) {
+                    return $this->hydrator->hydrate(LogLineInterface::HYDRATE_CLASS, $event->getData());
+                })->subscribeCallback(
+                    function (LogLineInterface $line) use ($observer, &$subscription) {
+                        $observer->onNext($line);
+
+                        if ($line->final()) {
+                            $subscription->dispose();
+                        }
+                    },
+                    function ($error) use ($observer) {
+                        $observer->onError($error);
+                    },
+                    function () use ($observer) {
+                        $observer->onComplete();
+                    },
+                    $scheduler
+                );
+            }));
+        });
+    }
+}

src/CommandBus/Handler/JobsHandler.php

@@ -33,10 +33,10 @@ public function __construct(FetchAndIterateService $service)
      */
     public function handle(JobsCommand $command): PromiseInterface
     {
-        return $this->service->handle(
+        return resolve($this->service->iterate(
             'builds/' . (string)$command->getBuildId(),
             'jobs',
             JobInterface::HYDRATE_CLASS
-        );
+        ));
     }
 }