diff --git a/CHANGELOG.md b/CHANGELOG.md
index 1c706a9..cfc9741 100644
--- a/CHANGELOG.md
+++ b/CHANGELOG.md
@@ -8,6 +8,11 @@ CHANGELOG
---
+* wamp-topics (devel branch)
+
+ * BC: Renamed "WampServer" to "ServerProtocol"
+ * BC: New "WampServer" component manages Topic containers of subscribed Connections
+
* 0.2 (2012-TBD)
* Ratchet passes every non-binary-frame test from the Autobahn Testsuite
diff --git a/composer.lock b/composer.lock
index a3f6490..dbdf524 100644
--- a/composer.lock
+++ b/composer.lock
@@ -25,15 +25,7 @@
},
{
"package": "symfony/event-dispatcher",
- "version": "dev-master",
- "alias-pretty-version": "2.1.x-dev",
- "alias-version": "2.1.9999999.9999999-dev"
- },
- {
- "package": "symfony/event-dispatcher",
- "version": "dev-master",
- "source-reference": "v2.1.0-BETA3",
- "commit-date": "1342347231"
+ "version": "v2.1.0-BETA3"
},
{
"package": "symfony/http-foundation",
diff --git a/src/Ratchet/Wamp/ServerProtocol.php b/src/Ratchet/Wamp/ServerProtocol.php
new file mode 100644
index 0000000..128ab88
--- /dev/null
+++ b/src/Ratchet/Wamp/ServerProtocol.php
@@ -0,0 +1,153 @@
+<?php
+namespace Ratchet\Wamp;
+use Ratchet\MessageComponentInterface;
+use Ratchet\WebSocket\WsServerInterface;
+use Ratchet\ConnectionInterface;
+
+/**
+ * WebSocket Application Messaging Protocol
+ *
+ * @link http://wamp.ws/spec
+ * @link https://github.com/oberstet/AutobahnJS
+ *
+ * +--------------+----+------------------+
+ * | Message Type | ID | DIRECTION |
+ * |--------------+----+------------------+
+ * | WELCOME | 0 | Server-to-Client |
+ * | PREFIX | 1 | Bi-Directional |
+ * | CALL | 2 | Client-to-Server |
+ * | CALL RESULT | 3 | Server-to-Client |
+ * | CALL ERROR | 4 | Server-to-Client |
+ * | SUBSCRIBE | 5 | Client-to-Server |
+ * | UNSUBSCRIBE | 6 | Client-to-Server |
+ * | PUBLISH | 7 | Client-to-Server |
+ * | EVENT | 8 | Server-to-Client |
+ * +--------------+----+------------------+
+ */
+class ServerProtocol implements MessageComponentInterface, WsServerInterface {
+ const MSG_WELCOME = 0;
+ const MSG_PREFIX = 1;
+ const MSG_CALL = 2;
+ const MSG_CALL_RESULT = 3;
+ const MSG_CALL_ERROR = 4;
+ const MSG_SUBSCRIBE = 5;
+ const MSG_UNSUBSCRIBE = 6;
+ const MSG_PUBLISH = 7;
+ const MSG_EVENT = 8;
+
+ /**
+ * @var WampServerInterface
+ */
+ protected $_decorating;
+
+ /**
+ * @var SplObjectStorage
+ */
+ protected $connections;
+
+ /**
+ * @param WampServerInterface An class to propagate calls through
+ */
+ public function __construct(WampServerInterface $server_component) {
+ $this->_decorating = $server_component;
+ $this->connections = new \SplObjectStorage;
+ }
+
+ /**
+ * {@inheritdoc}
+ */
+ public function getSubProtocols() {
+ if ($this->_decorating instanceof WsServerInterface) {
+ $subs = $this->_decorating->getSubProtocols();
+ $subs[] = 'wamp';
+
+ return $subs;
+ } else {
+ return array('wamp');
+ }
+ }
+
+ /**
+ * {@inheritdoc}
+ */
+ public function onOpen(ConnectionInterface $conn) {
+ $decor = new WampConnection($conn);
+ $this->connections->attach($conn, $decor);
+
+ $this->_decorating->onOpen($decor);
+ }
+
+ /**
+ * {@inheritdoc}
+ * @throws Exception
+ * @throws JsonException
+ */
+ public function onMessage(ConnectionInterface $from, $msg) {
+ $from = $this->connections[$from];
+
+ if (null === ($json = @json_decode($msg, true))) {
+ throw new JsonException;
+ }
+
+ switch ($json[0]) {
+ case static::MSG_PREFIX:
+ $from->WAMP->prefixes[$json[1]] = $json[2];
+ break;
+
+ case static::MSG_CALL:
+ array_shift($json);
+ $callID = array_shift($json);
+ $procURI = array_shift($json);
+
+ if (count($json) == 1 && is_array($json[0])) {
+ $json = $json[0];
+ }
+
+ $this->_decorating->onCall($from, $callID, $procURI, $json);
+ break;
+
+ case static::MSG_SUBSCRIBE:
+ $this->_decorating->onSubscribe($from, $from->getUri($json[1]));
+ break;
+
+ case static::MSG_UNSUBSCRIBE:
+ $this->_decorating->onUnSubscribe($from, $from->getUri($json[1]));
+ break;
+
+ case static::MSG_PUBLISH:
+ $exclude = (array_key_exists(3, $json) ? $json[3] : null);
+ if (!is_array($exclude)) {
+ if (true === (boolean)$exclude) {
+ $exclude = array($from->WAMP->sessionId);
+ } else {
+ $exclude = array();
+ }
+ }
+
+ $eligible = (array_key_exists(4, $json) ? $json[4] : array());
+
+ $this->_decorating->onPublish($from, $from->getUri($json[1]), $json[2], $exclude, $eligible);
+ break;
+
+ default:
+ throw new Exception('Invalid message type');
+ }
+ }
+
+ /**
+ * {@inheritdoc}
+ */
+ public function onClose(ConnectionInterface $conn) {
+ $decor = $this->connections[$conn];
+ $this->connections->detach($conn);
+
+ $this->_decorating->onClose($decor);
+ }
+
+ /**
+ * {@inheritdoc}
+ */
+ public function onError(ConnectionInterface $conn, \Exception $e) {
+ return $this->_decorating->onError($this->connections[$conn], $e);
+ }
+}
\ No newline at end of file
diff --git a/src/Ratchet/Wamp/Topic.php b/src/Ratchet/Wamp/Topic.php
new file mode 100644
index 0000000..5af61fa
--- /dev/null
+++ b/src/Ratchet/Wamp/Topic.php
@@ -0,0 +1,66 @@
+<?php
+namespace Ratchet\Wamp;
+
+/**
+ * A topic/channel containing connections that have subscribed to it
+ */
+class Topic implements \IteratorAggregate, \Countable {
+ private $id;
+
+ private $subscribers;
+
+ /**
+ * @param string Unique ID for this object
+ */
+ public function __construct($topicId) {
+ $this->id = $topicId;
+ $this->subscribers = new \SplObjectStorage;
+ }
+
+ /**
+ * @return string
+ */
+ public function getId() {
+ return $this->id;
+ }
+
+ /**
+ * Send a message to all the connectiosn in this topic
+ * @param string
+ */
+ public function broadcast($msg) {
+ foreach ($thisi->subscribers as $client) {
+ $client->event($this->id, $msg);
+ }
+ }
+
+ /**
+ * @param WampConnection
+ */
+ public function subscribed(WampConnection $conn) {
+ $this->subscribers->attach($conn);
+ }
+
+ /**
+ * @param WampConnection
+ */
+ public function unsubscribed(WampConnection $conn) {
+ if ($this->subscribers->contains($conn)) {
+ $this->subscribers->detach($conn);
+ }
+ }
+
+ /**
+ * {@inheritdoc}
+ */
+ public function getIterator() {
+ return $this->subscribers;
+ }
+
+ /**
+ * {@inheritdoc}
+ */
+ public function count() {
+ return $this->subscribers->count();
+ }
+}
\ No newline at end of file
diff --git a/src/Ratchet/Wamp/WampConnection.php b/src/Ratchet/Wamp/WampConnection.php
index e99b8e5..e625060 100644
--- a/src/Ratchet/Wamp/WampConnection.php
+++ b/src/Ratchet/Wamp/WampConnection.php
@@ -2,7 +2,7 @@
namespace Ratchet\Wamp;
use Ratchet\ConnectionInterface;
use Ratchet\AbstractConnectionDecorator;
-use Ratchet\Wamp\WampServer as WAMP;
+use Ratchet\Wamp\ServerProtocol as WAMP;
/**
* A ConnectionInterface object wrapper that is passed to your WAMP application
diff --git a/src/Ratchet/Wamp/WampServer.php b/src/Ratchet/Wamp/WampServer.php
index 6b6b8be..a63bbdc 100644
--- a/src/Ratchet/Wamp/WampServer.php
+++ b/src/Ratchet/Wamp/WampServer.php
@@ -1,153 +1,100 @@
<?php
namespace Ratchet\Wamp;
-use Ratchet\MessageComponentInterface;
-use Ratchet\WebSocket\WsServerInterface;
use Ratchet\ConnectionInterface;
-/**
- * WebSocket Application Messaging Protocol
- *
- * @link http://wamp.ws/spec
- * @link https://github.com/oberstet/AutobahnJS
- *
- * +--------------+----+------------------+
- * | Message Type | ID | DIRECTION |
- * |--------------+----+------------------+
- * | WELCOME | 0 | Server-to-Client |
- * | PREFIX | 1 | Bi-Directional |
- * | CALL | 2 | Client-to-Server |
- * | CALL RESULT | 3 | Server-to-Client |
- * | CALL ERROR | 4 | Server-to-Client |
- * | SUBSCRIBE | 5 | Client-to-Server |
- * | UNSUBSCRIBE | 6 | Client-to-Server |
- * | PUBLISH | 7 | Client-to-Server |
- * | EVENT | 8 | Server-to-Client |
- * +--------------+----+------------------+
- */
-class WampServer implements MessageComponentInterface, WsServerInterface {
- const MSG_WELCOME = 0;
- const MSG_PREFIX = 1;
- const MSG_CALL = 2;
- const MSG_CALL_RESULT = 3;
- const MSG_CALL_ERROR = 4;
- const MSG_SUBSCRIBE = 5;
- const MSG_UNSUBSCRIBE = 6;
- const MSG_PUBLISH = 7;
- const MSG_EVENT = 8;
+class WampServer implements MessageComponentInterface, WsServerInterface, WampServerInterface {
+ /**
+ * @var ServerProtocol
+ */
+ protected $protocol;
/**
* @var WampServerInterface
*/
- protected $_decorating;
+ protected $app;
/**
- * @var SplObjectStorage
+ * @var array
*/
- protected $connections;
+ protected $topicLookup = array();
- /**
- * @param WampServerInterface An class to propagate calls through
- */
- public function __construct(WampServerInterface $server_component) {
- $this->_decorating = $server_component;
- $this->connections = new \SplObjectStorage;
- }
-
- /**
- * {@inheritdoc}
- */
- public function getSubProtocols() {
- if ($this->_decorating instanceof WsServerInterface) {
- $subs = $this->_decorating->getSubProtocols();
- $subs[] = 'wamp';
-
- return $subs;
- } else {
- return array('wamp');
- }
+ public function __construct(WampServerInterface $app) {
+ $this->protocol = new ServerProtocol($this);
}
/**
* {@inheritdoc}
*/
public function onOpen(ConnectionInterface $conn) {
- $decor = new WampConnection($conn);
- $this->connections->attach($conn, $decor);
-
- $this->_decorating->onOpen($decor);
+ if ($conn instanceof WampConnection) {
+ $this->app->onOpen($conn);
+ } else {
+ $conn->WAMP->topics = new \SplObjectStorage;
+ $this->protocol->onOpen($conn);
+ }
}
/**
* {@inheritdoc}
- * @throws Exception
- * @throws JsonException
*/
- public function onMessage(ConnectionInterface $from, $msg) {
- $from = $this->connections[$from];
+ public function onCall(ConnectionInterface $conn, $id, $topic, array $params) {
+ $this->app->onCall($conn, $id, $this->getTopic($topic), $params);
+ }
- if (null === ($json = @json_decode($msg, true))) {
- throw new JsonException;
+ /**
+ * {@inheritdoc}
+ */
+ public function onSubscribe(ConnectionInterface $conn, $topic) {
+ $topicObj = $this->getTopic($topic);
+
+ $conn->WAMP->topics->attach($topicObj);
+ $this->app->onSubscribe($conn, $topicObj);
+ }
+
+ /**
+ * {@inheritdoc}
+ */
+ public function onUnsubscribe(ConnectionInterface $conn, $topic) {
+ $topicObj = $this->getTopic($topic);
+
+ if ($conn->WAMP->topics->contains($topicobj) {
+ $conn->WAMP->topics->detach($topicObj);
}
- switch ($json[0]) {
- case static::MSG_PREFIX:
- $from->WAMP->prefixes[$json[1]] = $json[2];
- break;
+ $this->topicLookup[$topic]->remove($conn);
+ $this->app->onUnsubscribe($conn, $topicObj);
+ }
- case static::MSG_CALL:
- array_shift($json);
- $callID = array_shift($json);
- $procURI = array_shift($json);
-
- if (count($json) == 1 && is_array($json[0])) {
- $json = $json[0];
- }
-
- $this->_decorating->onCall($from, $callID, $procURI, $json);
- break;
-
- case static::MSG_SUBSCRIBE:
- $this->_decorating->onSubscribe($from, $from->getUri($json[1]));
- break;
-
- case static::MSG_UNSUBSCRIBE:
- $this->_decorating->onUnSubscribe($from, $from->getUri($json[1]));
- break;
-
- case static::MSG_PUBLISH:
- $exclude = (array_key_exists(3, $json) ? $json[3] : null);
- if (!is_array($exclude)) {
- if (true === (boolean)$exclude) {
- $exclude = array($from->WAMP->sessionId);
- } else {
- $exclude = array();
- }
- }
-
- $eligible = (array_key_exists(4, $json) ? $json[4] : array());
-
- $this->_decorating->onPublish($from, $from->getUri($json[1]), $json[2], $exclude, $eligible);
- break;
-
- default:
- throw new Exception('Invalid message type');
- }
+ /**
+ * {@inheritdoc}
+ */
+ public function onPublish(ConnectionInterface $conn, $topic, $event, array $exclude = array(), array $eligible = array()) {
+ $this->app->onPublish($conn, $this->getTopic($topic), $event, $exclude, $eligible);
}
/**
* {@inheritdoc}
*/
public function onClose(ConnectionInterface $conn) {
- $decor = $this->connections[$conn];
- $this->connections->detach($conn);
+ $this->app->onClose($conn);
- $this->_decorating->onClose($decor);
+ foreach ($this->topicLookup as $topic) {
+ $topic->remove($conn);
+ }
}
/**
* {@inheritdoc}
*/
public function onError(ConnectionInterface $conn, \Exception $e) {
- return $this->_decorating->onError($this->connections[$conn], $e);
+ $this->app->onError($conn, $e);
+ }
+
+ protected function getTopic($topic) {
+ if (!array_key_exists($topic, $this->topicLookup)) {
+ $this->topicLookup[$topic] = new Topic($topic);
+ }
+
+ return $this->topicLookup[$topic];
}
}
\ No newline at end of file
diff --git a/tests/Ratchet/Tests/Wamp/WampServerTest.php b/tests/Ratchet/Tests/Wamp/ServerProtocolTest.php
similarity index 97%
rename from tests/Ratchet/Tests/Wamp/WampServerTest.php
rename to tests/Ratchet/Tests/Wamp/ServerProtocolTest.php
index ae46143..4dcb28e 100644
--- a/tests/Ratchet/Tests/Wamp/WampServerTest.php
+++ b/tests/Ratchet/Tests/Wamp/ServerProtocolTest.php
@@ -1,23 +1,23 @@
<?php
namespace Ratchet\Tests\Wamp;
-use Ratchet\Wamp\WampServer;
+use Ratchet\Wamp\ServerProtocol;
use Ratchet\Wamp\WampConnection;
use Ratchet\Tests\Mock\Connection;
use Ratchet\Tests\Mock\WampComponent as TestComponent;
/**
- * @covers Ratchet\Wamp\WampServer
+ * @covers Ratchet\Wamp\ServerProtocol
* @covers Ratchet\Wamp\WampServerInterface
* @covers Ratchet\Wamp\WampConnection
*/
-class WampServerTest extends \PHPUnit_Framework_TestCase {
+class ServerProtocolTest extends \PHPUnit_Framework_TestCase {
protected $_comp;
protected $_app;
public function setUp() {
$this->_app = new TestComponent;
- $this->_comp = new WampServer($this->_app);
+ $this->_comp = new ServerProtocol($this->_app);
}
protected function newConn() {