Updated example with a cleaner and clearer output

Author: WyriHaximus

examples/queues-subscribe-rates-async.php

@@ -11,11 +11,43 @@
 $loop = Factory::create();
 $config = require 'resolve_config.php';
 
+$height = 0;
+$length = 0;
+$queues = [];
+
 $client = AsyncClient::create($loop, $config['baseUrl'], $config['username'], $config['password']);
-$queues = $client->queues(1)->filter(function (QueueInterface $queue) {
+$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;
+})->subscribe(new CallbackObserver(function (QueueInterface $queue) use (&$queues) {
+    $queues[$queue->name()] = $queue->messageStats()->deliverDetails()->rate();
 }));
 
+$loop->addPeriodicTimer(1, function () use (&$height, &$length, &$queues) {
+    for ($i = 0; $i < $height; $i++) {
+        echo "\033[1A";
+    }
+
+    if ($height < count($queues)) {
+        $height = count($queues);
+    }
+
+    ksort($queues);
+
+    $i = 0;
+    foreach ($queues as $key => $value) {
+        $line = $key . ': ' . $value . ' sec';
+        if ($length < strlen($line)) {
+            $length = strlen($line);
+        }
+        $line = str_pad($line, $length, ' ');
+        echo $line, PHP_EOL;
+        $i++;
+    }
+    for (; $i < $height; $i++) {
+        echo "\r";
+    }
+
+    $queues = [];
+});
+
 $loop->run();