P42/P42_UI
Public Access
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

threetobus now uses [UID] template, MessageBus upgraded to cnxId

Browse Source
This commit is contained in:
STEINNI
2026-06-26 21:56:19 +00:00
parent aefa5ab147
commit cf292031fd
4 changed files with 81 additions and 89 deletions
Download Patch File Download Diff File Expand all files Collapse all files
core/libs/MessageBus.js
+76 -69
View File
@@ -84,26 +84,26 @@ class MessageBus {
*/
constructor(config, userInfo){
this.config = config
if(this.config.debug) console.log('Lauching Websocket worker...');
if(this.config.debug) console.log('Lauching Websocket worker...')
this.config.hostname = (('host' in this.config) && ( this.config.host!='')) ? this.config.host : document.location.hostname
this.userInfo = userInfo
this.createWorker();
this.createWorker()
this.activeSubscriptions = [];
this.promisesRegister = { };
this.promisesRegister = { }
this.bus2jsEventsRegister = []; // items: { eventType:'string', RegisteredCb: function, realCb: function }
this.whenConnectedQ = [];
this.connected = false;
this.whenConnectedQ = []
this.connected = false
}
/**
*
*/
createWorker() {
if(!this.config.pathToWorker.endsWith('.js')) this.config.pathToWorker+='.js';
this.MessageBusWorker = new Worker(this.config.pathToWorker+'?'+crypto.randomUUID());
if(!this.config.pathToWorker.endsWith('.js')) this.config.pathToWorker+='.js'
this.MessageBusWorker = new Worker(this.config.pathToWorker+'?'+crypto.randomUUID())
this.MessageBusWorker.postMessage({ 'action':'start', 'config': this.config, 'userInfo': this.userInfo });
this.MessageBusWorker.onmessage = this.receiveFromWorker.bind(this);
if(this.config.debug) console.log('Websocket worker launched.');
this.MessageBusWorker.onmessage = this.receiveFromWorker.bind(this)
if(this.config.debug) console.log('Websocket worker launched.')
}
/**
@@ -111,9 +111,9 @@ class MessageBus {
* @param {*} callBack
*/
whenConnected(callBack){
if(typeof(callBack) != 'function') return;
if(this.connected) callBack();
else this.whenConnectedQ.push(callBack);
if(typeof(callBack) != 'function') return
if(this.connected) callBack()
else this.whenConnectedQ.push(callBack)
}
/**
@@ -135,8 +135,8 @@ class MessageBus {
* @param {*} callBack
*/
ifConnected(callBack){
if(typeof(callBack) != 'function') return;
if(this.connected) callBack();
if(typeof(callBack) != 'function') return
if(this.connected) callBack()
}
/**
@@ -152,17 +152,17 @@ class MessageBus {
* This method gives (and resolves) a promise, taking care of all lower-level details
*/
requestWssGwAction(action, payload=null, timeOut=5000){
if(!action) return;
let request = {'action':action, 'payload':payload};
request.reqid = crypto.randomUUID();
if(!action) return
let request = {'action':action, 'payload':payload}
request.reqid = crypto.randomUUID()
return(new Promise((resolve, fail) => {
let timeOutID = setTimeout(() => {
fail(`Timeout (>${timeOut}ms) for action ${action}`);
if(this.promisesRegister[request.reqid]) delete(this.promisesRegister[request.reqid])
}, timeOut);
this.promisesRegister[request.reqid] = [resolve, fail, timeOutID];
this.MessageBusWorker.postMessage(request);
}));
}, timeOut)
this.promisesRegister[request.reqid] = [resolve, fail, timeOutID]
this.MessageBusWorker.postMessage(request)
}))
}
/**
@@ -172,18 +172,18 @@ class MessageBus {
* This method gives (and resolves) a promise, taking care of all lower-level details
*/
requestBusAction(chan, action, payload=null, timeOut=5000){
if(!action) return;
let request = {'action':action, 'payload':payload};
request.reqid = crypto.randomUUID();
if(!action) return
let request = {'action':action, 'payload':payload}
request.reqid = crypto.randomUUID()
return(new Promise((resolve, fail) => {
let timeOutID = setTimeout(() => {
fail(`Timeout (>${timeOut}ms) for action ${action}`);
if(this.promisesRegister[request.reqid]) delete(this.promisesRegister[request.reqid])
}, timeOut);
this.promisesRegister[request.reqid] = [resolve, fail, timeOutID];
}, timeOut)
this.promisesRegister[request.reqid] = [resolve, fail, timeOutID]
if(!chan.startsWith(this.config.frontBusPrefix)) chan = this.config.frontBusPrefix+chan
this.send(chan, JSON.stringify(request))
}));
}))
}
/**
@@ -193,18 +193,18 @@ class MessageBus {
* This method gives (and resolves) a promise, taking care of all lower-level details
*/
requestMidasAction(chan, action, data=null, timeOut=5000){
if(!action) return;
if(!action) return
let request = {payload: {'action':action, 'data':data}}
request.reqid = crypto.randomUUID();
request.reqid = crypto.randomUUID()
return(new Promise((resolve, fail) => {
let timeOutID = setTimeout(() => {
fail(`Timeout (>${timeOut}ms) for action ${action}`);
if(this.promisesRegister[request.reqid]) delete(this.promisesRegister[request.reqid])
}, timeOut);
this.promisesRegister[request.reqid] = [resolve, fail, timeOutID];
}, timeOut)
this.promisesRegister[request.reqid] = [resolve, fail, timeOutID]
if(!chan.startsWith(this.config.frontBusPrefix)) chan = this.config.frontBusPrefix+chan
this.send(chan, JSON.stringify(request))
}));
}))
}
/**
@@ -230,9 +230,9 @@ class MessageBus {
*/
send(chan, msg){
// You can publish to an unsubscribed chan, userchans are the best example !
// if(this.activeSubscriptions.indexOf(chan)<0) return;
var request = {'action':'PUB', 'payload': { 'chan':chan, 'msg': msg}};
this.MessageBusWorker.postMessage(request);
// if(this.activeSubscriptions.indexOf(chan)<0) return
var request = {'action':'PUB', 'payload': { 'chan':chan, 'msg': msg}}
this.MessageBusWorker.postMessage(request)
}
/**
@@ -322,11 +322,13 @@ class MessageBus {
/**
* Helper method to match a chan with globbing
*
* @param {string} myChan (no glob)
* @param {string} targetChan PATTERN (possible glob)
* @param {string} myChan (no glob, no user expansion)
* @param {string} targetChan PATTERN (possible glob and user expansion)
* @returns {boolean}
*/
chanMatch(myChan, targetChan) {
targetChan = targetChan.replace(/\[UID\]/g, this.userInfo.uuid)
targetChan = targetChan.replace(/\[CNXID\]/g, this.cnxId)
let re = new RegExp('^'+targetChan.replace(/\*/g,'(.+)')+'$','g')
return(myChan.match(re)!=null)
}
@@ -339,28 +341,28 @@ class MessageBus {
var workermsg = e.data;
if('event' in workermsg){
// event "ReceiveFromServer" is the general case of a message from server, found in data, with its own struct.
// other type og event are generated by the worker, about the connection
// other type of event are generated by the worker, about the connection
switch(workermsg.event){
case 'ReceiveFromServer':
this.receiveFromServer(JSON.parse(workermsg.data));
break;
this.receiveFromServer(JSON.parse(workermsg.data))
break
case 'connected':
this.connected = true;
if(this.config.debug) console.log('received connected event from worker !');
this.executewhenConnectedQ();
app.events.trigger('MessageBus.Connected');
break;
this.connected = true
if(this.config.debug) console.log('received connected event from worker !')
this.executewhenConnectedQ()
app.events.trigger('MessageBus.Connected')
break
case 'closed':
if(this.config.debug) console.log('received closed event from worker!');
this.activeSubscriptions = [];
this.callBacksRegister = { };
this.whenConnectedQ = [];
if(this.config.debug) console.log('received closed event from worker!')
this.activeSubscriptions = []
this.callBacksRegister = { }
this.whenConnectedQ = []
this.connected = false;
app.events.trigger('MessageBus.Closed');
break;
break
default:
if(this.config.debug) console.warn('Unknown Websocket Worker message:', workermsg);
if(this.config.debug) console.warn('Unknown Websocket Worker message:', workermsg)
}
}
}
@@ -377,35 +379,40 @@ class MessageBus {
receiveFromServer(srvdata) {
// See protocol reminder comment at the bottom
if('action' in srvdata){ // Reply to a request
let action = srvdata.action;
let payload = ('payload' in srvdata) ? srvdata.payload : null;
let action = srvdata.action
let payload = ('payload' in srvdata) ? srvdata.payload : null
// Piggyback on the results of some actions for this module internal use
switch(action){
case 'WELCOME':
console.log('Received WSS welcome', srvdata)
this.cnxId = srvdata.cnxId
this.serverTimeDelta = Date.now() - srvdata.serverTime
break
case 'SUB':
if(this.activeSubscriptions.indexOf(payload)<0) this.activeSubscriptions = this.activeSubscriptions.concat(payload);
break;
if(this.activeSubscriptions.indexOf(payload)<0) this.activeSubscriptions = this.activeSubscriptions.concat(payload)
break
case 'SUBLST':
if(this.activeSubscriptions.indexOf(payload)<0) this.activeSubscriptions = this.activeSubscriptions.concat(payload);
break;
if(this.activeSubscriptions.indexOf(payload)<0) this.activeSubscriptions = this.activeSubscriptions.concat(payload)
break
}
app.events.trigger('MessageBus.anyAction', srvdata);
} else { // Low-level event : Redis Event, contrary to requ/reply with wssGateway, or other later
if(('event' in srvdata) && (srvdata.event == 'REDISMSG')){
var payload = ('payload' in srvdata) ? srvdata.payload : null;
var payload = ('payload' in srvdata) ? srvdata.payload : null
if(payload && payload.msg && (payload.msg.eventType || payload.msg.action)) {
if(payload.msg.eventType){
var eventType = payload.msg.eventType;
var eventType = payload.msg.eventType
app.events.trigger('MessageBus.event.'+eventType, {
chan: payload.chan,
sender: payload.msg.sender,
eventType: payload.msg.eventType,
payload: payload.msg.payload,
});
})
} else if(payload.msg.action && payload.msg.reqid) {
let reqid = payload.msg.reqid;
let action = payload.msg.action;
let actionPayload = ('payload' in payload.msg) ? payload.msg.payload : null;
let err = ('err' in payload.msg) ? payload.msg.err : null;
let reqid = payload.msg.reqid
let action = payload.msg.action
let actionPayload = ('payload' in payload.msg) ? payload.msg.payload : null
let err = ('err' in payload.msg) ? payload.msg.err : null
let success = payload.msg.success;
if(reqid in this.promisesRegister) {
clearTimeout(this.promisesRegister[reqid][2]); // Stop timeout timer
@@ -416,12 +423,12 @@ class MessageBus {
app.events.trigger('MessageBus.anyMessage', {
chan: payload.chan,
msg : payload.msg,
});
})
} else if(payload && payload.bmsg){
app.events.trigger('MessageBus.promiscuousMessage', { // Repill msg : decapsulate & use spcific event
chan: payload.bmsg.chan,
msg : payload.bmsg.msg,
});
})
}
else {
console.warn('Weird bus message (discarted) :', srvdata)
@@ -431,9 +438,9 @@ class MessageBus {
// For request-reply, settle promise
if(srvdata.reqid && (srvdata.reqid in this.promisesRegister)) {
let payload = ('payload' in srvdata) ? srvdata.payload : null;
let err = ('err' in srvdata) ? srvdata.err : null;
let success = srvdata.success;
let payload = ('payload' in srvdata) ? srvdata.payload : null
let err = ('err' in srvdata) ? srvdata.err : null
let success = srvdata.success
clearTimeout(this.promisesRegister[srvdata.reqid][2]); // Stop timeout timer
if(success) this.promisesRegister[srvdata.reqid][0](payload); // resolve
else this.promisesRegister[srvdata.reqid][1](`MsgBus action failed.\nError: ${err}`); // Fail