Add unsubscribe feature for client

src/Thruway/ClientSession.php

@@ -45,7 +45,18 @@ public function subscribe($topicName, callable $callback, $options = null)
     {
         return $this->peer->getSubscriber()->subscribe($this, $topicName, $callback, $options);
     }
-
+
+    /**
+     * Unsubscribe
+     * 
+     * @param string $topicName
+     * @return Promise
+     */
+    public function unsubscribe($topicName)
+    {
+        return $this->peer->getSubscriber()->unsubscribe($this, $topicName);
+    }
+
     /**
      * Publish
      *
@@ -129,7 +140,6 @@ public function close()
      */
     public function onClose()
     {
-
+        $this->state = static::STATE_DOWN;
     }
-
 }

src/Thruway/Role/Subscriber.php

@@ -15,6 +15,7 @@
 use Thruway\Message\SubscribedMessage;
 use Thruway\Message\SubscribeMessage;
 use Thruway\Message\UnsubscribedMessage;
+use Thruway\Message\UnsubscribeMessage;
 
 /**
  * Class Subscriber
@@ -28,14 +29,19 @@ class Subscriber extends AbstractRole
      * @var array
      */
     private $subscriptions;
+
+    /**
+     * @var array
+     */
+    private $unsubscriptionsPromises;
 
     /**
      * Constructor
      */
     public function __construct()
     {
-
         $this->subscriptions = [];
+        $this->unsubscriptionsPromises = [];
     }
 
     /**
@@ -86,7 +92,7 @@ protected function processError(AbstractSession $session, ErrorMessage $msg)
                 $this->processSubscribeError($session, $msg);
                 break;
             case Message::MSG_UNSUBSCRIBE:
-                // TODO
+                $this->processUnsubscribeError($session, $msg);
                 break;
             default:
                 Logger::critical($this, "Unhandled error");
@@ -128,6 +134,32 @@ protected function processSubscribed(ClientSession $session, SubscribedMessage $
             }
         }
     }
+
+    /**
+     * Process unsubscribe error
+     * 
+     * @param \Thruway\AbstractSession $session
+     * @param \Thruway\Message\ErrorMessage $msg
+     */
+    protected function processUnsubscribeError(AbstractSession $session, ErrorMessage $msg)
+    {
+        foreach ($this->subscriptions as $key => $subscription) {
+            if ($subscription["unsubscribed_request_id"] === $msg->getErrorRequestId()) {
+                // reject the promise
+                $this->subscriptions[$key]['deferred']->reject($msg);
+                return;
+            }
+        }
+
+        // Execution continues up here in case the original unsubscribe request has not been found
+        foreach ($this->unsubscriptionsPromises as $key => $unsubscriptionPromise) {
+            if ($unsubscriptionPromise["request_id"] === $msg->getErrorRequestId()) {
+                $unsubscriptionPromise["deferred"]->reject($msg);
+                unset($this->unsubscriptionPromises[$key]);
+                return;
+            }
+        }
+    }
 
     /**
      * process unsubscribed
@@ -219,13 +251,49 @@ public function subscribe(ClientSession $session, $topicName, callable $callback
             "options"    => $options,
             "deferred"   => $deferred
         ];
-
+        
         array_push($this->subscriptions, $subscription);
-
+        
         $subscribeMsg = new SubscribeMessage($requestId, $options, $topicName);
         $session->sendMessage($subscribeMsg);
 
         return $deferred->promise();
     }
-
+
+    /**
+     * process unsubscribe
+     * @param ClientSession $session
+     * @param type $topicName
+     * @return boolean
+     */
+    public function unsubscribe(ClientSession $session, $topicName)
+    {
+        $requestId = Utils::getUniqueId();
+        $subscriptionId = false;
+        $deferred  = new Deferred();
+
+        foreach ($this->subscriptions as $i => $subscription) {
+            if ($subscription["topic_name"] == $topicName) {
+                $subscriptionId = $subscription["subscription_id"];
+
+                $this->subscriptions[$i]["unsubscribed_request_id"] = $requestId;
+                $this->subscriptions[$i]["unsubscribed_deferred"] = $deferred;
+            }
+        }
+
+        // In case the client never subscribed to this topic before
+        if ($subscriptionId === false) {
+            $unsubscriptionPromise = [
+                "request_id" => $requestId,
+                "deferred" => $deferred
+            ];
+
+            array_push($this->unsubscriptionsPromises, $unsubscriptionPromise);
+        }
+
+        $unsubscribeMessage = new UnsubscribeMessage($requestId, $subscriptionId);
+        $session->sendMessage($unsubscribeMessage);
+
+        return $deferred->promise();
+    }
 }