From 3ca46452313615caf98c9d428a31bc8870914535 Mon Sep 17 00:00:00 2001
From: Daniele Alessandri <suppakilla@gmail.com>
Date: Sun, 24 Apr 2011 13:57:19 +0200
Subject: [PATCH] Implement Predis\DispatcherLoop on top of
 Predis\PubSubContext.

---
 lib/Predis/DispatcherLoop.php | 81 +++++++++++++++++++++++++++++++++++
 1 file changed, 81 insertions(+)
 create mode 100644 lib/Predis/DispatcherLoop.php

diff --git a/lib/Predis/DispatcherLoop.php b/lib/Predis/DispatcherLoop.php
new file mode 100644
index 00000000..23cce6c5
--- /dev/null
+++ b/lib/Predis/DispatcherLoop.php
@@ -0,0 +1,81 @@
+<?php
+
+namespace Predis;
+
+class DispatcherLoop {
+    private $_client;
+    private $_pubSubContext;
+    private $_callbacks;
+    private $_defaultCallback;
+    private $_subscriptionCallback;
+
+    public function __construct(Client $client) {
+        $this->_callbacks = array();
+        $this->_client = $client;
+        $this->_pubSubContext = $client->pubSubContext();
+    }
+
+    protected function validateCallback($callback) {
+        if (!is_callable($callback)) {
+            throw new ClientException(
+                "The callback parameter must be a valid callable object"
+            );
+        }
+    }
+
+    public function getPubSubContext() {
+        return $this->_pubSubContext;
+    }
+
+    public function subscriptionCallback($callback = null) {
+        if (isset($callback)) {
+            $this->validateCallback($callback);
+        }
+        $this->_subscriptionCallback = $callback;
+    }
+
+    public function defaultCallback($callback = null) {
+        if (isset($callback)) {
+            $this->validateCallback($callback);
+        }
+        $this->_subscriptionCallback = $callback;
+    }
+
+    public function attachCallback($channel, $callback) {
+        $this->validateCallback($callback);
+        $this->_callbacks[$channel] = $callback;
+        $this->_pubSubContext->subscribe($channel);
+    }
+
+    public function detachCallback($channel) {
+        if (isset($this->_callbacks[$channel])) {
+            unset($this->_callbacks[$channel]);
+            $this->_pubSubContext->unsubscribe($channel);
+        }
+    }
+
+    public function run() {
+        foreach ($this->_pubSubContext as $message) {
+            $kind = $message->kind;
+            if ($kind !== PubSubContext::MESSAGE && $kind !== PubSubContext::PMESSAGE) {
+                if (isset($this->_subscriptionCallback)) {
+                    $callback = $this->_subscriptionCallback;
+                    $callback($message);
+                }
+                continue;
+            }
+            if (isset($this->_callbacks[$message->channel])) {
+                $callback = $this->_callbacks[$message->channel];
+                $callback($message->payload);
+            }
+            else if (isset($this->_defaultCallback)) {
+                $callback = $this->_defaultCallback;
+                $callback($message);
+            }
+        }
+    }
+
+    public function stop() {
+        $this->_pubSubContext->closeContext();
+    }
+}