Deliver rates example

Author: WyriHaximus

examples/queues-subscribe-rates-async.php

@@ -0,0 +1,21 @@
+<?php declare(strict_types=1);
+
+use ApiClients\Client\RabbitMQ\Management\AsyncClient;
+use function ApiClients\Foundation\resource_pretty_print;
+use ApiClients\Client\RabbitMQ\Management\Resource\QueueInterface;
+use React\EventLoop\Factory;
+use Rx\Observer\CallbackObserver;
+
+require dirname(__DIR__) . DIRECTORY_SEPARATOR . 'vendor/autoload.php';
+
+$loop = Factory::create();
+$config = require 'resolve_config.php';
+
+$client = AsyncClient::create($loop, $config['baseUrl'], $config['username'], $config['password']);
+$queues = $client->queues(1)->filter(function (QueueInterface $queue) {
+    return strpos($queue->name(), 'amq.gen') !== 0;
+})->subscribe(new CallbackObserver(function (QueueInterface $queue) {
+    echo $queue->name(), ': ', $queue->messageStats()->deliverDetails()->rate(), PHP_EOL;
+}));
+
+$loop->run();

src/AsyncClient.php

@@ -14,6 +14,7 @@
 use Rx\ObservableInterface;
 use Rx\React\Promise;
 use function React\Promise\resolve;
+use Rx\Scheduler\EventLoopScheduler;
 
 final class AsyncClient implements AsyncClientInterface
 {
@@ -79,18 +80,26 @@ public function overview(): PromiseInterface
     }
 
     /**
+     * @param int|null $interval
      * @return ObservableInterface
      */
-    public function queues(): ObservableInterface
+    public function queues(int $interval = null): ObservableInterface
     {
-        return Promise::toObservable($this->client->handle(
-            new SimpleRequestCommand('queues')
-        ))->flatMap(function (ResponseInterface $response) {
-            return Observable::fromArray($response->getBody()->getJson());
-        })->flatMap(function ($queue) {
+        if ($interval === null) {
             return Promise::toObservable($this->client->handle(
-                new HydrateCommand('Queue', $queue)
-            ));
+                new SimpleRequestCommand('queues')
+            ))->flatMap(function (ResponseInterface $response) {
+                return Observable::fromArray($response->getBody()->getJson());
+            })->flatMap(function ($queue) {
+                return Promise::toObservable($this->client->handle(
+                    new HydrateCommand('Queue', $queue)
+                ));
+            });
+        }
+
+        $scheduler = new EventLoopScheduler($this->client->getFromContainer(LoopInterface::class));
+        return Observable::interval($interval * 1000, $scheduler)->flatMap(function () {
+            return $this->queues();
         });
     }
 

src/AsyncClientInterface.php

@@ -16,9 +16,10 @@ interface AsyncClientInterface
     public function overview(): PromiseInterface;
 
     /**
+     * @param int|null $interval
      * @return ObservableInterface
      */
-    public function queues(): ObservableInterface;
+    public function queues(int $interval = null): ObservableInterface;
 
     /**
      * @return ObservableInterface