Compare commits

..

No commits in common. "b5bdae461184ebe6ec9984947409fe798fd7635d" and "68590cae33fc45f977d3771f761a959b06524bc9" have entirely different histories.

3 changed files with 144 additions and 64 deletions

View File

@ -1,26 +1,53 @@
import * as core from './core.js';
let g_next_id = 1;
let g_calls = {};
let gSessionIndex = 0;
/**
* TODOC
* @returns
*/
function makeSessionId() {
return 'session_' + (gSessionIndex++).toString();
}
/**
* TODOC
* @returns
*/
function App() {
this._on_output = null;
this._send_queue = [];
this.calls = {};
this._next_call_id = 1;
return this;
}
/**
* TODOC
* @param {*} callback
*/
App.prototype.readOutput = function (callback) {
this._on_output = callback;
};
/**
* TODOC
* @param {*} api
* @returns
*/
App.prototype.makeFunction = function (api) {
let self = this;
let result = function () {
let id = self._next_call_id++;
while (!id || self.calls[id]) {
id = self._next_call_id++;
let id = g_next_id++;
while (!id || g_calls[id]) {
id = g_next_id++;
}
let promise = new Promise(function (resolve, reject) {
self.calls[id] = {resolve: resolve, reject: reject};
g_calls[id] = {resolve: resolve, reject: reject};
});
let message = {
action: 'tfrpc',
message: 'tfrpc',
method: api[0],
params: [...arguments],
id: id,
@ -32,6 +59,10 @@ App.prototype.makeFunction = function (api) {
return result;
};
/**
* TODOC
* @param {*} message
*/
App.prototype.send = function (message) {
if (this._send_queue) {
if (this._on_output) {
@ -46,6 +77,11 @@ App.prototype.send = function (message) {
}
};
/**
* TODOC
* @param {*} request
* @param {*} response
*/
exports.app_socket = async function socket(request, response) {
let process;
let options = {};
@ -66,16 +102,10 @@ exports.app_socket = async function socket(request, response) {
try {
message = JSON.parse(event.data);
} catch (error) {
print(
'WebSocket error:',
error,
event.data,
event.data.length,
event.opCode
);
print('ERROR', error, event.data, event.data.length, event.opCode);
return;
}
if (!process && message.action == 'hello') {
if (message.action == 'hello') {
let packageOwner;
let packageName;
let blobId;
@ -92,7 +122,7 @@ exports.app_socket = async function socket(request, response) {
if (!blobId) {
response.send(
JSON.stringify({
action: 'tfrpc',
message: 'tfrpc',
method: 'error',
params: [message.path + ' not found'],
id: -1,
@ -133,7 +163,7 @@ exports.app_socket = async function socket(request, response) {
options.packageOwner = packageOwner;
options.packageName = packageName;
options.url = message.url;
let sessionId = 'session_' + (gSessionIndex++).toString();
let sessionId = makeSessionId();
if (blobId) {
if (message.edit_only) {
response.send(
@ -145,24 +175,9 @@ exports.app_socket = async function socket(request, response) {
}
}
if (process) {
process.client_api.tfrpc = function (message) {
if (message.id) {
let calls = process?.app?.calls;
if (calls) {
let call = calls[message.id];
if (call) {
if (message.error !== undefined) {
call.reject(message.error);
} else {
call.resolve(message.result);
}
delete calls[message.id];
}
}
}
};
process.app._on_output = (message) =>
process.app.readOutput(function (message) {
response.send(JSON.stringify(message), 0x1);
});
process.app.send();
}
@ -191,13 +206,26 @@ exports.app_socket = async function socket(request, response) {
if (process && process.timeout > 0) {
setTimeout(ping, process.timeout);
}
} else {
} else if (message.action == 'resetPermission') {
if (process) {
if (process.client_api[message.action]) {
process.client_api[message.action](message);
} else if (process.eventHandlers['message']) {
await core.invoke(process.eventHandlers['message'], [message]);
process.resetPermission(message.permission);
}
} else if (message.action == 'setActiveIdentity') {
process.setActiveIdentity(message.identity);
} else if (message.action == 'createIdentity') {
await process.createIdentity();
} else if (message.message == 'tfrpc') {
if (message.id && g_calls[message.id]) {
if (message.error !== undefined) {
g_calls[message.id].reject(message.error);
} else {
g_calls[message.id].resolve(message.result);
}
delete g_calls[message.id];
}
} else {
if (process && process.eventHandlers['message']) {
await core.invoke(process.eventHandlers['message'], [message]);
}
}
} else if (event.opCode == 0x8) {

View File

@ -1325,7 +1325,7 @@ function _receive_websocket_message(message) {
line.append(key, message.stats[key]);
}
}
} else if (message && message.action === 'tfrpc' && message.method) {
} else if (message && message.message === 'tfrpc' && message.method) {
let api = k_api[message.method];
let id = message.id;
let params = message.params;
@ -1333,14 +1333,14 @@ function _receive_websocket_message(message) {
Promise.resolve(api.func(...params))
.then(function (result) {
send({
action: 'tfrpc',
message: 'tfrpc',
id: id,
result: result,
});
})
.catch(function (error) {
send({
action: 'tfrpc',
message: 'tfrpc',
id: id,
error: error,
});

View File

@ -3,10 +3,12 @@ import * as http from './http.js';
let gProcesses = {};
let gStatsTimer = false;
let g_handler_index = 0;
const k_ping_interval = 60 * 1000;
let kPingInterval = 60 * 1000;
/**
* TODOC
* @param {*} error
*/
function printError(error) {
if (error.stackTrace) {
print(error.fileName + ':' + error.lineNumber + ': ' + error.message);
@ -19,6 +21,12 @@ function printError(error) {
}
}
/**
* TODOC
* @param {*} handlers
* @param {*} argv
* @returns
*/
function invoke(handlers, argv) {
let promises = [];
if (handlers) {
@ -39,6 +47,12 @@ function invoke(handlers, argv) {
return Promise.all(promises);
}
/**
* TODOC
* @param {*} eventName
* @param {*} argv
* @returns
*/
function broadcastEvent(eventName, argv) {
let promises = [];
for (let process of Object.values(gProcesses)) {
@ -49,6 +63,11 @@ function broadcastEvent(eventName, argv) {
return Promise.all(promises);
}
/**
* TODOC
* @param {*} message
* @returns
*/
function broadcast(message) {
let sender = this;
let promises = [];
@ -65,6 +84,12 @@ function broadcast(message) {
return Promise.all(promises);
}
/**
* TODOC
* @param {String} eventName
* @param {*} argv
* @returns
*/
function broadcastAppEventToUser(
user,
packageOwner,
@ -87,6 +112,12 @@ function broadcastAppEventToUser(
return Promise.all(promises);
}
/**
* TODOC
* @param {*} caller
* @param {*} process
* @returns
*/
function getUser(caller, process) {
return {
key: process.key,
@ -97,6 +128,12 @@ function getUser(caller, process) {
};
}
/**
* TODOC
* @param {*} user
* @param {*} process
* @returns
*/
async function getApps(user, process) {
if (
process.credentials &&
@ -123,13 +160,28 @@ async function getApps(user, process) {
return {};
}
/**
* TODOC
* @param {*} from
* @param {*} to
* @param {*} message
* @returns
*/
function postMessageInternal(from, to, message) {
if (to.eventHandlers['message']) {
return invoke(to.eventHandlers['message'], [getUser(from, from), message]);
}
}
/**
* TODOC
* @param {*} blobId
* @param {*} key
* @param {*} options
* @returns
*/
async function getProcessBlob(blobId, key, options) {
// TODO(tasiaiso): break this down ?
let process = gProcesses[key];
if (!process && !(options && 'create' in options && !options.create)) {
let resolveReady;
@ -148,7 +200,7 @@ async function getProcessBlob(blobId, key, options) {
}
process.lastActive = Date.now();
process.lastPing = null;
process.timeout = k_ping_interval;
process.timeout = kPingInterval;
process.ready = new Promise(function (resolve, reject) {
resolveReady = resolve;
rejectReady = reject;
@ -596,24 +648,16 @@ async function getProcessBlob(blobId, key, options) {
permissions: await imports.core.permissionsGranted(),
});
};
process.client_api = {
createIdentity: function () {
return process.createIdentity();
},
resetPermission: async function resetPermission(message) {
let user = process?.credentials?.session?.name;
await ssb.setUserPermission(
user,
options?.packageOwner,
options?.packageName,
message.permission,
undefined
);
return process.sendPermissions();
},
setActiveIdentity: function setActiveIdentity(message) {
return process.setActiveIdentity(message.identity);
},
process.resetPermission = async function resetPermission(permission) {
let user = process?.credentials?.session?.name;
await ssb.setUserPermission(
user,
options?.packageOwner,
options?.packageName,
permission,
undefined
);
return process.sendPermissions();
};
ssb.registerImports(imports, process);
process.task.setImports(imports);
@ -679,6 +723,9 @@ ssb.addEventListener('connections', function () {
broadcastEvent('onConnectionsChanged', []);
});
/**
* TODOC
*/
async function loadSettings() {
let data = {};
try {
@ -697,6 +744,9 @@ async function loadSettings() {
return data;
}
/**
* TODOC
*/
function sendStats() {
let apps = Object.values(gProcesses)
.filter((process) => process.app)
@ -712,6 +762,8 @@ function sendStats() {
}
}
let g_handler_index = 0;
exports.callAppHandler = async function callAppHandler(
response,
app_blob_id,