First push of async task framework with import tags
ADD
src/async-task/AsyncTask.ts
+import AsyncTask from '../types/AsyncTask';
+import Constants from '../utils/Constants';
+import Logging from '../utils/Logging';
+import { ServerAction } from '../types/Server';
+import moment from 'moment';
+
+const MODULE_NAME = 'AbstractAsyncTask';
+
+export default abstract class AbstractAsyncTask {
+ protected asyncTask: AsyncTask;
+
+ public constructor(asyncTask: AsyncTask) {
+ this.asyncTask = asyncTask;
+ }
+
+ public async run(): Promise<void> {
+ const startTime = moment();
+ await Logging.logInfo({
+ tenantID: Constants.DEFAULT_TENANT,
+ action: ServerAction.ASYNC_TASK,
+ module: MODULE_NAME, method: 'run',
+ message: `The async task '${this.asyncTask.name}' is running...`
+ });
+ // Execute
+ await this.executeAsyncTask();
+ // Log Total Processing Time
+ const totalMigrationTimeSecs = moment.duration(moment().diff(startTime)).asSeconds();
+ await Logging.logInfo({
+ tenantID: Constants.DEFAULT_TENANT,
+ action: ServerAction.SCHEDULER,
+ module: MODULE_NAME, method: 'run',
+ message: `The task '${this.asyncTask.name}' has been run in ${totalMigrationTimeSecs} secs`
+ });
+ }
+
+ protected abstract executeAsyncTask(): Promise<void>;
+}

ADD
src/async-task/AsyncTaskManager.ts
+import AsyncTask, { AsyncTaskStatus, AsyncTasks } from '../types/AsyncTask';
+
+import AbstractAsyncTask from './AsyncTask';
+import { ActionsResponse } from '../types/GlobalType';
+import AsyncTaskStorage from '../storage/mongodb/AsyncTaskStorage';
+import Configuration from '../utils/Configuration';
+import Constants from '../utils/Constants';
+import ImportTagsAsyncTask from './tasks/TagsImportAsyncTask';
+import LockingHelper from '../locking/LockingHelper';
+import LockingManager from '../locking/LockingManager';
+import Logging from '../utils/Logging';
+import { ServerAction } from '../types/Server';
+import User from '../types/User';
+import UserToken from '../types/UserToken';
+import Utils from '../utils/Utils';
+
+const MODULE_NAME = 'AsyncTaskManager';
+
+export default class AsyncTaskManager {
+ private static asyncTaskConfig = Configuration.getAsyncTaskConfig();
+
+ public static async handleAsyncTasks(): Promise<void> {
+ // Active?
+ if (AsyncTaskManager.asyncTaskConfig.active) {
+ await Logging.logDebug({
+ tenantID: Constants.DEFAULT_TENANT,
+ action: ServerAction.ASYNC_TASK,
+ module: MODULE_NAME, method: 'handleAsyncTasks',
+ message: 'Checking if there is async task to process...'
+ });
+ const processedTask: ActionsResponse = {
+ inError: 0,
+ inSuccess: 0,
+ };
+ const startTime = new Date().getTime();
+ // Handle number of instances
+ let nbrTasksInParallel = 1;
+ if (this.asyncTaskConfig.nbrTasksInParallel > 0) {
+ nbrTasksInParallel = this.asyncTaskConfig.nbrTasksInParallel;
+ }
+ // Get the tasks
+ const asyncTasks = await AsyncTaskStorage.getAsyncTasks( { status: AsyncTaskStatus.NEW }, Constants.DB_PARAMS_MAX_LIMIT);
+ // Process them
+ let abstractAsyncTask: AbstractAsyncTask;
+ if (!Utils.isEmptyArray(asyncTasks.result)) {
+ await Promise.map(asyncTasks.result,
+ async (asyncTask: AsyncTask) => {
+ // Tasks
+ switch (asyncTask.name) {
+ case AsyncTasks.TAGS_IMPORT:
+ abstractAsyncTask = new ImportTagsAsyncTask(asyncTask);
+ break;
+ default:
+ await Logging.logError({
+ tenantID: Constants.DEFAULT_TENANT,
+ action: ServerAction.ASYNC_TASK,
+ module: MODULE_NAME, method: 'handleAsyncTasks',
+ message: `The async task '${asyncTask.name}' is unknown`
+ });
+ }
+ if (abstractAsyncTask) {
+ // Get the lock
+ const asyncTaskLock = await LockingHelper.createAsyncTaskLock(Constants.DEFAULT_TENANT, asyncTask);
+ if (asyncTaskLock) {
+ try {
+ const startAsyncTaskTime = new Date().getTime();
+ asyncTask.execTimestamp = new Date();
+ asyncTask.execHost = Utils.getHostname();
+ // Log
+ await Logging.logInfo({
+ tenantID: Constants.DEFAULT_TENANT,
+ action: ServerAction.ASYNC_TASK,
+ module: MODULE_NAME, method: 'handleAsyncTasks',
+ message: `The task '${asyncTask.name}' is running...`
+ });
+ // Update the task
+ asyncTask.status = AsyncTaskStatus.RUNNING;
+ await AsyncTaskStorage.saveAsyncTask(asyncTask);
+ // Run
+ await abstractAsyncTask.run();
+ // Update the task
+ const asyncTaskTotalDurationSecs = Math.trunc((new Date().getTime() - startAsyncTaskTime) / 1000);
+ asyncTask.status = AsyncTaskStatus.SUCCESS;
+ asyncTask.execTimeSecs = asyncTaskTotalDurationSecs;
+ await AsyncTaskStorage.saveAsyncTask(asyncTask);
+ processedTask.inSuccess++;
+ // Log
+ await Logging.logInfo({
+ tenantID: Constants.DEFAULT_TENANT,
+ action: ServerAction.ASYNC_TASK,
+ module: MODULE_NAME, method: 'handleAsyncTasks',
+ message: `The task '${asyncTask.name}' has been processed in ${asyncTaskTotalDurationSecs} secs`
+ });
+ } catch (error) {
+ processedTask.inError++;
+ // Update the task
+ asyncTask.status = AsyncTaskStatus.ERROR;
+ asyncTask.message = error.message;
+ await AsyncTaskStorage.saveAsyncTask(asyncTask);
+ // Log error
+ Logging.logError({
+ tenantID: Constants.DEFAULT_TENANT,
+ module: MODULE_NAME, method: 'handleAsyncTasks',
+ action: ServerAction.ASYNC_TASK,
+ message: `Error while running the async task '${asyncTask.name}': ${error.message}`,
+ detailedMessages: { error: error.message, stack: error.stack, asyncTask }
+ });
+ } finally {
+ // Release lock
+ await LockingManager.release(asyncTaskLock);
+ }
+ }
+ }
+ },
+ { concurrency: nbrTasksInParallel }).then(() => {
+ // Log result
+ const totalDurationSecs = Math.trunc((new Date().getTime() - startTime) / 1000);
+ void Logging.logActionsResponse(Constants.DEFAULT_TENANT, ServerAction.ASYNC_TASK,
+ MODULE_NAME, 'handleAsyncTasks', processedTask,
+ `{{inSuccess}} async task(s) were successfully processed in ${totalDurationSecs} secs`,
+ `{{inError}} async task(s) failed to be processed in ${totalDurationSecs} secs`,
+ `{{inSuccess}} async task(s) were successfully processed in ${totalDurationSecs} secs and {{inError}} failed`,
+ `No async task to process`
+ );
+ });
+ }
+ }
+ }
+
+ public static async createAndSaveAsyncTasks(asyncTask: Omit<AsyncTask, 'id'>, user?: UserToken|User|string): Promise<void> {
+ // Check
+ if (Utils.isNullOrUndefined(asyncTask)) {
+ throw new Error("The async task must not be null");
+ }
+ // Check
+ if (Utils.isNullOrUndefined(asyncTask.name)) {
+ throw new Error("The Name of the async task is mandatory");
+ }
+ if (Utils.isNullOrUndefined(asyncTask.tenantID)) {
+ throw new Error("The Tenant ID of the async task is mandatory");
+ }
+ if (!Utils.isNullOrUndefined(asyncTask.parameters) && (typeof asyncTask.parameters !== 'object')) {
+ throw new Error("The Parameters of the async task must be a Json document");
+ }
+ // Set
+ asyncTask.status = AsyncTaskStatus.NEW;
+ asyncTask.createdBy = user as User ?? null;
+ asyncTask.createdOn = new Date();
+ // Save
+ await AsyncTaskStorage.saveAsyncTask(asyncTask as AsyncTask);
+ // Log
+ await Logging.logInfo({
+ tenantID: Constants.DEFAULT_TENANT,
+ action: ServerAction.ASYNC_TASK,
+ module: MODULE_NAME, method: 'createAndSaveAsyncTasks',
+ message: `The task '${asyncTask.name}' has been saved successfully and will be processed soon`
+ });
+ }
+}

ADD
src/async-task/tasks/TagsImportAsyncTask.ts
+import { ActionsResponse, ImportStatus } from '../../types/GlobalType';
+import Tag, { ImportedTag } from '../../types/Tag';
+
+import AbstractAsyncTask from '../AsyncTask';
+import Constants from '../../utils/Constants';
+import { DataResult } from '../../types/DataResult';
+import DbParams from '../../types/database/DbParams';
+import LockingHelper from '../../locking/LockingHelper';
+import LockingManager from '../../locking/LockingManager';
+import Logging from '../../utils/Logging';
+import { ServerAction } from '../../types/Server';
+import TagStorage from '../../storage/mongodb/TagStorage';
+import TenantStorage from '../../storage/mongodb/TenantStorage';
+import Utils from '../../utils/Utils';
+
+const MODULE_NAME = 'TagsImportAsyncTask';
+
+export default class TagsImportAsyncTask extends AbstractAsyncTask {
+ protected async executeAsyncTask(): Promise<void> {
+ const importTagsLock = await LockingHelper.createImportTagsLock(this.asyncTask.tenantID);
+ if (importTagsLock) {
+ const tenant = await TenantStorage.getTenant(this.asyncTask.tenantID);
+ try {
+ const dbParams: DbParams = { limit: Constants.IMPORT_PAGE_SIZE, skip: 0 };
+ let importedTags: DataResult<ImportedTag>;
+ const result: ActionsResponse = {
+ inError: 0,
+ inSuccess: 0,
+ };
+ const startTime = new Date().getTime();
+ // Get total number of Tags to import
+ const totalTagsToImport = await TagStorage.getImportedTagsCount(tenant.id);
+ if (totalTagsToImport > 0) {
+ await Logging.logInfo({
+ tenantID: tenant.id,
+ action: ServerAction.TAGS_IMPORT,
+ module: MODULE_NAME, method: 'processTenant',
+ message: `${totalTagsToImport} Tag(s) are going to be imported...`
+ });
+ }
+ do {
+ // Get the imported tags
+ importedTags = await TagStorage.getImportedTags(tenant.id, { status: ImportStatus.READY }, dbParams);
+ for (const importedTag of importedTags.result) {
+ try {
+ // Existing tags
+ const foundTag = await TagStorage.getTag(tenant.id, importedTag.id);
+ if (foundTag) {
+ // Check tag is already in use
+ if (!foundTag.issuer) {
+ throw new Error('Tag is not local to the organization');
+ }
+ if (foundTag.deleted) {
+ throw new Error('Tag is deleted');
+ }
+ if (foundTag.userID) {
+ throw new Error('Tag is already assigned to an user');
+ }
+ if (foundTag.active) {
+ throw new Error('Tag is already active');
+ }
+ // Update it
+ await TagStorage.saveTag(tenant.id, { ...foundTag, ...importedTag });
+ // Remove the imported Tag
+ await TagStorage.deleteImportedTag(tenant.id, importedTag.id);
+ result.inSuccess++;
+ continue;
+ }
+ // New Tag
+ const tag: Tag = {
+ id: importedTag.id,
+ description: importedTag.description,
+ issuer: true,
+ deleted: false,
+ active: false,
+ createdBy: { id: importedTag.importedBy },
+ createdOn: importedTag.importedOn,
+ };
+ // Save the new Tag
+ await TagStorage.saveTag(tenant.id, tag);
+ // Remove the imported Tag
+ await TagStorage.deleteImportedTag(tenant.id, importedTag.id);
+ result.inSuccess++;
+ } catch (error) {
+ // Update the imported Tag
+ importedTag.status = ImportStatus.ERROR;
+ importedTag.errorDescription = error.message;
+ result.inError++;
+ // Update it
+ await TagStorage.saveImportedTag(tenant.id, importedTag);
+ // Log
+ await Logging.logError({
+ tenantID: tenant.id,
+ action: ServerAction.TAGS_IMPORT,
+ module: MODULE_NAME, method: 'processTenant',
+ message: `Error when importing Tag ID '${importedTag.id}': ${error.message}`,
+ detailedMessages: { tag: importedTag, error: error.message, stack: error.stack }
+ });
+ }
+ }
+ // Log
+ if (importedTags.result.length > 0 && (result.inError + result.inSuccess) > 0) {
+ const intermediateDurationSecs = Math.round((new Date().getTime() - startTime) / 1000);
+ await Logging.logDebug({
+ tenantID: tenant.id,
+ action: ServerAction.TAGS_IMPORT,
+ module: MODULE_NAME, method: 'processTenant',
+ message: `${result.inError + result.inSuccess}/${totalTagsToImport} Tag(s) have been processed in ${intermediateDurationSecs}s...`
+ });
+ }
+ } while (!Utils.isEmptyArray(importedTags?.result));
+ // Log final results
+ const executionDurationSecs = Math.round((new Date().getTime() - startTime) / 1000);
+ await Logging.logActionsResponse(tenant.id, ServerAction.TAGS_IMPORT, MODULE_NAME, 'processTenant', result,
+ `{{inSuccess}} Tag(s) have been imported successfully in ${executionDurationSecs}s in Tenant ${Utils.buildTenantName(tenant)}`,
+ `{{inError}} Tag(s) failed to be imported in ${executionDurationSecs}s in Tenant ${Utils.buildTenantName(tenant)}`,
+ `{{inSuccess}} Tag(s) have been imported successfully but {{inError}} failed in ${executionDurationSecs}s in Tenant ${Utils.buildTenantName(tenant)}`,
+ `Not Tag has been imported in ${executionDurationSecs}s in Tenant ${Utils.buildTenantName(tenant)}`
+ );
+ } catch (error) {
+ // Log error
+ await Logging.logActionExceptionMessage(tenant.id, ServerAction.TAGS_IMPORT, error);
+ } finally {
+ // Release the lock
+ await LockingManager.release(importTagsLock);
+ }
+ }
+ }
+}

MODIFY
src/locking/LockingHelper.ts
src/locking/LockingHelper.ts
import Lock, { LockEntity } from '../types/Locking';
import Asset from '../types/Asset';
+import AsyncTask from '../types/AsyncTask';
import LockingManager from './LockingManager';
import OCPIEndpoint from '../types/ocpi/OCPIEndpoint';
import OICPEndpoint from '../types/oicp/OICPEndpoint';
@@ -15,6 +16,14 @@ export default class LockingHelper {
return lock;
}
+ public static async createAsyncTaskLock(tenantID: string, asyncTask: AsyncTask): Promise<Lock | null> {
+ const lock = LockingManager.createExclusiveLock(tenantID, LockEntity.ASYNC_TASK, `${asyncTask.id}`);
+ if (!(await LockingManager.acquire(lock))) {
+ return null;
+ }
+ return lock;
+ }
+
public static async createSiteAreaSmartChargingLock(tenantID: string, siteArea: SiteArea): Promise<Lock | null> {
const lock = LockingManager.createExclusiveLock(tenantID, LockEntity.SITE_AREA, `${siteArea.id}-smart-charging`);
if (!(await LockingManager.acquire(lock))) {

MODIFY
src/migration/tasks/RecomputeAllTransactionsConsumptionsTask.ts
src/migration/tasks/RecomputeAllTransactionsConsumptionsTask.ts
@@ -28,7 +28,7 @@ export default class RecomputeAllTransactionsConsumptionsTask extends MigrationT
inError: 0,
inSuccess: 0,
};
- const timeTotalFrom = new Date().getTime();
+ const startTime = new Date().getTime();
// Get transactions
const transactionsMDB = await global.database.getCollection<any>(tenant.id, 'transactions')
.aggregate([
@@ -88,7 +88,7 @@ export default class RecomputeAllTransactionsConsumptionsTask extends MigrationT
});
}
}, { concurrency: 5 }).then(() => {
- const totalDurationSecs = Math.trunc((new Date().getTime() - timeTotalFrom) / 1000);
+ const totalDurationSecs = Math.trunc((new Date().getTime() - startTime) / 1000);
// Log in the default tenant
void Logging.logActionsResponse(Constants.DEFAULT_TENANT, ServerAction.MIGRATION,
MODULE_NAME, 'migrateTenant', consumptionsUpdated,

MODIFY
src/server/rest/CentralRestServer.ts
src/server/rest/CentralRestServer.ts
@@ -432,6 +432,22 @@ export default class CentralRestServer {
});
}
+ public notifyAsyncTaskEndpoint(tenantID: string, action: Action, data: NotificationData): void {
+ // Add in buffer
+ this.addSingleChangeNotificationInBuffer({
+ 'tenantID': tenantID,
+ 'entity': Entity.ASYNC_TASK,
+ 'action': action,
+ 'data': data
+ });
+ // Add in buffer
+ this.addChangeNotificationInBuffer({
+ 'tenantID': tenantID,
+ 'entity': Entity.ASYNC_TASKS,
+ 'action': action
+ });
+ }
+
private addChangeNotificationInBuffer(notification: ChangeNotification) {
let dups = false;
if (this.hasSocketIOClients(notification.tenantID)) {

MODIFY
src/server/rest/v1/service/TagService.ts
src/server/rest/v1/service/TagService.ts
import { Action, Entity } from '../../../../types/Authorization';
import { ActionsResponse, ImportStatus } from '../../../../types/GlobalType';
+import { AsyncTaskStatus, AsyncTaskType, AsyncTasks } from '../../../../types/AsyncTask';
import { HTTPAuthError, HTTPError } from '../../../../types/HTTPError';
import { NextFunction, Request, Response } from 'express';
import { OCPITokenType, OCPITokenWhitelist } from '../../../../types/ocpi/OCPIToken';
@@ -7,6 +8,7 @@ import Tag, { ImportedTag, TagRequiredImportProperties } from '../../../../types
import AppAuthError from '../../../../exception/AppAuthError';
import AppError from '../../../../exception/AppError';
+import AsyncTaskManager from '../../../../async-task/AsyncTaskManager';
import AuthorizationService from './AuthorizationService';
import Authorizations from '../../../../authorization/Authorizations';
import Busboy from 'busboy';
@@ -582,8 +584,15 @@ export default class TagService {
`{{inSuccess}} Tag(s) were successfully uploaded in ${executionDurationSecs}s and ready for asynchronous import and {{inError}} failed to be uploaded`,
`No Tag have been uploaded in ${executionDurationSecs}s`, req.user
);
- // Trigger manually and asynchronously the job
- void new ImportTagsTask().processTenant(req.tenant);
+ // Create and Save async task
+ AsyncTaskManager.createAndSaveAsyncTasks({
+ name: AsyncTasks.TAGS_IMPORT,
+ action: ServerAction.TAGS_IMPORT,
+ type: AsyncTaskType.TASK,
+ tenantID: req.tenant.id,
+ module: MODULE_NAME,
+ method: 'handleImportTags',
+ });
// Respond
res.json({ ...result, ...Constants.REST_RESPONSE_SUCCESS });
next();

ADD
src/storage/mongodb/AsyncTaskStorage.ts
+import AsyncTask, { AsyncTaskStatus } from '../../types/AsyncTask';
+import global, { FilterParams } from '../../types/GlobalType';
+
+import Constants from '../../utils/Constants';
+import Cypher from '../../utils/Cypher';
+import { DataResult } from '../../types/DataResult';
+import DatabaseUtils from './DatabaseUtils';
+import DbParams from '../../types/database/DbParams';
+import Logging from '../../utils/Logging';
+import Utils from '../../utils/Utils';
+
+const MODULE_NAME = 'AsyncTaskStorage';
+
+export default class AsyncTaskStorage {
+ public static async getAsyncTask(id: string = Constants.UNKNOWN_OBJECT_ID,
+ params: {} = {}, projectFields?: string[]): Promise<AsyncTask> {
+ const asyncTasksMDB = await AsyncTaskStorage.getAsyncTasks({
+ asyncTaskIDs: [id],
+ }, Constants.DB_PARAMS_SINGLE_RECORD, projectFields);
+ return asyncTasksMDB.count === 1 ? asyncTasksMDB.result[0] : null;
+ }
+
+ public static async saveAsyncTask(asyncTaskToSave: AsyncTask): Promise<string> {
+ // Debug
+ const uniqueTimerID = Logging.traceStart(Constants.DEFAULT_TENANT, MODULE_NAME, 'saveAsyncTask');
+ // Set
+ const asyncTaskMDB: any = {
+ _id: asyncTaskToSave.id ?? Cypher.hash(`${asyncTaskToSave.tenantID}~${asyncTaskToSave.name}~${asyncTaskToSave.parameters ? JSON.stringify(asyncTaskToSave.parameters) : ''}`),
+ name: asyncTaskToSave.name,
+ action: asyncTaskToSave.action,
+ type: asyncTaskToSave.type,
+ tenantID: Utils.convertToObjectID(asyncTaskToSave.tenantID),
+ status: asyncTaskToSave.status,
+ parent: asyncTaskToSave.parent,
+ execTimeSecs: Utils.convertToInt(asyncTaskToSave.execTimeSecs),
+ execHost: asyncTaskToSave.execHost,
+ execTimestamp: Utils.convertToDate(asyncTaskToSave.execTimestamp),
+ module: asyncTaskToSave.module,
+ method: asyncTaskToSave.method,
+ message: asyncTaskToSave.message,
+ parameters: asyncTaskToSave.parameters,
+ };
+ // Add Last Changed/Created props
+ DatabaseUtils.addLastChangedCreatedProps(asyncTaskMDB, asyncTaskToSave);
+ // Modify
+ await global.database.getCollection<AsyncTask>(Constants.DEFAULT_TENANT, 'asynctasks').findOneAndUpdate(
+ { _id: asyncTaskMDB._id },
+ { $set: asyncTaskMDB },
+ { upsert: true }
+ );
+ // Debug
+ await Logging.traceEnd(Constants.DEFAULT_TENANT, MODULE_NAME, 'saveAsyncTask', uniqueTimerID, asyncTaskMDB);
+ return asyncTaskMDB._id;
+ }
+
+ public static async getAsyncTasks(params: { status?: AsyncTaskStatus, asyncTaskIDs?: string[] } = {},
+ dbParams?: DbParams, projectFields?: string[]): Promise<DataResult<AsyncTask>> {
+ // Debug
+ const uniqueTimerID = Logging.traceStart(Constants.DEFAULT_TENANT, MODULE_NAME, 'getAsyncTasks');
+ // Clone before updating the values
+ dbParams = Utils.cloneObject(dbParams);
+ // Check Limit
+ dbParams.limit = Utils.checkRecordLimit(dbParams.limit);
+ // Check Skip
+ dbParams.skip = Utils.checkRecordSkip(dbParams.skip);
+ // Create Aggregation
+ const aggregation = [];
+ // Set the filters
+ const filters: FilterParams = {};
+ // Async task IDs
+ if (!Utils.isEmptyArray(params.asyncTaskIDs)) {
+ filters._id = {
+ $in: params.asyncTaskIDs
+ };
+ }
+ // Status
+ if (params.status) {
+ filters.status = params.status;
+ }
+ // Limit records?
+ if (!dbParams.onlyRecordCount) {
+ // Always limit the nbr of record to avoid perfs issues
+ aggregation.push({ $limit: Constants.DB_RECORD_COUNT_CEIL });
+ }
+ // Count Records
+ const asyncTasksCountMDB = await global.database.getCollection<DataResult<AsyncTask>>(Constants.DEFAULT_TENANT, 'asynctasks')
+ .aggregate([...aggregation, { $count: 'count' }], { allowDiskUse: true })
+ .toArray();
+ // Check if only the total count is requested
+ if (dbParams.onlyRecordCount) {
+ // Return only the count
+ await Logging.traceEnd(Constants.DEFAULT_TENANT, MODULE_NAME, 'getAsyncTasks', uniqueTimerID, asyncTasksCountMDB);
+ return {
+ count: (asyncTasksCountMDB.length > 0 ? asyncTasksCountMDB[0].count : 0),
+ result: []
+ };
+ }
+ // Remove the limit
+ aggregation.pop();
+ // Sort
+ if (!dbParams.sort) {
+ dbParams.sort = { name: 1 };
+ }
+ aggregation.push({
+ $sort: dbParams.sort
+ });
+ // Skip
+ if (dbParams.skip > 0) {
+ aggregation.push({ $skip: dbParams.skip });
+ }
+ // Limit
+ aggregation.push({
+ $limit: (dbParams.limit > 0 && dbParams.limit < Constants.DB_RECORD_COUNT_CEIL) ? dbParams.limit : Constants.DB_RECORD_COUNT_CEIL
+ });
+ // Handle the ID
+ DatabaseUtils.pushRenameDatabaseID(aggregation);
+ // Add Created By / Last Changed By
+ DatabaseUtils.pushCreatedLastChangedInAggregation(Constants.DEFAULT_TENANT, aggregation);
+ // Project
+ DatabaseUtils.projectFields(aggregation, projectFields);
+ // Read DB
+ const asyncTasksMDB = await global.database.getCollection<any>(Constants.DEFAULT_TENANT, 'asynctasks')
+ .aggregate(aggregation, {
+ allowDiskUse: true
+ })
+ .toArray();
+ // Debug
+ await Logging.traceEnd(Constants.DEFAULT_TENANT, MODULE_NAME, 'getAsyncTasks', uniqueTimerID, asyncTasksMDB);
+ return {
+ count: (asyncTasksCountMDB.length > 0 ?
+ (asyncTasksCountMDB[0].count === Constants.DB_RECORD_COUNT_CEIL ? -1 : asyncTasksCountMDB[0].count) : 0),
+ result: asyncTasksMDB
+ };
+ }
+
+ public static async deleteAsyncTask(id: string): Promise<void> {
+ // Debug
+ const uniqueTimerID = Logging.traceStart(Constants.DEFAULT_TENANT, MODULE_NAME, 'deleteAsyncTask');
+ // Delete the AsyncTask
+ await global.database.getCollection<AsyncTask>(Constants.DEFAULT_TENANT, 'asynctasks')
+ .findOneAndDelete({ '_id': id });
+ // Debug
+ await Logging.traceEnd(Constants.DEFAULT_TENANT, MODULE_NAME, 'deleteAsyncTask', uniqueTimerID, { id });
+ }
+}

MODIFY
src/storage/mongodb/MongoDBStorageNotification.ts
src/storage/mongodb/MongoDBStorageNotification.ts
import { Action, Entity } from '../../types/Authorization';
+import AsyncTaskManager from '../../async-task/AsyncTaskManager';
import CentralRestServer from '../../server/rest/CentralRestServer';
import { ChangeStreamOptions } from 'mongodb';
import Constants from '../../utils/Constants';
import Logging from '../../utils/Logging';
import { ServerAction } from '../../types/Server';
import StorageConfiguration from '../../types/configuration/StorageConfiguration';
-import TenantStorage from './TenantStorage';
import { TransactionNotificationData } from '../../types/SingleChangeNotification';
import Utils from '../../utils/Utils';
import global from '../../types/GlobalType';
@@ -169,6 +169,10 @@ export default class MongoDBStorageNotification {
case 'ocpiendpoints':
this.centralRestServer?.notifyOcpiEndpoint(tenantID, action, { id: documentID });
break;
+ case 'asynctasks':
+ this.centralRestServer?.notifyAsyncTaskEndpoint(tenantID, action, { id: documentID });
+ void AsyncTaskManager.handleAsyncTasks();
+ break;
}
}

ADD
src/types/AsyncTask.ts
+import CreatedUpdatedProps from './CreatedUpdatedProps';
+import { ServerAction } from './Server';
+
+export default interface AsyncTask extends CreatedUpdatedProps {
+ id: string;
+ name: AsyncTasks;
+ action: ServerAction;
+ type: AsyncTaskType;
+ tenantID: string;
+ status?: AsyncTaskStatus;
+ parent?: string;
+ execTimeSecs?: number;
+ execHost?: string;
+ execTimestamp?: Date;
+ module: string;
+ method: string;
+ message?: string;
+ parameters?: Record<string, string>;
+}
+
+export enum AsyncTaskType {
+ TASK = 'T',
+ TASK_GROUP = 'G',
+}
+
+export enum AsyncTaskStatus {
+ NEW = 'N',
+ RUNNING = 'R',
+ ERROR = 'E',
+ SUCCESS = 'S',
+}
+
+export enum AsyncTasks {
+ TAGS_IMPORT = 'TagsImportAsyncTask',
+}
+

MODIFY
src/types/Authorization.ts
src/types/Authorization.ts
@@ -65,6 +65,8 @@ export enum Entity {
SETTINGS = 'Settings',
TOKENS = 'Tokens',
TOKEN = 'Token',
+ ASYNC_TASK = 'AsyncTask',
+ ASYNC_TASKS = 'AsyncTasks',
OCPI_ENDPOINT = 'OcpiEndpoint',
OCPI_ENDPOINTS = 'OcpiEndpoints',
OICP_ENDPOINT = 'OicpEndpoint',

MODIFY
src/types/Locking.ts
src/types/Locking.ts
@@ -29,4 +29,5 @@ export enum LockEntity {
OCPI_ENDPOINT = 'ocpi-endpoint',
TAG = 'tag',
OICP_ENDPOINT = 'oicp-endpoint',
+ ASYNC_TASK = 'async-task',
}

MODIFY
src/types/Server.ts
src/types/Server.ts
@@ -89,6 +89,7 @@ export enum ServerAction {
PERFORMANCES = 'Performances',
SCHEDULER = 'Scheduler',
+ ASYNC_TASK = 'AsyncTask',
REMOTE_PUSH_NOTIFICATION = 'RemotePushNotification',
EMAIL_NOTIFICATION = 'EmailNotification',

ADD
src/types/configuration/AsyncTaskConfiguration.ts
+export default interface AsyncTaskConfiguration {
+ active: boolean;
+ nbrTasksInParallel: number;
+}

MODIFY
src/types/configuration/Configuration.ts
src/types/configuration/Configuration.ts
@@ -50,6 +50,7 @@ export interface Configuration {
ChargingStation: ChargingStationConfiguration;
Locales?: LocalesConfiguration;
Scheduler: SchedulerConfiguration;
+ AsyncTask: AsyncTaskConfiguration;
Logging: LoggingConfiguration;
HealthCheck?: HealthCheckConfiguration;
Migration?: MigrationConfiguration;

MODIFY
src/utils/Configuration.ts
src/utils/Configuration.ts
import { AppEnv, getAppEnv } from 'cfenv';
import { CloudCredentials, CloudCredentialsKey } from '../types/Cloud';
+import AsyncTaskConfiguration from '../types/configuration/AsyncTaskConfiguration';
import AuthorizationConfiguration from '../types/configuration/AuthorizationConfiguration';
import AxiosConfiguration from '../types/configuration/AxiosConfiguration';
import CentralSystemConfiguration from '../types/configuration/CentralSystemConfiguration';
@@ -56,6 +57,12 @@ export default class Configuration {
return Configuration.getConfig().Scheduler;
}
+ // Async task config
+ public static getAsyncTaskConfig(): AsyncTaskConfiguration {
+ // Read conf
+ return Configuration.getConfig().AsyncTask;
+ }
+
// Firebase config
public static getFirebaseConfig(): FirebaseConfiguration {
// Read conf

First push of async task framework with import tags