New websocket message format for both clients

client/src/app/app.component.ts

@@ -1,5 +1,7 @@
 import { Component, OnInit } from '@angular/core';
 import { TranslateService } from '@ngx-translate/core';
+import { AutoupdateService } from './core/services/autoupdate.service';
+import { NotifyService } from './core/services/notify.service';
 
 /**
  * Angular's global App Component
@@ -12,11 +14,15 @@ import { TranslateService } from '@ngx-translate/core';
 export class AppComponent {
     /**
      * Initialises the operator, the auto update (and therefore a websocket) feature and the translation unit.
-     * @param operator
+     * @param autoupdateService
-     * @param autoupdate
+     * @param notifyService
      * @param translate
      */
-    constructor(private translate: TranslateService) {
+    constructor(
+        private autoupdateService: AutoupdateService,
+        private notifyService: NotifyService,
+        private translate: TranslateService
+    ) {
         // manually add the supported languages
         translate.addLangs(['en', 'de', 'fr']);
         // this language will be used as a fallback when a translation isn't found in the current language

client/src/app/core/services/autoupdate.service.ts

@@ -33,26 +33,13 @@ import { User } from 'app/shared/models/users/user';
     providedIn: 'root'
 })
 export class AutoupdateService extends OpenSlidesComponent {
-    /**
-     * Stores the to create the socket created using {@link WebsocketService}.
-     */
-    private socket;
-
     /**
      * Constructor to create the AutoupdateService. Calls the constructor of the parent class.
      * @param websocketService
      */
     constructor(private websocketService: WebsocketService) {
         super();
-    }
+        websocketService.getOberservable<any>('autoupdate').subscribe(response => {
-
-    /**
-     * Function to start the automatic update process
-     * will build up a websocket connection using {@link WebsocketService}
-     */
-    startAutoupdate(): void {
-        this.socket = this.websocketService.connect();
-        this.socket.subscribe(response => {
             this.storeResponse(response);
         });
     }
@@ -69,8 +56,6 @@ export class AutoupdateService extends OpenSlidesComponent {
         socketResponse.forEach(jsonObj => {
             const targetClass = this.getClassFromCollectionString(jsonObj.collection);
             if (jsonObj.action === 'deleted') {
-                console.log('storeResponse detect delete');
-
                 this.DS.remove(jsonObj.collection, jsonObj.id);
             } else {
                 this.DS.add(new targetClass().deserialize(jsonObj.data));

client/src/app/core/services/notify.service.spec.ts

@@ -0,0 +1,15 @@
+import { TestBed, inject } from '@angular/core/testing';
+
+import { NotifyService } from './notify.service';
+
+describe('WebsocketService', () => {
+    beforeEach(() => {
+        TestBed.configureTestingModule({
+            providers: [NotifyService]
+        });
+    });
+
+    it('should be created', inject([NotifyService], (service: NotifyService) => {
+        expect(service).toBeTruthy();
+    }));
+});

client/src/app/core/services/notify.service.ts

@@ -0,0 +1,42 @@
+import { Injectable } from '@angular/core';
+
+import { OpenSlidesComponent } from 'app/openslides.component';
+import { WebsocketService } from './websocket.service';
+
+interface NotifyFormat {
+    id: number; //Dummy
+}
+
+/**
+ * Handles all incoming and outgoing notify messages via {@link WebsocketService}.
+ */
+@Injectable({
+    providedIn: 'root'
+})
+export class NotifyService extends OpenSlidesComponent {
+    /**
+     * Constructor to create the NotifyService. Registers itself to the WebsocketService.
+     * @param websocketService
+     */
+    constructor(private websocketService: WebsocketService) {
+        super();
+        websocketService.getOberservable<any>('notify').subscribe(notify => {
+            this.receive(notify);
+        });
+    }
+
+    // TODO: Implement this
+    private receive(notify: NotifyFormat): void {
+        console.log('recv', notify);
+        // TODO: Use a Subject, so one can subscribe and get notifies.
+    }
+
+    // TODO: Make this api better: e.g. send(data, users?, projectors?, channel?, ...)
+    /**
+     * Sents a notify object to the server
+     * @param notify the notify objects
+     */
+    public send(notify: NotifyFormat): void {
+        this.websocketService.send('notify', notify);
+    }
+}

client/src/app/core/services/websocket.service.ts

@@ -1,6 +1,13 @@
 import { Injectable } from '@angular/core';
 import { Router } from '@angular/router';
 import { webSocket, WebSocketSubject } from 'rxjs/webSocket';
+import { Observable, Subject } from 'rxjs';
+
+interface WebsocketMessage {
+    type: string;
+    content: any;
+    id: string;
+}
 
 /**
  * Service that handles WebSocket connections.
@@ -20,21 +27,71 @@ export class WebsocketService {
     /**
      * Observable subject that might be `any` for simplicity, `MessageEvent` or something appropriate
      */
-    private subject: WebSocketSubject<any>;
+    private _websocketSubject: WebSocketSubject<WebsocketMessage>;
+
+    /**
+     * Subjects for types of websocket messages. A subscriber can get an Observable by {@function getOberservable}.
+     */
+    private _subjects: { [type: string]: Subject<any> } = {};
 
     /**
      * Creates a new WebSocket connection as WebSocketSubject
      *
      * Can return old Subjects to prevent multiple WebSocket connections.
      */
-    public connect(): WebSocketSubject<any> {
+    public connect(): void {
         const socketProtocol = this.getWebSocketProtocol();
         const socketPath = this.getWebSocketPath();
         const socketServer = window.location.hostname + ':' + window.location.port;
-        if (!this.subject) {
+        if (!this._websocketSubject) {
-            this.subject = webSocket(socketProtocol + socketServer + socketPath);
+            this._websocketSubject = webSocket(socketProtocol + socketServer + socketPath);
+            // directly subscribe. The messages are distributes below
+            this._websocketSubject.subscribe(message => {
+                const type: string = message.type;
+                if (type === 'error') {
+                    console.error('Websocket error', message.content);
+                } else if (this._subjects[type]) {
+                    this._subjects[type].next(message.content);
+                } else {
+                    console.log(`Got unknown websocket message type "${type}" with content`, message.content);
                 }
-        return this.subject;
+            });
+        }
+    }
+
+    /**
+     * Returns an observable for messages of the given type.
+     * @param type the message type
+     */
+    public getOberservable<T>(type: string): Observable<T> {
+        if (!this._subjects[type]) {
+            this._subjects[type] = new Subject<T>();
+        }
+        return this._subjects[type].asObservable();
+    }
+
+    /**
+     * Sends a message to the server with the content and the given type.
+     *
+     * @param type the message type
+     * @param content the actual content
+     */
+    public send<T>(type: string, content: T): void {
+        if (!this._websocketSubject) {
+            return;
+        }
+
+        const message: WebsocketMessage = {
+            type: type,
+            content: content,
+            id: ''
+        };
+
+        const possible = 'ABCDEFGHIJKLMNOPQRSTUVWXYZabcdefghijklmnopqrstuvwxyz';
+        for (let i = 0; i < 8; i++) {
+            message.id += possible.charAt(Math.floor(Math.random() * possible.length));
+        }
+        this._websocketSubject.next(message);
     }
 
     /**

client/src/app/site/site.component.ts

@@ -3,8 +3,8 @@ import { Router } from '@angular/router';
 import { BreakpointObserver, Breakpoints, BreakpointState } from '@angular/cdk/layout';
 
 import { AuthService } from 'app/core/services/auth.service';
-import { AutoupdateService } from 'app/core/services/autoupdate.service';
 import { OperatorService } from 'app/core/services/operator.service';
+import { WebsocketService } from 'app/core/services/websocket.service';
 
 import { TranslateService } from '@ngx-translate/core'; //showcase
 import { BaseComponent } from 'app/base.component';
@@ -33,7 +33,7 @@ export class SiteComponent extends BaseComponent implements OnInit {
      * Constructor
      *
      * @param authService
-     * @param autoupdateService
+     * @param websocketService
      * @param operator
      * @param router
      * @param breakpointObserver
@@ -42,7 +42,7 @@ export class SiteComponent extends BaseComponent implements OnInit {
      */
     constructor(
         private authService: AuthService,
-        private autoupdateService: AutoupdateService,
+        private websocketService: WebsocketService,
         private operator: OperatorService,
         private router: Router,
         public vp: ViewportService,
@@ -66,7 +66,7 @@ export class SiteComponent extends BaseComponent implements OnInit {
         // start autoupdate if the user is logged in:
         this.operator.whoAmI().subscribe(resp => {
             if (resp.user) {
-                this.autoupdateService.startAutoupdate();
+                this.websocketService.connect();
             } else {
                 //if whoami is not sucsessfull, forward to login again
                 this.operator.clear();

openslides/core/static/js/core/base.js

@@ -117,11 +117,11 @@ angular.module('OpenSlidesApp.core', [
                 $timeout(runRetryConnectCallbacks, getTimeoutTime());
             };
             socket.onmessage = function (event) {
-                var dataList = [];
+                var data;
                 try {
-                    dataList = JSON.parse(event.data);
+                    data = JSON.parse(event.data);
                     _.forEach(Autoupdate.messageReceivers, function (receiver) {
-                        receiver(dataList);
+                        receiver(data);
                     });
                 } catch(err) {
                     console.error(err);
@@ -133,10 +133,23 @@ angular.module('OpenSlidesApp.core', [
                 ErrorMessage.clearConnectionError();
             };
         };
-        Autoupdate.send = function (message) {
+        Autoupdate.send = function (type, content) {
-            if (socket) {
+            if (!socket) {
-                socket.send(JSON.stringify(message));
+                return;
             }
+
+            var message = {
+                type: type,
+                content: content,
+                id: '',
+            };
+
+            // Generate random id
+            var possible = 'ABCDEFGHIJKLMNOPQRSTUVWXYZabcdefghijklmnopqrstuvwxyz';
+            for (var i = 0; i < 8; i++) {
+                message.id += possible.charAt(Math.floor(Math.random() * possible.length));
+            }
+            socket.send(JSON.stringify(message));
         };
         Autoupdate.closeConnection = function () {
             if (socket) {
@@ -369,7 +382,12 @@ angular.module('OpenSlidesApp.core', [
     'dsEject',
     function (DS, autoupdate, dsEject) {
         // Handler for normal autoupdate messages.
-        autoupdate.onMessage(function(dataList) {
+        autoupdate.onMessage(function(data) {
+            if (data.type !== 'autoupdate') {
+                return;
+            }
+
+            var dataList = data.content;
             var dataListByCollection = _.groupBy(dataList, 'collection');
             _.forEach(dataListByCollection, function (list, key) {
                 var changedElements = [];
@@ -379,8 +397,6 @@ angular.module('OpenSlidesApp.core', [
                     // Uncomment this line for debugging to log all autoupdates:
                     // console.log("Received object: " + data.collection + ", " + data.id);
 
-                    // Now handle autoupdate message but do not handle notify messages.
-                    if (data.collection !== 'notify') {
                     // remove (=eject) object from local DS store
                     var instance = DS.get(data.collection, data.id);
                     if (instance) {
@@ -395,7 +411,6 @@ angular.module('OpenSlidesApp.core', [
                         console.error('Error: Undefined action for received object' +
                             '(' + data.collection + ', ' + data.id + ')');
                     }
-                    }
                 });
                 // add (=inject) all given objects into local DS store
                 if (changedElements.length > 0) {
@@ -418,7 +433,12 @@ angular.module('OpenSlidesApp.core', [
         var anonymousTrackId;
 
         // Handler for notify messages.
-        autoupdate.onMessage(function(dataList) {
+        autoupdate.onMessage(function(data) {
+            if (data.type !== 'notify') {
+                return;
+            }
+
+            var dataList = data.content;
             var dataListByCollection = _.groupBy(dataList, 'collection');
             _.forEach(dataListByCollection.notify, function (notifyItem) {
                 // Check, if this current user (or anonymous instance) has send this notify.
@@ -505,7 +525,7 @@ angular.module('OpenSlidesApp.core', [
                         }
                         notifyItem.anonymousTrackId = anonymousTrackId;
                     }
-                    autoupdate.send([notifyItem]);
+                    autoupdate.send('notify', [notifyItem]);
                 } else {
                     throw 'eventName should only consist of [a-zA-Z0-9_-]';
                 }
@@ -514,6 +534,18 @@ angular.module('OpenSlidesApp.core', [
     }
 ])
 
+.run([
+    'autoupdate',
+    function (autoupdate) {
+        // Handler for normal autoupdate messages.
+        autoupdate.onMessage(function (data) {
+            if (data.type === 'error') {
+                console.error("Websocket error", data.content);
+            }
+        });
+    }
+])
+
 // Save the server time to the rootscope.
 .run([
     '$http',