Made OCPI pull cdrs async
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 OCPIPullCdrsAsyncTask from './tasks/ocpi/OCPIPullCdrsAsyncTask';
import OCPIPullLocationsAsyncTask from './tasks/ocpi/OCPIPullLocationsAsyncTask';
import OCPIPullSessionsAsyncTask from './tasks/ocpi/OCPIPullSessionsAsyncTask';
import OCPIPushTokensAsyncTask from './tasks/ocpi/OCPIPushTokensAsyncTask';
@@ -83,6 +84,9 @@ export default class AsyncTaskManager {
case AsyncTasks.OCPI_PULL_SESSIONS:
abstractAsyncTask = new OCPIPullSessionsAsyncTask(asyncTask);
break;
+ case AsyncTasks.OCPI_PULL_CDRS:
+ abstractAsyncTask = new OCPIPullCdrsAsyncTask(asyncTask);
+ break;
default:
await Logging.logError({
tenantID: Constants.DEFAULT_TENANT,

ADD
src/async-task/tasks/ocpi/OCPIPullCdrsAsyncTask.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 OCPIPullCdrsAsyncTask 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 pullCdrsLock = await LockingHelper.createOCPIPullCdrsLock(tenant.id, ocpiEndpoint);
+ if (pullCdrsLock) {
+ try {
+ // 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}'`);
+ }
+ await ocpiClient.pullCdrs();
+ } finally {
+ // Release the lock
+ await LockingManager.release(pullCdrsLock);
+ }
+ }
+ } catch (error) {
+ await Logging.logActionExceptionMessage(tenant.id, ServerAction.OCPI_PULL_CDRS, error);
+ }
+ }
+ }
+}

MODIFY
src/client/ocpi/EmspOCPIClient.ts
src/client/ocpi/EmspOCPIClient.ts
@@ -279,7 +279,9 @@ export default class EmspOCPIClient extends OCPIClient {
Authorization: `Token ${this.ocpiEndpoint.token}`
},
});
- for (const cdr of response.data.data as OCPICdr[]) {
+ const cdrs = response.data.data as OCPICdr[];
+ if (!Utils.isEmptyArray(cdrs)) {
+ await Promise.map(cdrs, async (cdr: OCPICdr) => {
try {
await OCPIUtilsService.processCdr(this.tenant.id, cdr);
result.success++;
@@ -289,6 +291,8 @@ export default class EmspOCPIClient extends OCPIClient {
`Failed to update CDR ID '${cdr.id}': ${error.message}`
);
}
+ },
+ { concurrency: Constants.OCPI_MAX_PARALLEL_REQUESTS });
}
const nextUrl = OCPIUtils.getNextUrl(response.headers.link);
if (nextUrl && nextUrl.length > 0 && nextUrl !== cdrsUrl) {

MODIFY
src/server/rest/v1/service/OCPIEndpointService.ts
src/server/rest/v1/service/OCPIEndpointService.ts
@@ -429,8 +429,8 @@ export default class OCPIEndpointService {
MODULE_NAME, 'handlePullCdrsEndpoint', req.user);
const tenant = await TenantStorage.getTenant(req.user.tenantID);
// Get the lock
- const ocpiLock = await LockingHelper.createOCPIPullCdrsLock(tenant.id, ocpiEndpoint);
- if (!ocpiLock) {
+ const pullCdrsLock = await LockingHelper.createOCPIPullCdrsLock(tenant.id, ocpiEndpoint);
+ if (!pullCdrsLock) {
throw new AppError({
source: Constants.CENTRAL_SERVER,
action: ServerAction.OCPI_PULL_CDRS,
@@ -440,15 +440,25 @@ export default class OCPIEndpointService {
});
}
try {
- // Build OCPI Client
- const ocpiClient = await OCPIClientFactory.getEmspOcpiClient(tenant, ocpiEndpoint);
- const result = await ocpiClient.pullCdrs();
- // Return result
- res.json(result);
- next();
+ // Create and Save async task
+ AsyncTaskManager.createAndSaveAsyncTasks({
+ name: AsyncTasks.OCPI_PULL_CDRS,
+ action,
+ type: AsyncTaskType.TASK,
+ tenantID: req.tenant.id,
+ parameters: {
+ endpointID: filteredRequest.id,
+ },
+ module: MODULE_NAME,
+ method: 'handlePullCdrsEndpoint',
+ });
} finally {
- await LockingManager.release(ocpiLock);
+ // Release the lock
+ await LockingManager.release(pullCdrsLock);
}
+ // Return result
+ res.json(Constants.REST_RESPONSE_SUCCESS);
+ next();
}
static async handleCheckCdrsEndpoint(action: ServerAction, req: Request, res: Response, next: NextFunction): Promise<void> {

MODIFY
src/types/AsyncTask.ts
src/types/AsyncTask.ts
@@ -37,5 +37,6 @@ export enum AsyncTasks {
OCPI_PUSH_TOKENS = 'OCPIPushTokensAsyncTask',
OCPI_PULL_LOCATIONS = 'OCPIPullLocationsAsyncTask',
OCPI_PULL_SESSIONS = 'OCPIPullSessionsAsyncTask',
+ OCPI_PULL_CDRS = 'OCPIPullCdrsAsyncTask',
}

Made OCPI pull cdrs async