Add background task to handle data changes for corpora and jobs via socketio.

app/main/events.py

@@ -1,19 +1,77 @@
+from flask import request
 from flask_login import current_user, login_required
-from flask_socketio import emit
+from flask_socketio import send
-from .. import socketio
+from .. import create_app, db, socketio
+from ..models import User
 import json
+import os
+
+stop = []
 
 
 @socketio.on('connect')
 @login_required
 def connect():
+    print('{} connected'.format(current_user.username))
+    send('You entered the room {}'.format(request.sid),
+         room=request.sid)
+
+    if request.sid in stop:
+        stop.remove(request.sid)
+    socketio.start_background_task(background_task,
+                                   current_user.id,
+                                   request.sid)
+
+
+@socketio.on('disconnect')
+@login_required
+def disconnect():
+    stop.append(request.sid)
+    print('{} disconnected'.format(current_user.username))
+
+
+def background_task(user_id, session_id):
+    '''
+    ' Check user jobs every 3 seconds and emit a json patch on change
+    '''
+    app = create_app(os.getenv('FLASK_CONFIG') or 'default', main=False)
     corpora = []
     jobs = []
 
-    for corpus in current_user.corpora:
+    with app.app_context():
+        user = db.session.query(User).filter_by(id=user_id).first()
+        for corpus in user.corpora:
             corpora.append(corpus.to_dict())
-    for job in current_user.jobs:
+        socketio.emit('init-corpora',
+                      {'data': json.dumps(corpora)},
+                      room=session_id)
+        for job in user.jobs:
             jobs.append(job.to_dict())
+        socketio.emit('init-jobs',
+                      {'data': json.dumps(jobs)},
+                      room=session_id)
+        while True:
+            if session_id in stop:
+                break
+            print(session_id + ' running')
+            socketio.emit('message', 'heartbeat', room=session_id)
 
-    emit('corpora', {'data': json.dumps(corpora)})
+            new_corpora = []
-    emit('jobs', {'data': json.dumps(jobs)})
+            new_jobs = []
+            for corpus in user.corpora:
+                new_corpora.append(corpus.to_dict())
+            for job in user.jobs:
+                new_jobs.append(job.to_dict())
+
+            # socketio.send('new_corpora: ' + json.dumps(new_corpora), room=session_id)
+            # socketio.send('new_jobs: ' + json.dumps(new_jobs), room=session_id)
+            # TODO: calculate json patch: new_corpora <-> corpora
+            # TODO: calculate json patch: new_jobs <-> jobs
+            # socketio.emit('update-corpora', {data: corpora_patch}, room=session_id)
+            # socketio.emit('update-jobs', {data: jobs_patch}, room=session_id)
+
+            corpora = new_corpora
+            jobs = new_jobs
+            socketio.sleep(3)
+            # TODO: Implement maximum runtime for this thread.
+    print(session_id + ' stopped')

app/templates/base.html.j2

@@ -25,16 +25,21 @@
     <script>
       var socket = io();
 
-      socket.on('corpora', function(msg) {
+      socket.on('init-corpora', function(msg) {
         corpora = JSON.parse(msg.data);
         for (subscriber of corporaSubscribers) {subscriber.init();}
       });
 
 
-      socket.on('jobs', function(msg) {
+      socket.on('init-jobs', function(msg) {
         jobs = JSON.parse(msg.data);
         for (subscriber of jobsSubscribers) {subscriber.init();}
       });
+
+
+      socket.on('message', function(msg) {
+        console.log(msg);
+      });
     </script>
     <meta name="viewport" content="width=device-width, initial-scale=1.0"/>
   </head>