Add new classes for improve Performance of push notifications

components/device-mgt-extensions/org.wso2.carbon.device.mgt.extensions.device.type.deployer/src/main/java/org/wso2/carbon/device/mgt/extensions/device/type/deployer/config/PushNotificationProvider.java

@@ -59,6 +59,8 @@ public class PushNotificationProvider {
     protected ConfigProperties configProperties;
     @XmlAttribute(name = "type")
     protected String type;
+    @XmlAttribute(name = "isScheduled")
+    protected boolean isScheduled;
 
     /**
      * Gets the value of the fileBasedProperties property.
@@ -124,4 +126,30 @@ public class PushNotificationProvider {
         this.type = value;
     }
 
+    /**
+     * Gets the value of the isScheduled property.
+     * This property will be used to determine whether to use scheduler task to send push notification
+     * If true push notification will be sent using scheduler task
+     * If false push notifications will be sent immediately.
+     *
+     * @return
+     *     possible object is
+     *     {@link Boolean }
+     *
+     */
+    public boolean isScheduled() {
+        return isScheduled;
+    }
+
+    /**
+     * Sets the value of the isScheduled property.
+     *
+     * @return
+     *     possible object is
+     *     {@link Boolean }
+     *
+     */
+    public void setScheduled(boolean scheduled) {
+        isScheduled = scheduled;
+    }
 }

components/device-mgt/org.wso2.carbon.device.mgt.common/src/main/java/org/wso2/carbon/device/mgt/common/operation/mgt/OperationMapping.java

@@ -0,0 +1,54 @@
+/*
+*  Copyright (c) 2017, WSO2 Inc. (http://www.wso2.org) All Rights Reserved.
+*
+*  WSO2 Inc. licenses this file to you under the Apache License,
+*  Version 2.0 (the "License"); you may not use this file except
+*  in compliance with the License.
+*  You may obtain a copy of the License at
+*
+*    http://www.apache.org/licenses/LICENSE-2.0
+*
+* Unless required by applicable law or agreed to in writing,
+* software distributed under the License is distributed on an
+* "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+* KIND, either express or implied.  See the License for the
+* specific language governing permissions and limitations
+* under the License.
+*/
+package org.wso2.carbon.device.mgt.common.operation.mgt;
+
+import org.wso2.carbon.device.mgt.common.DeviceIdentifier;
+
+/**
+ * Class for represent operation mapping
+ */
+public class OperationMapping {
+
+    private DeviceIdentifier deviceIdentifier;
+    private int operationId;
+    private int tenantId;
+
+    public int getOperationId() {
+        return operationId;
+    }
+
+    public void setOperationId(int operationId) {
+        this.operationId = operationId;
+    }
+
+    public int getTenantId() {
+        return tenantId;
+    }
+
+    public void setTenantId(int tenantId) {
+        this.tenantId = tenantId;
+    }
+
+    public DeviceIdentifier getDeviceIdentifier() {
+        return deviceIdentifier;
+    }
+
+    public void setDeviceIdentifier(DeviceIdentifier deviceIdentifier) {
+        this.deviceIdentifier = deviceIdentifier;
+    }
+}

components/device-mgt/org.wso2.carbon.device.mgt.core/src/main/java/org/wso2/carbon/device/mgt/core/config/push/notification/PushNotificationConfiguration.java

@@ -0,0 +1,71 @@
+/*
+*  Copyright (c) 2017, WSO2 Inc. (http://www.wso2.org) All Rights Reserved.
+*
+*  WSO2 Inc. licenses this file to you under the Apache License,
+*  Version 2.0 (the "License"); you may not use this file except
+*  in compliance with the License.
+*  You may obtain a copy of the License at
+*
+*    http://www.apache.org/licenses/LICENSE-2.0
+*
+* Unless required by applicable law or agreed to in writing,
+* software distributed under the License is distributed on an
+* "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+* KIND, either express or implied.  See the License for the
+* specific language governing permissions and limitations
+* under the License.
+*/
+package org.wso2.carbon.device.mgt.core.config.push.notification;
+
+import javax.xml.bind.annotation.XmlElement;
+import javax.xml.bind.annotation.XmlElementWrapper;
+import javax.xml.bind.annotation.XmlRootElement;
+import java.util.List;
+
+/**
+ * This class is for Push notification related Configurations
+ */
+@XmlRootElement(name = "PushNotificationConfiguration")
+public class PushNotificationConfiguration {
+
+    private int SchedulerBatchSize;
+    private int SchedulerBatchDelayMills;
+    private boolean SchedulerTaskEnabled;
+    private List<String> pushNotificationProviders;
+
+    @XmlElement(name = "SchedulerBatchSize", required = true)
+    public int getSchedulerBatchSize() {
+        return SchedulerBatchSize;
+    }
+
+    public void setSchedulerBatchSize(int SchedulerBatchSize) {
+        this.SchedulerBatchSize = SchedulerBatchSize;
+    }
+
+    @XmlElement(name = "SchedulerBatchDelayMills", required = true)
+    public int getSchedulerBatchDelayMills() {
+        return SchedulerBatchDelayMills;
+    }
+
+    public void setSchedulerBatchDelayMills(int SchedulerBatchDelayMills) {
+        this.SchedulerBatchDelayMills = SchedulerBatchDelayMills;
+    }
+    @XmlElement(name = "SchedulerTaskEnabled", required = true)
+    public boolean isSchedulerTaskEnabled() {
+        return SchedulerTaskEnabled;
+    }
+
+    public void setSchedulerTaskEnabled(boolean schedulerTaskEnabled) {
+        SchedulerTaskEnabled = schedulerTaskEnabled;
+    }
+
+    @XmlElementWrapper(name = "PushNotificationProviders", required = true)
+    @XmlElement(name = "Provider", required = true)
+    public List<String> getPushNotificationProviders() {
+        return pushNotificationProviders;
+    }
+
+    public void setPushNotificationProviders(List<String> pushNotificationProviders) {
+        this.pushNotificationProviders = pushNotificationProviders;
+    }
+}

components/device-mgt/org.wso2.carbon.device.mgt.core/src/main/java/org/wso2/carbon/device/mgt/core/push/notification/mgt/task/PushNotificationJob.java

@@ -0,0 +1,122 @@
+/*
+*  Copyright (c) 2017, WSO2 Inc. (http://www.wso2.org) All Rights Reserved.
+*
+*  WSO2 Inc. licenses this file to you under the Apache License,
+*  Version 2.0 (the "License"); you may not use this file except
+*  in compliance with the License.
+*  You may obtain a copy of the License at
+*
+*    http://www.apache.org/licenses/LICENSE-2.0
+*
+* Unless required by applicable law or agreed to in writing,
+* software distributed under the License is distributed on an
+* "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+* KIND, either express or implied.  See the License for the
+* specific language governing permissions and limitations
+* under the License.
+*/
+package org.wso2.carbon.device.mgt.core.push.notification.mgt.task;
+
+import org.apache.commons.logging.Log;
+import org.apache.commons.logging.LogFactory;
+import org.wso2.carbon.context.PrivilegedCarbonContext;
+import org.wso2.carbon.device.mgt.common.DeviceManagementException;
+import org.wso2.carbon.device.mgt.common.TransactionManagementException;
+import org.wso2.carbon.device.mgt.common.operation.mgt.OperationManagementException;
+import org.wso2.carbon.device.mgt.common.operation.mgt.OperationMapping;
+import org.wso2.carbon.device.mgt.common.push.notification.NotificationContext;
+import org.wso2.carbon.device.mgt.common.push.notification.NotificationStrategy;
+import org.wso2.carbon.device.mgt.common.push.notification.PushNotificationExecutionFailedException;
+import org.wso2.carbon.device.mgt.core.config.DeviceConfigurationManager;
+import org.wso2.carbon.device.mgt.core.dto.operation.mgt.Operation;
+import org.wso2.carbon.device.mgt.core.internal.DeviceManagementDataHolder;
+import org.wso2.carbon.device.mgt.core.operation.mgt.dao.OperationDAO;
+import org.wso2.carbon.device.mgt.core.operation.mgt.dao.OperationManagementDAOException;
+import org.wso2.carbon.device.mgt.core.operation.mgt.dao.OperationManagementDAOFactory;
+import org.wso2.carbon.device.mgt.core.operation.mgt.dao.OperationMappingDAO;
+import org.wso2.carbon.device.mgt.core.service.DeviceManagementProviderService;
+import org.wso2.carbon.user.api.UserStoreException;
+import org.wso2.carbon.user.core.tenant.TenantManager;
+
+import java.util.LinkedList;
+import java.util.List;
+
+/**
+ * ${{@link PushNotificationJob}} is for sending push notifications for given device batch.
+ */
+public class PushNotificationJob implements Runnable {
+
+    private static Log log = LogFactory.getLog(PushNotificationJob.class);
+    private final OperationDAO operationDAO = OperationManagementDAOFactory.getOperationDAO();
+    private final OperationMappingDAO operationMappingDAO = OperationManagementDAOFactory.getOperationMappingDAO();
+    private final DeviceManagementProviderService provider = DeviceManagementDataHolder.getInstance()
+            .getDeviceManagementProvider();
+    private final TenantManager tenantManager = DeviceManagementDataHolder.getInstance().getRealmService()
+            .getTenantManager();
+
+    @Override
+    public void run() {
+        List<OperationMapping> operationsCompletedList = new LinkedList<>();
+        try {
+            if (log.isDebugEnabled()) {
+                log.debug("Push notification job started");
+            }
+            //Get next available operation list per device batch
+            List<OperationMapping> operationMappings = operationDAO.getOperationMappingsByStatus(Operation.Status
+                    .PENDING, Operation.PushStatus.SCHEDULED, DeviceConfigurationManager.getInstance()
+                    .getDeviceManagementConfig().getPushNotificationConfiguration().getSchedulerBatchSize());
+            // Sending push notification to each device
+            for (OperationMapping operationMapping : operationMappings) {
+                try {
+                    if (log.isDebugEnabled()) {
+                        log.debug("Sending push notification for operationId :" + operationMapping.getOperationId() +
+                                "to deviceId : " + operationMapping.getDeviceIdentifier().getId());
+                    }
+                    // Set tenant id and domain
+                    PrivilegedCarbonContext.startTenantFlow();
+                    PrivilegedCarbonContext.getThreadLocalCarbonContext().setTenantId(operationMapping.getTenantId());
+                    PrivilegedCarbonContext.getThreadLocalCarbonContext().setTenantDomain(tenantManager.getDomain
+                            (operationMapping.getTenantId()));
+                    // Get notification strategy for given device type
+                    NotificationStrategy notificationStrategy = provider.getNotificationStrategyByDeviceType
+                            (operationMapping.getDeviceIdentifier().getType());
+                    // Send the push notification on given strategy
+                    notificationStrategy.execute(new NotificationContext(operationMapping.getDeviceIdentifier(),
+                            provider.getOperation(operationMapping.getDeviceIdentifier().getType(), operationMapping
+                                    .getOperationId())));
+                    operationsCompletedList.add(operationMapping);
+                } 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);
+                } catch (UserStoreException e) {
+                    log.error("Tenant domain cannot be found for given tenant id:  " + operationMapping.getTenantId()
+                            , e);
+                } finally {
+                    PrivilegedCarbonContext.endTenantFlow();
+                }
+            }
+            if (log.isDebugEnabled()) {
+                log.debug("Push notification job running completed.");
+            }
+        } catch (OperationManagementDAOException e) {
+            log.error("Unable to retrieve scheduled pending operations for task. ", e);
+        } finally {
+            // Update push notification status to competed for operations which already sent
+            try {
+                OperationManagementDAOFactory.beginTransaction();
+                operationMappingDAO.updateOperationMapping(operationsCompletedList, Operation.PushStatus.COMPLETED);
+                OperationManagementDAOFactory.commitTransaction();
+            } catch (TransactionManagementException | OperationManagementDAOException e) {
+                OperationManagementDAOFactory.rollbackTransaction();
+                log.error("Error occurred while updating operation mappings for sent notifications ", e);
+            } finally {
+                OperationManagementDAOFactory.closeConnection();
+            }
+        }
+    }
+}

components/device-mgt/org.wso2.carbon.device.mgt.core/src/main/java/org/wso2/carbon/device/mgt/core/push/notification/mgt/task/PushNotificationSchedulerTask.java

@@ -0,0 +1,55 @@
+/*
+*  Copyright (c) 2017, WSO2 Inc. (http://www.wso2.org) All Rights Reserved.
+*
+*  WSO2 Inc. licenses this file to you under the Apache License,
+*  Version 2.0 (the "License"); you may not use this file except
+*  in compliance with the License.
+*  You may obtain a copy of the License at
+*
+*    http://www.apache.org/licenses/LICENSE-2.0
+*
+* Unless required by applicable law or agreed to in writing,
+* software distributed under the License is distributed on an
+* "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+* KIND, either express or implied.  See the License for the
+* specific language governing permissions and limitations
+* under the License.
+*/
+package org.wso2.carbon.device.mgt.core.push.notification.mgt.task;
+
+import org.apache.commons.logging.Log;
+import org.apache.commons.logging.LogFactory;
+
+import java.util.concurrent.ExecutorService;
+import java.util.concurrent.Executors;
+import java.util.concurrent.Future;
+
+/**
+ * ${{@link PushNotificationSchedulerTask}} for scheduling push notification job.
+ */
+public class PushNotificationSchedulerTask implements Runnable {
+
+    private static Log log = LogFactory.getLog(PushNotificationSchedulerTask.class);
+    private final ExecutorService executor = Executors.newSingleThreadExecutor();
+    private Future<?> lastExecution;
+    final PushNotificationJob pushNotificationJob = new PushNotificationJob();
+
+    @Override
+    public void run() {
+        /*
+          There will be only one push notification job submit to thread pool.
+          Scheduler  will submit new job only if last execution is completed.
+         */
+        if (lastExecution != null && !lastExecution.isDone()) {
+            if (log.isDebugEnabled()) {
+                log.debug("Previous push notification job is already running. New notification job will start " +
+                        "after existing job completed.");
+            }
+            return;
+        }
+        if (log.isDebugEnabled()) {
+            log.debug("Submitting new notification job.");
+        }
+        lastExecution = executor.submit(pushNotificationJob);
+    }
+}