PushNotificationSchedulerTask implements Runnable { @Override public void run() { try { Map<Integer, List<OperationMapping>> operationMappingsTenantMap = new HashMap<>(); List<OperationMapping> operationsCompletedList = new LinkedList<>(); if (log.isDebugEnabled()) { log.debug("Push notification job started"); } try { OperationManagementDAOFactory.openConnection(); operationMappingsTenantMap = operationDAO.getOperationMappingsByStatus(Operation.Status .PENDING, Operation.PushNotificationStatus.SCHEDULED, DeviceConfigurationManager.getInstance() .getDeviceManagementConfig().getPushNotificationConfiguration().getSchedulerBatchSize()); } catch (OperationManagementDAOException e) { log.error("Unable to retrieve scheduled pending operations for task.", e); } finally { OperationManagementDAOFactory.closeConnection(); } for (List<OperationMapping> operationMappings : operationMappingsTenantMap.values()) { for (OperationMapping operationMapping : operationMappings) { try { if (log.isDebugEnabled()) { log.debug("Sending push notification for operationId :" + operationMapping.getOperationId() + " to deviceId : " + operationMapping.getDeviceIdentifier().getId()); } PrivilegedCarbonContext.startTenantFlow(); PrivilegedCarbonContext.getThreadLocalCarbonContext().setTenantId(operationMapping.getTenantId(), true); NotificationStrategy notificationStrategy = provider.getNotificationStrategyByDeviceType (operationMapping.getDeviceIdentifier().getType()); if (notificationStrategy != null) { notificationStrategy.execute(new NotificationContext(operationMapping.getDeviceIdentifier(), provider.getOperation(operationMapping.getDeviceIdentifier().getType(), operationMapping .getOperationId()))); operationMapping.setPushNotificationStatus(Operation.PushNotificationStatus.COMPLETED); operationsCompletedList.add(operationMapping); } else { if (log.isDebugEnabled()) { log.debug("Tenant '" + PrivilegedCarbonContext.getThreadLocalCarbonContext() .getTenantDomain() + "' does not have push notification strategy."); } } } catch (DeviceManagementException e) { log.error("Error occurred while getting notification strategy for operation mapping " + operationMapping.getDeviceIdentifier().getType(), e); } catch (OperationManagementException e) { log.error("Unable to get the operation for operation " + operationMapping.getOperationId(), e); } catch (PushNotificationExecutionFailedException e) { log.error("Error occurred while sending push notification to operation: " + operationMapping .getOperationId(), e); } finally { PrivilegedCarbonContext.endTenantFlow(); } } } if (operationsCompletedList.size() > 0) { try { OperationManagementDAOFactory.beginTransaction(); operationMappingDAO.updateOperationMapping(operationsCompletedList); OperationManagementDAOFactory.commitTransaction(); } catch (TransactionManagementException | OperationManagementDAOException e) { OperationManagementDAOFactory.rollbackTransaction(); log.error("Error occurred while updating operation mappings for sent notifications ", e); } finally { OperationManagementDAOFactory.closeConnection(); } } if (log.isDebugEnabled()) { log.debug("Push notification job running completed."); } } catch (Throwable cause) { log.error("PushNotificationSchedulerTask failed due to " + cause.getMessage(), cause); } }  @Override void run();  }
@Test(description = "Tests the push notification scheduling for devices") public void testPushNotificationScheduler() throws DeviceManagementException, OperationManagementException, SQLException, OperationManagementDAOException { try { log.info("Attempting to execute push notification task scheduler"); Mockito.doReturn(new TestNotificationStrategy()).when(this.deviceMgtProviderService) .getNotificationStrategyByDeviceType(Mockito.anyString()); Mockito.doReturn(new org.wso2.carbon.device.mgt.common.operation.mgt.Operation()) .when(this.deviceMgtProviderService).getOperation(Mockito.anyString(), Mockito.anyInt()); this.pushNotificationSchedulerTask.run(); OperationManagementDAOFactory.openConnection(); Map<Integer, List<OperationMapping>> operationMappingsTenantMap = operationDAO .getOperationMappingsByStatus(Operation.Status.PENDING, Operation.PushNotificationStatus.SCHEDULED, DeviceConfigurationManager.getInstance().getDeviceManagementConfig() .getPushNotificationConfiguration().getSchedulerBatchSize()); Assert.assertEquals(operationMappingsTenantMap.size(), 0); log.info("Push notification task execution complete."); } finally { OperationManagementDAOFactory.closeConnection(); } }