fix(core): update

This commit is contained in:
Philipp Kunz 2023-08-16 12:08:27 +02:00
parent 54f529b0a7
commit 7f9983382a
3 changed files with 33 additions and 6 deletions

View File

@ -55,6 +55,7 @@ tap.test('SmartdataDistributedCoordinator should handle distributed task request
const mockTaskRequest: taskbuffer.distributedCoordination.IDistributedTaskRequest = { const mockTaskRequest: taskbuffer.distributedCoordination.IDistributedTaskRequest = {
submitterId: "mockSubmitter12345", // Some unique mock submitter ID submitterId: "mockSubmitter12345", // Some unique mock submitter ID
requestResponseId: 'uni879873462hjhfkjhsdf', // Some unique ID for the request-response
taskName: "SampleTask", taskName: "SampleTask",
taskVersion: "1.0.0", // Assuming it's a version string taskVersion: "1.0.0", // Assuming it's a version string
taskExecutionTime: Date.now(), taskExecutionTime: Date.now(),
@ -105,7 +106,7 @@ tap.test('should elect only one leader amongst multiple instances', async (tools
tap.test('should clean up', async () => { tap.test('should clean up', async () => {
await smartmongoInstance.stopAndDumpToDir(`.nogit/testdata/`); await smartmongoInstance.stopAndDumpToDir(`.nogit/testdata/`);
setTimeout(() => process.exit(0), 1000); setTimeout(() => process.exit(), 2000);
}) })
tap.start({ throwOnError: true }); tap.start({ throwOnError: true });

View File

@ -3,6 +3,6 @@
*/ */
export const commitinfo = { export const commitinfo = {
name: '@push.rocks/smartdata', name: '@push.rocks/smartdata',
version: '5.0.30', version: '5.0.31',
description: 'do more with data' description: 'do more with data'
} }

View File

@ -2,6 +2,7 @@ import * as plugins from './smartdata.plugins.js';
import { SmartdataDb } from './smartdata.classes.db.js'; import { SmartdataDb } from './smartdata.classes.db.js';
import { Manager, setDefaultManagerForDoc } from './smartdata.classes.collection.js'; import { Manager, setDefaultManagerForDoc } from './smartdata.classes.collection.js';
import { SmartDataDbDoc, svDb, unI } from './smartdata.classes.doc.js'; import { SmartDataDbDoc, svDb, unI } from './smartdata.classes.doc.js';
import { SmartdataDbWatcher } from './smartdata.classes.watcher.js';
@Manager() @Manager()
export class DistributedClass extends SmartDataDbDoc<DistributedClass, DistributedClass> { export class DistributedClass extends SmartDataDbDoc<DistributedClass, DistributedClass> {
@ -39,6 +40,7 @@ export class SmartdataDistributedCoordinator extends plugins.taskbuffer.distribu
public db: SmartdataDb; public db: SmartdataDb;
private asyncExecutionStack = new plugins.lik.AsyncExecutionStack(); private asyncExecutionStack = new plugins.lik.AsyncExecutionStack();
public ownInstance: DistributedClass; public ownInstance: DistributedClass;
public distributedWatcher: SmartdataDbWatcher<DistributedClass>;
constructor(dbArg: SmartdataDb) { constructor(dbArg: SmartdataDb) {
super(); super();
@ -55,6 +57,7 @@ export class SmartdataDistributedCoordinator extends plugins.taskbuffer.distribu
public async stop() { public async stop() {
await this.asyncExecutionStack.getExclusiveExecutionSlot(async () => { await this.asyncExecutionStack.getExclusiveExecutionSlot(async () => {
if (this.ownInstance?.data.elected) { if (this.ownInstance?.data.elected) {
await this.distributedWatcher.close();
this.ownInstance.data.elected = false; this.ownInstance.data.elected = false;
} }
if (this.ownInstance?.data.status === 'stopped') { if (this.ownInstance?.data.status === 'stopped') {
@ -208,12 +211,25 @@ export class SmartdataDistributedCoordinator extends plugins.taskbuffer.distribu
* the leading is implemented here * the leading is implemented here
*/ */
public async leadFunction() { public async leadFunction() {
const watcher = await DistributedClass.watch({}); this.distributedWatcher = await DistributedClass.watch({});
watcher.changeSubject.subscribe({
const currentTaskRequests: Array<{
taskName: string;
taskExecutionTime: number;
/**
* all instances that requested this task
*/
requestingDistibutedInstanceIds: string[];
responseTimeout: plugins.smartdelay.Timeout<any>;
}> = [];
this.distributedWatcher.changeSubject.subscribe({
next: async (distributedDoc) => { next: async (distributedDoc) => {
console.log(`registered change for distributed doc ${distributedDoc.id}`);
distributedDoc; distributedDoc;
}, },
}); });
while (this.ownInstance.data.status !== 'stopped' && this.ownInstance.data.elected) { while (this.ownInstance.data.status !== 'stopped' && this.ownInstance.data.elected) {
const allInstances = await DistributedClass.getInstances({}); const allInstances = await DistributedClass.getInstances({});
for (const instance of allInstances) { for (const instance of allInstances) {
@ -239,9 +255,19 @@ export class SmartdataDistributedCoordinator extends plugins.taskbuffer.distribu
await this.ownInstance.save(); await this.ownInstance.save();
}); });
await plugins.smartdelay.delayFor(10000); await plugins.smartdelay.delayFor(10000);
const result = await this.asyncExecutionStack.getExclusiveExecutionSlot(async () => {
await this.ownInstance.updateFromDb();
const taskRequestResult = this.ownInstance.data.taskRequestResults.find((resultItem) => {
return resultItem.requestResponseId === taskRequestArg.requestResponseId;
});
return taskRequestResult;
});
if (!result) {
console.warn('no result found for task request...');
return null; return null;
} }
return result;
}
public async updateDistributedTaskRequest( public async updateDistributedTaskRequest(
infoBasisArg: plugins.taskbuffer.distributedCoordination.IDistributedTaskRequest infoBasisArg: plugins.taskbuffer.distributedCoordination.IDistributedTaskRequest