src/charging-station/ocpp/OCPPRequestService.ts

   1 import { JSONSchemaType } from 'ajv';
   2 import Ajv from 'ajv-draft-04';
   3 import ajvFormats from 'ajv-formats';
   4
   5 import OCPPError from '../../exception/OCPPError';
   6 import PerformanceStatistics from '../../performance/PerformanceStatistics';
   7 import { EmptyObject } from '../../types/EmptyObject';
   8 import { HandleErrorParams } from '../../types/Error';
   9 import { JsonObject, JsonType } from '../../types/JsonType';
  10 import { ErrorType } from '../../types/ocpp/ErrorType';
  11 import { MessageType } from '../../types/ocpp/MessageType';
  12 import {
  13   IncomingRequestCommand,
  14   OutgoingRequest,
  15   RequestCommand,
  16   RequestParams,
  17   ResponseType,
  18 } from '../../types/ocpp/Requests';
  19 import { ErrorResponse, Response } from '../../types/ocpp/Responses';
  20 import Constants from '../../utils/Constants';
  21 import logger from '../../utils/Logger';
  22 import Utils from '../../utils/Utils';
  23 import type ChargingStation from '../ChargingStation';
  24 import type OCPPResponseService from './OCPPResponseService';
  25 import { OCPPServiceUtils } from './OCPPServiceUtils';
  26
  27 const moduleName = 'OCPPRequestService';
  28
  29 export default abstract class OCPPRequestService {
  30   private static instance: OCPPRequestService | null = null;
  31   private ajv: Ajv;
  32
  33   private readonly ocppResponseService: OCPPResponseService;
  34
  35   protected constructor(ocppResponseService: OCPPResponseService) {
  36     this.ocppResponseService = ocppResponseService;
  37     this.ajv = new Ajv();
  38     ajvFormats(this.ajv);
  39     this.requestHandler.bind(this);
  40     this.sendResponse.bind(this);
  41     this.sendError.bind(this);
  42     this.internalSendMessage.bind(this);
  43     this.buildMessageToSend.bind(this);
  44     this.validateRequestPayload.bind(this);
  45   }
  46
  47   public static getInstance<T extends OCPPRequestService>(
  48     this: new (ocppResponseService: OCPPResponseService) => T,
  49     ocppResponseService: OCPPResponseService
  50   ): T {
  51     if (OCPPRequestService.instance === null) {
  52       OCPPRequestService.instance = new this(ocppResponseService);
  53     }
  54     return OCPPRequestService.instance as T;
  55   }
  56
  57   public async sendResponse(
  58     chargingStation: ChargingStation,
  59     messageId: string,
  60     messagePayload: JsonType,
  61     commandName: IncomingRequestCommand
  62   ): Promise<ResponseType> {
  63     try {
  64       // Send response message
  65       return await this.internalSendMessage(
  66         chargingStation,
  67         messageId,
  68         messagePayload,
  69         MessageType.CALL_RESULT_MESSAGE,
  70         commandName
  71       );
  72     } catch (error) {
  73       this.handleRequestError(chargingStation, commandName, error as Error);
  74     }
  75   }
  76
  77   public async sendError(
  78     chargingStation: ChargingStation,
  79     messageId: string,
  80     ocppError: OCPPError,
  81     commandName: RequestCommand | IncomingRequestCommand
  82   ): Promise<ResponseType> {
  83     try {
  84       // Send error message
  85       return await this.internalSendMessage(
  86         chargingStation,
  87         messageId,
  88         ocppError,
  89         MessageType.CALL_ERROR_MESSAGE,
  90         commandName
  91       );
  92     } catch (error) {
  93       this.handleRequestError(chargingStation, commandName, error as Error);
  94     }
  95   }
  96
  97   protected async sendMessage(
  98     chargingStation: ChargingStation,
  99     messageId: string,
 100     messagePayload: JsonType,
 101     commandName: RequestCommand,
 102     params: RequestParams = {
 103       skipBufferingOnError: false,
 104       triggerMessage: false,
 105     }
 106   ): Promise<ResponseType> {
 107     try {
 108       return await this.internalSendMessage(
 109         chargingStation,
 110         messageId,
 111         messagePayload,
 112         MessageType.CALL_MESSAGE,
 113         commandName,
 114         params
 115       );
 116     } catch (error) {
 117       this.handleRequestError(chargingStation, commandName, error as Error, { throwError: false });
 118     }
 119   }
 120
 121   protected validateRequestPayload<T extends JsonType>(
 122     chargingStation: ChargingStation,
 123     commandName: RequestCommand,
 124     schema: JSONSchemaType<T>,
 125     payload: T
 126   ): boolean {
 127     if (!chargingStation.getPayloadSchemaValidation()) {
 128       return true;
 129     }
 130     const validate = this.ajv.compile(schema);
 131     if (validate(payload)) {
 132       return true;
 133     }
 134     logger.error(
 135       `${chargingStation.logPrefix()} ${moduleName}.validateRequestPayload: Request PDU is invalid: %j`,
 136       validate.errors
 137     );
 138     throw new OCPPError(
 139       OCPPServiceUtils.ajvErrorsToErrorType(validate.errors),
 140       'Request PDU is invalid',
 141       commandName,
 142       JSON.stringify(validate.errors, null, 2)
 143     );
 144   }
 145
 146   private async internalSendMessage(
 147     chargingStation: ChargingStation,
 148     messageId: string,
 149     messagePayload: JsonType | OCPPError,
 150     messageType: MessageType,
 151     commandName?: RequestCommand | IncomingRequestCommand,
 152     params: RequestParams = {
 153       skipBufferingOnError: false,
 154       triggerMessage: false,
 155     }
 156   ): Promise<ResponseType> {
 157     if (
 158       (chargingStation.isInUnknownState() && commandName === RequestCommand.BOOT_NOTIFICATION) ||
 159       (!chargingStation.getOcppStrictCompliance() && chargingStation.isInUnknownState()) ||
 160       chargingStation.isInAcceptedState() ||
 161       (chargingStation.isInPendingState() &&
 162         (params.triggerMessage || messageType === MessageType.CALL_RESULT_MESSAGE))
 163     ) {
 164       // eslint-disable-next-line @typescript-eslint/no-this-alias
 165       const self = this;
 166       // Send a message through wsConnection
 167       return Utils.promiseWithTimeout(
 168         new Promise((resolve, reject) => {
 169           const messageToSend = this.buildMessageToSend(
 170             chargingStation,
 171             messageId,
 172             messagePayload,
 173             messageType,
 174             commandName,
 175             responseCallback,
 176             errorCallback
 177           );
 178           if (chargingStation.getEnableStatistics()) {
 179             chargingStation.performanceStatistics.addRequestStatistic(commandName, messageType);
 180           }
 181           // Check if wsConnection opened
 182           if (chargingStation.isWebSocketConnectionOpened()) {
 183             // Yes: Send Message
 184             const beginId = PerformanceStatistics.beginMeasure(commandName);
 185             // FIXME: Handle sending error
 186             chargingStation.wsConnection.send(messageToSend);
 187             PerformanceStatistics.endMeasure(commandName, beginId);
 188             logger.debug(
 189               `${chargingStation.logPrefix()} >> Command '${commandName}' sent ${this.getMessageTypeString(
 190                 messageType
 191               )} payload: ${messageToSend}`
 192             );
 193           } else if (!params.skipBufferingOnError) {
 194             // Buffer it
 195             chargingStation.bufferMessage(messageToSend);
 196             const ocppError = new OCPPError(
 197               ErrorType.GENERIC_ERROR,
 198               `WebSocket closed for buffered message id '${messageId}' with content '${messageToSend}'`,
 199               commandName,
 200               (messagePayload as JsonObject)?.details ?? {}
 201             );
 202             if (messageType === MessageType.CALL_MESSAGE) {
 203               // Reject it but keep the request in the cache
 204               return reject(ocppError);
 205             }
 206             return errorCallback(ocppError, false);
 207           } else {
 208             // Reject it
 209             return errorCallback(
 210               new OCPPError(
 211                 ErrorType.GENERIC_ERROR,
 212                 `WebSocket closed for non buffered message id '${messageId}' with content '${messageToSend}'`,
 213                 commandName,
 214                 (messagePayload as JsonObject)?.details ?? {}
 215               ),
 216               false
 217             );
 218           }
 219           // Response?
 220           if (messageType !== MessageType.CALL_MESSAGE) {
 221             // Yes: send Ok
 222             return resolve(messagePayload);
 223           }
 224
 225           /**
 226            * Function that will receive the request's response
 227            *
 228            * @param payload
 229            * @param requestPayload
 230            */
 231           async function responseCallback(
 232             payload: JsonType,
 233             requestPayload: JsonType
 234           ): Promise<void> {
 235             if (chargingStation.getEnableStatistics()) {
 236               chargingStation.performanceStatistics.addRequestStatistic(
 237                 commandName,
 238                 MessageType.CALL_RESULT_MESSAGE
 239               );
 240             }
 241             // Handle the request's response
 242             try {
 243               await self.ocppResponseService.responseHandler(
 244                 chargingStation,
 245                 commandName as RequestCommand,
 246                 payload,
 247                 requestPayload
 248               );
 249               resolve(payload);
 250             } catch (error) {
 251               reject(error);
 252             } finally {
 253               chargingStation.requests.delete(messageId);
 254             }
 255           }
 256
 257           /**
 258            * Function that will receive the request's error response
 259            *
 260            * @param error
 261            * @param requestStatistic
 262            */
 263           function errorCallback(error: OCPPError, requestStatistic = true): void {
 264             if (requestStatistic && chargingStation.getEnableStatistics()) {
 265               chargingStation.performanceStatistics.addRequestStatistic(
 266                 commandName,
 267                 MessageType.CALL_ERROR_MESSAGE
 268               );
 269             }
 270             logger.error(
 271               `${chargingStation.logPrefix()} Error %j occurred when calling command %s with message data %j`,
 272               error,
 273               commandName,
 274               messagePayload
 275             );
 276             chargingStation.requests.delete(messageId);
 277             reject(error);
 278           }
 279         }),
 280         Constants.OCPP_WEBSOCKET_TIMEOUT,
 281         new OCPPError(
 282           ErrorType.GENERIC_ERROR,
 283           `Timeout for message id '${messageId}'`,
 284           commandName,
 285           (messagePayload as JsonObject)?.details ?? {}
 286         ),
 287         () => {
 288           messageType === MessageType.CALL_MESSAGE && chargingStation.requests.delete(messageId);
 289         }
 290       );
 291     }
 292     throw new OCPPError(
 293       ErrorType.SECURITY_ERROR,
 294       `Cannot send command ${commandName} PDU when the charging station is in ${chargingStation.getRegistrationStatus()} state on the central server`,
 295       commandName
 296     );
 297   }
 298
 299   private buildMessageToSend(
 300     chargingStation: ChargingStation,
 301     messageId: string,
 302     messagePayload: JsonType | OCPPError,
 303     messageType: MessageType,
 304     commandName?: RequestCommand | IncomingRequestCommand,
 305     responseCallback?: (payload: JsonType, requestPayload: JsonType) => Promise<void>,
 306     errorCallback?: (error: OCPPError, requestStatistic?: boolean) => void
 307   ): string {
 308     let messageToSend: string;
 309     // Type of message
 310     switch (messageType) {
 311       // Request
 312       case MessageType.CALL_MESSAGE:
 313         // Build request
 314         chargingStation.requests.set(messageId, [
 315           responseCallback,
 316           errorCallback,
 317           commandName,
 318           messagePayload as JsonType,
 319         ]);
 320         messageToSend = JSON.stringify([
 321           messageType,
 322           messageId,
 323           commandName,
 324           messagePayload,
 325         ] as OutgoingRequest);
 326         break;
 327       // Response
 328       case MessageType.CALL_RESULT_MESSAGE:
 329         // Build response
 330         messageToSend = JSON.stringify([messageType, messageId, messagePayload] as Response);
 331         break;
 332       // Error Message
 333       case MessageType.CALL_ERROR_MESSAGE:
 334         // Build Error Message
 335         messageToSend = JSON.stringify([
 336           messageType,
 337           messageId,
 338           (messagePayload as OCPPError)?.code ?? ErrorType.GENERIC_ERROR,
 339           (messagePayload as OCPPError)?.message ?? '',
 340           (messagePayload as OCPPError)?.details ?? { commandName },
 341         ] as ErrorResponse);
 342         break;
 343     }
 344     return messageToSend;
 345   }
 346
 347   private getMessageTypeString(messageType: MessageType): string {
 348     switch (messageType) {
 349       case MessageType.CALL_MESSAGE:
 350         return 'request';
 351       case MessageType.CALL_RESULT_MESSAGE:
 352         return 'response';
 353       case MessageType.CALL_ERROR_MESSAGE:
 354         return 'error';
 355     }
 356   }
 357
 358   private handleRequestError(
 359     chargingStation: ChargingStation,
 360     commandName: RequestCommand | IncomingRequestCommand,
 361     error: Error,
 362     params: HandleErrorParams<EmptyObject> = { throwError: true }
 363   ): void {
 364     logger.error(chargingStation.logPrefix() + ' Request command %s error: %j', commandName, error);
 365     if (params?.throwError) {
 366       throw error;
 367     }
 368   }
 369
 370   // eslint-disable-next-line @typescript-eslint/no-unused-vars
 371   public abstract requestHandler<Request extends JsonType, Response extends JsonType>(
 372     chargingStation: ChargingStation,
 373     commandName: RequestCommand,
 374     commandParams?: JsonType,
 375     params?: RequestParams
 376   ): Promise<Response>;
 377 }