From de09d1dbd2debe89afab58e3b5743f203ce801bd Mon Sep 17 00:00:00 2001 From: Yuri Kuznetsov Date: Fri, 12 Feb 2021 14:30:46 +0200 Subject: [PATCH] cs fix --- application/Espo/Core/WebSocket/Pusher.php | 168 ++++++++++++++---- .../Espo/Core/WebSocket/Submission.php | 21 ++- 2 files changed, 148 insertions(+), 41 deletions(-) diff --git a/application/Espo/Core/WebSocket/Pusher.php b/application/Espo/Core/WebSocket/Pusher.php index 5015f335fe..d6733c1f38 100644 --- a/application/Espo/Core/WebSocket/Pusher.php +++ b/application/Espo/Core/WebSocket/Pusher.php @@ -32,6 +32,10 @@ namespace Espo\Core\WebSocket; use Ratchet\ConnectionInterface; use Ratchet\Wamp\WampServerInterface; +use Symfony\Component\Process\PhpExecutableFinder; + +use Exception; + class Pusher implements WampServerInterface { private $categoryList; @@ -52,42 +56,62 @@ class Pusher implements WampServerInterface private $phpExecutablePath; - public function __construct(array $categoriesData = [], ?string $phpExecutablePath = null, bool $isDebugMode = false) - { + public function __construct( + array $categoriesData = [], ?string $phpExecutablePath = null, bool $isDebugMode = false + ) { $this->categoryList = array_keys($categoriesData); $this->categoriesData = $categoriesData; - $this->phpExecutablePath = $phpExecutablePath ?: (new \Symfony\Component\Process\PhpExecutableFinder)->find(); + $this->phpExecutablePath = $phpExecutablePath ?: (new PhpExecutableFinder)->find(); $this->isDebugMode = $isDebugMode; } public function onSubscribe(ConnectionInterface $connection, $topic) { $topicId = $topic->getId(); - if (!$topicId) return; - if (!$this->isTopicAllowed($topicId)) return; + if (!$topicId) { + return; + } + + if (!$this->isTopicAllowed($topicId)) { + return; + } $connectionId = $connection->resourceId; $userId = $this->getUserIdByConnection($connection); - if (!$userId) return; - if (!isset($this->connectionIdTopicIdListMap[$connectionId])) $this->connectionIdTopicIdListMap[$connectionId] = []; + if (!$userId) { + return; + } + + if (!isset($this->connectionIdTopicIdListMap[$connectionId])) { + $this->connectionIdTopicIdListMap[$connectionId] = []; + } $checkCommand = $this->getAccessCheckCommandForTopic($connection, $topic); if ($checkCommand) { $checkResult = shell_exec($checkCommand); + if ($checkResult !== 'true') { - if ($this->isDebugMode) $this->log("{$connectionId}: check access failed for topic {$topicId} for user {$userId}"); + if ($this->isDebugMode) { + $this->log("{$connectionId}: check access failed for topic {$topicId} for user {$userId}"); + } + return; } - if ($this->isDebugMode) $this->log("{$connectionId}: check access succeed for topic {$topicId} for user {$userId}"); + if ($this->isDebugMode) { + $this->log("{$connectionId}: check access succeed for topic {$topicId} for user {$userId}"); + } } if (!in_array($topicId, $this->connectionIdTopicIdListMap[$connectionId])) { - if ($this->isDebugMode) $this->log("{$connectionId}: add topic {$topicId} for user {$userId}"); + if ($this->isDebugMode) { + $this->log("{$connectionId}: add topic {$topicId} for user {$userId}"); + } + $this->connectionIdTopicIdListMap[$connectionId][] = $topicId; } @@ -97,21 +121,36 @@ class Pusher implements WampServerInterface public function onUnSubscribe(ConnectionInterface $connection, $topic) { $topicId = $topic->getId(); - if (!$topicId) return; - if (!$this->isTopicAllowed($topicId)) return; + if (!$topicId) { + return; + } + + if (!$this->isTopicAllowed($topicId)) { + return; + } $connectionId = $connection->resourceId; $userId = $this->getUserIdByConnection($connection); - if (!$userId) return; + + if (!$userId) { + return; + } if (isset($this->connectionIdTopicIdListMap[$connectionId])) { $index = array_search($topicId, $this->connectionIdTopicIdListMap[$connectionId]); + if ($index !== false) { - if ($this->isDebugMode) $this->log("{$connectionId}: remove topic {$topicId} for user {$userId}"); + if ($this->isDebugMode) { + $this->log("{$connectionId}: remove topic {$topicId} for user {$userId}"); + } + unset($this->connectionIdTopicIdListMap[$connectionId][$index]); - $this->connectionIdTopicIdListMap[$connectionId] = array_values($this->connectionIdTopicIdListMap[$connectionId]); + + $this->connectionIdTopicIdListMap[$connectionId] = array_values( + $this->connectionIdTopicIdListMap[$connectionId] + ); } } } @@ -119,13 +158,16 @@ class Pusher implements WampServerInterface protected function getCategoryData(string $topicId) : array { $arr = explode('.', $topicId); + $category = $arr[0]; if (array_key_exists($category, $this->categoriesData)) { $data = $this->categoriesData[$category]; - } else if (array_key_exists($topicId, $this->categoriesData)) { + } + else if (array_key_exists($topicId, $this->categoriesData)) { $data = $this->categoriesData[$topicId]; - } else { + } + else { $data = []; } @@ -144,7 +186,8 @@ class Pusher implements WampServerInterface foreach ($data['paramList'] as $i => $item) { if (isset($arr[$i + 1])) { $params[$item] = $arr[$i + 1]; - } else { + } + else { $params[$item] = ''; } } @@ -159,14 +202,18 @@ class Pusher implements WampServerInterface $params = $this->getParamsFromTopicId($topicId); $params['userId'] = $this->getUserIdByConnection($connection); + if (!$params['userId']) { $connection->close(); + return null; } $data = $this->getCategoryData($topic->getId()); - if (!array_key_exists('accessCheckCommand', $data)) return null; + if (!array_key_exists('accessCheckCommand', $data)) { + return null; + } $command = $this->phpExecutablePath . " command.php " . $data['accessCheckCommand']; @@ -180,6 +227,7 @@ class Pusher implements WampServerInterface protected function getTopicCategory($topic) { list($category) = explode('.', $topic->getId()); + return $category; } @@ -192,13 +240,19 @@ class Pusher implements WampServerInterface protected function getConnectionIdListByUserId($userId) { - if (!isset($this->userIdConnectionIdListMap[$userId])) return []; + if (!isset($this->userIdConnectionIdListMap[$userId])) { + return []; + } + return $this->userIdConnectionIdListMap[$userId]; } protected function getUserIdByConnection(ConnectionInterface $connection) { - if (!isset($this->connectionIdUserIdMap[$connection->resourceId])) return; + if (!isset($this->connectionIdUserIdMap[$connection->resourceId])) { + return; + } + return $this->connectionIdUserIdMap[$connection->resourceId]; } @@ -208,7 +262,9 @@ class Pusher implements WampServerInterface $this->connectionIdUserIdMap[$resourceId] = $userId; - if (!isset($this->userIdConnectionIdListMap[$userId])) $this->userIdConnectionIdListMap[$userId] = []; + if (!isset($this->userIdConnectionIdListMap[$userId])) { + $this->userIdConnectionIdListMap[$userId] = []; + } if (!in_array($resourceId, $this->userIdConnectionIdListMap[$userId])) { $this->userIdConnectionIdListMap[$userId][] = $resourceId; @@ -216,7 +272,9 @@ class Pusher implements WampServerInterface $this->connections[$resourceId] = $connection; - if ($this->isDebugMode) $this->log("{$resourceId}: user {$userId} subscribed"); + if ($this->isDebugMode) { + $this->log("{$resourceId}: user {$userId} subscribed"); + } } protected function unsubscribeUser(ConnectionInterface $connection, $userId) @@ -227,38 +285,49 @@ class Pusher implements WampServerInterface if (isset($this->userIdConnectionIdListMap[$userId])) { $index = array_search($resourceId, $this->userIdConnectionIdListMap[$userId]); + if ($index !== false) { unset($this->userIdConnectionIdListMap[$userId][$index]); $this->userIdConnectionIdListMap[$userId] = array_values($this->userIdConnectionIdListMap[$userId]); } } - if ($this->isDebugMode) $this->log("{$resourceId}: user {$userId} unsubscribed"); + if ($this->isDebugMode) { + $this->log("{$resourceId}: user {$userId} unsubscribed"); + } } public function onOpen(ConnectionInterface $connection) { - if ($this->isDebugMode) $this->log("{$connection->resourceId}: open"); + if ($this->isDebugMode) { + $this->log("{$connection->resourceId}: open"); + } $query = $connection->httpRequest->getUri()->getQuery(); + $params = \GuzzleHttp\Psr7\parse_query($query ?: ''); + if (empty($params['userId']) || empty($params['authToken'])) { $this->closeConnection($connection); + return; } $authToken = preg_replace('/[^a-zA-Z0-9]+/', '', $params['authToken']); + $userId = $params['userId']; $result = $this->getUserIdByAuthToken($authToken); if (empty($result)) { $this->closeConnection($connection); + return; } if ($result !== $userId) { $this->closeConnection($connection); + return; } @@ -273,6 +342,7 @@ class Pusher implements WampServerInterface protected function closeConnection(ConnectionInterface $connection) { $userId = $this->getUserIdByConnection($connection); + if ($userId) { $this->unsubscribeUser($connection, $userId); } @@ -282,7 +352,9 @@ class Pusher implements WampServerInterface public function onClose(ConnectionInterface $connection) { - if ($this->isDebugMode) $this->log("{$connection->resourceId}: close"); + if ($this->isDebugMode) { + $this->log("{$connection->resourceId}: close"); + } $userId = $this->getUserIdByConnection($connection); @@ -301,10 +373,11 @@ class Pusher implements WampServerInterface public function onPublish(ConnectionInterface $connection, $topic, $event, array $exclude, array $eligible) { $topicId = $topic->getId(); + $connection->close(); } - public function onError(ConnectionInterface $connection, \Exception $e) + public function onError(ConnectionInterface $connection, Exception $e) { } @@ -312,36 +385,59 @@ class Pusher implements WampServerInterface { $data = json_decode($dataString); - if (!property_exists($data, 'topicId')) return; + if (!property_exists($data, 'topicId')) { + return; + } $userId = $data->userId ?? null; $topicId = $data->topicId; - if (!$topicId) return; - if (!$this->isTopicAllowed($topicId)) return; + if (!$topicId) { + return; + } + + if (!$this->isTopicAllowed($topicId)) { + return; + } if ($userId) { foreach ($this->getConnectionIdListByUserId($userId) as $connectionId) { - if (!isset($this->connections[$connectionId])) continue; - if (!isset($this->connectionIdTopicIdListMap[$connectionId])) continue; + if (!isset($this->connections[$connectionId])) { + continue; + } + + if (!isset($this->connectionIdTopicIdListMap[$connectionId])) { + continue; + } $connection = $this->connections[$connectionId]; if (in_array($topicId, $this->connectionIdTopicIdListMap[$connectionId])) { - if ($this->isDebugMode) $this->log("send {$topicId} for connection {$connectionId}"); + if ($this->isDebugMode) { + $this->log("send {$topicId} for connection {$connectionId}"); + } + $connection->event($topicId, $data); } } - if ($this->isDebugMode) $this->log("message {$topicId} for user {$userId}"); + if ($this->isDebugMode) { + $this->log("message {$topicId} for user {$userId}"); + } } else { $topic = $this->topicHash[$topicId] ?? null; + if ($topic) { $topic->broadcast($data); - if ($this->isDebugMode) $this->log("send {$topicId} to all"); + + if ($this->isDebugMode) { + $this->log("send {$topicId} to all"); + } } - if ($this->isDebugMode) $this->log("message {$topicId} for all"); + if ($this->isDebugMode) { + $this->log("message {$topicId} for all"); + } } } diff --git a/application/Espo/Core/WebSocket/Submission.php b/application/Espo/Core/WebSocket/Submission.php index f5c929c1c3..c0097b4db3 100644 --- a/application/Espo/Core/WebSocket/Submission.php +++ b/application/Espo/Core/WebSocket/Submission.php @@ -31,6 +31,10 @@ namespace Espo\Core\WebSocket; use Espo\Core\Utils\Config; +use ZMQContext; +use ZMQ; +use Throwable; + class Submission { protected $config; @@ -42,25 +46,32 @@ class Submission public function submit(string $topic, ?string $userId = null, $data = null) { - if (!$data) $data = (object) []; + if (!$data) { + $data = (object) []; + } $dsn = $this->config->get('webSocketSubmissionDsn', 'tcp://localhost:5555'); if ($userId) { $data->userId = $userId; } + $data->topicId = $topic; try { - $context = new \ZMQContext(); - $socket = $context->getSocket(\ZMQ::SOCKET_PUSH, 'my pusher'); + $context = new ZMQContext(); + + $socket = $context->getSocket(ZMQ::SOCKET_PUSH, 'my pusher'); + $socket->connect($dsn); $socket->send(json_encode($data)); - $socket->setSockOpt(\ZMQ::SOCKOPT_LINGER, 1000); + $socket->setSockOpt(ZMQ::SOCKOPT_LINGER, 1000); + $socket->disconnect($dsn); - } catch (\Throwable $e) { + } + catch (Throwable $e) { $GLOBALS['log']->error("WebSocketSubmission: " . $e->getMessage()); } }