diff --git a/application/Espo/Core/WebSocket/Pusher.php b/application/Espo/Core/WebSocket/Pusher.php index 3f170a9257..0a199e20c5 100644 --- a/application/Espo/Core/WebSocket/Pusher.php +++ b/application/Espo/Core/WebSocket/Pusher.php @@ -48,9 +48,11 @@ class Pusher implements WampServerInterface protected $connections = []; + protected $topicHash = []; + private $phpExecutablePath; - public function __construct(array $categoriesData = [], $phpExecutablePath = null, $isDebugMode = false) + public function __construct(array $categoriesData = [], ?string $phpExecutablePath = null, bool $isDebugMode = false) { $this->categoryList = array_keys($categoriesData); $this->categoriesData = $categoriesData; @@ -74,9 +76,11 @@ class Pusher implements WampServerInterface 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) echo "check access faied for topic {$topicId} for user {$userId}\n"; return; } } @@ -85,6 +89,8 @@ class Pusher implements WampServerInterface if ($this->isDebugMode) echo "add topic {$topicId} for user {$userId}\n"; $this->connectionIdTopicIdListMap[$connectionId][] = $topicId; } + + $this->topicHash[$topicId] = $topic; } public function onUnSubscribe(ConnectionInterface $connection, $topic) @@ -119,6 +125,8 @@ class Pusher implements WampServerInterface foreach ($this->categoriesData[$category]['paramList'] as $i => $item) { if (isset($arr[$i + 1])) { $params[$item] = $arr[$i + 1]; + } else { + $params[$item] = ''; } } } @@ -142,7 +150,7 @@ class Pusher implements WampServerInterface $command = $this->phpExecutablePath . " command.php " . $this->categoriesData[$category]['accessCheckCommand']; foreach ($params as $key => $value) { - str_replace(':' . $key, $value, $command); + $command = str_replace(':' . $key, $value, $command); } return $command; @@ -281,28 +289,34 @@ class Pusher implements WampServerInterface { $data = json_decode($dataString); - if (!property_exists($data, 'category')) return; - if (!property_exists($data, 'userId')) return; + if (!property_exists($data, 'topicId')) return; - $userId = $data->userId; - $category = $data->category; + $userId = $data->userId ?? null; + $topicId = $data->topicId; - if (!$userId || !$category) return; + if (!$topicId) return; + if (!$this->isTopicAllowed($topicId)) return; - if (!in_array($category, $this->categoryList)) return; + if ($userId) { + foreach ($this->getConnectionIdListByUserId($userId) as $connectionId) { + if (!isset($this->connections[$connectionId])) continue; + if (!isset($this->connectionIdTopicIdListMap[$connectionId])) continue; - foreach ($this->getConnectionIdListByUserId($userId) as $connectionId) { - if (!isset($this->connections[$connectionId])) continue; - if (!isset($this->connectionIdTopicIdListMap[$connectionId])) continue; + $connection = $this->connections[$connectionId]; - $connection = $this->connections[$connectionId]; - - if (in_array($category, $this->connectionIdTopicIdListMap[$connectionId])) { - if ($this->isDebugMode) echo "send {$category} for connection {$connectionId}\n"; - $connection->event($category, $data); + if (in_array($topicId, $this->connectionIdTopicIdListMap[$connectionId])) { + if ($this->isDebugMode) echo "send {$topicId} for connection {$connectionId}\n"; + $connection->event($topicId, $data); + } + } + } else { + $topic = $this->topicHash[$topicId] ?? null; + if ($topic) { + $topic->broadcast($data); + if ($this->isDebugMode) echo "send {$topicId} to all\n"; } } - if ($this->isDebugMode) echo "onMessage {$category} for {$userId}\n"; + if ($this->isDebugMode) echo "onMessage {$topicId} for {$userId}\n"; } } diff --git a/application/Espo/Core/WebSocket/Submission.php b/application/Espo/Core/WebSocket/Submission.php index c337c6902f..56bd842200 100644 --- a/application/Espo/Core/WebSocket/Submission.php +++ b/application/Espo/Core/WebSocket/Submission.php @@ -38,14 +38,16 @@ class Submission $this->config = $config; } - public function submit(string $category, ?string $userId = null, $data = null) + public function submit(string $topic, ?string $userId = null, $data = null) { if (!$data) $data = (object) []; $dsn = $this->config->get('webSocketSubmissionDsn', 'tcp://localhost:5555'); - $data->userId = $userId; - $data->category = $category; + if ($userId) { + $data->userId = $userId; + } + $data->topicId = $topic; try { $context = new \ZMQContext(); diff --git a/application/Espo/Hooks/Note/WebSocketSubmit.php b/application/Espo/Hooks/Note/WebSocketSubmit.php new file mode 100644 index 0000000000..cef764564a --- /dev/null +++ b/application/Espo/Hooks/Note/WebSocketSubmit.php @@ -0,0 +1,59 @@ +addDependency('webSocketSubmission'); + } + + public function afterSave(Entity $entity, array $options = []) + { + if (!$this->getConfig()->get('useWebSocket')) return; + if (!$entity->isNew()) return; + $parentId = $entity->get('parentId'); + $parentType = $entity->get('parentType'); + if (!$parentId) return; + if (!$parentType) return; + + $data = (object) [ + 'createdById' => $entity->get('createdById'), + ]; + + $topic = "streamUpdate.{$parentType}.{$parentId}"; + $this->getInjection('webSocketSubmission')->submit($topic, null, $data); + } +} diff --git a/application/Espo/Services/Notification.php b/application/Espo/Services/Notification.php index 397404e675..4fef5e1b6d 100644 --- a/application/Espo/Services/Notification.php +++ b/application/Espo/Services/Notification.php @@ -92,7 +92,7 @@ class Notification extends \Espo\Services\Record $pdo->query($sql); if ($this->getConfig()->get('useWebSocket')) { - foreach ($userIdList as $useId) { + foreach ($userIdList as $userId) { $this->getInjection('container')->get('webSocketSubmission')->submit('newNotification', $userId); } } diff --git a/client/src/views/record/detail-bottom.js b/client/src/views/record/detail-bottom.js index e629086b54..41b60b7c8c 100644 --- a/client/src/views/record/detail-bottom.js +++ b/client/src/views/record/detail-bottom.js @@ -60,6 +60,7 @@ define('views/record/detail-bottom', 'views/record/panels-container', function ( if (streamAllowed) { this.showPanel('stream', function () { this.getView('stream').collection.fetch(); + this.getView('stream').subscribeToWebSocket(); }); } }, this); diff --git a/client/src/views/stream/panel.js b/client/src/views/stream/panel.js index 2126ed1174..52a0a4e9c3 100644 --- a/client/src/views/stream/panel.js +++ b/client/src/views/stream/panel.js @@ -174,6 +174,37 @@ Espo.define('views/stream/panel', ['views/record/panels/relationship', 'lib!Text this.wait(false); }, this); }, this); + + if (!this.defs.hidden) { + this.subscribeToWebSocket(); + } + + this.once('remove', function () { + if (this.isSubscribedToWebSocked) { + this.unsubscribeFromWebSocket(); + } + }.bind(this)); + }, + + subscribeToWebSocket: function () { + if (!this.getConfig().get('useWebSocket')) return; + + var topic = 'streamUpdate.' + this.model.entityType + '.' + this.model.id; + if (this.getUser().isPortal()) { + topic += '.' + this.getUser().get('portalId'); + } + this.streamUpdateWebSocketTopic = topic; + + this.isSubscribedToWebSocked = true; + + this.getHelper().webSocketManager.subscribe(topic, function (t, data) { + if (data.createdById === this.getUser().id) return; + this.collection.fetchNew(); + }.bind(this)) + }, + + unsubscribeFromWebSocket: function () { + this.getHelper().webSocketManager.unsubscribe(this.streamUpdateWebSocketTopic); }, setupTitle: function () { diff --git a/websocket.php b/websocket.php index 8bf656eb8f..0ae8ab0897 100644 --- a/websocket.php +++ b/websocket.php @@ -36,9 +36,10 @@ $app = new \Espo\Core\Application(); $categoriesData = $app->getContainer()->get('metadata')->get(['app', 'webSocket', 'categories'], []); $phpExecutablePath = $app->getContainer()->get('config')->get('phpExecutablePath'); +$isDebugMode = (bool) $app->getContainer()->get('config')->get('webSocketDebugMode'); $loop = \React\EventLoop\Factory::create(); -$pusher = new \Espo\Core\WebSocket\Pusher($categoriesData, $phpExecutablePath); +$pusher = new \Espo\Core\WebSocket\Pusher($categoriesData, $phpExecutablePath, $isDebugMode); $context = new \React\ZMQ\Context($loop); $pull = $context->getSocket(\ZMQ::SOCKET_PULL);