acs/requestprocessor/src/main/java/org/commscope/tr069adapter/acs/requestprocessor/impl/TR069RequestProcessEngine.java

   1 /*\r
   2  * ============LICENSE_START========================================================================\r
   3  * ONAP : tr-069-adapter\r
   4  * =================================================================================================\r
   5  * Copyright (C) 2020 CommScope Inc Intellectual Property.\r
   6  * =================================================================================================\r
   7  * This tr-069-adapter software file is distributed by CommScope Inc under the Apache License,\r
   8  * Version 2.0 (the "License"); you may not use this file except in compliance with the License. You\r
   9  * may obtain a copy of the License at\r
  10  *\r
  11  * http://www.apache.org/licenses/LICENSE-2.0\r
  12  *\r
  13  * This file is distributed on an "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND,\r
  14  * either express or implied. See the License for the specific language governing permissions and\r
  15  * limitations under the License.\r
  16  * ===============LICENSE_END=======================================================================\r
  17  */\r
  18 \r
  19 package org.commscope.tr069adapter.acs.requestprocessor.impl;\r
  20 \r
  21 import static org.commscope.tr069adapter.acs.common.utils.AcsConstants.SESSION_ID;\r
  22 \r
  23 import java.util.List;\r
  24 \r
  25 import org.commscope.tr069adapter.acs.common.DeviceDetails;\r
  26 import org.commscope.tr069adapter.acs.common.DeviceInform;\r
  27 import org.commscope.tr069adapter.acs.common.DeviceRPCRequest;\r
  28 import org.commscope.tr069adapter.acs.common.DeviceRPCResponse;\r
  29 import org.commscope.tr069adapter.acs.common.OperationCode;\r
  30 import org.commscope.tr069adapter.acs.common.OperationDetails;\r
  31 import org.commscope.tr069adapter.acs.common.dto.CustomOperationCode;\r
  32 import org.commscope.tr069adapter.acs.common.dto.DeviceOperationRequestDetails;\r
  33 import org.commscope.tr069adapter.acs.common.dto.TR069DeviceDetails;\r
  34 import org.commscope.tr069adapter.acs.common.dto.TR069InformType;\r
  35 import org.commscope.tr069adapter.acs.common.dto.TR069OperationCode;\r
  36 import org.commscope.tr069adapter.acs.common.dto.TR069OperationDetails;\r
  37 import org.commscope.tr069adapter.acs.common.exception.DeviceOperationException;\r
  38 import org.commscope.tr069adapter.acs.common.exception.SessionConcurrentAccessException;\r
  39 import org.commscope.tr069adapter.acs.common.exception.SessionManagerException;\r
  40 import org.commscope.tr069adapter.acs.common.exception.TR069EventProcessingException;\r
  41 import org.commscope.tr069adapter.acs.common.faults.AcsFaultCode;\r
  42 import org.commscope.tr069adapter.acs.common.inform.BootInform;\r
  43 import org.commscope.tr069adapter.acs.common.inform.BootstrapInform;\r
  44 import org.commscope.tr069adapter.acs.common.response.DeviceInformResponse;\r
  45 import org.commscope.tr069adapter.acs.common.utils.ErrorCode;\r
  46 import org.commscope.tr069adapter.acs.requestprocessor.DeviceOperationInterface;\r
  47 import org.commscope.tr069adapter.acs.requestprocessor.dao.DeviceRPCRequestRepositoryHelper;\r
  48 import org.commscope.tr069adapter.acs.requestprocessor.dto.CustomOperationData;\r
  49 import org.commscope.tr069adapter.acs.requestprocessor.dto.SessionDTO;\r
  50 import org.commscope.tr069adapter.acs.requestprocessor.dto.SessionState;\r
  51 import org.commscope.tr069adapter.acs.requestprocessor.dto.TR069RequestProcessorData;\r
  52 import org.commscope.tr069adapter.acs.requestprocessor.entity.TR069DeviceRPCRequestEntity;\r
  53 import org.commscope.tr069adapter.acs.requestprocessor.helper.TR069RequestProcessEngineHelper;\r
  54 import org.commscope.tr069adapter.acs.requestprocessor.util.TR069RequestProcessorUtility;\r
  55 import org.slf4j.Logger;\r
  56 import org.slf4j.LoggerFactory;\r
  57 import org.springframework.beans.factory.annotation.Autowired;\r
  58 import org.springframework.stereotype.Component;\r
  59 \r
  60 @Component\r
  61 public class TR069RequestProcessEngine extends TR069RequestProcessEngineHelper {\r
  62 \r
  63   private static final String PENDING_RPC_CHECK =\r
  64       "Checking if any pending Device RPC requests exists for the device";\r
  65 \r
  66   private static final Logger logger = LoggerFactory.getLogger(TR069RequestProcessEngine.class);\r
  67 \r
  68   @Autowired\r
  69   TR069EventNotificationService tr069EventNotificationService;\r
  70 \r
  71   @Autowired\r
  72   DeviceOperationInterface deviceOperationInterface;\r
  73 \r
  74   @Autowired\r
  75   protected DeviceRPCRequestRepositoryHelper deviceRPCRequestRepositoryHelper;\r
  76 \r
  77   /**\r
  78    * @param deviceRPCRequest\r
  79    * @throws TR069EventProcessingException\r
  80    * @throws SessionConcurrentAccessException\r
  81    */\r
  82   public void processDeviceRPCRequest(DeviceRPCRequest deviceRPCRequest)\r
  83       throws TR069EventProcessingException, SessionConcurrentAccessException {\r
  84 \r
  85     DeviceRPCResponse deviceRPCResponse = null;\r
  86     String deviceId = null;\r
  87 \r
  88     try {\r
  89       if (deviceRPCRequest == null) {\r
  90         TR069EventProcessingException ex =\r
  91             new TR069EventProcessingException(ErrorCode.INVALID_NBI_REQUEST);\r
  92         logger.error(ex.getMessage());\r
  93         throw ex;\r
  94       }\r
  95 \r
  96       Long operationId = deviceRPCRequest.getOperationId();\r
  97       logger.info("A Mapper request is received with operationID: {}", operationId);\r
  98       TR069DeviceDetails tr069DeviceDetails = null;\r
  99       deviceId = deviceRPCRequest.getDeviceDetails().getDeviceId();\r
 100       try {\r
 101         tr069DeviceDetails = deviceOperationInterface.getDeviceDetails(deviceId);\r
 102       } catch (DeviceOperationException deo) {\r
 103         logger.error(deo.getMessage());\r
 104         deviceRPCResponse = tr069RequestProcessEngineUtility.buildAbortedOperationresult(\r
 105             tr069DeviceDetails, deviceRPCRequest, AcsFaultCode.FAULT_CODE_8000);\r
 106         return;\r
 107       }\r
 108 \r
 109       try {\r
 110         SessionDTO sessionDTO = acquireSessionLockWithRetryOnFailure(deviceId, operationId);\r
 111 \r
 112         logger.debug("Persisting the Device RPC request, with operation ID: {}", operationId);\r
 113         List<TR069DeviceRPCRequestEntity> tr069DeviceRPCRequestEntities =\r
 114             TR069RequestProcessorUtility.convertToEntity(deviceRPCRequest);\r
 115         deviceRPCRequestRepositoryHelper.saveAll(tr069DeviceRPCRequestEntities);\r
 116         logger.info("Successfully persisted the Device RPC request");\r
 117 \r
 118         if (sessionDTO != null) {\r
 119           if (SessionState.TERMINATED.equals(sessionDTO.getSessionState())) {\r
 120             logger.debug("No active session exists, hence requesting for Connection request");\r
 121             requestForConnectionRequestInform(tr069DeviceDetails);\r
 122 \r
 123             // Start Request Timer\r
 124             startDeviceRPCRequestTimer(deviceId, deviceRPCRequest.getOperationId(),\r
 125                 deviceRPCRequest.getOptions().getExecutionTimeout());\r
 126           } else {\r
 127             logger.debug(\r
 128                 "Session is in processing state, Will be notified to session manager to pick the request on session availability");\r
 129           }\r
 130         } else {\r
 131           logger.warn(\r
 132               "The device is not activated yet, hence the NBI Operation result cannot be processed!");\r
 133           deviceRPCResponse = tr069RequestProcessEngineUtility.buildAbortedOperationresult(\r
 134               tr069DeviceDetails, deviceRPCRequest, AcsFaultCode.FAULT_CODE_8001);\r
 135         }\r
 136       } catch (SessionConcurrentAccessException scae) {\r
 137         throw scae;\r
 138       } catch (Exception e) {\r
 139         logger.error("An unknown exception occurred while processing the NBI request, Reason: {}",\r
 140             e.getMessage());\r
 141         deviceRPCResponse = tr069RequestProcessEngineUtility.buildAbortedOperationresult(\r
 142             tr069DeviceDetails, deviceRPCRequest, AcsFaultCode.FAULT_CODE_8004);\r
 143       }\r
 144     } finally {\r
 145       if (deviceRPCResponse != null) {\r
 146         logger.debug("Sending failed operation result for this NBI request");\r
 147         tr069EventNotificationService.sendOperationResultToNBI(deviceRPCResponse);\r
 148         // Marking the NBI Request as processed\r
 149         deviceRPCRequestRepositoryHelper.markDeviceRPCRequestAsProcessed(deviceId,\r
 150             deviceRPCRequest.getOperationId());\r
 151       }\r
 152     }\r
 153   }\r
 154 \r
 155   /**\r
 156    * Common Step 1. Since there can exist only one Inform from any device, which will be the\r
 157    * initiator of the session, following steps to be followed a. Stop the session timer for this\r
 158    * device if any running. b. Get the lock by reading the session record for this device from DB\r
 159    * and load the session object in Thread Local Cache c. Create a new session id from\r
 160    * SessionManager, and update the session object with new Session id and change the state to\r
 161    * PROCESSING. 2. Read the Device record from the DB, and load the device DTO also in the\r
 162    * ThreadLocalCache\r
 163    * \r
 164    * Common Notification Specific Step 1. Take the connection request URL from the\r
 165    * deviceNotification object 2. Update the Device DTO object with connection request URL,\r
 166    * swVersion, hwVersion if there is a difference. Update the last updated time if updated 3. Send\r
 167    * this Inform to the NBI, by calling the ProcessDeviceInform method on TR069NBIService module. 4.\r
 168    * Create the Inform response Object and update the sessionID used for this transaction in the\r
 169    * response which will be used in the cookie of the HTTP response 5. Post the message into\r
 170    * Response Queue 6. Change the session state to LOCKED 7. Save the Session and device records\r
 171    * with the values in ThreadLocalCache 8. Start the session timer and default request timer (As\r
 172    * configured for all the requests from device) 8. All the above steps to be executed in a single\r
 173    * transaction.\r
 174    * \r
 175    * @param deviceNotification\r
 176    * @return\r
 177    * @throws SessionConcurrentAccessException\r
 178    */\r
 179   public DeviceInformResponse processDeviceInform(DeviceInform deviceNotification)\r
 180       throws SessionConcurrentAccessException {\r
 181 \r
 182     String deviceId = deviceNotification.getDeviceDetails().getDeviceId();\r
 183     TR069InformType notificationType = (TR069InformType) deviceNotification.getInformType();\r
 184     logger.info("Processing the Device Inform Event: '{}'", notificationType.getNotificationCode());\r
 185     String newSessionId = null;\r
 186     SessionDTO session = null;\r
 187     TR069DeviceDetails deviceDetails = null;\r
 188     DeviceInformResponse informResponse = null;\r
 189 \r
 190     try {\r
 191 \r
 192       SessionDTO sessionDTO = getSession(deviceId);\r
 193       if (sessionDTO != null && !SessionState.TERMINATED.equals(sessionDTO.getSessionState())) {\r
 194         String sessionId = sessionDTO.getSessionId();\r
 195         logger.debug(\r
 196             "The session with session id {} is not terminated, hence stopping the associated timer",\r
 197             sessionId);\r
 198         stopSessionTimer(sessionDTO.getSessionId());\r
 199       }\r
 200 \r
 201       // To stop the request timer if any running for this device, and send failed operation\r
 202       // result for any such pending cases. Requests pending in DB should not be cleared\r
 203 \r
 204       /*\r
 205        * Read any pending records in TR069_NBI_REQUEST table for this device. Send abort operation\r
 206        * result for all the pending requests Delete the records from the TR069_NBI_Request table for\r
 207        * this device.\r
 208        */\r
 209       if (deviceNotification instanceof BootstrapInform\r
 210           || deviceNotification instanceof BootInform) {\r
 211         sendAbortedOperationResultForPendingRPCRequests(deviceNotification.getDeviceDetails(),\r
 212             notificationType);\r
 213       }\r
 214 \r
 215       session = acquireSessionLock(deviceId, notificationType, true);\r
 216       deviceDetails = getPersistedDeviceDetails(deviceId, deviceNotification);\r
 217 \r
 218       newSessionId = session.getSessionId();\r
 219       logger.debug("The session id generated to process the device notification request is: {} ",\r
 220           newSessionId);\r
 221 \r
 222       initThreadLocalCache(deviceDetails, session);\r
 223       TR069RequestProcessorData tr069RequestProcessorData = getTR069RequestProcessorData();\r
 224       updateSessionOnDeviceNotification(tr069RequestProcessorData, newSessionId);\r
 225 \r
 226       logger.debug("Sending the Device Inform Event to the Mapper");\r
 227       tr069EventNotificationService.sendDeviceInformToNBI(deviceNotification);\r
 228 \r
 229       updateDeviceDetailsFromInform(tr069RequestProcessorData, deviceNotification);\r
 230 \r
 231       logger.debug("Updating the session for the device with newly generated session id");\r
 232       changeSessionState(tr069RequestProcessorData, SessionState.LOCKED);\r
 233       updateSession(session);\r
 234 \r
 235       // Start session timer - Get a default timeout for session.\r
 236       startSessionTimer(session.getSessionId());\r
 237 \r
 238       informResponse =\r
 239           new DeviceInformResponse(newSessionId, deviceNotification.getDeviceDetails());\r
 240     } catch (\r
 241 \r
 242     DeviceOperationException doe) {\r
 243       logger.error(doe.getMessage());\r
 244     } catch (SessionConcurrentAccessException scae) {\r
 245       throw scae;\r
 246     } catch (Exception e) {\r
 247       throw new SessionConcurrentAccessException(ErrorCode.UNKNOWN_ERROR, e.getMessage());\r
 248     }\r
 249     return informResponse;\r
 250   }\r
 251 \r
 252   /**\r
 253    * \r
 254    * 1. Stop the request timer for this device using the session ID received in the cookie of the\r
 255    * request 2. Read the session record from the session table into ThreadLocalCache 3. Move the\r
 256    * session state to PROCESSING and update the OPERATION_ID as null. 4a. Session Manager to check\r
 257    * if any pending request being notified for the device using in memory cache - Not planned for\r
 258    * this Drop 4. As an interim solution for drop1, Check the TR069_NBI_REQUEST table if any request\r
 259    * is pending.\r
 260    * \r
 261    * if (anyPendingRequestExists) { 1. Pick the request with the least created_time 2. Create the\r
 262    * response object and update the sessionID used for this transaction in the response which will\r
 263    * be used in the cookie of the HTTP response 3. Post the response object into Response Queue 4.\r
 264    * Change the session state to LOCKED and update the OPERATION_ID in the session table with the\r
 265    * NBI requests OPERATION_ID. 5. Save the Session record with the values in ThreadLocalCache 6.\r
 266    * Start the session timer and request timer (available in the request) } else { 1. Move the\r
 267    * session state to PROCESSING and update the OPERATION_ID as null. 2. Save the Session record\r
 268    * with the values in ThreadLocalCache. 3. Start the session timer. }\r
 269    * \r
 270    * 6. All the above steps to be executed in a single transaction\r
 271    * \r
 272    * @param deviceRPCResponse\r
 273    * @return\r
 274    * @throws SessionConcurrentAccessException\r
 275    */\r
 276   public DeviceRPCRequest processDeviceRPCResponse(DeviceRPCResponse deviceRPCResponse)\r
 277       throws SessionConcurrentAccessException {\r
 278     TR069DeviceDetails deviceDetails = (TR069DeviceDetails) deviceRPCResponse.getDeviceDetails();\r
 279     String deviceId = deviceDetails.getDeviceId();\r
 280     logger.info("Processing the operation response from device");\r
 281 \r
 282     SessionDTO session;\r
 283     DeviceRPCRequest deviceRPCRequest = null;\r
 284     try {\r
 285       session = acquireSessionLock(deviceId, null, false);\r
 286       String newSessionId = session.getSessionId();\r
 287       logger.debug("The session id used to process the device RPC response is: {}", newSessionId);\r
 288 \r
 289       initThreadLocalCache(deviceDetails, session);\r
 290       TR069RequestProcessorData tr069RequestProcessorData = getTR069RequestProcessorData();\r
 291       updateSessionOnDeviceNotification(tr069RequestProcessorData, newSessionId);\r
 292 \r
 293       Long operationId = deviceRPCResponse.getOperationId();\r
 294 \r
 295       deviceRPCRequest = getNextRPCRequest(deviceId, operationId);\r
 296 \r
 297       if (deviceRPCRequest != null) {\r
 298         OperationDetails opDetails = deviceRPCRequest.getOpDetails();\r
 299         if (opDetails != null) {\r
 300           OperationCode opCode = opDetails.getOpCode();\r
 301           if (opCode instanceof CustomOperationCode) {\r
 302             CustomOperationCode customOperationCode = (CustomOperationCode) opCode;\r
 303             String customOperationCodeName = customOperationCode.name();\r
 304             logger.info(\r
 305                 "The Device RPC request is of custom type, the custom operation to be performed is: {}",\r
 306                 customOperationCodeName);\r
 307             String jndiName = customOperationCode.getJndiName();\r
 308             CustomOperationData customOperationData =\r
 309                 new CustomOperationData(deviceDetails, deviceRPCResponse, deviceRPCRequest);\r
 310             customOperationData = executeCustomOperation(jndiName, customOperationData);\r
 311 \r
 312             DeviceRPCRequest operationRequest = customOperationData.getDeviceRPCRequest();\r
 313             deviceRPCResponse = customOperationData.getDeviceRPCResponse();\r
 314             if (operationRequest != null) {\r
 315               operationRequest.addContextParam(SESSION_ID, newSessionId);\r
 316               updateSessionCurOpId(tr069RequestProcessorData, deviceRPCRequest.getOperationId());\r
 317               changeSessionState(tr069RequestProcessorData, SessionState.LOCKED);\r
 318               updateSession(session);\r
 319               if (deviceRPCResponse != null && operationRequest.getOperationId() != null\r
 320                   && !operationRequest.getOperationId()\r
 321                       .equals(deviceRPCResponse.getOperationId())) {\r
 322                 logger.debug(\r
 323                     "Sending the Device RPC response for a configure Multiple object prior operation");\r
 324                 // Sending the operation response to NBI\r
 325                 tr069EventNotificationService.sendOperationResultToNBI(deviceRPCResponse);\r
 326               }\r
 327               return operationRequest;\r
 328             } else {\r
 329               logger.debug(PENDING_RPC_CHECK);\r
 330               deviceRPCRequest =\r
 331                   deviceRPCRequestRepositoryHelper.findOldestDeviceRPCRequest(deviceId);\r
 332             }\r
 333           }\r
 334         }\r
 335       }\r
 336 \r
 337       if (deviceRPCRequest != null) {\r
 338         logger.info("A pending Device RPC request exists for the device with operation ID: {}",\r
 339             deviceRPCRequest.getOperationId());\r
 340         updateSessionCurOpId(tr069RequestProcessorData, deviceRPCRequest.getOperationId());\r
 341         changeSessionState(tr069RequestProcessorData, SessionState.LOCKED);\r
 342 \r
 343         deviceRPCRequest.addContextParam(SESSION_ID, newSessionId);\r
 344 \r
 345       } else {\r
 346         logger.info(\r
 347             "No pending Device RPC request exists for the device, hence empty response will be sent to the device");\r
 348         logger.debug("Updating the session to terminated state");\r
 349         // To stop the session timer if any running for this device.\r
 350         stopSessionTimer(newSessionId);\r
 351         changeSessionState(tr069RequestProcessorData, SessionState.TERMINATED);\r
 352       }\r
 353 \r
 354       // To stop the request timer if any running for this operation.\r
 355       stopDeviceRPCRequestTimer(deviceId, deviceRPCResponse.getOperationId());\r
 356 \r
 357       // Sending the operation response to NBI\r
 358       logger.debug("Sending the Device RPC Response to the Mapper");\r
 359       tr069EventNotificationService.sendOperationResultToNBI(deviceRPCResponse);\r
 360 \r
 361       updateSession(session);\r
 362     } catch (DeviceOperationException doe) {\r
 363       logger.error(doe.getMessage());\r
 364     } catch (SessionConcurrentAccessException scae) {\r
 365       throw scae;\r
 366     }\r
 367 \r
 368     return deviceRPCRequest;\r
 369   }\r
 370 \r
 371   /**\r
 372    * \r
 373    * 1. Stop the request timer for this device using the session ID received in the cookie of the\r
 374    * request 2. Read the session record from the session table into ThreadLocalCache 3. Move the\r
 375    * session state to PROCESSING and update the OPERATION_ID as null. 4a. Session Manager to check\r
 376    * if any pending request being notified for the device using in memory cache - Not planned for\r
 377    * this Drop 4. As an interim solution for drop1, Check the TR069_NBI_REQUEST table if any request\r
 378    * is pending.\r
 379    * \r
 380    * if (anyPendingRequestExists) { 1. Pick the request with the least created_time 2. Create the\r
 381    * response object and update the sessionID used for this transaction in the response which will\r
 382    * be used in the cookie of the HTTP response 3. Post the response object into Response Queue 4.\r
 383    * Change the session state to LOCKED and update the OPERATION_ID in the session table with the\r
 384    * NBI requests OPERATION_ID. 5. Save the Session record with the values in ThreadLocalCache 6.\r
 385    * Start the session timer and request timer (available in the request) } else { 1. Move the\r
 386    * session state to PROCESSING and update the OPERATION_ID as null. 2. Save the Session record\r
 387    * with the values in ThreadLocalCache. 3. Start the session timer. }\r
 388    * \r
 389    * 6. All the above steps to be executed in a single transaction\r
 390    * \r
 391    * @param deviceDetails\r
 392    * @return\r
 393    * @throws SessionConcurrentAccessException\r
 394    */\r
 395   public DeviceRPCRequest processEmptyDeviceRequest(TR069DeviceDetails deviceDetails)\r
 396       throws SessionConcurrentAccessException {\r
 397     String deviceId = deviceDetails.getDeviceId();\r
 398     logger.info("Processing the empty request from device");\r
 399 \r
 400     SessionDTO session;\r
 401     DeviceRPCRequest nbiDeviceOperationRequest = null;\r
 402     try {\r
 403       session = acquireSessionLock(deviceId, null, false);\r
 404       String newSessionId = session.getSessionId();\r
 405       logger.debug("The session id used to process the empty device request is: {}", newSessionId);\r
 406 \r
 407       initThreadLocalCache(deviceDetails, session);\r
 408       TR069RequestProcessorData tr069RequestProcessorData = getTR069RequestProcessorData();\r
 409       updateSessionOnDeviceNotification(tr069RequestProcessorData, newSessionId);\r
 410 \r
 411       logger.debug(PENDING_RPC_CHECK);\r
 412       nbiDeviceOperationRequest =\r
 413           deviceRPCRequestRepositoryHelper.findOldestDeviceRPCRequest(deviceId);\r
 414       if (nbiDeviceOperationRequest != null) {\r
 415         Long operationId = nbiDeviceOperationRequest.getOperationId();\r
 416         OperationDetails opDetails = nbiDeviceOperationRequest.getOpDetails();\r
 417         if (opDetails != null) {\r
 418           OperationCode opCode = opDetails.getOpCode();\r
 419           if (opCode instanceof CustomOperationCode) {\r
 420             CustomOperationCode customOperationCode = (CustomOperationCode) opCode;\r
 421             String customOperationCodeName = customOperationCode.name();\r
 422             logger.info(\r
 423                 "The Device RPC operation request is of custom type, the custom operation to be performed is: {}",\r
 424                 customOperationCodeName);\r
 425             String jndiName = customOperationCode.getJndiName();\r
 426             CustomOperationData customOperationData =\r
 427                 new CustomOperationData(deviceDetails, null, nbiDeviceOperationRequest);\r
 428             customOperationData = executeCustomOperation(jndiName, customOperationData);\r
 429 \r
 430             DeviceRPCRequest operationRequest = customOperationData.getDeviceRPCRequest();\r
 431             if (operationRequest != null) {\r
 432               operationRequest.addContextParam(SESSION_ID, newSessionId);\r
 433               updateSessionCurOpId(tr069RequestProcessorData, operationId);\r
 434               changeSessionState(tr069RequestProcessorData, SessionState.LOCKED);\r
 435               updateSession(session);\r
 436               return operationRequest;\r
 437             } else {\r
 438               logger.debug(PENDING_RPC_CHECK);\r
 439               nbiDeviceOperationRequest =\r
 440                   deviceRPCRequestRepositoryHelper.findOldestDeviceRPCRequest(deviceId);\r
 441             }\r
 442           }\r
 443         }\r
 444       }\r
 445 \r
 446       if (nbiDeviceOperationRequest != null) {\r
 447         Long operationId = nbiDeviceOperationRequest.getOperationId();\r
 448         logger.info("A pending Device RPC request exists for the device with operation ID: {}",\r
 449             operationId);\r
 450         updateSessionCurOpId(tr069RequestProcessorData, operationId);\r
 451         changeSessionState(tr069RequestProcessorData, SessionState.LOCKED);\r
 452 \r
 453         nbiDeviceOperationRequest.addContextParam(SESSION_ID, newSessionId);\r
 454 \r
 455       } else {\r
 456         logger.info(\r
 457             "No pending Device RPC request exists for the device, hence empty device response will be sent to the device");\r
 458         logger.debug("Updating the session to terminated state");\r
 459         // To stop the session timer if any running for this device.\r
 460         stopSessionTimer(newSessionId);\r
 461         changeSessionState(tr069RequestProcessorData, SessionState.TERMINATED);\r
 462       }\r
 463 \r
 464       updateSession(session);\r
 465     } catch (DeviceOperationException doe) {\r
 466       logger.error(doe.getMessage());\r
 467     } catch (SessionConcurrentAccessException scae) {\r
 468       throw scae;\r
 469     }\r
 470 \r
 471     return nbiDeviceOperationRequest;\r
 472   }\r
 473 \r
 474   /**\r
 475    * @param sessionId\r
 476    * @return\r
 477    * @throws SessionManagerException\r
 478    */\r
 479   public DeviceOperationRequestDetails getOpRequestDetailsBySessionId(String sessionId)\r
 480       throws SessionManagerException {\r
 481     DeviceOperationRequestDetails deviceOperationRequestDetails =\r
 482         new DeviceOperationRequestDetails();\r
 483 \r
 484     logger.debug("Fetching Operation request details for session: {}", sessionId);\r
 485     SessionDTO session = getSessionBySessionId(sessionId);\r
 486     String deviceId = session.getDeviceId();\r
 487     TR069DeviceDetails deviceDetails = null;\r
 488     try {\r
 489       deviceDetails = deviceOperationInterface.getDeviceDetails(deviceId);\r
 490       if (session.getCurrentOperationId() == null) {\r
 491         logger.debug("There exists no pending operation request for the session: {}", sessionId);\r
 492       } else {\r
 493         logger.debug("There exists pending operation request for the session: {}", sessionId);\r
 494         List<TR069DeviceRPCRequestEntity> tr069DeviceRPCRequestEntityList =\r
 495             deviceRPCRequestRepositoryHelper.findByDeviceIdAndOperationId(deviceId,\r
 496                 session.getCurrentOperationId());\r
 497         if (tr069DeviceRPCRequestEntityList == null) {\r
 498           SessionManagerException ex =\r
 499               new SessionManagerException(ErrorCode.SESSION_EXPIRED, sessionId);\r
 500           logger.error(ex.getMessage());\r
 501           throw ex;\r
 502         }\r
 503         DeviceRPCRequest deviceRPCRequest =\r
 504             TR069RequestProcessorUtility.convertToDTO(tr069DeviceRPCRequestEntityList);\r
 505         OperationCode opCode = deviceRPCRequest.getOpDetails().getOpCode();\r
 506 \r
 507         String operationName = null;\r
 508 \r
 509         if (opCode instanceof TR069OperationCode) {\r
 510           operationName = ((TR069OperationCode) opCode).name();\r
 511         } else {\r
 512           operationName = ((CustomOperationCode) opCode).name();\r
 513           TR069OperationDetails tr069OperationDetails =\r
 514               (TR069OperationDetails) deviceRPCRequest.getOpDetails();\r
 515           opCode = getCustomOperationCode(tr069OperationDetails);\r
 516         }\r
 517         logger.info("The pending operation request for the session is of operation {}",\r
 518             operationName);\r
 519         deviceOperationRequestDetails.setOpCode(opCode);\r
 520         deviceOperationRequestDetails.setOperationId(deviceRPCRequest.getOperationId());\r
 521       }\r
 522       deviceOperationRequestDetails.setDeviceDetails(deviceDetails);\r
 523     } catch (DeviceOperationException e) {\r
 524       SessionManagerException ex =\r
 525           new SessionManagerException(ErrorCode.DEVICE_NOT_EXISTS, deviceId);\r
 526       logger.error(ex.getMessage());\r
 527       throw ex;\r
 528     } catch (Exception e) {\r
 529       logger.error(e.getMessage());\r
 530       SessionManagerException ex =\r
 531           new SessionManagerException(ErrorCode.SESSION_EXPIRED, sessionId);\r
 532       logger.error(ex.getMessage());\r
 533       throw ex;\r
 534     }\r
 535     return deviceOperationRequestDetails;\r
 536   }\r
 537 \r
 538   /**\r
 539    * @param deviceId\r
 540    * @return\r
 541    * @throws DeviceOperationException\r
 542    */\r
 543   public TR069DeviceDetails getDeviceDetails(String deviceId) throws DeviceOperationException {\r
 544     return deviceOperationInterface.getDeviceDetails(deviceId);\r
 545   }\r
 546 \r
 547   /**\r
 548    * @param deviceId\r
 549    * @param operationId\r
 550    * @return\r
 551    * @throws DeviceOperationException\r
 552    * @throws SessionConcurrentAccessException\r
 553    * @throws InterruptedException\r
 554    */\r
 555   private SessionDTO acquireSessionLockWithRetryOnFailure(String deviceId, Long operationId)\r
 556       throws DeviceOperationException, SessionConcurrentAccessException, InterruptedException {\r
 557     int sessionLockAcquireRetryCount = 0;\r
 558     SessionDTO sessionDTO = null;\r
 559     do {\r
 560       try {\r
 561         sessionDTO = acquireSessionLock(deviceId, null, false);\r
 562         logger.info(\r
 563             "Successfully acquired the session lock for processing NBI request with operation ID: {}",\r
 564             operationId);\r
 565         break;\r
 566       } catch (SessionConcurrentAccessException ex) {\r
 567         sessionLockAcquireRetryCount++;\r
 568         if (sessionLockAcquireRetryCount == 3) {\r
 569           logger.error("Failed acquiring the lock after retry, rolling back the transaction");\r
 570           throw ex;\r
 571         }\r
 572         logger.warn(\r
 573             "Session lock acquiring failed with SessionConcurrentAccessException, hence retrying");\r
 574         Thread.sleep(1000L);\r
 575       }\r
 576     } while (sessionLockAcquireRetryCount < 3);\r
 577 \r
 578     return sessionDTO;\r
 579   }\r
 580 \r
 581   /**\r
 582    * @param deviceId\r
 583    * @param deviceNotification\r
 584    * @return\r
 585    * @throws DeviceOperationException\r
 586    */\r
 587   private TR069DeviceDetails getPersistedDeviceDetails(String deviceId,\r
 588       DeviceInform deviceNotification) throws DeviceOperationException {\r
 589     TR069DeviceDetails deviceDetails = null;\r
 590     try {\r
 591       deviceDetails = deviceOperationInterface.getDeviceDetails(deviceId);\r
 592     } catch (DeviceOperationException doe) {\r
 593       if (ErrorCode.DEVICE_NOT_EXISTS.equals(doe.getErrorCode())) {\r
 594         logger.info(\r
 595             "Creating the device record in TR069_DEVICE_ table, as the device is authenticated successfully.");\r
 596         createDevice(deviceNotification.getDeviceDetails());\r
 597         deviceDetails = deviceOperationInterface.getDeviceDetails(deviceId);\r
 598       }\r
 599     }\r
 600 \r
 601     return deviceDetails;\r
 602   }\r
 603 \r
 604   /**\r
 605    * @param deviceDetails\r
 606    * @param notificationType\r
 607    * @throws TR069EventProcessingException\r
 608    */\r
 609   private void sendAbortedOperationResultForPendingRPCRequests(DeviceDetails deviceDetails,\r
 610       TR069InformType notificationType) throws TR069EventProcessingException {\r
 611     String deviceId = deviceDetails.getDeviceId();\r
 612     String notificationName = notificationType.name();\r
 613     logger.debug(\r
 614         "Device Inform event received is {}, hence aborting all the pending operations if any exists",\r
 615         notificationName);\r
 616 \r
 617     List<DeviceRPCRequest> deviceRPCRequestList =\r
 618         deviceRPCRequestRepositoryHelper.findAllDeviceRPCRequests(deviceId);\r
 619 \r
 620     for (DeviceRPCRequest pendingDeviceRPCRequest : deviceRPCRequestList) {\r
 621       DeviceRPCResponse deviceOpResult =\r
 622           tr069RequestProcessEngineUtility.buildAbortedOperationresult(deviceDetails,\r
 623               pendingDeviceRPCRequest, AcsFaultCode.FAULT_CODE_8002);\r
 624       String operationName = null;\r
 625       if (pendingDeviceRPCRequest.getOpDetails().getOpCode() instanceof CustomOperationCode) {\r
 626         CustomOperationCode operationCode =\r
 627             (CustomOperationCode) pendingDeviceRPCRequest.getOpDetails().getOpCode();\r
 628         operationName = operationCode.name();\r
 629       } else {\r
 630         TR069OperationCode operationCode =\r
 631             (TR069OperationCode) pendingDeviceRPCRequest.getOpDetails().getOpCode();\r
 632         operationName = operationCode.name();\r
 633       }\r
 634       Long operationId = pendingDeviceRPCRequest.getOperationId();\r
 635       logger.debug("Aborting the NBI Operation request with operation Id : {} for operation: {}",\r
 636           operationId, operationName);\r
 637       tr069EventNotificationService.sendOperationResultToNBI(deviceOpResult);\r
 638       // Marking the NBI Request as processed\r
 639       deviceRPCRequestRepositoryHelper.markDeviceRPCRequestAsProcessed(deviceId, operationId);\r
 640       stopDeviceRPCRequestTimer(deviceId, operationId);\r
 641     }\r
 642   }\r
 643 \r
 644   /**\r
 645    * @param tr069RequestProcessorData\r
 646    * @param deviceNotification\r
 647    */\r
 648   private void updateDeviceDetailsFromInform(TR069RequestProcessorData tr069RequestProcessorData,\r
 649       DeviceInform deviceNotification) {\r
 650     Boolean isDeviceDataChanged =\r
 651         isDeviceUpdateExists(tr069RequestProcessorData, deviceNotification);\r
 652     if (isDeviceDataChanged.booleanValue()) {\r
 653       updateDeviceDetails(tr069RequestProcessorData, deviceNotification);\r
 654       try {\r
 655         logger.info(\r
 656             "The device data like connection request URL/ Device SW/HW version has changed, hence updating the device details");\r
 657         deviceOperationInterface\r
 658             .updateDeviceDetails(tr069RequestProcessorData.getTr069DeviceDetails());\r
 659       } catch (DeviceOperationException e) {\r
 660         logger.error("Updating the device details with the notification details failed, Reason: {}",\r
 661             e.getMessage());\r
 662         logger.error(e.getMessage());\r
 663       }\r
 664     }\r
 665   }\r
 666 \r
 667   /**\r
 668    * @param deviceId\r
 669    * @param operationId\r
 670    * @return\r
 671    */\r
 672   private DeviceRPCRequest getNextRPCRequest(String deviceId, Long operationId) {\r
 673     DeviceRPCRequest deviceRPCRequest = null;\r
 674     try {\r
 675       List<TR069DeviceRPCRequestEntity> entityList =\r
 676           deviceRPCRequestRepositoryHelper.findByDeviceIdAndOperationId(deviceId, operationId);\r
 677       Integer operationCode = entityList.get(0).getOpCode();\r
 678       if (CustomOperationCode.getByOperationCode(operationCode) == null) {\r
 679         logger.info("Marking the Device RPC request with operation id - {} as processed.",\r
 680             operationId);\r
 681         deviceRPCRequestRepositoryHelper.markDeviceRPCRequestAsProcessed(deviceId, operationId);\r
 682       }\r
 683       logger.debug(PENDING_RPC_CHECK);\r
 684       deviceRPCRequest = deviceRPCRequestRepositoryHelper.findOldestDeviceRPCRequest(deviceId);\r
 685     } catch (TR069EventProcessingException e) {\r
 686       logger.error("An unknown exception occurred while fetching the NBI request, Reason: {}",\r
 687           e.getMessage());\r
 688     }\r
 689 \r
 690     return deviceRPCRequest;\r
 691   }\r
 692 \r
 693   /**\r
 694    * Creates the device in the DM module if factory imported already\r
 695    * \r
 696    * @param deviceDetails\r
 697    * @throws DeviceOperationException\r
 698    */\r
 699   private void createDevice(DeviceDetails deviceDetails) throws DeviceOperationException {\r
 700     deviceOperationInterface.updateDeviceDetails(deviceDetails);\r
 701   }\r
 702 }\r