A notification stream with process monitoring

gns3server/handlers/api/project_handler.py

@@ -15,14 +15,23 @@
 # You should have received a copy of the GNU General Public License
 # along with this program.  If not, see <http://www.gnu.org/licenses/>.
 
+import asyncio
+import json
+
 from ...web.route import Route
 from ...schemas.project import PROJECT_OBJECT_SCHEMA, PROJECT_CREATE_SCHEMA, PROJECT_UPDATE_SCHEMA
 from ...modules.project_manager import ProjectManager
 from ...modules import MODULES
 
+import logging
+log = logging.getLogger()
+
 
 class ProjectHandler:
 
+    # How many clients has subcribe to notifications
+    _notifications_listening = 0
+
     @classmethod
     @Route.post(
         r"/projects",
@@ -123,8 +132,9 @@ class ProjectHandler:
 
         pm = ProjectManager.instance()
         project = pm.get_project(request.match_info["project_id"])
-        yield from project.close()
-        pm.remove_project(project.id)
+        if ProjectHandler._notifications_listening == 0:
+            yield from project.close()
+            pm.remove_project(project.id)
         response.set_status(204)
 
     @classmethod
@@ -145,3 +155,45 @@ class ProjectHandler:
         yield from project.delete()
         pm.remove_project(project.id)
         response.set_status(204)
+
+    @classmethod
+    @Route.get(
+        r"/projects/{project_id}/notifications",
+        description="Receive notifications about the projects",
+        parameters={
+            "project_id": "The UUID of the project",
+        },
+        status_codes={
+            200: "End of stream",
+            404: "The project doesn't exist"
+        })
+    def notification(request, response):
+
+        pm = ProjectManager.instance()
+        project = pm.get_project(request.match_info["project_id"])
+
+        response.content_type = "application/json"
+        response.set_status(200)
+        response.enable_chunked_encoding()
+        # Very important: do not send a content lenght otherwise QT close the connection but curl can consume the Feed
+        response.content_length = None
+
+        response.start(request)
+        queue = project.get_listen_queue()
+        ProjectHandler._notifications_listening += 1
+        response.write("{\"action\": \"ping\"}\n".encode("utf-8"))
+        while True:
+            try:
+                (action, msg) = yield from asyncio.wait_for(queue.get(), 5)
+                if hasattr(msg, "__json__"):
+                    msg = json.dumps({"action": action, "event": msg.__json__()}, sort_keys=True)
+                else:
+                    msg = json.dumps({"action": action, "event": msg}, sort_keys=True)
+                log.debug("Send notification: %s", msg)
+                response.write(("{}\n".format(msg)).encode("utf-8"))
+            except asyncio.futures.CancelledError as e:
+                break
+            except asyncio.futures.TimeoutError as e:
+                response.write("{\"action\": \"ping\"}\n".encode("utf-8"))
+        project.stop_listen_queue(queue)
+        ProjectHandler._notifications_listening -= 1