Merge Collector into ActionSender

Author: clue

README.md

@@ -208,7 +208,16 @@ $sender = new ActionSender($client);
 All public methods resemble their respective AMI actions.
 
 ```php
+$sender->login($name, $pass);
+$sender->logoff();
 $sender->ping();
+$sender->command($command);
+$sender->events($eventMask);
+
+$sender->coreShowChannels();
+$sender->sipPeers();
+$sender->agents();
+
 // many more…
 ```
 
@@ -239,6 +248,10 @@ $sender->ping()->then(
 });
 ```
 
+All actions resolve with a [`Response`](#response) object on success,
+some actions are documented to return the specialized [`Collection`](#collection)
+object to contain a list of entries.
+
 #### Custom actions
 
 Using the `ActionSender` is not strictly necessary, but is the recommended way to execute common actions.
@@ -305,6 +318,7 @@ to access the list entries and completion event.
 
 ```
 Action: CoreShowChannels
+
 Response: Success
 EventList: start
 Message: Channels will follow

examples/peers.php

@@ -1,9 +1,8 @@
 <?php
 
-use Clue\React\Ami\Factory;
-use Clue\React\Ami\Client;
 use Clue\React\Ami\ActionSender;
-use Clue\React\Ami\Collector;
+use Clue\React\Ami\Client;
+use Clue\React\Ami\Factory;
 use Clue\React\Ami\Protocol\Collection;
 
 require __DIR__ . '/../vendor/autoload.php';
@@ -16,7 +15,7 @@
 $factory->createClient($target)->then(function (Client $client) use ($loop) {
     echo 'Successfully connected' . PHP_EOL;
 
-    $collector = new Collector($client);
+    $collector = new ActionSender($client);
 
     $collector->sipPeers()->then(function (Collection $collection) {
         var_dump('result', $collection);

src/ActionSender.php

@@ -3,10 +3,10 @@
 namespace Clue\React\Ami;
 
 use Clue\React\Ami\Client;
-use Clue\React\Ami\Protocol\Response;
-use Clue\React\Ami\Protocol\Action;
-use UnexpectedValueException;
+use Clue\React\Ami\Protocol\Collection;
 use Clue\React\Ami\Protocol\Event;
+use Clue\React\Ami\Protocol\Response;
+use React\Promise\Deferred;
 
 class ActionSender
 {
@@ -39,11 +39,6 @@ public function ping()
         return $this->request('Ping');
     }
 
-    public function coreShowChannels()
-    {
-        return $this->request('CoreShowChannels');
-    }
-
     public function command($command)
     {
         return $this->request('Command', array('Command' => $command));
@@ -62,11 +57,6 @@ public function events($eventMask)
         return $this->request('Events', array('EventMask' => $eventMask));
     }
 
-    public function sipPeers()
-    {
-        return $this->request('SIPPeers');
-    }
-
     public function sipShowPeer($peerName)
     {
         return $this->request('SIPshowpeer', array('Peer' => $peerName));
@@ -97,6 +87,32 @@ public function getConfig($filename, $category = null)
         return $this->request('GetConfig', array('Filename' => $filename, 'Category' => $category));
     }
 
+    /**
+     * @return \React\Promise\PromiseInterface Promise<Collection> collection with "Event: CoreShowChannel"
+     */
+    public function coreShowChannels()
+    {
+        return $this->collectEvents('CoreShowChannels', 'CoreShowChannelsComplete');
+    }
+
+    /**
+     *
+     * @return \React\Promise\PromiseInterface Promise<Collection> collection with "Event: PeerEntry"
+     */
+    public function sipPeers()
+    {
+        return $this->collectEvents('SIPPeers', 'PeerlistComplete');
+    }
+
+    /**
+     *
+     * @return \React\Promise\PromiseInterface Promise<Collection> collection with "Event: Agents"
+     */
+    public function agents()
+    {
+        return $this->collectEvents('Agents', 'AgentsComplete');
+    }
+
     private function boolParam($value)
     {
         if ($value === true) {
@@ -112,4 +128,44 @@ private function request($name, array $args = array())
     {
         return $this->client->request($this->client->createAction($name, $args));
     }
+
+    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/Collector.php

@@ -1,13 +1,12 @@
 <?php
 
-use Clue\React\Ami\Collector;
-use Clue\React\Ami\Protocol\Collection;
-use Clue\React\Ami\Protocol\Response;
+use Clue\React\Ami\ActionSender;
 use Clue\React\Ami\Client;
-use Clue\React\Ami\Protocol\Event;
 use Clue\React\Ami\Protocol\Action;
+use Clue\React\Ami\Protocol\Event;
+use Clue\React\Ami\Protocol\Response;
 
-class CollectorTest extends TestCase
+class ActionSenderTest extends TestCase
 {
     public function testCollectingSIPEvents()
     {
@@ -19,7 +18,7 @@ public function testCollectingSIPEvents()
                  ->with($this->equalTo('SIPPeers'), $this->equalTo(array()))
                  ->will($this->returnValue(new Action(array('Action' => 'SIPPeers', 'ActionID' => '123'))));
 
-        $collector = new Collector($client);
+        $collector = new ActionSender($client);
 
         $promise = $collector->sipPeers();