hazza/Phraseanet
1
0
Fork 0
mirror of https://github.com/alchemy-fr/Phraseanet.git synced 2025-10-17 15:03:25 +00:00
Code Issues Packages Projects Releases Wiki Activity

Remove Websockets dependencies

Browse Source
This commit is contained in:
Nicolas Le Goff
2015-03-09 20:22:50 +01:00
parent 6b89c123e3
commit bd4836aaa7
42 changed files with 21 additions and 1819 deletions
Download Patch File Download Diff File Expand all files Collapse all files

11
Gruntfile.js
View File

@@ -323,16 +323,6 @@ module.exports = function(grunt) {
"dest": "<%= path.asset %>/underscore-amd/",
"flatten": true
},
"web-socket-js": {
"expand": true,
"src": [
"<%= path.bower %>/web-socket-js/LICENSE.txt",
"<%= path.bower %>/web-socket-js/WebSocketMain.swf",
"<%= path.bower %>/web-socket-js/web_socket.js"
],
"dest": "<%= path.asset %>/web-socket-js",
"flatten": true
},
"zxcvbn": {
"expand": true,
"src": [
@@ -430,7 +420,6 @@ module.exports = function(grunt) {
"copy:swfobject",
"copy:tinymce",
"copy:underscore",
"copy:web-socket-js",
"copy:zxcvbn"
]);
grunt.registerTask("install-assets", [

1
bower.json
View File

@@ -27,7 +27,6 @@
"jquery.cookie": "~1.4",
"autobahn": "http://autobahn.s3.amazonaws.com/js/autobahn.min.js",
"when": "~2.7.0",
"web-socket-js": "~1.0.1",
"jquery.treeview": "1.4.1",
"joyride": "https://github.com/zurb/joyride/archive/v2.0.0.zip"
},

2
composer.json
View File

@@ -29,7 +29,6 @@
"alchemy/phlickr" : "0.2.9",
"alchemy/task-manager" : "2.0.x-dev@dev",
"alchemy/zippy" : "0.2.x-dev@dev",
"cboden/ratchet" : "~0.3",
"dailymotion/sdk" : "~1.5",
"data-uri/data-uri" : "~0.1.0",
"doctrine/orm" : "~2.4.0",
@@ -60,7 +59,6 @@
"php-ffmpeg/php-ffmpeg" : "~0.5.0",
"php-xpdf/php-xpdf" : "~0.2.1",
"phpexiftool/phpexiftool" : "dev-0.4.1-mwg-metadata-copy as 0.4.1",
"react/zmq" : "~0.2",
"rhumsaa/uuid" : "~2.7",
"silex/silex" : "1.1.x-dev@dev",
"silex/web-profiler" : "~1.0.0@dev",

18
config/configuration.sample.yml
View File

@@ -32,23 +32,15 @@ main:
task-manager:
status: started
enabled: true
logger:
max-files: 10
enabled: true
level: INFO
listener:
options:
protocol: tcp
host: 127.0.0.1
port: 6660
linger: 500
websocket-server:
host: local.phrasea
port: 9090
ip: 0.0.0.0
subscriber:
protocol: tcp
host: 127.0.0.1
port: 13598
logger:
max-files: 10
enabled: true
level: INFO
session:
type: 'file'
options: []

6
lib/Alchemy/Phrasea/Authentication/Authenticator.php
View File

@@ -88,13 +88,7 @@ class Authenticator
{
$user = $session->getUser();
$rights = [];
if ($this->app['acl']->get($user)->has_right('taskmanager')) {
$rights[] = 'task-manager';
}
$this->session->set('usr_id', $user->getId());
$this->session->set('websockets_rights', $rights);
$this->session->set('session_id', $session->getId());
}

2
lib/Alchemy/Phrasea/CLI.php
View File

@@ -13,7 +13,6 @@ namespace Alchemy\Phrasea;
use Alchemy\Phrasea\Command\CommandInterface;
use Alchemy\Phrasea\Core\CLIProvider\TranslationExtractorServiceProvider;
use Alchemy\Phrasea\Core\CLIProvider\WebsocketServerServiceProvider;
use Alchemy\Phrasea\Core\Event\Subscriber\BridgeSubscriber;
use Alchemy\Phrasea\Core\PhraseaCLIExceptionHandler;
use Alchemy\Phrasea\Exception\RuntimeException;
@@ -67,7 +66,6 @@ class CLI extends Application
);
$this->register(new PluginServiceProvider());
$this->register(new WebsocketServerServiceProvider());
$this->register(new ComposerSetupServiceProvider());
$this->register(new CLIDriversServiceProvider());
$this->register(new LessBuilderServiceProvider());

37
lib/Alchemy/Phrasea/Command/WebsocketServer.php
View File

@@ -1,37 +0,0 @@
<?php
/*
* This file is part of Phraseanet
*
* (c) 2005-2014 Alchemy
*
* For the full copyright and license information, please view the LICENSE
* file that was distributed with this source code.
*/
namespace Alchemy\Phrasea\Command;
use Alchemy\Phrasea\Exception\RuntimeException;
use Symfony\Component\Console\Input\InputInterface;
use Symfony\Component\Console\Output\OutputInterface;
class WebsocketServer extends Command
{
public function __construct($name = null)
{
parent::__construct($name);
$this->setDescription("Runs the websocket server");
}
public function doExecute(InputInterface $input, OutputInterface $output)
{
$sessionConf = $this->container['conf']->get(['main', 'session', 'type'], 'file');
if (!in_array($sessionConf, ['memcached', 'memcache', 'redis'])) {
throw new RuntimeException(sprintf('Running the websocket server requires a server session storage, type `%s` provided', $sessionConf));
}
$this->container['ws.server']->run();
}
}

7
lib/Alchemy/Phrasea/Core/CLIProvider/TaskManagerServiceProvider.php
View File

@@ -31,9 +31,9 @@ class TaskManagerServiceProvider implements ServiceProviderInterface
});
$app['task-manager'] = $app->share(function (Application $app) {
$options = $app['task-manager.listener.options'];
$options = $app['task-manager.options'];
$manager = TaskManager::create(
return TaskManager::create(
$app['dispatcher'],
$app['task-manager.logger'],
$app['task-manager.task-list'],
@@ -44,9 +44,6 @@ class TaskManagerServiceProvider implements ServiceProviderInterface
'tick_period' => 1,
]
);
$manager->addSubscriber($app['ws.task-manager.broadcaster']);
return $manager;
});
$app['task-manager.logger.configuration'] = $app->share(function (Application $app) {

108
lib/Alchemy/Phrasea/Core/CLIProvider/WebsocketServerServiceProvider.php
View File

@@ -1,108 +0,0 @@
<?php
/*
* This file is part of Phraseanet
*
* (c) 2005-2014 Alchemy
*
* For the full copyright and license information, please view the LICENSE
* file that was distributed with this source code.
*/
namespace Alchemy\Phrasea\Core\CLIProvider;
use Alchemy\Phrasea\Websocket\Consumer\ConsumerManager;
use Alchemy\Phrasea\Websocket\Topics\Directive;
use Alchemy\Phrasea\Websocket\Topics\DirectivesManager;
use Alchemy\Phrasea\Websocket\Subscriber\TaskManagerBroadcasterSubscriber;
use Alchemy\Phrasea\Websocket\PhraseanetWampServer;
use Alchemy\Phrasea\Websocket\Topics\Plugin\TaskManagerSubscriberPlugin;
use Alchemy\Phrasea\Websocket\Topics\TopicsManager;
use Ratchet\App;
use Ratchet\Session\SessionProvider;
use Ratchet\Wamp\WampServer;
use Silex\Application;
use Silex\ServiceProviderInterface;
use React\EventLoop\Factory as EventLoopFactory;
class WebsocketServerServiceProvider implements ServiceProviderInterface
{
public function register(Application $app)
{
$app['ws.publisher.options'] = $app->share(function (Application $app) {
return array_replace([
'protocol' => 'tcp',
'host' => '127.0.0.1',
'port' => 13598,
], $app['conf']->get(['main', 'websocket-server', 'subscriber'], []));
});
$app['ws.task-manager.broadcaster'] = $app->share(function (Application $app) {
return TaskManagerBroadcasterSubscriber::create($app['ws.publisher.options']);
});
$app['ws.event-loop'] = $app->share(function () {
return EventLoopFactory::create();
});
$app['ws.server.subscriber'] = $app->share(function (Application $app) {
return new TaskManagerSubscriberPlugin($app['ws.publisher.options'], $app['ws.event-loop'], $app['ws.server.logger']);
});
$app['ws.server.application'] = $app->share(function (Application $app) {
return new SessionProvider(
new WampServer($app['ws.server.phraseanet-server']), $app['session.storage.handler']
);
});
$app['ws.server.phraseanet-server'] = $app->share(function (Application $app) {
return new PhraseanetWampServer($app['ws.server.topics-manager'], $app['ws.server.logger']);
});
$app['ws.server.logger'] = $app->share(function (Application $app) {
return $app['task-manager.logger'];
});
$app['ws.server.topics-manager.directives.conf'] = $app->share(function (Application $app) {
return [
new Directive(TopicsManager::TOPIC_TASK_MANAGER, true, ['task-manager']),
];
});
$app['ws.server.topics-manager.directives'] = $app->share(function (Application $app) {
return new DirectivesManager($app['ws.server.topics-manager.directives.conf']);
});
$app['ws.server.consumer-manager'] = $app->share(function (Application $app) {
return new ConsumerManager();
});
$app['ws.server.topics-manager'] = $app->share(function (Application $app) {
$manager = new TopicsManager($app['ws.server.topics-manager.directives'], $app['ws.server.consumer-manager']);
$manager->attach($app['ws.server.subscriber']);
return $manager;
});
$app['ws.server.options'] = $app->share(function (Application $app) {
return array_replace([
'host' => 'localhost',
'port' => 9090,
'ip' => '127.0.0.1',
], $app['conf']->get(['main', 'websocket-server'], []));
});
$app['ws.server'] = $app->share(function (Application $app) {
$options = $app['ws.server.options'];
$server = new App($options['host'], $options['port'], $options['ip'], $app['ws.event-loop']);
$server->route('/websockets', $app['ws.server.application']);
return $server;
});
}
public function boot(Application $app)
{
}
}

6
lib/Alchemy/Phrasea/Core/Provider/TasksServiceProvider.php
View File

@@ -33,16 +33,16 @@ class TasksServiceProvider implements ServiceProviderInterface
public function register(Application $app)
{
$app['task-manager.notifier'] = $app->share(function (Application $app) {
return Notifier::create($app['monolog'], $app['task-manager.listener.options']);
return Notifier::create($app['monolog'], $app['task-manager.options']);
});
$app['task-manager.listener.options'] = $app->share(function (Application $app) {
$app['task-manager.options'] = $app->share(function (Application $app) {
return array_replace([
'protocol' => 'tcp',
'host' => '127.0.0.1',
'port' => 6660,
'linger' => 500,
], $app['conf']->get(['main', 'task-manager', 'listener'], []));
], $app['conf']->get(['main', 'task-manager', 'options'], []));
});
$app['task-manager.job-factory'] = $app->share(function (Application $app) {

6
lib/Alchemy/Phrasea/Setup/Requirements/SystemRequirements.php
View File

@@ -242,12 +242,6 @@ class SystemRequirements extends RequirementCollection implements RequirementInt
'Install and enable the <strong>twig</strong> extension.'
);
$this->addRecommendation(
function_exists('event_base_new'),
'LibEvent extension is strongly recommended in production',
'Install and enable the <strong>LibEvent</strong> extension.'
);
$this->addRequirement(
extension_loaded('zmq'),
'ZMQ extension is required.',

43
lib/Alchemy/Phrasea/Websocket/Consumer/Consumer.php
View File

@@ -1,43 +0,0 @@
<?php
/*
* This file is part of Phraseanet
*
* (c) 2005-2014 Alchemy
*
* For the full copyright and license information, please view the LICENSE
* file that was distributed with this source code.
*/
namespace Alchemy\Phrasea\Websocket\Consumer;
/**
* Websocket consumer
*/
class Consumer implements ConsumerInterface
{
private $usrId;
private $rights;
public function __construct($usrId, array $rights)
{
$this->usrId = $usrId;
$this->rights = $rights;
}
/**
* {@inheritdoc}
*/
public function isAuthenticated()
{
return $this->usrId !== null;
}
/**
* {@inheritdoc}
*/
public function hasRights($rights)
{
return count(array_intersect($this->rights, (array) $rights)) === count($rights);
}
}

31
lib/Alchemy/Phrasea/Websocket/Consumer/ConsumerInterface.php
View File

@@ -1,31 +0,0 @@
<?php
/*
* This file is part of Phraseanet
*
* (c) 2005-2014 Alchemy
*
* For the full copyright and license information, please view the LICENSE
* file that was distributed with this source code.
*/
namespace Alchemy\Phrasea\Websocket\Consumer;
interface ConsumerInterface
{
/**
* Return true if the consumer is authenticated in Phraseanet
*
* @return Boolean
*/
public function isAuthenticated();
/**
* Return true if the user has the given rights
*
* @param string\array $rights A right or an array of rights
*
* @return Boolean
*/
public function hasRights($rights);
}

32
lib/Alchemy/Phrasea/Websocket/Consumer/ConsumerManager.php
View File

@@ -1,32 +0,0 @@
<?php
/*
* This file is part of Phraseanet
*
* (c) 2005-2014 Alchemy
*
* For the full copyright and license information, please view the LICENSE
* file that was distributed with this source code.
*/
namespace Alchemy\Phrasea\Websocket\Consumer;
use Symfony\Component\HttpFoundation\Session\SessionInterface;
class ConsumerManager
{
/**
* Creates a consumer given a Session
*
* @param Session $session
*
* @return Consumer
*/
public function create(SessionInterface $session)
{
$usrId = $session->has('usr_id') ? $session->get('usr_id') : null;
$rights = $session->has('websockets_rights') ? $session->get('websockets_rights') : [];
return new Consumer($usrId, $rights);;
}
}

94
lib/Alchemy/Phrasea/Websocket/PhraseanetWampServer.php
View File

@@ -1,94 +0,0 @@
<?php
/*
* This file is part of Phraseanet
*
* (c) 2005-2014 Alchemy
*
* For the full copyright and license information, please view the LICENSE
* file that was distributed with this source code.
*/
namespace Alchemy\Phrasea\Websocket;
use Alchemy\Phrasea\Websocket\Topics\TopicsManager;
use Psr\Log\LoggerInterface;
use Ratchet\ConnectionInterface as Conn;
use Ratchet\Wamp\WampServerInterface;
class PhraseanetWampServer implements WampServerInterface
{
private $logger;
private $manager;
public function __construct(TopicsManager $manager, LoggerInterface $logger)
{
$this->logger = $logger;
$this->manager = $manager;
}
/**
* {@inheritdoc}
*/
public function onPublish(Conn $conn, $topic, $event, array $exclude, array $eligible)
{
$this->logger->error(sprintf('Publishing on topic %s', $topic->getId()), ['event' => $event, 'topic' => $topic]);
$topic->broadcast($event);
}
/**
* {@inheritdoc}
*/
public function onCall(Conn $conn, $id, $topic, array $params)
{
$this->logger->error(sprintf('Received RPC call on topic %s', $topic->getId()), ['topic' => $topic]);
$conn->callError($id, $topic, 'RPC not supported on this demo');
}
/**
* {@inheritdoc}
*/
public function onSubscribe(Conn $conn, $topic)
{
if ($this->manager->subscribe($conn, $topic)) {
$this->logger->debug(sprintf('Subscription received on topic %s', $topic->getId()), ['topic' => $topic]);
} else {
$this->logger->error(sprintf('Subscription received on topic %s, user is not allowed', $topic->getId()), ['topic' => $topic]);
}
}
/**
* {@inheritdoc}
*/
public function onUnSubscribe(Conn $conn, $topic)
{
$this->logger->debug(sprintf('Unsubscription received on topic %s', $topic->getId()), ['topic' => $topic]);
$this->manager->unsubscribe($conn, $topic);
}
/**
* {@inheritdoc}
*/
public function onOpen(Conn $conn)
{
$this->logger->debug('[WS] Connection request accepted');
$this->manager->openConnection($conn);
}
/**
* {@inheritdoc}
*/
public function onClose(Conn $conn)
{
$this->logger->debug('[WS] Connection closed');
$this->manager->closeConnection($conn);
}
/**
* {@inheritdoc}
*/
public function onError(Conn $conn, \Exception $e)
{
$this->logger->error('[WS] Connection error', ['exception' => $e]);
}
}

88
lib/Alchemy/Phrasea/Websocket/Subscriber/TaskManagerBroadcasterSubscriber.php
View File

@@ -1,88 +0,0 @@
<?php
/*
* This file is part of Phraseanet
*
* (c) 2005-2014 Alchemy
*
* For the full copyright and license information, please view the LICENSE
* file that was distributed with this source code.
*/
namespace Alchemy\Phrasea\Websocket\Subscriber;
use Alchemy\Phrasea\Websocket\Topics\TopicsManager;
use Alchemy\TaskManager\Event\StateFormater;
use Alchemy\TaskManager\Event\TaskManagerEvent;
use Alchemy\TaskManager\Event\TaskManagerRequestEvent;
use Alchemy\TaskManager\Event\TaskManagerEvents;
use Symfony\Component\EventDispatcher\EventSubscriberInterface;
use Alchemy\TaskManager\ZMQSocket;
class TaskManagerBroadcasterSubscriber implements EventSubscriberInterface
{
private $broadcaster;
private $formater;
public function __construct(ZMQSocket $broadcaster)
{
$this->formater = new StateFormater();
$this->broadcaster = $broadcaster;
$this->broadcaster->bind();
usleep(300000);
}
public function onManagerStart(TaskManagerEvent $event)
{
$this->broadcaster->send(json_encode([
'topic' => TopicsManager::TOPIC_TASK_MANAGER,
'event' => TaskManagerEvents::MANAGER_START,
]));
}
public function onManagerStop(TaskManagerEvent $event)
{
$this->broadcaster->send(json_encode([
'topic' => TopicsManager::TOPIC_TASK_MANAGER,
'event' => TaskManagerEvents::MANAGER_STOP,
]));
}
public function onManagerRequest(TaskManagerRequestEvent $event)
{
$this->broadcaster->send(json_encode([
'topic' => TopicsManager::TOPIC_TASK_MANAGER,
'event' => TaskManagerEvents::MANAGER_REQUEST,
'request' => $event->getRequest(),
'response' => $event->getResponse(),
]));
}
public function onManagerTick(TaskManagerEvent $event)
{
$this->broadcaster->send(json_encode([
'topic' => TopicsManager::TOPIC_TASK_MANAGER,
'event' => TaskManagerEvents::MANAGER_TICK,
'message' => $this->formater->toArray(
$event->getManager()->getProcessManager()->getManagedProcesses()
),
]));
}
public static function getSubscribedEvents()
{
return [
TaskManagerEvents::MANAGER_START => 'onManagerStart',
TaskManagerEvents::MANAGER_STOP => 'onManagerStop',
TaskManagerEvents::MANAGER_REQUEST => 'onManagerRequest',
TaskManagerEvents::MANAGER_TICK => 'onManagerTick',
];
}
public static function create(array $options)
{
return new static(ZMQSocket::create(new \ZMQContext(), \ZMQ::SOCKET_PUB, $options['protocol'], $options['host'], $options['port']));
}
}

75
lib/Alchemy/Phrasea/Websocket/Topics/Directive.php
View File

@@ -1,75 +0,0 @@
<?php
/*
* This file is part of Phraseanet
*
* (c) 2005-2014 Alchemy
*
* For the full copyright and license information, please view the LICENSE
* file that was distributed with this source code.
*/
namespace Alchemy\Phrasea\Websocket\Topics;
use Alchemy\Phrasea\Websocket\Consumer\ConsumerInterface;
/**
* Stores consumer required settings for a topic
*/
class Directive
{
private $topic;
private $requireAuthentication;
private $requiredRights;
public function __construct($topic, $requireAuthentication, array $requiredRights)
{
$this->topic = $topic;
$this->requireAuthentication = (Boolean) $requireAuthentication;
$this->requiredRights = $requiredRights;
}
/**
* @return string
*/
public function getTopic()
{
return $this->topic;
}
/**
* Returns true if the topic requires an authenticated consumer
*
* @return Boolean
*/
public function requireAuthentication()
{
return $this->requireAuthentication;
}
/**
* Returns an array of required rights for the authenticated consumer
*
* @return array
*/
public function getRequiredRights()
{
return $this->requiredRights;
}
/**
* Returns true if the consumer satisfies the directive
*
* @param ConsumerInterface $consumer
*
* @return Boolean
*/
public function isStatisfiedBy(ConsumerInterface $consumer)
{
if ($this->requireAuthentication() && !$consumer->isAuthenticated()) {
return false;
}
return $consumer->hasRights($this->getRequiredRights());
}
}

61
lib/Alchemy/Phrasea/Websocket/Topics/DirectivesManager.php
View File

@@ -1,61 +0,0 @@
<?php
/*
* This file is part of Phraseanet
*
* (c) 2005-2014 Alchemy
*
* For the full copyright and license information, please view the LICENSE
* file that was distributed with this source code.
*/
namespace Alchemy\Phrasea\Websocket\Topics;
use Alchemy\Phrasea\Websocket\Consumer\ConsumerInterface;
use Ratchet\Wamp\Topic;
class DirectivesManager
{
private $directives;
public function __construct(array $directives)
{
array_walk($directives, function ($directive) {
if (!$directive instanceof Directive) {
throw new \InvalidArgumentException('Websocket configuration only accepts configuration directives.');
}
});
$this->directives = $directives;
}
/**
* Returns true if the consumer has access to the given topic
*
* @param ConsumerInterface $consumer
* @param Topic $topic
*
* @return Boolean
*/
public function hasAccess(ConsumerInterface $consumer, Topic $topic)
{
foreach ($this->getDirectives($topic) as $directive) {
if (!$directive->isStatisfiedBy($consumer)) {
return false;
}
}
return true;
}
/**
* @param Topic $topic
*
* @return Directive[]
*/
private function getDirectives(Topic $topic)
{
return array_filter($this->directives, function (Directive $directive) use ($topic) {
return $directive->getTopic() === $topic->getId();
});
}
}

24
lib/Alchemy/Phrasea/Websocket/Topics/Plugin/PluginInterface.php
View File

@@ -1,24 +0,0 @@
<?php
/*
* This file is part of Phraseanet
*
* (c) 2005-2014 Alchemy
*
* For the full copyright and license information, please view the LICENSE
* file that was distributed with this source code.
*/
namespace Alchemy\Phrasea\Websocket\Topics\Plugin;
use Alchemy\Phrasea\Websocket\Topics\TopicsManager;
interface PluginInterface
{
/**
* Attaches a Plugin to the TopicsManager
*
* @param TopicsManager $manager
*/
public function attach(TopicsManager $manager);
}

63
lib/Alchemy/Phrasea/Websocket/Topics/Plugin/TaskManagerSubscriberPlugin.php
View File

@@ -1,63 +0,0 @@
<?php
/*
* This file is part of Phraseanet
*
* (c) 2005-2014 Alchemy
*
* For the full copyright and license information, please view the LICENSE
* file that was distributed with this source code.
*/
namespace Alchemy\Phrasea\Websocket\Topics\Plugin;
use Alchemy\Phrasea\Websocket\Topics\TopicsManager;
use Psr\Log\LoggerInterface;
use React\EventLoop\LoopInterface;
use React\ZMQ\Context;
class TaskManagerSubscriberPlugin implements PluginInterface
{
private $logger;
private $pull;
public function __construct($options, LoopInterface $loop, LoggerInterface $logger)
{
$this->logger = $logger;
$context = new Context($loop);
$this->pull = $context->getSocket(\ZMQ::SOCKET_SUB);
$this->pull->setSockOpt(\ZMQ::SOCKOPT_SUBSCRIBE, "");
$this->pull->connect(sprintf('%s://%s:%s', $options['protocol'], $options['host'], $options['port']));
$this->pull->on('error', function ($e) use ($logger) {
$logger->error('TaskManager Subscriber received an error.', ['exception' => $e]);
});
}
/**
* {@inheritdoc}
*/
public function attach(TopicsManager $manager)
{
$this->pull->on('message', function ($msg) use ($manager) {
$data = @json_decode($msg, true);
if (json_last_error() !== JSON_ERROR_NONE) {
$this->logger->error(sprintf('[WS] Received invalid message %s : invalid json', $msg));
return;
}
if (!isset($data['topic'])) {
$this->logger->error(sprintf('[WS] Received invalid message %s : no topic', $msg));
return;
}
$this->logger->debug(sprintf('[WS] Received message %s', $msg));
$manager->broadcast($data['topic'], json_encode($msg));
});
}
}

163
lib/Alchemy/Phrasea/Websocket/Topics/TopicsManager.php
View File

@@ -1,163 +0,0 @@
<?php
/*
* This file is part of Phraseanet
*
* (c) 2005-2014 Alchemy
*
* For the full copyright and license information, please view the LICENSE
* file that was distributed with this source code.
*/
namespace Alchemy\Phrasea\Websocket\Topics;
use Alchemy\Phrasea\Websocket\Consumer\Consumer;
use Alchemy\Phrasea\Websocket\Consumer\ConsumerManager;
use Alchemy\Phrasea\Websocket\Topics\Plugin\PluginInterface;
use Ratchet\ConnectionInterface as Conn;
use Ratchet\Wamp\Topic;
class TopicsManager
{
const TOPIC_TASK_MANAGER = 'http://phraseanet.com/topics/admin/task-manager';
private $topics = [];
private $directives;
private $consumerManager;
public function __construct(DirectivesManager $directives, ConsumerManager $consumerManagaer)
{
$this->directives = $directives;
$this->consumerManager = $consumerManagaer;
}
/**
* Attaches a plugin to the TopicsManager
*
* @param PluginInterface $plugin
*
* @return TopicsManager
*/
public function attach(PluginInterface $plugin)
{
$plugin->attach($this);
return $this;
}
/**
* Checks if the consumer related to the connection has access to the topic,
* removes the connection from topic if the consumer is not granted.
*
* @param Conn $conn
* @param Topic $topic
*
* @return Boolean Return true if the consumer is granted, false otherwise
*/
public function subscribe(Conn $conn, Topic $topic)
{
if (!$this->directives->hasAccess($conn->User, $topic)) {
$topic->remove($conn);
return false;
}
$this->topics[$topic->getId()] = $topic;
return true;
}
/**
* Triggered on unsubscription.
*
* Removes internal references to the topic if no more consumers are listening.
*
* @param Conn $conn
* @param Topic $topic
*
* @return TopicsManager
*/
public function unsubscribe(Conn $conn, Topic $topic)
{
$this->cleanupReferences($conn, $topic);
return $this;
}
/**
* Triggered on connection, populates the connection with a consumer.
*
* @param Conn $conn
*
* @return TopicsManager
*/
public function openConnection(Conn $conn)
{
try {
$conn->User = $this->consumerManager->create($conn->Session);
} catch (\RuntimeException $e) {
$conn->close();
}
return $this;
}
/**
* Triggered on deconnexion.
*
* Removes internal references to topics if no more consumers are listening.
*
* @param Conn $conn
*
* @return TopicsManager
*/
public function closeConnection(Conn $conn)
{
$this->cleanupReferences($conn);
return $this;
}
/**
* Brodcasts a message to a topic, if it exists
*
* @param $topicId string
* @param $message string
*
* @return TopicsManager
*/
public function broadcast($topicId, $message)
{
if (isset($this->topics[$topicId])) {
$this->topics[$topicId]->broadcast($message);
}
return $this;
}
/**
* Removes internal references to topics if they do not contains any reference to an active connection.
*
* @param Conn $conn
* @param null|Topic $topic Restrict to this topic, if provided
*/
private function cleanupReferences(Conn $conn, Topic $topic = null)
{
$storage = $this->topics;
$updated = [];
foreach ($storage as $id => $storedTopic) {
if (null !== $topic && $id !== $topic->getId()) {
continue;
}
if ($storedTopic->has($conn)) {
$storedTopic->remove($conn);
}
if (count($storedTopic) > 0) {
$updated[] = $storedTopic;
}
}
$this->topics = $updated;
}
}

13
lib/classes/patch/390alpha11a.php
View File

@@ -58,21 +58,12 @@ class patch_390alpha11a extends patchAbstract
public function apply(base $appbox, Application $app)
{
$app['conf']->set(['main', 'task-manager', 'status'], 'started');
$app['conf']->set(['main', 'websocket-server'], [
'host' => parse_url($app['conf']->get('servername'), PHP_URL_HOST),
'port' => 9090,
'ip' => '0.0.0.0',
]);
$app['conf']->set(['main', 'task-manager', 'listener'], [
$app['conf']->set(['main', 'task-manager', 'options'], [
'protocol' => 'tcp',
'host' => '127.0.0.1',
'port' => 6660,
'linger' => 500,
]);
$app['conf']->set(['main', 'websocket-server', 'subscriber'], [
'protocol' => 'tcp',
'host' => '127.0.0.1',
'port' => 13598,
]);
}
}

18
lib/conf.d/configuration.yml
View File

@@ -31,23 +31,15 @@ main:
task-manager:
status: started
enabled: true
logger:
max-files: 10
enabled: true
level: INFO
listener:
options:
protocol: tcp
host: 127.0.0.1
port: 6660
linger: 500
websocket-server:
host: local.phrasea
port: 9090
ip: 0.0.0.0
subscriber:
protocol: tcp
host: 127.0.0.1
port: 13598
logger:
max-files: 10
enabled: true
level: INFO
session:
type: 'file'
options: []

1
lib/conf.d/minifyGroupsConfig.php
View File

@@ -56,7 +56,6 @@ $groups = [
, '//assets/blueimp-load-image/load-image.js'
, '//assets/jquery-file-upload/jquery.iframe-transport.js'
, '//assets/jquery-file-upload/jquery.fileupload.js'
, '//assets/autobahn/autobahn.js'
],
'report' => [
'//assets/jquery.ui/i18n/jquery-ui-i18n.js'

2
templates/web/admin/tree.html.twig
View File

@@ -62,7 +62,7 @@
{% if app['acl'].get(app['authentication'].getUser()).has_right('taskmanager') %}
<li class="{% if feature == 'taskmanager' %}selected{% endif %}">
<a target="right" href="{{ path('admin_tasks_list') }}" class="ajax" data-ws-topic="http://phraseanet.com/topics/admin/task-manager">
<a target="right" href="{{ path('admin_tasks_list') }}" class="ajax">
<img src="/skins/admin/TaskManager.png" />
<span>{{ 'admin::utilisateurs: gestionnaire de taches' | trans }}</span>
</a>

29
tests/Alchemy/Tests/Phrasea/Command/WebsocketServerTest.php
View File

@@ -1,29 +0,0 @@
<?php
namespace Alchemy\Tests\Phrasea\Command;
use Alchemy\Phrasea\Command\WebsocketServer;
class WebsocketServerTest extends \PhraseanetTestCase
{
public function testRunWithoutProblems()
{
$input = $this->getMock('Symfony\Component\Console\Input\InputInterface');
$output = $this->getMock('Symfony\Component\Console\Output\OutputInterface');
$sessionType = self::$DI['cli']['conf']->get(['main', 'session', 'type'], 'file');
self::$DI['cli']['conf']->set(['main', 'session', 'type'], 'memcached');
self::$DI['cli']['ws.server'] = $this->getMockBuilder('Ratchet\App')
->disableOriginalConstructor()
->getMock();
self::$DI['cli']['ws.server']->expects($this->once())
->method('run');
$command = new WebsocketServer('websocketserver');
$command->setContainer(self::$DI['cli']);
$command->execute($input, $output);
self::$DI['cli']['conf']->set(['main', 'session', 'type'], $sessionType);
}
}

57
tests/Alchemy/Tests/Phrasea/Core/CLIProvider/WebsocketServerServiceProviderTest.php
View File

@@ -1,57 +0,0 @@
<?php
namespace Alchemy\Tests\Phrasea\Core\CLIProvider;
class WebsocketServerServiceProviderTest extends ServiceProviderTestCase
{
public function provideServiceDescription()
{
return [
[
'Alchemy\Phrasea\Core\CLIProvider\WebsocketServerServiceProvider',
'ws.task-manager.broadcaster',
'Alchemy\Phrasea\Websocket\Subscriber\TaskManagerBroadcasterSubscriber',
],
[
'Alchemy\Phrasea\Core\CLIProvider\WebsocketServerServiceProvider',
'ws.event-loop',
'React\EventLoop\LoopInterface',
],
[
'Alchemy\Phrasea\Core\CLIProvider\WebsocketServerServiceProvider',
'ws.server.subscriber',
'Alchemy\Phrasea\Websocket\Topics\Plugin\TaskManagerSubscriberPlugin',
],
[
'Alchemy\Phrasea\Core\CLIProvider\WebsocketServerServiceProvider',
'ws.server.application',
'Ratchet\WebSocket\WsServerInterface',
],
[
'Alchemy\Phrasea\Core\CLIProvider\WebsocketServerServiceProvider',
'ws.server',
'Ratchet\App',
],
[
'Alchemy\Phrasea\Core\CLIProvider\WebsocketServerServiceProvider',
'ws.server.phraseanet-server',
'Alchemy\Phrasea\Websocket\PhraseanetWampServer',
],
[
'Alchemy\Phrasea\Core\CLIProvider\WebsocketServerServiceProvider',
'ws.server.logger',
'Psr\Log\LoggerInterface',
],
[
'Alchemy\Phrasea\Core\CLIProvider\WebsocketServerServiceProvider',
'ws.server.topics-manager.directives',
'Alchemy\Phrasea\Websocket\Topics\DirectivesManager',
],
[
'Alchemy\Phrasea\Core\CLIProvider\WebsocketServerServiceProvider',
'ws.server.consumer-manager',
'Alchemy\Phrasea\Websocket\Consumer\ConsumerManager',
],
];
}
}

58
tests/Alchemy/Tests/Phrasea/Websocket/Consumer/ConsumerManagerTest.php
View File

@@ -1,58 +0,0 @@
<?php
namespace Alchemy\Tests\Phrasea\Websocket\Consumer;
use Alchemy\Phrasea\Websocket\Consumer\ConsumerManager;
class ConsumerManagerTest extends \PhraseanetTestCase
{
/**
* @dataProvider provideConsumerManagerData
*/
public function testCreate($usrId, $rights, $authenticated, $checkedRights, $hasRights)
{
$manager = new ConsumerManager();
$consumer = $manager->create($this->createSessionMock($usrId, $rights));
$this->assertSame($authenticated, $consumer->isAuthenticated());
$this->assertSame($hasRights, $consumer->hasRights($checkedRights));
}
public function provideConsumerManagerData()
{
return [
[25, ['task-manager'], true, [], true],
[25, ['task-manager'], true, ['task-manager'], true],
[null, ['task-manager'], false, ['task-manager', 'neutron'], false],
[null, ['neutron', 'task-manager'], false, ['task-manager', 'neutron'], true],
[42, ['neutron', 'task-manager', 'romain'], true, ['task-manager', 'neutron'], true],
];
}
private function createSessionMock($usrId, $rights)
{
$session = $this->getMock('Symfony\Component\HttpFoundation\Session\SessionInterface');
$session->expects($this->any())
->method('has')
->will($this->returnCallback(function ($prop) use ($usrId, $rights) {
switch ($prop) {
case 'usr_id':
return $usrId !== null;
case 'websockets_rights':
return $rights !== null;
}
}));
$session->expects($this->any())
->method('get')
->will($this->returnCallback(function ($prop) use ($usrId, $rights) {
switch ($prop) {
case 'usr_id':
return $usrId;
case 'websockets_rights':
return $rights;
}
}));
return $session;
}
}

32
tests/Alchemy/Tests/Phrasea/Websocket/Consumer/ConsumerTest.php
View File

@@ -1,32 +0,0 @@
<?php
namespace Alchemy\Tests\Phrasea\Websocket\Consumer;
use Alchemy\Phrasea\Websocket\Consumer\Consumer;
class ConsumerTest extends \PhraseanetTestCase
{
public function testIsAuthenticated()
{
$consumer = new Consumer(42, []);
$this->assertTrue($consumer->isAuthenticated());
$consumer = new Consumer(null, []);
$this->assertFalse($consumer->isAuthenticated());
}
public function testHasRights()
{
$consumer = new Consumer(42, ['neutron']);
$this->assertTrue($consumer->hasRights('neutron'));
$consumer = new Consumer(42, ['neutron']);
$this->assertTrue($consumer->hasRights(['neutron']));
$consumer = new Consumer(42, ['romainneutron']);
$this->assertFalse($consumer->hasRights('neutron'));
$consumer = new Consumer(42, ['romainneutron']);
$this->assertFalse($consumer->hasRights(['neutron']));
$consumer = new Consumer(42, ['neutron']);
$this->assertFalse($consumer->hasRights(['neutron', 'romain']));
$consumer = new Consumer(42, ['romain', 'neutron', 'bouteille']);
$this->assertTrue($consumer->hasRights(['neutron', 'romain']));
}
}

27
tests/Alchemy/Tests/Phrasea/Websocket/PhraseanetWampServerTest.php
View File

@@ -1,27 +0,0 @@
<?php
namespace Alchemy\Tests\Phrasea\Websocket;
use Alchemy\Phrasea\Websocket\PhraseanetWampServer;
class PhraseanetWampServerTest extends \PhraseanetTestCase
{
public function testOpenConnectionConnected()
{
$topicsManager = $this->createTopicsManagerMock();
$conn = $this->getMock('Ratchet\ConnectionInterface');
$topicsManager->expects($this->once())
->method('openConnection')
->with($conn);
$server = new PhraseanetWampServer($topicsManager, $this->createLoggerMock());
$server->onOpen($conn);
}
private function createTopicsManagerMock()
{
return $this->getMockBuilder('Alchemy\Phrasea\Websocket\Topics\TopicsManager')
->disableOriginalConstructor()
->getMock();
}
}

127
tests/Alchemy/Tests/Phrasea/Websocket/Subscriber/TaskManagerBroadcasterSubscriberTest.php
View File

@@ -1,127 +0,0 @@
<?php
namespace Alchemy\Tests\Phrasea\Websocket\Subscriber;
use Alchemy\Phrasea\Websocket\Subscriber\TaskManagerBroadcasterSubscriber;
use Alchemy\Phrasea\Websocket\Topics\TopicsManager;
use Alchemy\TaskManager\Event\TaskManagerEvent;
use Alchemy\TaskManager\Event\TaskManagerEvents;
use Alchemy\TaskManager\Event\TaskManagerRequestEvent;
class TaskManagerBroadcasterSubscriberTest extends \PhraseanetTestCase
{
public function testOnManagerStart()
{
$socket = $this->createZMQSocketMock();
$socket->expects($this->once())
->method('send')
->will($this->jsonCapture($json));
$subscriber = new TaskManagerBroadcasterSubscriber($socket);
$subscriber->onManagerStart($this->createTaskManagerEvent());
$this->assertValidJson($json, TopicsManager::TOPIC_TASK_MANAGER, TaskManagerEvents::MANAGER_START);
}
public function testOnManagerStop()
{
$socket = $this->createZMQSocketMock();
$socket->expects($this->once())
->method('send')
->will($this->jsonCapture($json));
$subscriber = new TaskManagerBroadcasterSubscriber($socket);
$subscriber->onManagerStop($this->createTaskManagerEvent());
$this->assertValidJson($json, TopicsManager::TOPIC_TASK_MANAGER, TaskManagerEvents::MANAGER_STOP);
}
public function testOnManagerRequest()
{
$socket = $this->createZMQSocketMock();
$socket->expects($this->once())
->method('send')
->will($this->jsonCapture($json));
$subscriber = new TaskManagerBroadcasterSubscriber($socket);
$subscriber->onManagerRequest(new TaskManagerRequestEvent($this->createTaskManagerMock(), 'PING', 'PONG'));
$data = $this->assertValidJson($json, TopicsManager::TOPIC_TASK_MANAGER, TaskManagerEvents::MANAGER_REQUEST);
$this->assertEquals('PING', $data['request']);
$this->assertEquals('PONG', $data['response']);
}
public function testOnManagerTick()
{
$socket = $this->createZMQSocketMock();
$socket->expects($this->once())
->method('send')
->will($this->jsonCapture($json));
$subscriber = new TaskManagerBroadcasterSubscriber($socket);
$subscriber->onManagerTick($this->createTaskManagerEvent());
$data = $this->assertValidJson($json, TopicsManager::TOPIC_TASK_MANAGER, TaskManagerEvents::MANAGER_TICK);
$this->assertArrayHasKey('message', $data);
$this->assertInternalType('array', $data['message']);
}
private function assertValidJson($json, $topic, $event)
{
$data = json_decode($json, true);
$this->assertTrue(json_last_error() === JSON_ERROR_NONE);
$this->assertArrayHasKey('event', $data);
$this->assertArrayHasKey('topic', $data);
$this->assertEquals($event, $data['event']);
$this->assertEquals($topic, $data['topic']);
return $data;
}
private function jsonCapture(&$json)
{
return $this->returnCallback(function ($arg) use (&$json) { $json = $arg; return 'lala'; });
}
private function createZMQSocketMock()
{
$socket = $this->getMockBuilder('Alchemy\TaskManager\ZMQSocket')
->setMethods(['send', 'bind'])
->disableOriginalConstructor()
->getMock();
$socket->expects($this->once())
->method('bind');
return $socket;
}
private function createTaskManagerMock()
{
$manager = $this->getMockBuilder('Alchemy\TaskManager\TaskManager')
->disableOriginalConstructor()
->getMock();
$processManager = $this->getMockBuilder('Neutron\ProcessManager\ProcessManager')
->disableOriginalConstructor()
->getMock();
$processManager->expects($this->any())
->method('getManagedProcesses')
->will($this->returnValue([]));
$manager->expects($this->any())
->method('getProcessManager')
->will($this->returnValue($processManager));
return $manager;
}
private function createTaskManagerEvent()
{
return new TaskManagerEvent($this->createTaskManagerMock());
}
}

50
tests/Alchemy/Tests/Phrasea/Websocket/Topics/DirectiveTest.php
View File

@@ -1,50 +0,0 @@
<?php
namespace Alchemy\Tests\Phrasea\Websocket\Topics;
use Alchemy\Phrasea\Websocket\Topics\Directive;
class DirectiveTest extends \PhraseanetTestCase
{
public function testGetters()
{
$directive = new Directive('http://topic', true, ['neutron']);
$this->assertSame('http://topic', $directive->getTopic());
$this->assertTrue($directive->requireAuthentication());
$this->assertSame(['neutron'], $directive->getRequiredRights());
}
/**
* @dataProvider provideStatisfiedByCombinaisons
*/
public function testIsSatisfiedBy($authenticationRequired, $requiredRights, $authenticated, $hasRights, $satisfied)
{
$consumer = $this->createConsumerMock($authenticated, $hasRights, $requiredRights);
$directive = new Directive('http://topic', $authenticationRequired, $requiredRights);
$this->assertEquals($satisfied, $directive->isStatisfiedBy($consumer));
}
public function provideStatisfiedByCombinaisons()
{
return [
[true, ['neutron'], true, true, true],
[true, [], false, true, false],
[false, ['neutron'], true, false, false],
[false, ['neutron'], false, false, false],
];
}
private function createConsumerMock($authenticated, $hasRights, $requiredRights)
{
$consumer = $this->getMock('Alchemy\Phrasea\Websocket\Consumer\ConsumerInterface');
$consumer->expects($this->any())
->method('isAuthenticated')
->will($this->returnValue($authenticated));
$consumer->expects($this->any())
->method('hasRights')
->with($requiredRights)
->will($this->returnValue($hasRights));
return $consumer;
}
}

46
tests/Alchemy/Tests/Phrasea/Websocket/Topics/DirectivesManagerTest.php
View File

@@ -1,46 +0,0 @@
<?php
namespace Alchemy\Tests\Phrasea\Websocket\Topics;
use Alchemy\Phrasea\Websocket\Consumer\Consumer;
use Alchemy\Phrasea\Websocket\Topics\Directive;
use Alchemy\Phrasea\Websocket\Topics\DirectivesManager;
use Ratchet\Wamp\Topic;
class DirectivesManagerTest extends \PhraseanetTestCase
{
public function testHasAccess()
{
$manager = new DirectivesManager([
new Directive('http://topic', false, []),
new Directive('http://topic2', true, []),
new Directive('http://topic3', true, ['neutron']),
new Directive('http://topic4', true, ['bouteille']),
new Directive('http://topic4', true, ['neutron', 'romain']),
]);
$consumer = new Consumer(42, []);
$this->assertTrue($manager->hasAccess($consumer, new Topic('http://topic')));
$this->assertTrue($manager->hasAccess($consumer, new Topic('http://topic2')));
$this->assertFalse($manager->hasAccess($consumer, new Topic('http://topic3')));
$this->assertFalse($manager->hasAccess($consumer, new Topic('http://topic4')));
$consumer = new Consumer(null, []);
$this->assertTrue($manager->hasAccess($consumer, new Topic('http://topic')));
$this->assertFalse($manager->hasAccess($consumer, new Topic('http://topic2')));
$this->assertFalse($manager->hasAccess($consumer, new Topic('http://topic3')));
$this->assertFalse($manager->hasAccess($consumer, new Topic('http://topic4')));
$consumer = new Consumer(42, ['neutron']);
$this->assertTrue($manager->hasAccess($consumer, new Topic('http://topic')));
$this->assertTrue($manager->hasAccess($consumer, new Topic('http://topic2')));
$this->assertTrue($manager->hasAccess($consumer, new Topic('http://topic3')));
$this->assertFalse($manager->hasAccess($consumer, new Topic('http://topic4')));
$consumer = new Consumer(42, ['neutron', 'bouteille', 'romain']);
$this->assertTrue($manager->hasAccess($consumer, new Topic('http://topic')));
$this->assertTrue($manager->hasAccess($consumer, new Topic('http://topic2')));
$this->assertTrue($manager->hasAccess($consumer, new Topic('http://topic3')));
$this->assertTrue($manager->hasAccess($consumer, new Topic('http://topic4')));
}
}

187
tests/Alchemy/Tests/Phrasea/Websocket/Topics/TopicsManagerTest.php
View File

@@ -1,187 +0,0 @@
<?php
namespace Alchemy\Tests\Phrasea\Websocket\Topics;
use Alchemy\Phrasea\Websocket\Topics\TopicsManager;
use Ratchet\Wamp\Topic;
class TopicsManagerTest extends \PhraseanetTestCase
{
public function testAttach()
{
$directivesManager = $this->createDirectivesManagerMock();
$consumerManager = $this->createConsumerManagerMock();
$manager = new TopicsManager($directivesManager, $consumerManager);
$plugin = $this->getMock('Alchemy\Phrasea\Websocket\Topics\Plugin\PluginInterface');
$plugin->expects($this->once())
->method('attach')
->with($manager);
$this->assertSame($manager, $manager->attach($plugin));
}
public function testSubscribeWithAccess()
{
$directivesManager = $this->createDirectivesManagerMock();
$consumerManager = $this->createConsumerManagerMock();
$manager = new TopicsManager($directivesManager, $consumerManager);
$conn = $this->getMock('Ratchet\ConnectionInterface');
$conn->User = $this->getMock('Alchemy\Phrasea\Websocket\Consumer\ConsumerInterface');
$topic = $this->getMockBuilder('Ratchet\Wamp\Topic')
->disableOriginalConstructor()
->getMock();
$topic->expects($this->never())
->method('remove');
$directivesManager->expects($this->once())
->method('hasAccess')
->will($this->returnValue(true));
$manager->subscribe($conn, $topic);
}
public function testSubscribeWithoutAccess()
{
$directivesManager = $this->createDirectivesManagerMock();
$consumerManager = $this->createConsumerManagerMock();
$manager = new TopicsManager($directivesManager, $consumerManager);
$conn = $this->getMock('Ratchet\ConnectionInterface');
$conn->User = $this->getMock('Alchemy\Phrasea\Websocket\Consumer\ConsumerInterface');
$topic = $this->getMockBuilder('Ratchet\Wamp\Topic')
->disableOriginalConstructor()
->getMock();
$topic->expects($this->once())
->method('remove')
->with($conn);
$directivesManager->expects($this->once())
->method('hasAccess')
->will($this->returnValue(false));
$manager->subscribe($conn, $topic);
}
public function testUnsubscribe()
{
$directivesManager = $this->createDirectivesManagerMock();
$consumerManager = $this->createConsumerManagerMock();
$directivesManager->expects($this->once())
->method('hasAccess')
->will($this->returnValue(true));
$manager = new TopicsManager($directivesManager, $consumerManager);
$conn = $this->getMock('Ratchet\ConnectionInterface');
$conn->User = $this->getMock('Alchemy\Phrasea\Websocket\Consumer\ConsumerInterface');
$topic = new Topic('http://topic');
$topic->add($conn);
// should be subscribed to be unsubscribed
$manager->subscribe($conn, $topic);
$manager->unsubscribe($conn, $topic);
$this->assertFalse($topic->has($conn));
}
public function testOpenConnection()
{
$consumer = $this->getMock('Alchemy\Phrasea\Websocket\Consumer\ConsumerInterface');
$session = $this->getMock('Symfony\Component\HttpFoundation\Session\SessionInterface');
$directivesManager = $this->createDirectivesManagerMock();
$consumerManager = $this->createConsumerManagerMock();
$consumerManager->expects($this->once())
->method('create')
->with($session)
->will($this->returnValue($consumer));
$manager = new TopicsManager($directivesManager, $consumerManager);
$conn = $this->getMock('Ratchet\ConnectionInterface');
$conn->Session = $session;
$manager->openConnection($conn);
$this->assertSame($consumer, $conn->User);
}
public function testCloseConnection()
{
$directivesManager = $this->createDirectivesManagerMock();
$consumerManager = $this->createConsumerManagerMock();
$directivesManager->expects($this->once())
->method('hasAccess')
->will($this->returnValue(true));
$manager = new TopicsManager($directivesManager, $consumerManager);
$conn = $this->getMock('Ratchet\ConnectionInterface');
$conn->User = $this->getMock('Alchemy\Phrasea\Websocket\Consumer\ConsumerInterface');
$topic = new Topic('http://topic');
$topic->add($conn);
// should be subscribed to be unsubscribed
$manager->subscribe($conn, $topic);
$manager->closeConnection($conn);
$this->assertFalse($topic->has($conn));
}
public function testBroadcast()
{
$directivesManager = $this->createDirectivesManagerMock();
$consumerManager = $this->createConsumerManagerMock();
$directivesManager->expects($this->once())
->method('hasAccess')
->will($this->returnValue(true));
$manager = new TopicsManager($directivesManager, $consumerManager);
$conn = $this->getMock('Ratchet\ConnectionInterface');
$conn->User = $this->getMock('Alchemy\Phrasea\Websocket\Consumer\ConsumerInterface');
$topic = $this->getMockBuilder('Ratchet\Wamp\Topic')
->disableOriginalConstructor()
->getMock();
$topic->expects($this->any())
->method('getId')
->will($this->returnValue('http://topic'));
$topic->expects($this->once())
->method('broadcast')
->with('hello world !');
// should be subscribed to be unsubscribed
$manager->subscribe($conn, $topic);
$manager->broadcast('http://topic', 'hello world !');
$manager->broadcast('http://topic2', 'nothing');
}
private function createDirectivesManagerMock()
{
return $this->getMockBuilder('Alchemy\Phrasea\Websocket\Topics\DirectivesManager')
->disableOriginalConstructor()
->getMock();
}
private function createConsumerManagerMock()
{
return $this->getMockBuilder('Alchemy\Phrasea\Websocket\Consumer\ConsumerManager')
->disableOriginalConstructor()
->getMock();
}
}

7
www/scripts/apps/admin/main/app.js
View File

@@ -11,10 +11,9 @@ define([
"jquery",
"underscore",
"backbone",
"common/websockets/connection",
"apps/admin/main/views/leftPanel",
"apps/admin/main/views/rightPanel"
], function ($, _, Backbone, WSConnection, LeftPanel, RightPanel) {
], function ($, _, Backbone, LeftPanel, RightPanel) {
window.AdminApp = {
$scope: $("#admin-app"),
$leftView : $(".left-view", this.$scope),
@@ -68,10 +67,6 @@ define([
throw "You must define a websocket url";
}
if (false === WSConnection.isConnected()) {
WSConnection.connect(options.wsurl);
}
create();
AdminApp.LeftView.activeTree();

3
www/scripts/apps/admin/main/views/leftPanel.js
View File

@@ -11,9 +11,8 @@ define([
"jquery",
"underscore",
"backbone",
"common/websockets/subscriberManager",
"jquery.treeview"
], function ($, _, Backbone, SubscriberManager) {
], function ($, _, Backbone) {
var LeftPanelView = Backbone.View.extend({
initialize: function (options) {
options = options || {};

21
www/scripts/apps/admin/tasks-manager/app.js
View File

@@ -12,14 +12,12 @@ define([
"underscore",
"backbone",
"models/scheduler",
"common/websockets/connection",
"common/websockets/subscriberManager",
"apps/admin/tasks-manager/views/scheduler",
"apps/admin/tasks-manager/views/tasks",
"apps/admin/tasks-manager/views/ping",
"apps/admin/tasks-manager/views/refresh",
"apps/admin/tasks-manager/collections/tasks"
], function ($, _, Backbone, Scheduler, WSConnection, SubscriberManager, SchedulerView, TasksView, PingView, RefreshView, TasksCollection) {
], function ($, _, Backbone, Scheduler, SchedulerView, TasksView, PingView, RefreshView, TasksCollection) {
var create = function() {
window.TaskManagerApp = {
$scope: $("#task-manager-app"),
@@ -55,23 +53,6 @@ define([
TaskManagerApp.tasksView.render();
TaskManagerApp.schedulerView.render();
SubscriberManager.pushCallback(function(topic, msg) {
// double encoded string
var msg = JSON.parse(JSON.parse(msg));
WSConnection.trigger("ws:"+msg.event, msg);
});
// On ticks re-render ping view, update tasks & scheduler model
WSConnection.on("ws:manager-tick", function(response) {
TaskManagerApp.pingView.render();
TaskManagerApp.Scheduler.set({"actual": "started", "process-id": response.message.manager["process-id"]});
_.each(response.message.jobs, function(data, id) {
var jobModel = TaskManagerApp.tasksCollection.get(id);
if ("undefined" !== typeof jobModel) {
jobModel.set({"actual": data["status"], "process-id": data["process-id"]});
}
});
});
}
);
};

53
www/scripts/common/websockets/connection.js
View File

@@ -1,53 +0,0 @@
define([
"underscore",
"backbone"
], function (_, Backbone) {
var activeSession = null;
return _.extend({
connect: function(url) {
if (this.isConnected()) {
throw "Connection is already active";
}
var that = this;
// autobahn js is defined as a global object there is no way to load
// it as a UMD module
ab.connect(url, function (session) {
activeSession = session;
that.trigger("ws:connect", activeSession);
},
function (code, reason) {
that.trigger("ws:session-gone", code, reason);
});
},
close: function() {
if (false === this.isConnected()) {
throw "Not connected to websocket";
}
activeSession.close();
activeSession = null;
this.trigger("ws:session-close");
},
isConnected: function() {
return activeSession !== null;
},
subscribe: function(topic, callback) {
if (false === this.isConnected()) {
this.on("ws:connect", function(session) {
session.subscribe(topic, callback);
this.trigger("ws:session-subscribe", topic);
});
return;
}
activeSession.subscribe(topic, callback);
this.trigger("ws:session-subscribe", topic);
},
unsubscribe: function(topic, callback) {
if (false === this.isConnected()) {
return;
}
activeSession.unsubscribe(topic, callback);
this.trigger("ws:session-unsubscribe", topic);
}
}, Backbone.Events);
});

39
www/scripts/common/websockets/subscriberManager.js
View File

@@ -1,39 +0,0 @@
define([
"underscore",
"backbone",
"common/websockets/connection"
], function (_, Backbone, WSConnection) {
var currentTopic = null;
var callbackStack = [];
var callbackHandler = function (topic, msg) {
_.each(callbackStack, function(cb) {
cb(topic, msg);
});
};
var reset = function () {
callbackStack = [];
currentTopic = null;
};
return {
'register': function (topic) {
this.unregister();
WSConnection.subscribe(topic, callbackHandler);
currentTopic = topic;
},
'unregister': function () {
if (currentTopic !== null) {
WSConnection.unsubscribe(currentTopic);
reset();
}
},
'pushCallback': function (callback) {
callbackStack.push(callback);
},
'hasCallbacks' : function() {
return callbackStack.length > 0;
}
}
});

86
www/scripts/tests/specs/websockets/connection.js
View File

@@ -1,86 +0,0 @@
define([
'chai',
'sinonchai',
'underscore',
'common/websockets/connection'
], function (chai, sinonchai, _, connection) {
var expect = chai.expect;
var assert = chai.assert;
var should = chai.should();
chai.use(sinonchai);
describe("Connection", function () {
describe("Functionnal", function () {
beforeEach(function () {
this.session = {"hello":"session"};
this.session.close = sinon.spy();
this.session.subscribe = sinon.spy();
this.session.unsubscribe = sinon.spy();
this.wsConnection = connection;
var $this = this;
var cbSuccess = function (session) {
activeSession = $this.session;
};
window.ab = {
connect: function(url, cbSuccess, cbError) {
cbSuccess($this.session);
}
}
});
afterEach(function () {
if (this.wsConnection.isConnected()) {
this.wsConnection.close();
}
});
it("should have a session", function () {
this.wsConnection.connect();
assert.ok(this.wsConnection.isConnected());
});
it("should close the session", function () {
this.wsConnection.connect();
assert.ok(this.wsConnection.isConnected());
this.wsConnection.close();
assert.ok(!this.wsConnection.isConnected());
});
it("should warn if you close the session and you are not connected", function () {
var throws = false;
try {
this.wsConnection.close();
} catch (e) {
throws = true;
}
assert.ok(throws);
});
it("should not connect anymore after first connect", function () {
var throws = false;
this.wsConnection.connect();
try {
this.wsConnection.connect();
} catch (e) {
throws = true;
}
assert.ok(throws);
});
it("should call session subscribe once", function () {
this.wsConnection.connect();
this.wsConnection.subscribe();
expect(this.session.subscribe.should.have.callCount(1)).to.be.ok;
});
it("should call session unsubscribe once", function () {
this.wsConnection.connect();
this.wsConnection.unsubscribe();
expect(this.session.unsubscribe.should.have.callCount(1)).to.be.ok;
});
});
});
});

73
www/scripts/tests/specs/websockets/subscriberManager.js
View File

@@ -1,73 +0,0 @@
define([
'chai',
'sinonchai',
'underscore',
'squire'
], function(chai, sinonchai, _, Squire) {
var expect = chai.expect;
var assert = chai.assert;
var should = chai.should();
chai.use(sinonchai);
(function () {
describe("SubscriberManager", function () {
beforeEach(function () {
var $this = this;
$this.connection = {};
$this.connection.subscribe = sinon.spy();
$this.connection.unsubscribe = sinon.spy();
});
it("should call subscribe", function () {
var $this = this;
var injector = new Squire();
injector.mock(
["common/websockets/connection"], $this.connection
).require(['common/websockets/subscriberManager'], function(manager) {
manager.register('topic');
expect($this.connection.subscribe.should.have.callCount(1)).to.be.ok;
assert.ok(manager.hasCallbacks());
});
try{
injector.remove();
} catch(e) {
}
});
it("should call unsubscribe", function () {
var $this = this;
var injector = new Squire();
injector.mock(
["common/websockets/connection"], $this.connection
).require(['common/websockets/subscriberManager'], function(manager) {
manager.register('topic');
manager.unregister();
expect($this.connection.unsubscribe.should.have.callCount(1)).to.be.ok;
assert.ok(!manager.hasCallbacks());
});
try{
injector.remove();
} catch(e) {
}
});
it("should add callbacks", function () {
var $this = this;
var injector = new Squire();
injector.mock(
["common/websockets/connection"], $this.connection
).require(['common/websockets/subscriberManager'], function(manager) {
assert.ok(!manager.hasCallbacks());
manager.pushCallback(function(){return null;});
assert.ok(manager.hasCallbacks());
});
try{
injector.remove();
} catch(e) {
}
});
});
})();
});

33
www/scripts/tests/websocket.html
View File

@@ -1,33 +0,0 @@
<!doctype html>
<head>
<meta charset="utf-8">
<meta http-equiv="X-UA-Compatible" content="IE=edge,chrome=1">
<title>Mocha Spec Runner</title>
<link rel="stylesheet" href="../../assets/mocha/mocha.css">
</head>
<body>
<div id="mocha"></div>
<script src="../../assets/mocha/mocha.js"></script>
<script src="../../assets/sinonjs/sinon.js"></script>
<script src="../../assets/requirejs/require.js"></script>
<script src="common.js"></script>
<script>
mocha.setup({
ui: "bdd",
ignoreLeaks: false,
globals: ['ab']
});
require([
'tests/specs/websockets/connection',
'tests/specs/websockets/subscriberManager'
], function () {
if (window.mochaPhantomJS) {
mochaPhantomJS.run();
} else {
mocha.run();
}
});
</script>
</body>
</html>