Made push of evse statuses async
MODIFY
src/async-task/AsyncTaskManager.ts
src/async-task/AsyncTaskManager.ts
@@ -15,6 +15,7 @@ import OCPIPullCdrsAsyncTask from './tasks/ocpi/OCPIPullCdrsAsyncTask';
import OCPIPullLocationsAsyncTask from './tasks/ocpi/OCPIPullLocationsAsyncTask';
import OCPIPullSessionsAsyncTask from './tasks/ocpi/OCPIPullSessionsAsyncTask';
import OCPIPullTokensAsyncTask from './tasks/ocpi/OCPIPullTokensAsyncTask';
+import OCPIPushEVSEStatusesAsyncTask from './tasks/ocpi/OCPIPushEVSEStatusesAsyncTask';
import OCPIPushTokensAsyncTask from './tasks/ocpi/OCPIPushTokensAsyncTask';
import { ServerAction } from '../types/Server';
import TagsImportAsyncTask from './tasks/TagsImportAsyncTask';
@@ -99,9 +100,13 @@ export default class AsyncTaskManager {
break;
case AsyncTasks.OCPI_CHECK_LOCATIONS:
abstractAsyncTask = new OCPICheckLocationsAsyncTask(asyncTask);
+ break;
case AsyncTasks.OCPI_PULL_TOKENS:
abstractAsyncTask = new OCPIPullTokensAsyncTask(asyncTask);
break;
+ case AsyncTasks.OCPI_PUSH_EVSE_STATUSES:
+ abstractAsyncTask = new OCPIPushEVSEStatusesAsyncTask(asyncTask);
+ break;
default:
await Logging.logError({
tenantID: Constants.DEFAULT_TENANT,

ADD
src/async-task/tasks/ocpi/OCPIPushEVSEStatusesAsyncTask.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 OCPIPushEVSEStatusesAsyncTask 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)) {
+ 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}'`);
+ }
+ const patchStatusesLock = await LockingHelper.createOCPIPatchEVSEStatusesLock(tenant.id, ocpiEndpoint);
+ if (patchStatusesLock) {
+ try {
+ // Get the OCPI Client
+ const ocpiClient = await OCPIClientFactory.getCpoOcpiClient(tenant, ocpiEndpoint);
+ if (!ocpiClient) {
+ throw new Error(`OCPI Client not found in Endpoint ID '${this.asyncTask.parameters.endpointID}'`);
+ }
+ await ocpiClient.sendEVSEStatuses(true);
+ } finally {
+ // Release the lock
+ await LockingManager.release(patchStatusesLock);
+ }
+ }
+ } catch (error) {
+ await Logging.logActionExceptionMessage(tenant.id, ServerAction.OCPI_PATCH_STATUS, error);
+ }
+ }
+ }
+}

MODIFY
src/client/ocpi/CpoOCPIClient.ts
src/client/ocpi/CpoOCPIClient.ts
@@ -684,7 +684,8 @@ export default class CpoOCPIClient extends OCPIClient {
{ skip: currentSkip, limit: Constants.DB_RECORD_COUNT_DEFAULT },
{ chargingStationIDs } )
// Loop through EVSE
- for (const evse of evses) {
+ if (!Utils.isEmptyArray(evses)) {
+ await Promise.map(evses, async (evse) => {
// Total amount of EVSEs
result.total++;
// Process it if not empty
@@ -711,6 +712,8 @@ export default class CpoOCPIClient extends OCPIClient {
);
}
}
+ },
+ { concurrency: Constants.OCPI_MAX_PARALLEL_REQUESTS });
}
currentSkip += Constants.DB_RECORD_COUNT_DEFAULT;
} while (!Utils.isEmptyArray(evses));

MODIFY
src/locking/LockingHelper.ts
src/locking/LockingHelper.ts
@@ -132,6 +132,10 @@ export default class LockingHelper {
return LockingHelper.createOCPIEndpointActionLock(tenantID, ocpiEndpoint, 'patch-locations');
}
+ public static async createOCPIPatchEVSEStatusesLock(tenantID: string, ocpiEndpoint: OCPIEndpoint): Promise<Lock|null> {
+ return LockingHelper.createOCPIEndpointActionLock(tenantID, ocpiEndpoint, 'patch-evse-statuses');
+ }
+
public static async createOICPPatchEVSEsLock(tenantID: string, oicpEndpoint: OICPEndpoint): Promise<Lock|null> {
return LockingHelper.createOICPEndpointActionLock(tenantID, oicpEndpoint, 'patch-evses');
}

MODIFY
src/server/rest/CentralRestServerService.ts
src/server/rest/CentralRestServerService.ts
@@ -92,7 +92,7 @@ class RequestMapper {
[ServerAction.OCPI_ENDPOINT_PULL_LOCATIONS]: OCPIEndpointService.handlePullLocationsEndpoint.bind(this),
[ServerAction.OCPI_ENDPOINT_PULL_SESSIONS]: OCPIEndpointService.handlePullSessionsEndpoint.bind(this),
[ServerAction.OCPI_ENDPOINT_PULL_TOKENS]: OCPIEndpointService.handlePullTokensEndpoint.bind(this),
- [ServerAction.OCPI_ENDPOINT_SEND_EVSE_STATUSES]: OCPIEndpointService.handleSendEVSEStatusesOcpiEndpoint.bind(this),
+ [ServerAction.OCPI_ENDPOINT_SEND_EVSE_STATUSES]: OCPIEndpointService.handlePushEVSEStatusesOcpiEndpoint.bind(this),
[ServerAction.OCPI_ENDPOINT_SEND_TOKENS]: OCPIEndpointService.handlePushTokensOcpiEndpoint.bind(this),
[ServerAction.OCPI_ENDPOINT_GENERATE_LOCAL_TOKEN]: OCPIEndpointService.handleGenerateLocalTokenOcpiEndpoint.bind(this),
[ServerAction.OICP_ENDPOINT_CREATE]: OICPEndpointService.handleCreateOicpEndpoint.bind(this),

MODIFY
src/server/rest/v1/service/OCPIEndpointService.ts
src/server/rest/v1/service/OCPIEndpointService.ts
@@ -633,49 +633,58 @@ export default class OCPIEndpointService {
next();
}
- static async handleSendEVSEStatusesOcpiEndpoint(action: ServerAction, req: Request, res: Response, next: NextFunction): Promise<void> {
+ static async handlePushEVSEStatusesOcpiEndpoint(action: ServerAction, req: Request, res: Response, next: NextFunction): Promise<void> {
// Check if component is active
UtilsService.assertComponentIsActiveFromToken(req.user, TenantComponents.OCPI,
- Action.READ, Entity.OCPI_ENDPOINT, MODULE_NAME, 'handleSendEVSEStatusesOcpiEndpoint');
+ Action.READ, Entity.OCPI_ENDPOINT, MODULE_NAME, 'handlePushEVSEStatusesOcpiEndpoint');
// Check auth
if (!Authorizations.canTriggerJobOcpiEndpoint(req.user)) {
throw new AppAuthError({
errorCode: HTTPAuthError.FORBIDDEN,
user: req.user,
action: Action.TRIGGER_JOB, entity: Entity.OCPI_ENDPOINT,
- module: MODULE_NAME, method: 'handleSendEVSEStatusesOcpiEndpoint'
+ module: MODULE_NAME, method: 'handlePushEVSEStatusesOcpiEndpoint'
});
}
// Filter
const filteredRequest = OCPIEndpointSecurity.filterOcpiEndpointSendEVSEStatusesRequest(req.body);
- UtilsService.assertIdIsProvided(action, filteredRequest.id, MODULE_NAME, 'handleSendEVSEStatusesOcpiEndpoint', req.user);
+ UtilsService.assertIdIsProvided(action, filteredRequest.id, MODULE_NAME, 'handlePushEVSEStatusesOcpiEndpoint', req.user);
// Get ocpiEndpoint
const ocpiEndpoint = await OCPIEndpointStorage.getOcpiEndpoint(req.user.tenantID, filteredRequest.id);
UtilsService.assertObjectExists(action, ocpiEndpoint, `OCPIEndpoint with ID '${filteredRequest.id}' does not exist`,
- MODULE_NAME, 'handleSendEVSEStatusesOcpiEndpoint', req.user);
+ MODULE_NAME, 'handlePushEVSEStatusesOcpiEndpoint', req.user);
const tenant = await TenantStorage.getTenant(req.user.tenantID);
// Get the lock
- const ocpiLock = await LockingHelper.createOCPIPatchLocationsLock(tenant.id, ocpiEndpoint);
- if (!ocpiLock) {
+ const patchStatusesLock = await LockingHelper.createOCPIPatchEVSEStatusesLock(tenant.id, ocpiEndpoint);
+ if (!patchStatusesLock) {
throw new AppError({
source: Constants.CENTRAL_SERVER,
action: ServerAction.OCPI_PATCH_LOCATIONS,
errorCode: HTTPError.CANNOT_ACQUIRE_LOCK,
- message: 'Error in patching the OCPI Locations: cannot acquire the lock',
- module: MODULE_NAME, method: 'handleSendEVSEStatusesOcpiEndpoint',
+ message: 'Error in pushing the OCPI EVSE Statuses: cannot acquire the lock',
+ module: MODULE_NAME, method: 'handlePushEVSEStatusesOcpiEndpoint',
});
}
try {
- // Build OCPI Client
- const ocpiClient = await OCPIClientFactory.getCpoOcpiClient(tenant, ocpiEndpoint);
- // Send EVSE statuses
- const sendResult = await ocpiClient.sendEVSEStatuses();
- // Return result
- res.json(sendResult);
- next();
+ // Create and Save async task
+ AsyncTaskManager.createAndSaveAsyncTasks({
+ name: AsyncTasks.OCPI_PUSH_EVSE_STATUSES,
+ action,
+ type: AsyncTaskType.TASK,
+ tenantID: req.tenant.id,
+ parameters: {
+ endpointID: filteredRequest.id,
+ },
+ module: MODULE_NAME,
+ method: 'handlePushEVSEStatusesOcpiEndpoint',
+ });
} finally {
- await LockingManager.release(ocpiLock);
+ // Release the lock
+ await LockingManager.release(patchStatusesLock);
}
+ // Return result
+ res.json(Constants.REST_RESPONSE_SUCCESS);
+ next();
}
static async handlePushTokensOcpiEndpoint(action: ServerAction, req: Request, res: Response, next: NextFunction): Promise<void> {

MODIFY
src/types/AsyncTask.ts
src/types/AsyncTask.ts
@@ -42,5 +42,6 @@ export enum AsyncTasks {
OCPI_CHECK_SESSIONS = 'OCPICheckSessionsAsyncTask',
OCPI_CHECK_LOCATIONS = 'OCPICheckLocationsAsyncTask',
OCPI_PULL_TOKENS = 'OCPIPullTokensAsyncTask',
+ OCPI_PUSH_EVSE_STATUSES = 'OCPIPushEVSEStatusesAsyncTask',
}

Made push of evse statuses async