websocket changes

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";
     }
 }

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();

application/Espo/Hooks/Note/WebSocketSubmit.php

@@ -0,0 +1,59 @@
+<?php
+/************************************************************************
+ * This file is part of EspoCRM.
+ *
+ * EspoCRM - Open Source CRM application.
+ * Copyright (C) 2014-2018 Yuri Kuznetsov, Taras Machyshyn, Oleksiy Avramenko
+ * Website: http://www.espocrm.com
+ *
+ * EspoCRM is free software: you can redistribute it and/or modify
+ * it under the terms of the GNU General Public License as published by
+ * the Free Software Foundation, either version 3 of the License, or
+ * (at your option) any later version.
+ *
+ * EspoCRM is distributed in the hope that it will be useful,
+ * but WITHOUT ANY WARRANTY; without even the implied warranty of
+ * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
+ * GNU General Public License for more details.
+ *
+ * You should have received a copy of the GNU General Public License
+ * along with EspoCRM. If not, see http://www.gnu.org/licenses/.
+ *
+ * The interactive user interfaces in modified source and object code versions
+ * of this program must display Appropriate Legal Notices, as required under
+ * Section 5 of the GNU General Public License version 3.
+ *
+ * In accordance with Section 7(b) of the GNU General Public License version 3,
+ * these Appropriate Legal Notices must retain the display of the "EspoCRM" word.
+ ************************************************************************/
+
+namespace Espo\Hooks\Note;
+
+use Espo\ORM\Entity;
+
+class WebSocketSubmit extends \Espo\Core\Hooks\Base
+{
+    public static $order = 20;
+
+    protected function init()
+    {
+        $this->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);
+    }
+}

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);
             }
         }

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);

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 () {

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);