Made check sessions async
MODIFY
src/async-task/AsyncTaskManager.ts
src/async-task/AsyncTaskManager.ts
@@ -9,6 +9,7 @@ import LockingHelper from '../locking/LockingHelper';
import LockingManager from '../locking/LockingManager';
import Logging from '../utils/Logging';
import OCPICheckCdrsAsyncTask from './tasks/ocpi/OCPICheckCdrsAsyncTask';
+import OCPICheckSessionsAsyncTask from './tasks/ocpi/OCPICheckSessionsAsyncTask';
import OCPIPullCdrsAsyncTask from './tasks/ocpi/OCPIPullCdrsAsyncTask';
import OCPIPullLocationsAsyncTask from './tasks/ocpi/OCPIPullLocationsAsyncTask';
import OCPIPullSessionsAsyncTask from './tasks/ocpi/OCPIPullSessionsAsyncTask';
@@ -91,6 +92,9 @@ export default class AsyncTaskManager {
case AsyncTasks.OCPI_CHECK_CDRS:
abstractAsyncTask = new OCPICheckCdrsAsyncTask(asyncTask);
break;
+ case AsyncTasks.OCPI_CHECK_SESSIONS:
+ abstractAsyncTask = new OCPICheckSessionsAsyncTask(asyncTask);
+ break;
default:
await Logging.logError({
tenantID: Constants.DEFAULT_TENANT,

ADD
src/async-task/tasks/ocpi/OCPICheckSessionsAsyncTask.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 OCPICheckSessionsAsyncTask 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 checkSessionsLock = await LockingHelper.createOCPICheckSessionsLock(tenant.id, ocpiEndpoint);
+ if (checkSessionsLock) {
+ 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.checkSessions();
+ } finally {
+ // Release the lock
+ await LockingManager.release(checkSessionsLock);
+ }
+ }
+ } catch (error) {
+ await Logging.logActionExceptionMessage(tenant.id, ServerAction.OCPI_CHECK_SESSIONS, error);
+ }
+ }
+ }
+}

MODIFY
src/client/ocpi/CpoOCPIClient.ts
src/client/ocpi/CpoOCPIClient.ts
@@ -505,7 +505,8 @@ export default class CpoOCPIClient extends OCPIClient {
issuer: true,
ocpiSessionChecked: false
}, Constants.DB_PARAMS_MAX_LIMIT);
- for (const transaction of transactions.result) {
+ if (!Utils.isEmptyArray(transactions.result)) {
+ await Promise.map(transactions.result, async (transaction) => {
if (transaction.stop && transaction.stop.timestamp) {
try {
if (await this.checkSession(transaction)) {
@@ -523,6 +524,8 @@ export default class CpoOCPIClient extends OCPIClient {
}
}
result.total++;
+ },
+ { concurrency: Constants.OCPI_MAX_PARALLEL_REQUESTS });
}
const executionDurationSecs = (new Date().getTime() - startTime) / 1000;
await Logging.logOcpiResult(this.tenant.id, ServerAction.OCPI_CHECK_SESSIONS,
@@ -597,7 +600,7 @@ export default class CpoOCPIClient extends OCPIClient {
// Perfs trace
const startTime = new Date().getTime();
const transactions = await TransactionStorage.getTransactions(this.tenant.id, {
- issuer: false,
+ issuer: true,
ocpiCdrChecked: false
}, Constants.DB_PARAMS_MAX_LIMIT);
if (!Utils.isEmptyArray(transactions.result)) {

MODIFY
src/server/rest/v1/service/OCPIEndpointService.ts
src/server/rest/v1/service/OCPIEndpointService.ts
@@ -537,8 +537,8 @@ export default class OCPIEndpointService {
MODULE_NAME, 'handleCheckSessionsEndpoint', req.user);
const tenant = await TenantStorage.getTenant(req.user.tenantID);
// Get the lock
- const ocpiLock = await LockingHelper.createOCPICheckSessionsLock(tenant.id, ocpiEndpoint);
- if (!ocpiLock) {
+ const checkSessionsLock = await LockingHelper.createOCPICheckSessionsLock(tenant.id, ocpiEndpoint);
+ if (!checkSessionsLock) {
throw new AppError({
source: Constants.CENTRAL_SERVER,
action: ServerAction.OCPI_CHECK_SESSIONS,
@@ -548,16 +548,25 @@ export default class OCPIEndpointService {
});
}
try {
- // Build OCPI Client
- const ocpiClient = await OCPIClientFactory.getCpoOcpiClient(tenant, ocpiEndpoint);
- // Send EVSE statuses
- const sendResult = await ocpiClient.checkSessions();
- // Return result
- res.json(sendResult);
- next();
+ // Create and Save async task
+ AsyncTaskManager.createAndSaveAsyncTasks({
+ name: AsyncTasks.OCPI_CHECK_SESSIONS,
+ action,
+ type: AsyncTaskType.TASK,
+ tenantID: req.tenant.id,
+ parameters: {
+ endpointID: filteredRequest.id,
+ },
+ module: MODULE_NAME,
+ method: 'handleCheckSessionsEndpoint',
+ });
} finally {
- await LockingManager.release(ocpiLock);
+ // Release the lock
+ await LockingManager.release(checkSessionsLock);
}
+ // Return result
+ res.json(Constants.REST_RESPONSE_SUCCESS);
+ next();
}
static async handleCheckLocationsEndpoint(action: ServerAction, req: Request, res: Response, next: NextFunction): Promise<void> {

MODIFY
src/types/AsyncTask.ts
src/types/AsyncTask.ts
@@ -39,5 +39,6 @@ export enum AsyncTasks {
OCPI_PULL_SESSIONS = 'OCPIPullSessionsAsyncTask',
OCPI_PULL_CDRS = 'OCPIPullCdrsAsyncTask',
OCPI_CHECK_CDRS = 'OCPICheckCdrsAsyncTask',
+ OCPI_CHECK_SESSIONS = 'OCPICheckSessionsAsyncTask',
}

Made check sessions async