fix(core): update
This commit is contained in:
parent
c99ec50853
commit
c7c9acf5bd
@ -31,7 +31,7 @@
|
||||
"@push.rocks/smartstring": "^4.0.7",
|
||||
"@push.rocks/smarttime": "^4.0.1",
|
||||
"@push.rocks/smartunique": "^3.0.3",
|
||||
"@push.rocks/taskbuffer": "^3.1.0",
|
||||
"@push.rocks/taskbuffer": "^3.1.2",
|
||||
"@tsclass/tsclass": "^4.0.42",
|
||||
"mongodb": "^5.7.0"
|
||||
},
|
||||
@ -40,8 +40,8 @@
|
||||
"@gitzone/tsrun": "^1.2.44",
|
||||
"@gitzone/tstest": "^1.0.77",
|
||||
"@push.rocks/qenv": "^6.0.2",
|
||||
"@push.rocks/tapbundle": "^5.0.8",
|
||||
"@types/node": "^20.4.9",
|
||||
"@push.rocks/tapbundle": "^5.0.15",
|
||||
"@types/node": "^20.4.10",
|
||||
"@types/shortid": "0.0.29"
|
||||
},
|
||||
"files": [
|
||||
|
1141
pnpm-lock.yaml
generated
1141
pnpm-lock.yaml
generated
File diff suppressed because it is too large
Load Diff
104
test/test.distributedcoordinator.ts
Normal file
104
test/test.distributedcoordinator.ts
Normal file
@ -0,0 +1,104 @@
|
||||
import { tap, expect } from '@push.rocks/tapbundle';
|
||||
import * as smartmongo from '@push.rocks/smartmongo';
|
||||
import type * as taskbuffer from '@push.rocks/taskbuffer';
|
||||
|
||||
import * as smartdata from '../ts/index.js';
|
||||
import { SmartdataDistributedCoordinator, DistributedClass } from '../ts/smartdata.classes.distributedcoordinator.js'; // path might need adjusting
|
||||
const totalInstances = 10;
|
||||
|
||||
// =======================================
|
||||
// Connecting to the database server
|
||||
// =======================================
|
||||
|
||||
let smartmongoInstance: smartmongo.SmartMongo;
|
||||
let testDb: smartdata.SmartdataDb;
|
||||
|
||||
tap.test('should create a testinstance as database', async () => {
|
||||
smartmongoInstance = await smartmongo.SmartMongo.createAndStart();
|
||||
testDb = new smartdata.SmartdataDb(await smartmongoInstance.getMongoDescriptor());
|
||||
await testDb.init();
|
||||
});
|
||||
|
||||
tap.test('should instantiate DistributedClass', async (tools) => {
|
||||
const instance = new DistributedClass();
|
||||
expect(instance).toBeInstanceOf(DistributedClass);
|
||||
});
|
||||
|
||||
tap.test('DistributedClass should update the time', async (tools) => {
|
||||
const distributedCoordinator = new SmartdataDistributedCoordinator(testDb);
|
||||
await distributedCoordinator.start();
|
||||
console.log('test: started');
|
||||
const initialTime = distributedCoordinator.ownInstance.data.lastUpdated;
|
||||
await distributedCoordinator.sendHeartbeat();
|
||||
console.log('test: sent heartbeat');
|
||||
const updatedTime = distributedCoordinator.ownInstance.data.lastUpdated;
|
||||
expect(updatedTime).toBeGreaterThan(initialTime);
|
||||
await distributedCoordinator.stop();
|
||||
});
|
||||
|
||||
tap.test('should instantiate SmartdataDistributedCoordinator', async (tools) => {
|
||||
const coordinator = new SmartdataDistributedCoordinator(testDb);
|
||||
await coordinator.start();
|
||||
expect(coordinator).toBeInstanceOf(SmartdataDistributedCoordinator);
|
||||
await coordinator.stop();
|
||||
});
|
||||
|
||||
tap.test('SmartdataDistributedCoordinator should update leader status', async (tools) => {
|
||||
const coordinator = new SmartdataDistributedCoordinator(testDb);
|
||||
await coordinator.start();
|
||||
await coordinator.checkAndMaybeLead();
|
||||
expect(coordinator.ownInstance.data.elected).toBeOneOf([true, false]);
|
||||
await coordinator.stop();
|
||||
process.exit(0);
|
||||
});
|
||||
|
||||
tap.test('SmartdataDistributedCoordinator should handle distributed task requests', async (tools) => {
|
||||
const coordinator = new SmartdataDistributedCoordinator(testDb);
|
||||
await coordinator.start();
|
||||
|
||||
const mockTaskRequest: taskbuffer.distributedCoordination.IDistributedTaskRequest = {
|
||||
submitterId: "mockSubmitter12345", // Some unique mock submitter ID
|
||||
taskName: "SampleTask",
|
||||
taskVersion: "1.0.0", // Assuming it's a version string
|
||||
taskExecutionTime: Date.now(),
|
||||
taskExecutionTimeout: 60000, // Let's say the timeout is 1 minute (60000 ms)
|
||||
taskExecutionParallel: 5, // Let's assume max 5 parallel executions
|
||||
status: 'requesting'
|
||||
};
|
||||
|
||||
const response = await coordinator.fireDistributedTaskRequest(mockTaskRequest);
|
||||
expect(response).toBeTruthy(); // based on your expected structure for the response
|
||||
await coordinator.stop();
|
||||
});
|
||||
|
||||
tap.test('SmartdataDistributedCoordinator should update distributed task requests', async (tools) => {
|
||||
const coordinator = new SmartdataDistributedCoordinator(testDb);
|
||||
|
||||
await coordinator.start();
|
||||
|
||||
const mockTaskRequest: taskbuffer.distributedCoordination.IDistributedTaskRequest = {
|
||||
submitterId: "mockSubmitter12345", // Some unique mock submitter ID
|
||||
taskName: "SampleTask",
|
||||
taskVersion: "1.0.0", // Assuming it's a version string
|
||||
taskExecutionTime: Date.now(),
|
||||
taskExecutionTimeout: 60000, // Let's say the timeout is 1 minute (60000 ms)
|
||||
taskExecutionParallel: 5, // Let's assume max 5 parallel executions
|
||||
status: 'requesting'
|
||||
};
|
||||
|
||||
|
||||
await coordinator.updateDistributedTaskRequest(mockTaskRequest);
|
||||
// Here, we can potentially check if a DB entry got updated or some other side-effect of the update method.
|
||||
});
|
||||
|
||||
tap.test('should elect only one leader amongst multiple instances', async (tools) => {
|
||||
const coordinators = Array.from({ length: totalInstances }).map(() => new SmartdataDistributedCoordinator(testDb));
|
||||
await Promise.all(coordinators.map(coordinator => coordinator.init()));
|
||||
|
||||
await Promise.all(coordinators.map(coordinator => coordinator.checkAndMaybeLead()));
|
||||
|
||||
const leaders = coordinators.filter(coordinator => coordinator.ownInstance.data.elected);
|
||||
expect(leaders.length).toEqual(1);
|
||||
});
|
||||
|
||||
tap.start({ throwOnError: true });
|
@ -30,7 +30,7 @@ tap.test('should create a testinstance as database', async () => {
|
||||
tap.skip.test('should connect to atlas', async (tools) => {
|
||||
const databaseName = `test-smartdata-${smartunique.shortId()}`;
|
||||
testDb = new smartdata.SmartdataDb({
|
||||
mongoDbUrl: testQenv.getEnvVarOnDemand('MONGO_URL'),
|
||||
mongoDbUrl: await testQenv.getEnvVarOnDemand('MONGO_URL'),
|
||||
mongoDbName: databaseName,
|
||||
});
|
||||
await testDb.init();
|
||||
|
@ -28,7 +28,7 @@ tap.test('should create a testinstance as database', async () => {
|
||||
tap.skip.test('should connect to atlas', async (tools) => {
|
||||
const databaseName = `test-smartdata-${smartunique.shortId()}`;
|
||||
testDb = new smartdata.SmartdataDb({
|
||||
mongoDbUrl: testQenv.getEnvVarOnDemand('MONGO_URL'),
|
||||
mongoDbUrl: await testQenv.getEnvVarOnDemand('MONGO_URL'),
|
||||
mongoDbName: databaseName,
|
||||
});
|
||||
await testDb.init();
|
||||
|
@ -3,6 +3,6 @@
|
||||
*/
|
||||
export const commitinfo = {
|
||||
name: '@push.rocks/smartdata',
|
||||
version: '5.0.24',
|
||||
version: '5.0.25',
|
||||
description: 'do more with data'
|
||||
}
|
||||
|
@ -4,7 +4,7 @@ import { Manager, setDefaultManagerForDoc } from './smartdata.classes.collection
|
||||
import { SmartDataDbDoc, svDb, unI } from './smartdata.classes.doc.js';
|
||||
|
||||
@Manager()
|
||||
class DistributedClass extends SmartDataDbDoc<DistributedClass, DistributedClass> {
|
||||
export class DistributedClass extends SmartDataDbDoc<DistributedClass, DistributedClass> {
|
||||
// INSTANCE
|
||||
@unI()
|
||||
public id: string;
|
||||
@ -37,14 +37,13 @@ export class SmartdataDistributedCoordinator extends plugins.taskbuffer.distribu
|
||||
public readyPromise: Promise<any>;
|
||||
public db: SmartdataDb;
|
||||
private asyncExecutionStack = new plugins.lik.AsyncExecutionStack();
|
||||
private ownInstance: DistributedClass;
|
||||
public ownInstance: DistributedClass;
|
||||
|
||||
constructor(dbArg?: SmartdataDb) {
|
||||
constructor(dbArg: SmartdataDb) {
|
||||
super();
|
||||
this.db = dbArg;
|
||||
setDefaultManagerForDoc(this, DistributedClass);
|
||||
this.readyPromise = this.db.statusConnectedDeferred.promise;
|
||||
this.init();
|
||||
}
|
||||
|
||||
// smartdata specific stuff
|
||||
@ -64,15 +63,19 @@ export class SmartdataDistributedCoordinator extends plugins.taskbuffer.distribu
|
||||
|
||||
private startHeartbeat = async () => {
|
||||
while (this.ownInstance.data.status !== 'stopped') {
|
||||
await this.sendHeartbeat();
|
||||
await plugins.smartdelay.delayFor(10000);
|
||||
}
|
||||
};
|
||||
|
||||
public async sendHeartbeat() {
|
||||
await this.asyncExecutionStack.getExclusiveExecutionSlot(async () => {
|
||||
await this.ownInstance.updateFromDb();
|
||||
this.ownInstance.data.lastUpdated = Date.now();
|
||||
await this.ownInstance.save();
|
||||
});
|
||||
await plugins.smartdelay.delayFor(10000);
|
||||
}
|
||||
};
|
||||
public async init() {
|
||||
private async init() {
|
||||
await this.readyPromise;
|
||||
if (!this.ownInstance) {
|
||||
await this.asyncExecutionStack.getExclusiveExecutionSlot(async () => {
|
||||
@ -86,7 +89,9 @@ export class SmartdataDistributedCoordinator extends plugins.taskbuffer.distribu
|
||||
taskRequestResults: [],
|
||||
};
|
||||
await this.ownInstance.save();
|
||||
console.log('saved own instance.')
|
||||
});
|
||||
console.log('yep, we are here')
|
||||
}
|
||||
|
||||
// lets enable the heartbeat
|
||||
|
Loading…
Reference in New Issue
Block a user