Merge pull request #6 from clue/collector

Add Collector to process a list of user-generated events

Author: clue

example/peers.php

@@ -0,0 +1,29 @@
+<?php
+
+use Clue\React\Ami\Factory;
+use Clue\React\Ami\Client;
+use Clue\React\Ami\Api;
+use Clue\React\Ami\Collector;
+use Clue\React\Ami\Protocol\Collection;
+
+require __DIR__ . '/../vendor/autoload.php';
+
+$loop = React\EventLoop\Factory::create();
+$factory = new Factory($loop);
+
+$target = isset($argv[1]) ? $argv[1] : 'name:password@localhost';
+
+$factory->createClient($target)->then(function (Client $client) use ($loop) {
+    echo 'Successfully connected' . PHP_EOL;
+
+    $collector = new Collector($client);
+
+    $collector->sipPeers()->then(function (Collection $collection) {
+        var_dump('result', $collection);
+        $peers = $collection->getEntryEvents();
+
+        echo 'found ' . count($peers) . ' peers' . PHP_EOL;
+    });
+}, 'var_dump');
+
+$loop->run();

src/Collector.php

@@ -0,0 +1,74 @@
+<?php
+
+namespace Clue\React\Ami;
+
+use Clue\React\Ami\Client;
+use Clue\React\Ami\Protocol\Response;
+use Clue\React\Ami\Protocol\Event;
+use Clue\React\Ami\Protocol\Collection;
+use React\Promise\Deferred;
+
+class Collector
+{
+    private $client;
+
+    public function __construct(Client $client)
+    {
+        $this->client = $client;
+    }
+
+    public function coreShowChannels()
+    {
+        return $this->collectEvents('CoreShowChannels', 'CoreShowChannelsComplete');
+    }
+
+    public function sipPeers()
+    {
+        return $this->collectEvents('SIPPeers', 'PeerlistComplete');
+    }
+
+    public function agents()
+    {
+        return $this->collectEvents('Agents', 'AgentsComplete');
+    }
+
+    private function collectEvents($command, $expectedEndEvent)
+    {
+        $req = $this->client->createAction($command);
+        $ret = $this->client->request($req);
+        $id = $req->getActionId();
+
+        $deferred = new Deferred();
+
+        // collect all intermediary channel events with this action ID
+        $collected = array();
+        $collector = function (Event $event) use ($id, &$collected, $deferred, $expectedEndEvent) {
+            if ($event->getActionId() === $id) {
+                $collected []= $event;
+
+                if ($event->getName() === $expectedEndEvent) {
+                    $deferred->resolve($collected);
+                }
+            }
+        };
+        $this->client->on('event', $collector);
+
+        // unregister collector if client fails
+        $client = $this->client;
+        $unregister = function () use ($client, $collector) {
+            $client->removeListener('event', $collector);
+        };
+        $ret->then(null, $unregister);
+
+        // stop waiting for events
+        $deferred->promise()->then($unregister);
+
+        return $ret->then(function (Response $response) use ($deferred) {
+            // final result has been received => merge all intermediary channel events
+            return $deferred->promise()->then(function ($collected) use ($response) {
+                $last = array_pop($collected);
+                return new Collection($response, $collected, $last);
+            });
+        });
+    }
+}

src/Protocol/Collection.php

@@ -0,0 +1,40 @@
+<?php
+
+namespace Clue\React\Ami\Protocol;
+
+use UnexpectedValueException;
+
+class Collection extends Message
+{
+    private $response;
+    private $entryEvents;
+    private $completeEvent;
+
+    public function __construct(Response $response, array $entryEvents, Event $completeEvent)
+    {
+//         if ($response->getPart('EventItems') !== count($events)) {
+//             var_dump($response, $events);
+//             throw new UnexpectedValueException('Expected ' . $response->getPart('EventItems') . ' entries, but collected ' . count($events));
+//         }
+
+        $this->parts = $response->getParts();
+        $this->response = $response;
+        $this->entryEvents = $entryEvents;
+        $this->completeEvent = $completeEvent;
+    }
+
+    public function getResponse()
+    {
+        return $this->response;
+    }
+
+    public function getEntryEvents()
+    {
+        return $this->entryEvents;
+    }
+
+    public function getCompleteEvent()
+    {
+        return $this->completeEvent;
+    }
+}

tests/CollectorTest.php

@@ -0,0 +1,60 @@
+<?php
+
+use Clue\React\Ami\Collector;
+use Clue\React\Ami\Protocol\Collection;
+use Clue\React\Ami\Protocol\Response;
+use Clue\React\Ami\Client;
+use Clue\React\Ami\Protocol\Event;
+use Clue\React\Ami\Protocol\Action;
+
+class CollectorTest extends TestCase
+{
+    public function testCollectingSIPEvents()
+    {
+        $client = $this->createClientMock();
+
+        // expect a single outgoing action request (and mock its ID)
+        $client->expects($this->once())
+                 ->method('createAction')
+                 ->with($this->equalTo('SIPPeers'), $this->equalTo(array()))
+                 ->will($this->returnValue(new Action('SIPPeers', array('ActionID' => '123'))));
+
+        $collector = new Collector($client);
+
+        $promise = $collector->sipPeers();
+
+        // save resolved result for comparisions
+        $resolved = null;
+        $promise->then(function($result) use (&$resolved) {
+            $resolved = $result;
+        });
+
+        // should not start out resolved
+        $this->assertNull($resolved);
+
+        $response = new Response(array('Response' => 'Success', 'ActionID' => '123'));
+
+        $client->handleMessage($response);
+        $client->handleMessage(new Event('PeerEntry', array('ActionID' => '123')));
+        $client->handleMessage(new Event('PeerEntry', array('ActionID' => '123')));
+
+        $this->assertNull($resolved);
+
+        $client->handleMessage(new Event('PeerlistComplete', array('EventList' => 'complete', 'ListItems' => '2', 'ActionID' => '123')));
+
+        $this->assertNotNull($resolved);
+
+        $promise->then(
+            $this->expectCallableOnce()
+        );
+    }
+
+    private function createClientMock()
+    {
+        $stream = $this->getMockBuilder('React\Stream\Stream')->disableOriginalConstructor()->getMock();
+
+        $client = $this->getMockBuilder('Clue\React\Ami\Client')->setMethods(array('createAction'))->setConstructorArgs(array($stream))->getMock();
+
+        return $client;
+    }
+}