Async call of OCPI pull token
MODIFY
src/async-task/AsyncTaskManager.ts
src/async-task/AsyncTaskManager.ts
@@ -8,6 +8,7 @@ import Constants from '../utils/Constants';
import LockingHelper from '../locking/LockingHelper';
import LockingManager from '../locking/LockingManager';
import Logging from '../utils/Logging';
+import OCPIEmspPushTokensAsyncTask from './tasks/ocpi/OCPIEmspPushTokensAsyncTask';
import { ServerAction } from '../types/Server';
import TagsImportAsyncTask from './tasks/TagsImportAsyncTask';
import UsersImportAsyncTask from './tasks/UsersImportAsyncTask';
@@ -71,6 +72,9 @@ export default class AsyncTaskManager {
case AsyncTasks.USERS_IMPORT:
abstractAsyncTask = new UsersImportAsyncTask(asyncTask);
break;
+ case AsyncTasks.OCPI_EMSP_PUSH_TOKENS:
+ abstractAsyncTask = new OCPIEmspPushTokensAsyncTask(asyncTask);
+ break;
default:
await Logging.logError({
tenantID: Constants.DEFAULT_TENANT,
@@ -138,7 +142,7 @@ export default class AsyncTaskManager {
}
}
},
- { concurrency: nbrTasksInParallel }).then(() => {
+ { concurrency: nbrTasksInParallel });
// Log result
const totalDurationSecs = Utils.truncTo((new Date().getTime() - startTime) / 1000, 2);
void Logging.logActionsResponse(Constants.DEFAULT_TENANT, ServerAction.ASYNC_TASK,
@@ -148,7 +152,6 @@ export default class AsyncTaskManager {
`{{inSuccess}} Async Task(s) were successfully processed in ${totalDurationSecs} secs and {{inError}} failed`,
`No Async Task to process`
);
- });
} else {
await Logging.logInfo({
tenantID: Constants.DEFAULT_TENANT,

ADD
src/async-task/tasks/ocpi/OCPIEmspPushTokensAsyncTask.ts
+import AbstractAsyncTask from '../../AsyncTask';
+import LockingHelper from '../../../locking/LockingHelper';
+import LockingManager from '../../../locking/LockingManager';
+import Logging from '../../../utils/Logging';
+import OCPIClientFactory from '../../../client/ocpi/OCPIClientFactory';
+import OCPIEndpointStorage from '../../../storage/mongodb/OCPIEndpointStorage';
+import { ServerAction } from '../../../types/Server';
+import TenantComponents from '../../../types/TenantComponents';
+import TenantStorage from '../../../storage/mongodb/TenantStorage';
+import Utils from '../../../utils/Utils';
+
+export default class OCPIEmspPushTokensAsyncTask extends AbstractAsyncTask {
+ protected async executeAsyncTask(): Promise<void> {
+ const tenant = await TenantStorage.getTenant(this.asyncTask.tenantID);
+ // Check if OCPI component is active
+ if (Utils.isTenantComponentActive(tenant, TenantComponents.OCPI)) {
+ const pushTokensLock = await LockingHelper.createOCPIPushTokensLock(tenant.id);
+ if (pushTokensLock) {
+ try {
+ // Get the OCPI Endpoint
+ const ocpiEndpoint = await OCPIEndpointStorage.getOcpiEndpoint(tenant.id, this.asyncTask.parameters.endpointID);
+ if (!ocpiEndpoint) {
+ throw new Error(`Unknown OCPI Endpoint ID '${this.asyncTask.parameters.endpointID}'`);
+ }
+ // Get the OCPI Client
+ const ocpiClient = await OCPIClientFactory.getEmspOcpiClient(tenant, ocpiEndpoint);
+ if (!ocpiClient) {
+ throw new Error(`OCPI Client not found in Endpoint ID '${this.asyncTask.parameters.endpointID}'`);
+ }
+ // Send Tokens
+ await ocpiClient.sendTokens();
+ } catch (error) {
+ // Log error
+ await Logging.logActionExceptionMessage(tenant.id, ServerAction.USERS_IMPORT, error);
+ } finally {
+ // Release the lock
+ await LockingManager.release(pushTokensLock);
+ }
+ }
+ }
+ }
+}

MODIFY
src/locking/LockingHelper.ts
src/locking/LockingHelper.ts
@@ -88,6 +88,14 @@ export default class LockingHelper {
return lock;
}
+ public static async createOCPIPushTokensLock(tenantID: string): Promise<Lock | null> {
+ const lock = LockingManager.createExclusiveLock(tenantID, LockEntity.TAG, 'push-tokens');
+ if (!(await LockingManager.acquire(lock))) {
+ return null;
+ }
+ return lock;
+ }
+
public static async createOCPIPushCpoCdrLock(tenantID: string, transactionID: number): Promise<Lock | null> {
const lock = LockingManager.createExclusiveLock(tenantID, LockEntity.TRANSACTION, `push-cdr-${transactionID}`);
if (!(await LockingManager.acquire(lock))) {

MODIFY
src/scheduler/tasks/ocpi/OCPIPushCdrsTask.ts
src/scheduler/tasks/ocpi/OCPIPushCdrsTask.ts
@@ -46,7 +46,7 @@ export default class OCPIPushCdrsTask extends SchedulerTask {
message: `${transactionsMDB.length} Transaction's CDRs are going to be pushed to OCPI`,
});
for (const transactionMDB of transactionsMDB) {
- // Get the lock
+ // Get the lock: Used to avoid collision with manual push or end of transaction push
const ocpiTransactionLock = await LockingHelper.createOCPIPushCpoCdrLock(tenant.id, transactionMDB._id);
if (ocpiTransactionLock) {
try {

MODIFY
src/server/rest/v1/service/OCPIEndpointService.ts
src/server/rest/v1/service/OCPIEndpointService.ts
import { Action, Entity } from '../../../../types/Authorization';
+import { AsyncTaskType, AsyncTasks } from '../../../../types/AsyncTask';
import { HTTPAuthError, HTTPError } from '../../../../types/HTTPError';
import { NextFunction, Request, Response } from 'express';
import AppAuthError from '../../../../exception/AppAuthError';
import AppError from '../../../../exception/AppError';
+import AsyncTaskManager from '../../../../async-task/AsyncTaskManager';
import Authorizations from '../../../../authorization/Authorizations';
import Constants from '../../../../utils/Constants';
import LockingHelper from '../../../../locking/LockingHelper';
@@ -629,13 +631,38 @@ export default class OCPIEndpointService {
const ocpiEndpoint = await OCPIEndpointStorage.getOcpiEndpoint(req.user.tenantID, filteredRequest.id);
UtilsService.assertObjectExists(action, ocpiEndpoint, `OCPIEndpoint with ID '${filteredRequest.id}' does not exist`,
MODULE_NAME, 'handleSendTokensOcpiEndpoint', req.user);
- const tenant = await TenantStorage.getTenant(req.user.tenantID);
- // Build OCPI Client
- const ocpiClient = await OCPIClientFactory.getEmspOcpiClient(tenant, ocpiEndpoint);
- // Send EVSE statuses
- const sendResult = await ocpiClient.sendTokens();
+ // Get the lock
+ const pushTokensLock = await LockingHelper.createOCPIPushTokensLock(req.tenant.id);
+ if (pushTokensLock) {
+ try {
+ // Create and Save async task
+ AsyncTaskManager.createAndSaveAsyncTasks({
+ name: AsyncTasks.OCPI_EMSP_PUSH_TOKENS,
+ action: ServerAction.OCPI_ENDPOINT_SEND_TOKENS,
+ type: AsyncTaskType.TASK,
+ tenantID: req.tenant.id,
+ parameters: {
+ endpointID: filteredRequest.id,
+ },
+ module: MODULE_NAME,
+ method: 'handleSendTokensOcpiEndpoint',
+ });
+ } finally {
+ // Release the lock
+ await LockingManager.release(pushTokensLock);
+ }
+ } else {
+ throw new AppError({
+ source: Constants.CENTRAL_SERVER,
+ action: action,
+ errorCode: HTTPError.CANNOT_ACQUIRE_LOCK,
+ module: MODULE_NAME, method: 'handleSendTokensOcpiEndpoint',
+ message: 'Error in pushing the Tags: cannot acquire the lock',
+ user: req.user
+ });
+ }
// Return result
- res.json(sendResult);
+ res.json(Constants.REST_RESPONSE_SUCCESS);
next();
}

MODIFY
src/types/AsyncTask.ts
src/types/AsyncTask.ts
@@ -34,5 +34,6 @@ export enum AsyncTaskStatus {
export enum AsyncTasks {
TAGS_IMPORT = 'TagsImportAsyncTask',
USERS_IMPORT = 'UsersImportAsyncTask',
+ OCPI_EMSP_PUSH_TOKENS = 'OCPIEmspPushTokensTask',
}

Async call of OCPI pull token