message bus

Author: phartenfeller

src/lib/sqlite/loadWorker.ts

@@ -1,15 +1,18 @@
-import { WorkerMessageTypes, type WorkerMessage } from './types';
+import { initMsgBus, sendMsgToWorker } from './messageBus';
+import { WorkerMessageTypes } from './types';
 
 const workerImp = await import('./worker/worker.ts?worker');
 
-export default function initWorker() {
+export default async function initWorker() {
 	const worker = new workerImp.default();
+	initMsgBus(worker);
 
-	const msg: WorkerMessage = { type: WorkerMessageTypes.INIT_DB };
-	console.log(`Sending message to worker:`, msg);
-	worker.postMessage(msg);
-
-	worker.addEventListener('message', async ({ data }: { data: WorkerMessage }) => {
-		console.log('Received message from worker:', data.type);
+	const res = await sendMsgToWorker({
+		storageId: 'db',
+		type: WorkerMessageTypes.INIT_DB,
+		expectedType: WorkerMessageTypes.INIT_DB_RESPONSE,
+		data: undefined
 	});
+
+	console.log('Init worker response:', res);
 }

src/lib/sqlite/messageBus.ts

@@ -0,0 +1,78 @@
+import type { WorkerMessage, WorkerMessageBase, WorkerMessageTypes } from './types';
+
+export type WorkerMsgCb = (data: WorkerMessage<unknown>) => unknown;
+const cbMap = new Map<string, WorkerMsgCb>();
+
+let worker: Worker;
+
+const characters = 'ABCDEFGHIJKLMNOPQRSTUVWXYZabcdefghijklmnopqrstuvwxyz0123456789';
+
+function randomId() {
+	let result = '';
+	for (let i = 0; i < 10; i++) {
+		result += characters.charAt(Math.floor(Math.random() * characters.length));
+	}
+	return result;
+}
+
+/**
+ * Initialize event listener for messages from worker
+ *
+ */
+export function initMsgBus(_worker: Worker) {
+	worker = _worker;
+
+	worker.addEventListener('message', ({ data }: { data: WorkerMessage<unknown> }) => {
+		console.log(`Message received from Worker: ${data.messageId} - ${data.type}`, data.data);
+
+		if (cbMap.has(data.messageId)) {
+			const cb = cbMap.get(data.messageId) as WorkerMsgCb;
+			cbMap.delete(data.messageId);
+			cb(data);
+		} else {
+			console.error(`Could not find callback for message type: ${data.messageId} - ${data.type}`);
+		}
+	});
+}
+
+function addCallback({ messageId, cb }: { messageId: string; cb: WorkerMsgCb }) {
+	cbMap.set(messageId, cb);
+}
+
+export function sendMsgToWorker({
+	storageId,
+	type,
+	data,
+	expectedType
+}: {
+	storageId: string;
+	type: WorkerMessageTypes;
+	data: unknown;
+	expectedType?: WorkerMessageTypes;
+}): Promise<WorkerMessage<unknown>> {
+	return new Promise((resolve, reject) => {
+		const messageId = `${storageId}-${randomId()}`;
+
+		const cb = (data: WorkerMessage<unknown>) => {
+			if (expectedType && data.type !== expectedType) {
+				const reason = `Excpected message type ${expectedType} but got: ${data.type}. MessageId: ${messageId}`;
+				console.error(reason);
+				reject(reason);
+			}
+
+			resolve(data);
+		};
+
+		addCallback({
+			messageId,
+			cb
+		});
+
+		const msgData: WorkerMessageBase = { messageId, storageId, type };
+		if (typeof data == 'object') {
+			Object.assign(msgData, data);
+		}
+
+		worker.postMessage(msgData);
+	});
+}

src/lib/sqlite/types.ts

@@ -3,6 +3,10 @@ export enum WorkerMessageTypes {
 	INIT_DB_RESPONSE
 }
 
-export type WorkerMessage = {
+export type WorkerMessageBase = {
+	messageId: string;
 	type: WorkerMessageTypes;
+	storageId: string;
 };
+
+export type WorkerMessage<T> = { data: T } & WorkerMessageBase;

src/lib/sqlite/worker/worker.ts

@@ -4,21 +4,29 @@ import { initDb } from './initDb';
 
 console.log('worker loaded');
 
+function sendMsgToMain(obj: WorkerMessage<unknown>) {
+	postMessage(obj);
+}
+
 (async function () {
-	addEventListener('message', async function ({ data }: { data: WorkerMessage }) {
+	addEventListener('message', async function ({ data }: { data: WorkerMessage<unknown> }) {
 		console.log('worker received message:', data.type);
 
-		let res: WorkerMessage;
-
 		switch (data.type) {
 			case WorkerMessageTypes.INIT_DB:
 				await import('../jswasm/sqlite3.mjs');
 
 				const initRes = await initDb();
 				console.log('worker initDb result:', initRes);
-				res = { type: WorkerMessageTypes.INIT_DB_RESPONSE };
-				console.log('worker sending message back to main:', res);
-				this.postMessage(res);
+
+				const initResult: WorkerMessage<undefined> = {
+					type: WorkerMessageTypes.INIT_DB_RESPONSE,
+					messageId: data.messageId,
+					storageId: data.storageId,
+					data: undefined
+				};
+				sendMsgToMain(initResult);
+
 				break;
 
 			default: