Implement JSON patches for corpus and job lists.

app/main/events.py

@@ -4,20 +4,22 @@ from flask_socketio import send
 from .. import create_app, db, socketio
 from ..models import User
 import json
+import jsonpatch
 import os
 
-stop = []
+'''
+' A list containing session ids of disconnected Socket.IO sessions. It is used
+' to signal associated background tasks to stop.
+'''
+disconnected = []
 
 
 @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)
+    # print('{} connected'.format(current_user.username))
+    # send('You entered the room {}'.format(request.sid),
+    #      room=request.sid)
     socketio.start_background_task(background_task,
                                    current_user.id,
                                    request.sid)
@@ -26,52 +28,55 @@ def connect():
 @socketio.on('disconnect')
 @login_required
 def disconnect():
-    stop.append(request.sid)
-    print('{} disconnected'.format(current_user.username))
+    disconnected.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
+    ' Sends initial corpus and job lists to the client. Afterwards it checks
+    ' every 3 seconds if changes to the initial values appeared. If changes are
+    ' detected, a RFC 6902 compliant JSON patch gets send.
+    '
+    ' NOTE: The initial values are send as a init-* events.
+    '       The JSON patches are send as update-* events.
     '''
     app = create_app(os.getenv('FLASK_CONFIG') or 'default', main=False)
-    corpora = []
-    jobs = []
 
     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())
+        ''' Get current values from the database. '''
+        corpora = list(map(lambda x: x.to_dict(), user.corpora))
+        jobs = list(map(lambda x: x.to_dict(), user.jobs))
+        ''' Send initial values. '''
         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
+        ''' TODO: Implement maximum runtime for this loop. '''
+        while session_id not in disconnected:
             print(session_id + ' running')
-            socketio.emit('message', 'heartbeat', room=session_id)
-
-            new_corpora = []
-            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)
-
+            # socketio.emit('message', 'heartbeat', room=session_id)
+            ''' Get current values from the database '''
+            new_corpora = list(map(lambda x: x.to_dict(), user.corpora))
+            new_jobs = list(map(lambda x: x.to_dict(), user.jobs))
+            ''' Compute JSON patches. '''
+            corpus_patch = jsonpatch.JsonPatch.from_diff(corpora, new_corpora)
+            jobs_patch = jsonpatch.JsonPatch.from_diff(jobs, new_jobs)
+            ''' In case there are patches, send them to the user. '''
+            if corpus_patch:
+                socketio.emit('update-corpora',
+                              corpus_patch.to_string(),
+                              room=session_id)
+            if jobs_patch:
+                socketio.emit('update-jobs',
+                              jobs_patch.to_string(),
+                              room=session_id)
+            ''' Set new values as a reference for the next iteration. '''
             corpora = new_corpora
             jobs = new_jobs
             socketio.sleep(3)
-            # TODO: Implement maximum runtime for this thread.
-    print(session_id + ' stopped')
+    disconnected.remove(session_id)
+    # print(session_id + ' stopped')

requirements.txt

@@ -9,5 +9,6 @@ Flask-SocketIO
 Flask-SQLAlchemy
 Flask-Table
 Flask-WTF
+jsonpatch
 python-dotenv
 redis