Task[hmi-server]: reimplement ssemq on hmi server (#2150)

.github/workflows/test-server.yaml

@@ -55,6 +55,23 @@ jobs:
           --health-timeout 5s
           --health-retries 5
 
+      rabbitmq:
+        # Docker Hub image
+        image: rabbitmq:3.9-alpine
+        ports:
+          - 5672:5672
+          - 15672:15672
+        env:
+          RABBITMQ_DEFAULT_USER: "terarium"
+          RABBITMQ_DEFAULT_PASS: "terarium123"
+          RABBITMQ_MANAGEMENT_ENABLED: "true"
+        # Set health checks to wait until rabbitmq has started
+        options: >-
+          --health-cmd "rabbitmq-diagnostics ping"
+          --health-interval 10s
+          --health-timeout 5s
+          --health-retries 5
+
     permissions:
       contents: read
 

containers/secrets.env.encrypted

@@ -1,37 +1,37 @@
 $ANSIBLE_VAULT;1.1;AES256
-33323637306661363065636332386337653538663562376634313061376630646335633561386561
-3963336136303636323766363666643333643639333963610a323035653133386630333938376237
-32613632393938363338356436616236636536376238343238343761376634303866666665376663
-3639623237396439620a323335336237313661653463343961366663656265633632343365636563
-36613934366333653237373230373333666164333465336462376465666531636135313165363266
-63383966343431316463623661643735346433336165373565636234666465666264636630343565
-32613063633636343061336665316666353235666664303136343961383537623662326562633235
-61636432356635646563633765383663626336633231356537376463653333393933653261323064
-30343938623633303632386661316136326439626339656239623335343033623366366132353831
-65326266363230653965623038376131653861333735663338356233666561336335653730663636
-63386161636262383966636239613465333866346332386235613835623937643434393466323766
-39393739613939396639613239376535646534383466626235386665393431643835333961656331
-39643937393435343464343733366466346137646531656533636333326164636133393264653232
-31396337623433616534383366643863376237353930393966313530313839396431636265303562
-37366139656437393139663565663638653837353432666264643664316131643139376130383439
-38336236303536646635623631313530306633336362333965653830653162663330373036666634
-38333030646535333333393930613265663231613264336466336330616237383931613033303635
-39366661613539613035386231306464623366613563316431616332393335666261653632626131
-30376666313737323566386564656638633764623130353338313135663431343339626233626665
-61393139366265323533303037396536306636316136656265333762323765373334373731383138
-39396333356565393063653862653534653563313165623734396265663834346636636438633432
-62306233626237666664643832333532613539623666393737356464623665643132363366333061
-35623936316232336666303038336139663934376537353461613339663637643736306165626536
-63386561643561383934366636643763626363633363343134396462626361656566656431313230
-61393635643532633033643430303361376539323237333036366537333832653764363263383631
-64656561376562613566653434613562353461386565356534323035323430313061343831623939
-33613736613335643937626539383664353230306662343461376363353634656435623161653364
-38333236333436663734616632336637626462366139653465303639333730393662653366336666
-39393964656166623862396662313262303935623161313834633736613162313465366336323564
-30323266643030303436353930376566303235346364333936396666653063303230353739373732
-64643161646662633739306234373935336431313466663366633136316464303231313631356538
-39626636643565643764643536333434376132303533616465343034666335653666383464323631
-35373839313032653737346436643938383961366633346662323139666635636637316232613032
-31333235303962346339353130363637303263303063646161386137393131393061613538383161
-35636631623465626634333663396165626664666463306434366363376330313165626663343364
-38393965653437396239
+62303464326266346461626465356232663335376337303539653837393366646230323234333436
+3363663635306539363834363464333631653864646162320a353665323033313833333862313462
+39363232636135303839363966316338343732326261633366356134313732633532303736343666
+3532633765383265620a386132376334613964396436633536643464323064646338636361373235
+35623831666465356262623664633866313233323936363034376361363439346435393836396630
+63613534343137363136323336323761366632343135363432326461366433646564633130613830
+38333530653530336536616635303836616663373839646335663133323361376261613433666366
+38376463323762643566616634616135306163386364316362643865333865656435393833623862
+65653737663261373039346264306361303734373539613639363435346238386566303438396637
+39636233373265333864376366326632383239616230393330303766353264633736333266316163
+38353038313535363066646234373831646661306337613065373338303465333863343263323466
+32386235316565306131653263323935353531613230663166393035303237366230333663346230
+34306639633630613532356137376130336461313166376164646234613337666565373530663934
+62346161373866316534643235666638393135303363303435656537323961323631663664663831
+31373361623338333839333162356334646331333634623834383264363630303936353536613566
+63663532663031323861663062616566643533353032636131313566633733663439346233643663
+31373863373236633836323561383163353130653730363761633233626632383438373132656536
+32633866396665353361386639656461336532623738633262363163623066323237383336633663
+32343032323030303734626531663034313466313163353931313538333266313062393461363665
+63653033363032353463383736346237353866303232373766356132376362363534353732343334
+66623430623963316463666262323439376239393564396163653434343766633736316565336637
+65653833333438396535633032323230656561383965393663623465366163393862383133303936
+33396639373438666336653731313266303035623762326639386439653936373364636263346232
+33636233353338363363643639623662633764373566653736376431333435383836376330343532
+62393764653830343937303838313631626634613536386236623063633832616333623261313839
+32623436613166613538363264356537373335333366653864653466343231373866313233633261
+38643465613561653930366435636566363861643963613465396536393135636262613334646165
+39616137373562313630306135323639303263356236323938343233376138653662306162396564
+34356165633064616263353337396334326661396163353136623031623038623464643635393432
+37663735393061343339666363343664373564323035366266383536386633613036386138313666
+35613131343765306536653764366133333039626133653061626232636431633162393730663139
+30356664303936613661383861643435616339396534636238633037323338646462363263663837
+39646263303463636361376637393731663930643034333866346464383333303239343536316231
+33323236376234363833363138303732363865633635316430313038613637643561653339323733
+37666163383932346461383463663262393064633233613637623466303365613232313261343137
+32656439373035633564

packages/client/hmi-client/package.json

@@ -33,6 +33,7 @@
 		"@lumino/commands": "2.1.3",
 		"@lumino/coreutils": "2.1.2",
 		"@lumino/widgets": "2.1.0",
+		"@microsoft/fetch-event-source": "^2.0.1",
 		"@uncharted.software/design-tokens": "1.1.0",
 		"@uncharted.software/facets-core": "3.2.6",
 		"@uncharted.software/facets-plugins": "3.2.5",

packages/client/hmi-client/src/main.ts

@@ -14,6 +14,7 @@ import { EventType } from '@/types/Types';
 import * as EventService from '@/services/event';
 import API from '@/api/api';
 import useAuthStore from '@/stores/auth';
+import { init } from '@/services/ClientEventService';
 import router from '@/router';
 import '@node_modules/katex/dist/katex.min.css';
 import App from '@/App.vue';
@@ -54,6 +55,9 @@ setInterval(async () => {
 	await window.keycloak.updateToken(70);
 }, 6000);
 
+// init sse
+init();
+
 // Set the hash value of the window.location to null
 // This is to prevent the Keycloak from redirecting to the hash value
 // after the authentication

packages/client/hmi-client/src/services/ClientEventService.ts

@@ -0,0 +1,130 @@
+import { fetchEventSource, EventSourceMessage } from '@microsoft/fetch-event-source';
+import { ClientEvent, ClientEventType } from '@/types/Types';
+import useAuthStore from '@/stores/auth';
+import getConfiguration from '@/services/ConfigService';
+
+/**
+ * A map of event types to message handlers
+ */
+const subscribers = new Map<ClientEventType, ((data: ClientEvent<any>) => void)[]>();
+
+/**
+ * The last time a heartbeat was received
+ */
+let lastHeartbeat = new Date().valueOf();
+
+/**
+ * The initial backoff time in milliseconds for resubscribing to the SSE endpoint
+ * in the event of a retriable error
+ */
+let backoffMs = 1000;
+
+/**
+ * Whether we are currently reconnecting to the SSE endpoint
+ */
+let reconnecting = false;
+
+/**
+ * An error that can be retried
+ */
+class RetriableError extends Error {}
+
+/**
+ * Connects to the SSE endpoint and adds a message handler to pass on the messages to the subscribers
+ */
+export async function init(): Promise<void> {
+	const authStore = useAuthStore();
+	const options = {
+		headers: {
+			Authorization: `Bearer ${authStore.token}`
+		},
+		onmessage(message: EventSourceMessage) {
+			// Parse the data as a ClientEvent and pass it on to the subscribers
+			const data = JSON.parse(message.data) as ClientEvent<any>;
+			if (data.type === ClientEventType.Heartbeat) {
+				lastHeartbeat = new Date().valueOf();
+				return;
+			}
+			const handlers = subscribers.get(data.type);
+			if (handlers) {
+				handlers.forEach((handler) => handler(data));
+			}
+		},
+		async onopen(response: Response) {
+			init();
+			if (response.status === 401) {
+				// redirect to login
+				authStore.keycloak?.login({
+					redirectUri: window.location.href
+				});
+			} else if (response.status === 500) {
+				throw new RetriableError('Internal server error');
+			} else {
+				// Reset the backoff time as we've made a connection successfully
+				backoffMs = 1000;
+			}
+		},
+		onerror(error: any) {
+			// If we get a retriable error, double the backoff time up to a maximum of 60 seconds
+			if (error instanceof RetriableError) {
+				backoffMs *= 2;
+				return Math.min(backoffMs, 60000);
+			}
+			throw error; // fatal
+		},
+		onclose() {
+			init();
+		},
+		openWhenHidden: true
+	};
+	await fetchEventSource('/api/client-event', options);
+}
+
+/**
+ * Periodically checks if we have received a heartbeat within the configured interval
+ * and reconnects if not
+ */
+setInterval(async () => {
+	if (!reconnecting) {
+		const config = await getConfiguration();
+		const heartbeatIntervalMillis = config?.sseHeartbeatIntervalMillis ?? 10000;
+		if (new Date().valueOf() - lastHeartbeat > heartbeatIntervalMillis) {
+			reconnecting = true;
+			await init();
+			reconnecting = false;
+		}
+	}
+}, 1000);
+
+/**
+ * Subscribes to a specific event type
+ * @param eventType       The event type to subscribe to
+ * @param messageHandler  The message handler
+ */
+export async function subscribe(
+	eventType: ClientEventType,
+	messageHandler: (data: ClientEvent<any>) => void
+): Promise<void> {
+	if (!subscribers.has(eventType)) {
+		subscribers.set(eventType, []);
+	}
+	subscribers.get(eventType)?.push(messageHandler);
+}
+
+/**
+ * Unsubscribes from a specific event type
+ * @param eventType
+ * @param messageHandler
+ */
+export async function unsubscribe(
+	eventType: ClientEventType,
+	messageHandler: (data: ClientEvent<any>) => void
+): Promise<void> {
+	const handlers = subscribers.get(eventType);
+	if (handlers) {
+		const index = handlers.indexOf(messageHandler);
+		if (index > -1) {
+			handlers.splice(index, 1);
+		}
+	}
+}

packages/client/hmi-client/src/services/ConfigService.ts

@@ -0,0 +1,16 @@
+import { ClientConfig } from '@/types/Types';
+import axios, { AxiosHeaders } from 'axios';
+import useAuthStore from '@/stores/auth';
+
+let configuration: ClientConfig | undefined;
+async function getConfiguration() {
+	if (!configuration) {
+		const response = await axios.get('/api/config', {
+			headers: new AxiosHeaders().setAuthorization(`Bearer ${useAuthStore().token}`)
+		});
+		configuration = response.data;
+	}
+	return configuration;
+}
+
+export default getConfiguration;

packages/client/hmi-client/src/temp/sse.vue

@@ -2,41 +2,35 @@
 	<main>
 		<header>
 			<h2>Test SSE</h2>
-			<Button label="Start listening" @click="listen" />
-			<Button label="Create empty models" @click="createEmptyModel" />
+			<Button label="Subscribe" @click="listen" />
+			<Button label="Unsubscribe" @click="stop" />
 		</header>
-		<ul>
-			<li v-for="modelId in models" :key="modelId">{{ modelId }}</li>
-		</ul>
+		<template v-for="message in messages" :key="message.id">
+			<div>{{ message.id }} : {{ message.data }}</div>
+		</template>
 	</main>
 </template>
 
 <script setup lang="ts">
 import { ref } from 'vue';
 import Button from 'primevue/button';
-import API from '@/api/api';
-import { EventSourcePolyfill } from 'event-source-polyfill';
-import useAuthStore from '../stores/auth';
+import { ClientEvent, ClientEventType } from '@/types/Types';
+import { subscribe, unsubscribe } from '@/services/ClientEventService';
+const messages = ref<ClientEvent<any>[]>([]);
 
-const models = ref<Array<string>>([]);
+function getMessageHandler(event: ClientEvent<any>) {
+	messages.value.push(event);
+	if (messages.value.length > 10) {
+		messages.value.shift();
+	}
+}
 
 function listen() {
-	const auth = useAuthStore();
-	const events = new EventSourcePolyfill('/api/user/server-sent-events', {
-		headers: {
-			Authorization: `Bearer ${auth.token}`
-		}
-	});
-	//const events = new EventSource("/api/server-sent-events", { withCredentials: true });
-	events.onmessage = (event) => {
-		const id: string = JSON.parse(event.data)?.id;
-		models.value.push(id);
-		console.log(id);
-	};
+	subscribe(ClientEventType.Notification, getMessageHandler);
 }
 
-async function createEmptyModel() {
-	await API.put('/dev-tests/user-event');
+function stop() {
+	unsubscribe(ClientEventType.Notification, getMessageHandler);
 }
 </script>