Fixes #348

Author: royrusso

elastichq/api/socket.py

@@ -99,4 +99,8 @@ def do_msg(message):
 @socketio.on('disconnect', namespace='/ws')
 def disconnect():
     LOG.debug('Client disconnected')
-    # TODO
+
+    # client disconnected
+    sid = request.sid
+    taskPool.diconnect_client(sid)
+

elastichq/common/TaskPool.py

@@ -10,12 +10,20 @@ def init_app(self, socketio):
         return self
 
     def add(self, task):
-        self.tasks.append(task)
+        # sanity check for unique tasks:
+        if self.get_task_by_room_name(task.room_name) is None:
+            self.tasks.append(task)
 
     def get_task_by_room_name(self, room_name):
+        """
+
+        :param room_name: unique ID of tasks
+        :return:
+        """
         for task in self.tasks:
             if task.room_name == room_name:
                 return task
+        return None
 
     def get_tasks_by_cluster_name(self, cluster_name):
         for task in self.tasks:
@@ -37,5 +45,16 @@ def create_task(self, task, sid):
             task.add_session(sid)
             self.socketio.start_background_task(target=task.run)
             self.add(task)
-        else:
+        else: # task already exists
             task.add_session(sid)
+
+    def diconnect_client(self, sid):
+        for task in self.tasks:
+            for session in task.sessions:
+                if session == sid:
+                    task.remove_session(sid)
+                    if len(task.sessions) == 0:
+                        task.stop()
+                        self.remove(task.room_name)
+                    break
+

elastichq/config/settings.py

@@ -56,7 +56,7 @@ class TestSettings(BaseSettings):
     # static
     HQ_SITE_URL = 'http://elastichq.org'
     HQ_GH_URL = 'https://github.com/ElasticHQ/elasticsearch-HQ'
-    API_VERSION = 'v3.0'
+    API_VERSION = 'v3.1.0'
     ES_V2_HOST = '127.0.0.1'
     ES_V2_PORT = '9200'
     ES_V5_HOST = '127.0.0.1'
@@ -116,7 +116,7 @@ class ProdSettings(BaseSettings):
     # static
     HQ_SITE_URL = 'http://elastichq.org'
     HQ_GH_URL = 'https://github.com/ElasticHQ/elasticsearch-HQ'
-    API_VERSION = '3.0.3'
+    API_VERSION = '3.1.0'
 
 
     SCHEDULER_EXECUTORS = {

elastichq/model/Task.py

@@ -12,9 +12,10 @@
 
 class Task:
     """
-
+    Websocket background thread that broadcasts to a room.
     """
 
+    # room_name == Task Name. Unique identifier
     room_name = None
     cluster_name = None
     metric = None
@@ -30,6 +31,11 @@ def __init__(self, room_name, cluster_name, metric):
         self.cluster_name = cluster_name
         self.metric = metric
 
+    def remove_session(self, session_id):
+        self.sessions.remove(session_id)
+        LOG.debug("Removing client: " + session_id)
+
+
     def add_session(self, session_id):
         self.sessions.append(session_id)
 
@@ -46,6 +52,12 @@ def run(self):
 
         # https://stackoverflow.com/questions/44371041/python-socketio-and-flask-how-to-stop-a-loop-in-a-background-thread
         while self.switch:
+
+            # automatically stop this task when the room is empty
+            if len(self.sessions) == 0:
+                self.stop()
+                # TODO: remove task from pool
+
             eventlet.sleep(5)
 
             LOG.debug('-----------------------------------------')
@@ -75,10 +87,11 @@ def run(self):
                             "docs_deleted": jmespath.search("indices.docs.deleted", node_dict),
                             "store_size": jmespath.search("indices.store.size_in_bytes", node_dict),
                             "cpu_percent": jmespath.search("process.cpu.percent", node_dict),
-                            "field_data_cache_in_bytes": jmespath.search("indices.fielddata.memory_size_in_bytes", node_dict),
-                            "fs_used_in_bytes" : total_in_bytes - available_in_bytes,
-                            "fs_free_in_bytes" : available_in_bytes,
-                            "index_total" : jmespath.search("indices.indexing.index_total", node_dict)
+                            "field_data_cache_in_bytes": jmespath.search("indices.fielddata.memory_size_in_bytes",
+                                                                         node_dict),
+                            "fs_used_in_bytes": total_in_bytes - available_in_bytes,
+                            "fs_free_in_bytes": available_in_bytes,
+                            "index_total": jmespath.search("indices.indexing.index_total", node_dict)
                         }
                     nodes.append(node)
                 LOG.debug("Broadcast to room: " + self.room_name)
@@ -88,3 +101,4 @@ def run(self):
 
     def stop(self):
         self.switch = False
+        LOG.debug("Task Stopped: " + self.room_name)

ui/src/components/cluster-nodes-line-graph/cluster-nodes-line-graph.controller.js

@@ -11,7 +11,7 @@ class clusterNodesLineGraphController {
         this.$window = $window;
         this.widthFrom = angular.element($element[0].children[0])[0];
 
-        this.h = 240;
+        this.h = 220;
         this.svgContainer = d3.select(angular.element($element[0].querySelector('.chart'))[0])
                             .append('svg').attr('height', this.h);