Websockets

Makefile

@@ -2,10 +2,26 @@
 # PRODUCTION COMMANDS #########################################
 ###############################################################
 # These are production commands which may be invoked in deployments
-altprodserver: NUM_PROCS:=3
-altprodserver: NUM_THREADS:=5
-altprodserver: collectstatic compilemessages
-	cd contentcuration/ && gunicorn contentcuration.wsgi:application --timeout=4000 --error-logfile=/var/log/gunicorn-error.log --workers=${NUM_PROCS} --threads=${NUM_THREADS} --bind=0.0.0.0:8081 --pid=/tmp/contentcuration.pid --log-level=debug || sleep infinity
+#altprodserver: NUM_PROCS:=3
+#altprodserver: NUM_THREADS:=5
+#altprodserver: collectstatic compilemessages
+#	cd contentcuration/ && gunicorn contentcuration.wsgi:application --timeout=4000 --error-logfile=/var/log/gunicorn-error.log --workers=${NUM_PROCS} --threads=${NUM_THREADS} --bind=0.0.0.0:8081 --pid=/tmp/contentcuration.pid --log-level=debug || sleep infinity
+
+altprodserver:
+	$(MAKE) -j 2 gunicornanddapneserver
+
+gunicornanddapneserver: gunicornserver daphneserver
+
+daphneserver:
+	cd contentcuration/ && daphne -b 0.0.0.0 -p 8082 contentcuration.asgi:application
+
+gunicornserver: NUM_PROCS:=1
+gunicornserver:
+	cd contentcuration/ && gunicorn contentcuration.wsgi:application --timeout=4000 --error-logfile=/var/log/gunicorn-error.log --workers=${NUM_PROCS} --bind=0.0.0.0:8081 --pid=/tmp/contentcuration.pid --log-level=debug || sleep infinity
+
+
+contentnodegc:
+	cd contentcuration/ && python manage.py garbage_collect
 
 prodceleryworkers:
 	cd contentcuration/ && celery -A contentcuration worker -l info --concurrency=3 --task-events
@@ -20,9 +36,6 @@ migrate:
 	python contentcuration/manage.py migrate || true
 	python contentcuration/manage.py loadconstants
 
-contentnodegc:
-	python contentcuration/manage.py garbage_collect
-
 filedurations:
 	python contentcuration/manage.py set_file_duration
 

contentcuration/contentcuration/apps.py

@@ -8,6 +8,9 @@ class ContentConfig(AppConfig):
     name = 'contentcuration'
 
     def ready(self):
+        # signals for websockets
+        import contentcuration.viewsets.websockets.signals # noqa
+
         if settings.AWS_AUTO_CREATE_BUCKET and not is_gcs_backend():
             from contentcuration.utils.minio_utils import ensure_storage_bucket_public
             ensure_storage_bucket_public()

contentcuration/contentcuration/asgi.py

@@ -0,0 +1,25 @@
+import django
+from channels.auth import AuthMiddlewareStack
+from channels.routing import ProtocolTypeRouter
+from channels.routing import URLRouter
+from django.conf import settings
+
+from contentcuration.viewsets.websockets.routing import http_urlpatterns
+from contentcuration.viewsets.websockets.routing import websocket_urlpatterns
+
+django.setup(set_prefix=False)
+
+protocol_config = {
+    "websocket":
+        AuthMiddlewareStack(
+            URLRouter(
+                websocket_urlpatterns
+            )
+        ),
+}
+
+# production settings to add healthcheck
+if not settings.DEBUG:
+    protocol_config.update(http=URLRouter(http_urlpatterns))
+
+application = ProtocolTypeRouter(protocol_config)

contentcuration/contentcuration/frontend/shared/data/applyRemoteChanges.js

@@ -1,8 +1,8 @@
 import Dexie from 'dexie';
 import sortBy from 'lodash/sortBy';
-import { CHANGE_TYPES, IGNORED_SOURCE, TABLE_NAMES } from './constants';
 import db from './db';
 import { INDEXEDDB_RESOURCES } from './registry';
+import { CHANGE_TYPES, IGNORED_SOURCE, TABLE_NAMES } from './constants';
 
 const { CREATED, DELETED, UPDATED, MOVED, PUBLISHED, SYNCED } = CHANGE_TYPES;
 

contentcuration/contentcuration/frontend/shared/data/resources.js

@@ -2,6 +2,7 @@ import Dexie from 'dexie';
 import Mutex from 'mutex-js';
 import findIndex from 'lodash/findIndex';
 import flatMap from 'lodash/flatMap';
+import intersection from 'lodash/intersection';
 import isArray from 'lodash/isArray';
 import isNumber from 'lodash/isNumber';
 import isString from 'lodash/isString';
@@ -14,29 +15,35 @@ import uniqBy from 'lodash/uniqBy';
 
 import { v4 as uuidv4 } from 'uuid';
 import {
-  CHANGE_TYPES,
+  ACTIVE_CHANNELS,
   CHANGES_TABLE,
+  CHANGE_TYPES,
+  CHANNEL_SYNC_KEEP_ALIVE_INTERVAL,
+  COPYING_FLAG,
+  CURRENT_USER,
+  CREATION_CHANGE_TYPES,
   IGNORED_SOURCE,
+  LAST_FETCHED,
+  MAX_REV_KEY,
   RELATIVE_TREE_POSITIONS,
   TABLE_NAMES,
-  COPYING_FLAG,
   TASK_ID,
-  CURRENT_USER,
-  ACTIVE_CHANNELS,
-  CHANNEL_SYNC_KEEP_ALIVE_INTERVAL,
-  MAX_REV_KEY,
-  LAST_FETCHED,
-  CREATION_CHANGE_TYPES,
   TREE_CHANGE_TYPES,
 } from './constants';
 import applyChanges, { applyMods, collectChanges } from './applyRemoteChanges';
 import mergeAllChanges from './mergeChanges';
 import db, { channelScope, CLIENTID, Collection } from './db';
 import { API_RESOURCES, INDEXEDDB_RESOURCES } from './registry';
 import { DELAYED_VALIDATION, fileErrors, NEW_OBJECT } from 'shared/constants';
-import client, { paramsSerializer } from 'shared/client';
-import { currentLanguage } from 'shared/i18n';
 import urls from 'shared/urls';
+import { currentLanguage } from 'shared/i18n';
+import client, { paramsSerializer } from 'shared/client';
+
+/**
+ * Task names for which it is only useful to keep the most recent task object
+ * @type {string[]}
+ */
+const SINGULAR_TASKS = ['export-channel', 'sync-channel'];
 
 // Number of seconds after which data is considered stale.
 const REFRESH_INTERVAL = 5;
@@ -1847,13 +1854,15 @@ export const Task = new IndexedDBResource({
       task.channel_id = task.channel_id.replace('-', '');
     }
     return this.transaction({ mode: 'rw', source: IGNORED_SOURCE }, () => {
-      return this.table
-        .where(this.idField)
-        .noneOf(tasks.map(t => t[this.idField]))
-        .delete()
-        .then(() => {
-          return this.table.bulkPut(tasks);
-        });
+      let deletePromise = Promise.resolve();
+      let taskDeletes = intersection(
+        SINGULAR_TASKS,
+        tasks.map(t => t.task_name)
+      );
+      if (taskDeletes.length) {
+        deletePromise = this.table.filter(t => taskDeletes.includes(t.task_name)).delete();
+      }
+      return deletePromise.then(() => this.table.bulkPut(tasks));
     });
   },
 });