change socket logic

app/corpora/events.py

@@ -13,58 +13,48 @@ import json
 ' A dictionary containing lists of, with corpus ids associated, Socket.IO
 ' session ids (sid). {<corpus_id>: [<sid>, ...], ...}
 '''
-analysis_sessions = {}
+corpus_analysis_sessions = {}
 '''
 ' A dictionary containing Socket.IO session id - CQi client pairs.
 ' {<sid>: CQiClient, ...}
 '''
-analysis_clients = {}
+corpus_analysis_clients = {}
 
 
-@socketio.on('init_corpus_analysis')
+@socketio.on('request_corpus_analysis')
 @login_required
-def init_corpus_analysis(corpus_id):
+def request_corpus_analysis(corpus_id):
     corpus = Corpus.query.get(corpus_id)
     if corpus is None:
-        socketio.emit('init_corpus_analysis', '[ERROR 404]: Not Found',
+        socketio.emit('init_corpus_analysis', '[404]: Not Found',
                       room=request.sid)
     elif not (corpus.creator == current_user
               or current_user.is_administrator()):
-        socketio.emit('init_corpus_analysis', '[ERROR 403]: Forbidden',
+        socketio.emit('init_corpus_analysis', '[403]: Forbidden',
                       room=request.sid)
     else:
-        if corpus_id not in analysis_sessions:
+        socketio.start_background_task(corpus_analysis_session_handler,
-            analysis_sessions[corpus_id] = [request.sid]
-        else:
-            analysis_sessions[corpus_id].append(request.sid)
-        while corpus.status != 'analysing':
-            db.session.refresh(corpus)
-            socketio.sleep(3)
-        analysis_clients[request.sid] = CQiWrapper(
-            host='analyse_corpus_{}'.format(corpus.id))
-        analysis_clients[request.sid].connect()
-        socketio.emit('init_corpus_analysis', 'Ready', room=request.sid)
-        socketio.start_background_task(observe_corpus_analysis_connection,
                                        current_app._get_current_object(),
                                        corpus_id, request.sid)
 
 
-@socketio.on('query')
+@socketio.on('corpus_analysis')
 @login_required
-def recv_query(message):
+def corpus_analysis(message):
-    analysis_client = analysis_clients.get(request.sid)
+    client = corpus_analysis_clients.get(request.sid)
-    if analysis_client is None:
+    if client is None:
-        socketio.emit('query', '[ERROR 424]: Failed Dependency',
+        socketio.emit('query', '[424]: Failed Dependency',
                       room=request.sid)
         return
     """ Prepare and execute a query """
     logger.warning('Payload: {}'.format(message))
     corpus_name = 'CORPUS'
     query = message['query']
-    analysis_client.select_corpus(corpus_name)
+    client.select_corpus(corpus_name)
-    analysis_client.query_subcorpus(query)
+    client.query_subcorpus(query)
-    results = analysis_client.show_query_results(result_len=int(message['hits_per_page']),
+    results = client.show_query_results(
-                                                 context_len=int(message['context']))
+        result_len=int(message['hits_per_page']),
+        context_len=int(message['context']))
     # logger.warning('RESULTS: {}'.format(results))
     size_internal_dict = sys.getsizeof(results) / 1000000
     size_dict_to_str = sys.getsizeof(str(results)) / 1000000
@@ -76,19 +66,35 @@ def recv_query(message):
     logger.warning('Size of dict as raw string for {} hits per page and context len {}: {} MB'.format(message['hits_per_page'], message['context'], size_dict_to_str))
     logger.warning('Size of gzip compressed dict to string for {} hits per page and context len {}: {} MB'.format(message['hits_per_page'], message['context'], size_dict_to_str_compressed))
     logger.warning('Size of zlib compressed and utf-8 encoded string for {} hits per page and context len {}: {} MB'.format(message['hits_per_page'], message['context'], size_zlib_compressed))
-    socketio.emit('query', zlib_compressed, room=request.sid)
+    socketio.emit('corpus_analysis', zlib_compressed, room=request.sid)
 
 
-def observe_corpus_analysis_connection(app, corpus_id, session_id):
+def corpus_analysis_session_handler(app, corpus_id, session_id):
     with app.app_context():
-        while session_id in connected_sessions:
+        ''' Setup analysis session '''
+        logger.warning('[{}] Setup analysis session'.format(session_id))
+        corpus = Corpus.query.get(corpus_id)
+        while corpus.status != 'analysing':
+            db.session.refresh(corpus)
             socketio.sleep(3)
-        analysis_client = analysis_clients.pop(session_id, None)
+        client = CQiWrapper(host='analyse_corpus_{}'.format(corpus_id))
-        if analysis_client is not None:
+        client.connect()
-            analysis_client.disconnect()
+        corpus_analysis_clients[session_id] = client
-        analysis_sessions[corpus_id].remove(session_id)
+        if corpus_id not in corpus_analysis_sessions:
-        if not analysis_sessions[corpus_id]:
+            corpus_analysis_sessions[corpus_id] = [session_id]
-            analysis_sessions.pop(corpus_id, None)
+        else:
-            corpus = Corpus.query.get(corpus_id)
+            corpus_analysis_sessions[corpus_id].append(session_id)
+        socketio.emit('request_corpus_analysis', '[201]: Created', room=session_id)
+        ''' Observe analysis session '''
+        while session_id in connected_sessions:
+            logger.warning('[{}] Observe analysis session'.format(session_id))
+            socketio.sleep(3)
+        ''' Teardown analysis session '''
+        logger.warning('[{}] Teardown analysis session'.format(session_id))
+        client.disconnect()
+        corpus_analysis_clients.pop(session_id, None)
+        corpus_analysis_sessions[corpus_id].remove(session_id)
+        if not corpus_analysis_sessions[corpus_id]:
+            corpus_analysis_sessions.pop(corpus_id, None)
             corpus.status = 'stop analysis'
             db.session.commit()

app/templates/corpora/analyse_corpus.html.j2

@@ -96,7 +96,10 @@
                                 {"dismissible": false});
     loadingModal.open();
   });
-  socket.emit('init_corpus_analysis', {{ corpus_id }});
+  socket.emit('request_corpus_analysis', {{ corpus_id }});
+  socket.on('request_corpus_analysis', function(msg) {
+    if (msg === '[201]: Created') {loadingModal.close();}
+  });
 
   var queryFormElement = document.getElementById("query-form");
   var queryFormSubmitElement = document.getElementById("query-form-submit");
@@ -108,13 +111,10 @@
     let queryData = {'context': formData.get('context'),
                      'hits_per_page': formData.get('hits_per_page'),
                      'query': formData.get('query')};
-    socket.emit('query', queryData);
+    socket.emit('corpus_analysis', queryData);
     M.toast({html: 'Query has been sent!'});
   });
 
-  socket.on('init_corpus_analysis', function(msg) {
-    if (msg === 'Ready') {loadingModal.close();}
-  });
   function decodeResults(resultsByteArray) {
     console.log(resultsByteArray);
     var decompressedData = pako.inflate(resultsByteArray); // decompresses the recieved ArrayBuffer holding the compressed Byte data
@@ -134,7 +134,7 @@
     }
     return infos
   }
-  socket.on('query', function(results) {
+  socket.on('corpus_analysis', function(results) {
     console.log(results);
     var decodedJSONStr = decodeResults(results);
     var results = JSON.parse(decodedJSONStr);