Add ActionManager class for object action tracking and websocket integration
- Implemented ActionManager to handle object updates using Etcd and broadcast events via websockets. - Added methods for subscribing to object actions, removing listeners, and sending object actions with callback handling. - Integrated logging for better traceability of actions and errors.
This commit is contained in:
parent
110f6d9a57
commit
75ccd91b50
100
src/actions/actionmanager.js
Normal file
100
src/actions/actionmanager.js
Normal file
@ -0,0 +1,100 @@
|
||||
import log4js from 'log4js';
|
||||
import { loadConfig } from '../config.js';
|
||||
import { etcdServer } from '../database/etcd.js';
|
||||
import { generateEtcId } from '../utils.js';
|
||||
|
||||
const config = loadConfig();
|
||||
|
||||
// Setup logger
|
||||
const logger = log4js.getLogger('Action Manager');
|
||||
logger.level = config.server.logLevel;
|
||||
|
||||
/**
|
||||
* ActionManager handles tracking object updates using Etcd and broadcasts update events via websockets.
|
||||
*/
|
||||
export class ActionManager {
|
||||
constructor(socketClient) {
|
||||
this.socketClient = socketClient;
|
||||
this.callbacks = new Map();
|
||||
}
|
||||
|
||||
async subscribeToObjectActions(id, objectType) {
|
||||
logger.debug('Subscribing to object actions...', id, objectType);
|
||||
await etcdServer.onPrefixPutEvent(
|
||||
`/${objectType}s/${id}/actions`,
|
||||
this.socketClient.id,
|
||||
(key, value) => {
|
||||
if (!value?.result) {
|
||||
logger.trace('Object action:', id);
|
||||
this.socketClient.socket.emit(
|
||||
'objectAction',
|
||||
{
|
||||
_id: id,
|
||||
objectType: objectType,
|
||||
action: { ...value }
|
||||
},
|
||||
result => {
|
||||
logger.trace('Got action result:', key);
|
||||
const actionId = key.split('/').pop();
|
||||
etcdServer.setKey(`/${objectType}s/${id}/actions/${actionId}`, {
|
||||
...value,
|
||||
result: { ...result }
|
||||
});
|
||||
}
|
||||
);
|
||||
}
|
||||
}
|
||||
);
|
||||
return { success: true };
|
||||
}
|
||||
|
||||
async removeObjectActionsListener(id, objectType) {
|
||||
await etcdServer.removePrefixWatcher(
|
||||
`/${objectType}s/${id}/actions`,
|
||||
this.socketClient.id,
|
||||
'put'
|
||||
);
|
||||
return { success: true };
|
||||
}
|
||||
|
||||
async sendObjectAction(id, objectType, action, callback) {
|
||||
try {
|
||||
const actionId = generateEtcId();
|
||||
this.callbacks.set(actionId, callback);
|
||||
logger.trace(
|
||||
`Calling action id: ${actionId}, object id: ${id}, object type: ${objectType} Action:`,
|
||||
action
|
||||
);
|
||||
await etcdServer.onKeyPutEvent(
|
||||
`/${objectType}s/${id}/actions/${actionId}`,
|
||||
this.socketClient.socketId,
|
||||
async (key, value) => {
|
||||
if (value.result) {
|
||||
logger.trace('Calling result callback...');
|
||||
const storedCallback = this.callbacks.get(actionId);
|
||||
await etcdServer.removeKeyWatcher(
|
||||
`/${objectType}s/${id}/actions/${actionId}`,
|
||||
this.socketClient.socketId,
|
||||
'put'
|
||||
);
|
||||
await etcdServer.deleteKey(
|
||||
`/${objectType}s/${id}/actions/${actionId}`
|
||||
);
|
||||
storedCallback(value.result);
|
||||
}
|
||||
}
|
||||
);
|
||||
await etcdServer.setKey(
|
||||
`/${objectType}s/${id}/actions/${actionId}`,
|
||||
action
|
||||
);
|
||||
return true;
|
||||
} catch (error) {
|
||||
logger.error(
|
||||
`Failed to set value for /${objectType}s/${id}/object:`,
|
||||
error
|
||||
);
|
||||
return false;
|
||||
}
|
||||
}
|
||||
}
|
||||
Loading…
x
Reference in New Issue
Block a user