espocrm-extensions/espocrm-base
Archived
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

cs fix

Browse Source
This commit is contained in:
Yuri Kuznetsov
2021-02-12 14:30:46 +02:00
parent b59e318444
commit de09d1dbd2
2 changed files with 148 additions and 41 deletions
Download Patch File Download Diff File Expand all files Collapse all files
application/Espo/Core/WebSocket/Pusher.php
+132 -36
View File
@@ -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");
}
}
}
application/Espo/Core/WebSocket/Submission.php
+16 -5
View File
@@ -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());
}
}