Add Scheduled for push notification sending

components/device-mgt/org.wso2.carbon.device.mgt.common/src/main/java/org/wso2/carbon/device/mgt/common/push/notification/NotificationStrategy.java

@@ -31,4 +31,10 @@ public interface NotificationStrategy {
      */
     void undeploy();
 
+    /**
+     * Provides push notification configuration
+     *
+     */
+    PushNotificationConfig getConfig();
+
 }

components/device-mgt/org.wso2.carbon.device.mgt.common/src/main/java/org/wso2/carbon/device/mgt/common/push/notification/PushNotificationConfig.java

@@ -26,11 +26,13 @@ import java.util.Map;
 public class PushNotificationConfig {
 
     private String type;
+    private boolean isScheduled;
     Map<String, String> properties;
 
-    public PushNotificationConfig(String type, Map<String, String> properties) {
+    public PushNotificationConfig(String type, boolean isScheduled, Map<String, String> properties) {
         this.type = type;
         this.properties = properties;
+        this.isScheduled = isScheduled;
     }
 
     @XmlElement(name = "Type", required = true)
@@ -38,6 +40,10 @@ public class PushNotificationConfig {
         return type;
     }
 
+    public boolean isScheduled() {
+        return isScheduled;
+    }
+
     public Map<String, String> getProperties() {
         return properties;
     }

components/device-mgt/org.wso2.carbon.device.mgt.core/src/main/java/org/wso2/carbon/device/mgt/core/DeviceManagementConstants.java

@@ -86,4 +86,12 @@ public final class DeviceManagementConstants {
         public static final String ACTIVITY = "ACTIVITY_";
     }
 
+    public static final class PushNotifications {
+        private PushNotifications() {
+            throw new AssertionError();
+        }
+        public static final int DEFAULT_BATCH_DELAY_MILLS = 60000;
+        public static final int DEFAULT_BATCH_SIZE = 1000;
+    }
+
 }

components/device-mgt/org.wso2.carbon.device.mgt.core/src/main/java/org/wso2/carbon/device/mgt/core/config/DeviceManagementConfig.java

@@ -20,6 +20,7 @@ package org.wso2.carbon.device.mgt.core.config;
 import org.wso2.carbon.device.mgt.core.config.identity.IdentityConfigurations;
 import org.wso2.carbon.device.mgt.core.config.pagination.PaginationConfiguration;
 import org.wso2.carbon.device.mgt.core.config.policy.PolicyConfiguration;
+import org.wso2.carbon.device.mgt.core.config.push.notification.PushNotificationConfiguration;
 import org.wso2.carbon.device.mgt.core.config.task.TaskConfiguration;
 
 import javax.xml.bind.annotation.XmlElement;
@@ -39,7 +40,7 @@ public final class DeviceManagementConfig {
     private IdentityConfigurations identityConfigurations;
     private PolicyConfiguration policyConfiguration;
     private PaginationConfiguration paginationConfiguration;
-    private List<String> pushNotificationProviders;
+    private PushNotificationConfiguration pushNotificationConfiguration;
 
 
     @XmlElement(name = "ManagementRepository", required = true)
@@ -79,16 +80,6 @@ public final class DeviceManagementConfig {
         this.taskConfiguration = taskConfiguration;
     }
 
-    @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;
-    }
-
     @XmlElement(name = "PaginationConfiguration", required = true)
     public PaginationConfiguration getPaginationConfiguration() {
         return paginationConfiguration;
@@ -98,5 +89,13 @@ public final class DeviceManagementConfig {
         this.paginationConfiguration = paginationConfiguration;
     }
 
+    @XmlElement(name = "PushNotificationConfiguration", required = true)
+    public PushNotificationConfiguration getPushNotificationConfiguration() {
+        return pushNotificationConfiguration;
+    }
+
+    public void setPushNotificationConfiguration(PushNotificationConfiguration pushNotificationConfiguration) {
+        this.pushNotificationConfiguration = pushNotificationConfiguration;
+    }
 }
 

components/device-mgt/org.wso2.carbon.device.mgt.core/src/main/java/org/wso2/carbon/device/mgt/core/dto/operation/mgt/Operation.java

@@ -35,6 +35,10 @@ public class Operation implements Serializable {
         REPEAT, NO_REPEAT, PAUSE_SEQUENCE, STOP_SEQUENCE
     }
 
+    public enum PushStatus {
+        SCHEDULED, IN_PROGRESS, COMPLETED
+    }
+
     private String code;
     private Properties properties;
     private Type type;

components/device-mgt/org.wso2.carbon.device.mgt.core/src/main/java/org/wso2/carbon/device/mgt/core/internal/DeviceManagementServiceComponent.java

@@ -48,6 +48,7 @@ import org.wso2.carbon.device.mgt.core.operation.mgt.OperationManagerImpl;
 import org.wso2.carbon.device.mgt.core.operation.mgt.dao.OperationManagementDAOFactory;
 import org.wso2.carbon.device.mgt.core.permission.mgt.PermissionManagerServiceImpl;
 import org.wso2.carbon.device.mgt.core.push.notification.mgt.PushNotificationProviderRepository;
+import org.wso2.carbon.device.mgt.core.push.notification.mgt.task.PushNotificationSchedulerTask;
 import org.wso2.carbon.device.mgt.core.service.DeviceManagementProviderService;
 import org.wso2.carbon.device.mgt.core.service.DeviceManagementProviderServiceImpl;
 import org.wso2.carbon.device.mgt.core.service.GroupManagementProviderService;
@@ -62,6 +63,11 @@ import org.wso2.carbon.utils.ConfigurationContextService;
 
 import java.util.ArrayList;
 import java.util.List;
+import java.util.concurrent.ExecutorService;
+import java.util.concurrent.Executors;
+import java.util.concurrent.Future;
+import java.util.concurrent.ScheduledExecutorService;
+import java.util.concurrent.TimeUnit;
 
 /**
  * @scr.component name="org.wso2.carbon.device.manager" immediate="true"
@@ -151,7 +157,8 @@ public class DeviceManagementServiceComponent {
             this.initOperationsManager();
 
             PushNotificationProviderRepository pushNotificationRepo = new PushNotificationProviderRepository();
-            List<String> pushNotificationProviders = config.getPushNotificationProviders();
+            List<String> pushNotificationProviders = config.getPushNotificationConfiguration()
+                    .getPushNotificationProviders();
             if (pushNotificationProviders != null) {
                 for (String pushNoteProvider : pushNotificationProviders) {
                     pushNotificationRepo.addProvider(pushNoteProvider);
@@ -179,6 +186,24 @@ public class DeviceManagementServiceComponent {
             if (log.isDebugEnabled()) {
                 log.debug("Device management core bundle has been successfully initialized");
             }
+            // Start Push Notification Scheduler Task
+            if (config.getPushNotificationConfiguration().isSchedulerTaskEnabled()) {
+                if (config.getPushNotificationConfiguration().getSchedulerBatchSize() <= 0) {
+                    log.error("Push notification batch size cannot be 0 or less than 0. Setting default batch size to:" +
+                            " " + DeviceManagementConstants.PushNotifications.DEFAULT_BATCH_SIZE);
+                    config.getPushNotificationConfiguration().setSchedulerBatchSize(DeviceManagementConstants.PushNotifications
+                            .DEFAULT_BATCH_SIZE);
+                }
+                if (config.getPushNotificationConfiguration().getSchedulerBatchDelayMills() <= 0) {
+                    log.error("Push notification batch delay cannot be 0 or less than 0. Setting default batch delay " +
+                            "milliseconds to" + DeviceManagementConstants.PushNotifications.DEFAULT_BATCH_DELAY_MILLS);
+                    config.getPushNotificationConfiguration().setSchedulerBatchDelayMills(DeviceManagementConstants.PushNotifications
+                            .DEFAULT_BATCH_DELAY_MILLS);
+                }
+                ScheduledExecutorService pushNotificationExecutor = Executors.newScheduledThreadPool(1);
+                pushNotificationExecutor.schedule(new PushNotificationSchedulerTask(), config.getPushNotificationConfiguration()
+                        .getSchedulerBatchDelayMills(), TimeUnit.MILLISECONDS);
+            }
         } catch (Throwable e) {
             log.error("Error occurred while initializing device management core bundle", e);
         }
@@ -269,7 +294,7 @@ public class DeviceManagementServiceComponent {
         try {
             if (log.isDebugEnabled()) {
                 log.debug("Setting Device Management Service Provider: '" +
-                                  deviceManagementService.getType() + "'");
+                        deviceManagementService.getType() + "'");
             }
             synchronized (LOCK) {
                 deviceManagers.add(deviceManagementService);
@@ -278,10 +303,10 @@ public class DeviceManagementServiceComponent {
                 }
             }
             log.info("Device Type deployed successfully : " + deviceManagementService.getType() + " for tenant "
-                             + deviceManagementService.getProvisioningConfig().getProviderTenantDomain());
+                    + deviceManagementService.getProvisioningConfig().getProviderTenantDomain());
         } catch (Throwable e) {
             log.error("Failed to register device management service for device type" + deviceManagementService.getType() +
-                            " for tenant " + deviceManagementService.getProvisioningConfig().getProviderTenantDomain(), e);
+                    " for tenant " + deviceManagementService.getProvisioningConfig().getProviderTenantDomain(), e);
         }
     }
 

components/device-mgt/org.wso2.carbon.device.mgt.core/src/main/java/org/wso2/carbon/device/mgt/core/operation/mgt/OperationManagerImpl.java

@@ -145,6 +145,11 @@ public class OperationManagerImpl implements OperationManager {
                 int operationId = this.lookupOperationDAO(operation).addOperation(operationDto);
                 boolean isScheduledOperation = this.isTaskScheduledOperation(operation, deviceIds);
                 boolean isNotRepeated = false;
+                boolean isScheduled = false;
+                if (notificationStrategy != null) {
+                    isScheduled = notificationStrategy.getConfig().isScheduled();
+                }
+
                 boolean hasExistingTaskOperation;
                 int enrolmentId;
                 if (org.wso2.carbon.device.mgt.core.dto.operation.mgt.Operation.Control.NO_REPEAT == operationDto.
@@ -161,7 +166,7 @@ public class OperationManagerImpl implements OperationManager {
                     if (isScheduledOperation) {
                         hasExistingTaskOperation = operationDAO.updateTaskOperation(enrolmentId, operationCode);
                         if (!hasExistingTaskOperation) {
-                            operationMappingDAO.addOperationMapping(operationId, enrolmentId);
+                            operationMappingDAO.addOperationMapping(operationId, enrolmentId, isScheduled);
                         }
                     } else if (isNotRepeated) {
                         operationDAO.updateEnrollmentOperationsStatus(enrolmentId, operationCode,
@@ -169,17 +174,22 @@ public class OperationManagerImpl implements OperationManager {
                                                                               Operation.Status.PENDING,
                                                                       org.wso2.carbon.device.mgt.core.dto.operation.mgt.
                                                                               Operation.Status.REPEATED);
-                        operationMappingDAO.addOperationMapping(operationId, enrolmentId);
+                        operationMappingDAO.addOperationMapping(operationId, enrolmentId, isScheduled);
                     } else {
-                        operationMappingDAO.addOperationMapping(operationId, enrolmentId);
+                        operationMappingDAO.addOperationMapping(operationId, enrolmentId, isScheduled);
                     }
-                    if (notificationStrategy != null) {
+                    if (notificationStrategy != null && !notificationStrategy.getConfig().isScheduled()) {
                         try {
                             notificationStrategy.execute(new NotificationContext(deviceId, operation));
+                            operationMappingDAO.updateOperationMapping(operationId, enrolmentId, org.wso2.carbon
+                                    .device.mgt.core.dto.operation.mgt.Operation.PushStatus.COMPLETED);
                         } catch (PushNotificationExecutionFailedException e) {
                             log.error("Error occurred while sending push notifications to " +
                                       deviceId.getType() + " device carrying id '" +
                                       deviceId + "'", e);
+                            // Reschedule if push notification failed.
+                            operationMappingDAO.updateOperationMapping(operationId, enrolmentId, org.wso2.carbon
+                                    .device.mgt.core.dto.operation.mgt.Operation.PushStatus.SCHEDULED);
                         }
                     }
                 }

components/device-mgt/org.wso2.carbon.device.mgt.core/src/main/java/org/wso2/carbon/device/mgt/core/operation/mgt/dao/OperationDAO.java

@@ -20,6 +20,7 @@ package org.wso2.carbon.device.mgt.core.operation.mgt.dao;
 
 import org.wso2.carbon.device.mgt.common.PaginationRequest;
 import org.wso2.carbon.device.mgt.common.operation.mgt.Activity;
+import org.wso2.carbon.device.mgt.common.operation.mgt.OperationMapping;
 import org.wso2.carbon.device.mgt.common.operation.mgt.OperationResponse;
 import org.wso2.carbon.device.mgt.core.dto.operation.mgt.Operation;
 
@@ -80,4 +81,7 @@ public interface OperationDAO {
 
     boolean resetAttemptCount(int enrolmentId) throws OperationManagementDAOException;
 
+    List<OperationMapping> getOperationMappingsByStatus(Operation.Status opStatus, Operation.PushStatus pushStatus,
+                                                        int limit) throws OperationManagementDAOException;;
+
 }

components/device-mgt/org.wso2.carbon.device.mgt.core/src/main/java/org/wso2/carbon/device/mgt/core/operation/mgt/dao/OperationMappingDAO.java

@@ -18,10 +18,20 @@
  */
 package org.wso2.carbon.device.mgt.core.operation.mgt.dao;
 
+import org.wso2.carbon.device.mgt.common.operation.mgt.OperationMapping;
+import org.wso2.carbon.device.mgt.core.dto.operation.mgt.Operation;
+
+import java.util.List;
+
 public interface OperationMappingDAO {
 
-    void addOperationMapping(int operationId, Integer deviceIds) throws OperationManagementDAOException;
+    void addOperationMapping(int operationId, Integer deviceId, boolean isScheduled) throws OperationManagementDAOException;
+
+    void removeOperationMapping(int operationId, Integer deviceId) throws OperationManagementDAOException;
 
-    void removeOperationMapping(int operationId, Integer deviceIds) throws OperationManagementDAOException;
+    void updateOperationMapping(int operationId, Integer deviceId, Operation.PushStatus pushStatus) throws
+            OperationManagementDAOException;
+    void updateOperationMapping(List<OperationMapping> operationMappingList, Operation.PushStatus pushStatus) throws
+            OperationManagementDAOException;
 
 }

components/device-mgt/org.wso2.carbon.device.mgt.core/src/main/java/org/wso2/carbon/device/mgt/core/operation/mgt/dao/impl/GenericOperationDAOImpl.java

@@ -24,6 +24,7 @@ import org.wso2.carbon.device.mgt.common.DeviceIdentifier;
 import org.wso2.carbon.device.mgt.common.PaginationRequest;
 import org.wso2.carbon.device.mgt.common.operation.mgt.Activity;
 import org.wso2.carbon.device.mgt.common.operation.mgt.ActivityStatus;
+import org.wso2.carbon.device.mgt.common.operation.mgt.OperationMapping;
 import org.wso2.carbon.device.mgt.common.operation.mgt.OperationResponse;
 import org.wso2.carbon.device.mgt.core.dto.operation.mgt.Operation;
 import org.wso2.carbon.device.mgt.core.operation.mgt.dao.OperationDAO;
@@ -32,11 +33,20 @@ import org.wso2.carbon.device.mgt.core.operation.mgt.dao.OperationManagementDAOF
 import org.wso2.carbon.device.mgt.core.operation.mgt.dao.OperationManagementDAOUtil;
 import org.wso2.carbon.device.mgt.core.operation.mgt.dao.util.OperationDAOUtil;
 
-import java.io.*;
+import java.io.ByteArrayInputStream;
-import java.sql.*;
+import java.io.ByteArrayOutputStream;
+import java.io.IOException;
+import java.io.ObjectInputStream;
+import java.io.ObjectOutputStream;
+import java.sql.Connection;
+import java.sql.PreparedStatement;
+import java.sql.ResultSet;
+import java.sql.SQLException;
+import java.sql.Timestamp;
 import java.util.ArrayList;
 import java.util.Calendar;
 import java.util.Date;
+import java.util.LinkedList;
 import java.util.List;
 
 /**
@@ -162,8 +172,8 @@ public class GenericOperationDAOImpl implements OperationDAO {
         try {
             Connection connection = OperationManagementDAOFactory.getConnection();
             String query = "SELECT EOM.ID FROM DM_ENROLMENT_OP_MAPPING AS EOM INNER JOIN DM_OPERATION DM " +
-                           "ON DM.ID = EOM.OPERATION_ID WHERE EOM.ENROLMENT_ID = ? AND DM.OPERATION_CODE = ? AND " +
+                    "ON DM.ID = EOM.OPERATION_ID WHERE EOM.ENROLMENT_ID = ? AND DM.OPERATION_CODE = ? AND " +
-                           "EOM.STATUS = ?;";
+                    "EOM.STATUS = ?;";
             stmt = connection.prepareStatement(query);
             stmt.setInt(1, enrolmentId);
             stmt.setString(2, operationCode);
@@ -176,7 +186,7 @@ public class GenericOperationDAOImpl implements OperationDAO {
             }
             if (id != 0) {
                 stmt = connection.prepareStatement("UPDATE DM_ENROLMENT_OP_MAPPING SET UPDATED_TIMESTAMP = ?  " +
-                                                   "WHERE ID = ?");
+                        "WHERE ID = ?");
                 stmt.setLong(1, System.currentTimeMillis() / 1000);
                 stmt.setInt(2, id);
                 stmt.executeUpdate();
@@ -184,7 +194,7 @@ public class GenericOperationDAOImpl implements OperationDAO {
             }
         } catch (SQLException e) {
             throw new OperationManagementDAOException("Error occurred while update device mapping operation status " +
-                                                      "metadata", e);
+                    "metadata", e);
         } finally {
             OperationManagementDAOUtil.cleanupResources(stmt);
         }
@@ -420,14 +430,13 @@ public class GenericOperationDAOImpl implements OperationDAO {
                     "WHERE opm.UPDATED_TIMESTAMP > ? \n" +
                     "AND de.TENANT_ID = ? \n";
 
-            if(timestamp == 0){
+            if (timestamp == 0) {
                 sql += "ORDER BY opm.OPERATION_ID LIMIT ? OFFSET ?;";
-            }else{
+            } else {
                 sql += "ORDER BY opm.UPDATED_TIMESTAMP asc LIMIT ? OFFSET ?";
             }
 
 
-
             stmt = conn.prepareStatement(sql);
 
             stmt.setLong(1, timestamp);
@@ -1068,7 +1077,7 @@ public class GenericOperationDAOImpl implements OperationDAO {
         try {
             conn = OperationManagementDAOFactory.getConnection();
             String query = "UPDATE DM_POLICY_COMPLIANCE_STATUS SET ATTEMPTS = 0, LAST_REQUESTED_TIME = ? " +
-                        "WHERE ENROLMENT_ID = ? AND TENANT_ID = ?";
+                    "WHERE ENROLMENT_ID = ? AND TENANT_ID = ?";
             stmt = conn.prepareStatement(query);
             stmt.setTimestamp(1, currentTimestamp);
             stmt.setInt(2, enrolmentId);
@@ -1082,4 +1091,41 @@ public class GenericOperationDAOImpl implements OperationDAO {
         }
         return status;
     }
+
+    @Override
+    public List<OperationMapping> getOperationMappingsByStatus(Operation.Status opStatus, Operation.PushStatus pushStatus,
+                                                               int limit) throws OperationManagementDAOException {
+
+        PreparedStatement stmt = null;
+        ResultSet rs = null;
+        OperationMapping operationMapping;
+        List<OperationMapping> operationMappings = new LinkedList<>();
+        try {
+            Connection conn = OperationManagementDAOFactory.getConnection();
+            String sql = "SELECT op_mappings.ENROLMENT_ID, op_mappings.OPERATION_ID, d_type.NAME ,d.TENANT_ID FROM " +
+                    "DM_DEVICE d, DM_ENROLMENT_OP_MAPPING op_mappings, DM_DEVICE_TYPE d_type  WHERE op_mappings" +
+                    ".STATUS = '?' AND op_mappings.PUSH_NOTIFICATION_STATUS = '?' AND d.DEVICE_TYPE_ID = d_type.ID " +
+                    "AND d.ID=op_mappings.ENROLMENT_ID ORDER BY op_mappings.OPERATION_ID LIMIT ?";
+
+            stmt = conn.prepareStatement(sql);
+            stmt.setString(1, opStatus.toString());
+            stmt.setString(2, pushStatus.toString());
+            stmt.setInt(3, limit);
+            rs = stmt.executeQuery();
+
+            while (rs.next()) {
+                operationMapping = new OperationMapping();
+                operationMapping.setOperationId(rs.getInt("OPERATION_ID"));
+                operationMapping.setDeviceIdentifier(new DeviceIdentifier(String.valueOf(rs.getInt("ENROLMENT_ID")),
+                        rs.getString("NAME")));
+                operationMapping.setTenantId(rs.getInt("TENANT_ID"));
+                operationMappings.add(operationMapping);
+            }
+        } catch (SQLException e) {
+            throw new OperationManagementDAOException("SQL error while getting operation mappings from database. ", e);
+        } finally {
+            OperationManagementDAOUtil.cleanupResources(stmt, rs);
+        }
+        return operationMappings;
+    }
 }

components/device-mgt/org.wso2.carbon.device.mgt.core/src/main/java/org/wso2/carbon/device/mgt/core/operation/mgt/dao/impl/OperationMappingDAOImpl.java

@@ -18,6 +18,7 @@
  */
 package org.wso2.carbon.device.mgt.core.operation.mgt.dao.impl;
 
+import org.wso2.carbon.device.mgt.common.operation.mgt.OperationMapping;
 import org.wso2.carbon.device.mgt.core.dto.operation.mgt.Operation;
 import org.wso2.carbon.device.mgt.core.operation.mgt.dao.OperationManagementDAOException;
 import org.wso2.carbon.device.mgt.core.operation.mgt.dao.OperationManagementDAOFactory;
@@ -27,23 +28,30 @@ import org.wso2.carbon.device.mgt.core.operation.mgt.dao.OperationMappingDAO;
 import java.sql.Connection;
 import java.sql.PreparedStatement;
 import java.sql.SQLException;
+import java.util.List;
 
 public class OperationMappingDAOImpl implements OperationMappingDAO {
 
     @Override
-    public void addOperationMapping(int operationId, Integer deviceId) throws OperationManagementDAOException {
+    public void addOperationMapping(int operationId, Integer deviceId, boolean isScheduled) throws
+            OperationManagementDAOException {
         PreparedStatement stmt = null;
         try {
-            long time = System.currentTimeMillis()/1000;
+            long time = System.currentTimeMillis() / 1000;
             Connection conn = OperationManagementDAOFactory.getConnection();
-            String sql = "INSERT INTO DM_ENROLMENT_OP_MAPPING(ENROLMENT_ID, OPERATION_ID, STATUS, CREATED_TIMESTAMP, " +
+            String sql = "INSERT INTO DM_ENROLMENT_OP_MAPPING(ENROLMENT_ID, OPERATION_ID, STATUS, " +
-                    "UPDATED_TIMESTAMP) VALUES (?, ?, ?, ?, ?)";
+                    "PUSH_NOTIFICATION_STATUS, CREATED_TIMESTAMP, UPDATED_TIMESTAMP) VALUES (?, ?, ?, ?, ?, ?)";
             stmt = conn.prepareStatement(sql);
             stmt.setInt(1, deviceId);
             stmt.setInt(2, operationId);
             stmt.setString(3, Operation.Status.PENDING.toString());
-            stmt.setLong(4, time);
+            if (isScheduled) {
+                stmt.setString(4, Operation.PushStatus.SCHEDULED.toString());
+            } else {
+                stmt.setString(4, Operation.PushStatus.IN_PROGRESS.toString());
+            }
             stmt.setLong(5, time);
+            stmt.setLong(6, time);
             stmt.executeUpdate();
         } catch (SQLException e) {
             throw new OperationManagementDAOException("Error occurred while persisting device operation mappings", e);
@@ -54,13 +62,13 @@ public class OperationMappingDAOImpl implements OperationMappingDAO {
 
     @Override
     public void removeOperationMapping(int operationId,
-                                       Integer deviceIds) throws OperationManagementDAOException {
+                                       Integer deviceId) throws OperationManagementDAOException {
         PreparedStatement stmt = null;
         try {
             Connection conn = OperationManagementDAOFactory.getConnection();
             String sql = "DELETE FROM DM_ENROLMENT_OP_MAPPING WHERE ENROLMENT_ID = ? AND OPERATION_ID = ?";
             stmt = conn.prepareStatement(sql);
-            stmt.setInt(1, 0);
+            stmt.setInt(1, deviceId);
             stmt.setInt(2, operationId);
             stmt.executeUpdate();
         } catch (SQLException e) {
@@ -70,4 +78,55 @@ public class OperationMappingDAOImpl implements OperationMappingDAO {
         }
     }
 
+    @Override
+    public void updateOperationMapping(int operationId, Integer deviceId, Operation.PushStatus pushStatus) throws OperationManagementDAOException {
+        PreparedStatement stmt = null;
+        try {
+            Connection conn = OperationManagementDAOFactory.getConnection();
+            String sql = "UPDATE DM_ENROLMENT_OP_MAPPING SET PUSH_NOTIFICATION_STATUS = ? WHERE ENROLMENT_ID = ? and " +
+                    "OPERATION_ID = ?";
+            stmt = conn.prepareStatement(sql);
+            stmt.setString(1, pushStatus.toString());
+            stmt.setInt(2, deviceId);
+            stmt.setInt(3, operationId);
+            stmt.executeUpdate();
+        } catch (SQLException e) {
+            throw new OperationManagementDAOException("Error occurred while updating device operation mappings", e);
+        } finally {
+            OperationManagementDAOUtil.cleanupResources(stmt, null);
+        }
+    }
+
+    @Override
+    public void updateOperationMapping(List<OperationMapping> operationMappingList, Operation.PushStatus pushStatus) throws
+            OperationManagementDAOException {
+        PreparedStatement stmt = null;
+        try {
+            Connection conn = OperationManagementDAOFactory.getConnection();
+            String sql = "UPDATE DM_ENROLMENT_OP_MAPPING SET PUSH_NOTIFICATION_STATUS = ? WHERE ENROLMENT_ID = ? and " +
+                    "OPERATION_ID = ?";
+            stmt = conn.prepareStatement(sql);
+            if (conn.getMetaData().supportsBatchUpdates()) {
+                for (OperationMapping operationMapping : operationMappingList) {
+                    stmt.setString(1, pushStatus.toString());
+                    stmt.setInt(2, Integer.parseInt(operationMapping.getDeviceIdentifier().getId()));
+                    stmt.setInt(3, operationMapping.getOperationId());
+                    stmt.addBatch();
+                }
+                stmt.executeBatch();
+            } else {
+                for (OperationMapping operationMapping : operationMappingList) {
+                    stmt.setString(1, pushStatus.toString());
+                    stmt.setInt(2, Integer.parseInt(operationMapping.getDeviceIdentifier().getId()));
+                    stmt.setInt(3, operationMapping.getOperationId());
+                    stmt.executeUpdate();
+                }
+            }
+        } catch (SQLException e) {
+            throw new OperationManagementDAOException("Error occurred while updating device operation mappings as " +
+                    "batch ", e);
+        } finally {
+            OperationManagementDAOUtil.cleanupResources(stmt, null);
+        }
+    }
 }

components/device-mgt/org.wso2.carbon.device.mgt.core/src/main/java/org/wso2/carbon/device/mgt/core/operation/mgt/dao/impl/operation/OracleOperationDAOImpl.java

@@ -23,6 +23,7 @@ import org.wso2.carbon.device.mgt.common.DeviceIdentifier;
 import org.wso2.carbon.device.mgt.common.PaginationRequest;
 import org.wso2.carbon.device.mgt.common.operation.mgt.Activity;
 import org.wso2.carbon.device.mgt.common.operation.mgt.ActivityStatus;
+import org.wso2.carbon.device.mgt.common.operation.mgt.OperationMapping;
 import org.wso2.carbon.device.mgt.common.operation.mgt.OperationResponse;
 import org.wso2.carbon.device.mgt.core.dto.operation.mgt.Operation;
 import org.wso2.carbon.device.mgt.core.operation.mgt.dao.OperationManagementDAOException;
@@ -37,6 +38,7 @@ import java.sql.PreparedStatement;
 import java.sql.ResultSet;
 import java.sql.SQLException;
 import java.util.ArrayList;
+import java.util.LinkedList;
 import java.util.List;
 
 /**
@@ -135,7 +137,7 @@ public class OracleOperationDAOImpl extends GenericOperationDAOImpl {
 
     @Override
     public void updateEnrollmentOperationsStatus(int enrolmentId, String operationCode,
-            Operation.Status existingStatus, Operation.Status newStatus) throws OperationManagementDAOException {
+                                                 Operation.Status existingStatus, Operation.Status newStatus) throws OperationManagementDAOException {
         PreparedStatement stmt = null;
         ResultSet rs = null;
         try {
@@ -361,4 +363,41 @@ public class OracleOperationDAOImpl extends GenericOperationDAOImpl {
         }
         return 0;
     }
+
+    @Override
+    public List<OperationMapping> getOperationMappingsByStatus(Operation.Status opStatus, Operation.PushStatus pushStatus,
+                                                               int limit) throws OperationManagementDAOException {
+
+        PreparedStatement stmt = null;
+        ResultSet rs = null;
+        OperationMapping operationMapping;
+        List<OperationMapping> operationMappings = new LinkedList<>();
+        try {
+            Connection conn = OperationManagementDAOFactory.getConnection();
+            String sql = "SELECT op_mappings.ENROLMENT_ID, op_mappings.OPERATION_ID, d_type.NAME ,d.TENANT_ID FROM " +
+                    "DM_DEVICE d, DM_ENROLMENT_OP_MAPPING op_mappings, DM_DEVICE_TYPE d_type  WHERE op_mappings" +
+                    ".STATUS = '?' AND op_mappings.PUSH_NOTIFICATION_STATUS = '?' AND d.DEVICE_TYPE_ID = d_type.ID " +
+                    "AND d.ID=op_mappings.ENROLMENT_ID AND ROWNUM <= ? ORDER BY op_mappings.OPERATION_ID";
+
+            stmt = conn.prepareStatement(sql);
+            stmt.setString(1, opStatus.toString());
+            stmt.setString(2, pushStatus.toString());
+            stmt.setInt(3, limit);
+            rs = stmt.executeQuery();
+
+            while (rs.next()) {
+                operationMapping = new OperationMapping();
+                operationMapping.setOperationId(rs.getInt("OPERATION_ID"));
+                operationMapping.setDeviceIdentifier(new DeviceIdentifier(String.valueOf(rs.getInt("ENROLMENT_ID")),
+                        rs.getString("NAME")));
+                operationMapping.setTenantId(rs.getInt("TENANT_ID"));
+                operationMappings.add(operationMapping);
+            }
+        } catch (SQLException e) {
+            throw new OperationManagementDAOException("SQL error while getting operation mappings from database. ", e);
+        } finally {
+            OperationManagementDAOUtil.cleanupResources(stmt, rs);
+        }
+        return operationMappings;
+    }
 }

components/device-mgt/org.wso2.carbon.device.mgt.core/src/main/java/org/wso2/carbon/device/mgt/core/operation/mgt/dao/impl/operation/SQLServerOperationDAOImpl.java

@@ -23,6 +23,7 @@ import org.wso2.carbon.device.mgt.common.DeviceIdentifier;
 import org.wso2.carbon.device.mgt.common.PaginationRequest;
 import org.wso2.carbon.device.mgt.common.operation.mgt.Activity;
 import org.wso2.carbon.device.mgt.common.operation.mgt.ActivityStatus;
+import org.wso2.carbon.device.mgt.common.operation.mgt.OperationMapping;
 import org.wso2.carbon.device.mgt.common.operation.mgt.OperationResponse;
 import org.wso2.carbon.device.mgt.core.dto.operation.mgt.Operation;
 import org.wso2.carbon.device.mgt.core.operation.mgt.dao.OperationManagementDAOException;
@@ -37,6 +38,7 @@ import java.sql.PreparedStatement;
 import java.sql.ResultSet;
 import java.sql.SQLException;
 import java.util.ArrayList;
+import java.util.LinkedList;
 import java.util.List;
 
 /**
@@ -264,4 +266,42 @@ public class SQLServerOperationDAOImpl extends GenericOperationDAOImpl {
         return activities;
     }
 
+    @Override
+    public List<OperationMapping> getOperationMappingsByStatus(Operation.Status opStatus, Operation.PushStatus pushStatus,
+                                                               int limit) throws OperationManagementDAOException {
+
+        PreparedStatement stmt = null;
+        ResultSet rs = null;
+        OperationMapping operationMapping;
+        List<OperationMapping> operationMappings = new LinkedList<>();
+        try {
+            Connection conn = OperationManagementDAOFactory.getConnection();
+            String sql = "SELECT op_mappings.ENROLMENT_ID, op_mappings.OPERATION_ID, d_type.NAME ,d.TENANT_ID FROM " +
+                    "DM_DEVICE d, DM_ENROLMENT_OP_MAPPING op_mappings, DM_DEVICE_TYPE d_type  WHERE op_mappings" +
+                    ".STATUS = '?' AND op_mappings.PUSH_NOTIFICATION_STATUS = '?' AND d.DEVICE_TYPE_ID = d_type.ID " +
+                    "AND d.ID=op_mappings.ENROLMENT_ID ORDER BY op_mappings.OPERATION_ID OFFSET ? ROWS FETCH NEXT ? ROWS ONLY";
+
+            stmt = conn.prepareStatement(sql);
+            stmt.setString(1, opStatus.toString());
+            stmt.setString(2, pushStatus.toString());
+            stmt.setInt(3, 0);
+            stmt.setInt(4, limit);
+            rs = stmt.executeQuery();
+
+            while (rs.next()) {
+                operationMapping = new OperationMapping();
+                operationMapping.setOperationId(rs.getInt("OPERATION_ID"));
+                operationMapping.setDeviceIdentifier(new DeviceIdentifier(String.valueOf(rs.getInt("ENROLMENT_ID")),
+                        rs.getString("NAME")));
+                operationMapping.setTenantId(rs.getInt("TENANT_ID"));
+                operationMappings.add(operationMapping);
+            }
+        } catch (SQLException e) {
+            throw new OperationManagementDAOException("SQL error while getting operation mappings from database. ", e);
+        } finally {
+            OperationManagementDAOUtil.cleanupResources(stmt, rs);
+        }
+        return operationMappings;
+    }
+
 }

components/device-mgt/org.wso2.carbon.device.mgt.core/src/main/java/org/wso2/carbon/device/mgt/core/service/DeviceManagementProviderService.java

@@ -32,6 +32,7 @@ import org.wso2.carbon.device.mgt.common.operation.mgt.Activity;
 import org.wso2.carbon.device.mgt.common.operation.mgt.Operation;
 import org.wso2.carbon.device.mgt.common.operation.mgt.OperationManagementException;
 import org.wso2.carbon.device.mgt.common.policy.mgt.PolicyMonitoringManager;
+import org.wso2.carbon.device.mgt.common.push.notification.NotificationStrategy;
 
 import java.util.Date;
 import java.util.HashMap;
@@ -130,7 +131,8 @@ public interface DeviceManagementProviderService {
 
     /**
      * This method returns the list of device owned by a user of given device type.
-     * @param userName user name.
+     *
+     * @param userName   user name.
      * @param deviceType device type name
      * @return
      * @throws DeviceManagementException
@@ -210,13 +212,21 @@ public interface DeviceManagementProviderService {
      * This method is used to check whether the device is enrolled with the give user.
      *
      * @param deviceId identifier of the device that needs to be checked against the user.
-     * @param user username of the device owner.
+     * @param user     username of the device owner.
-     *
      * @return true if the user owns the device else will return false.
      * @throws DeviceManagementException If some unusual behaviour is observed while fetching the device.
      */
     boolean isEnrolled(DeviceIdentifier deviceId, String user) throws DeviceManagementException;
 
+    /**
+     * This method is used to get notification strategy for given device type
+     *
+     * @param deviceType Device type
+     * @return Notification Strategy for device type
+     * @throws DeviceManagementException
+     */
+    NotificationStrategy getNotificationStrategyByDeviceType(String deviceType) throws DeviceManagementException;
+
     License getLicense(String deviceType, String languageCode) throws DeviceManagementException;
 
     void addLicense(String deviceType, License license) throws DeviceManagementException;
@@ -239,6 +249,7 @@ public interface DeviceManagementProviderService {
 
     /**
      * Returns the device of specified id.
+     *
      * @param deviceId device Id
      * @return Device returns null when device is not avaialble.
      * @throws DeviceManagementException

components/device-mgt/org.wso2.carbon.device.mgt.core/src/main/java/org/wso2/carbon/device/mgt/core/service/DeviceManagementProviderServiceImpl.java

@@ -23,7 +23,21 @@ import org.apache.commons.logging.LogFactory;
 import org.wso2.carbon.CarbonConstants;
 import org.wso2.carbon.context.CarbonContext;
 import org.wso2.carbon.context.PrivilegedCarbonContext;
-import org.wso2.carbon.device.mgt.common.*;
+import org.wso2.carbon.device.mgt.common.Device;
+import org.wso2.carbon.device.mgt.common.DeviceIdentifier;
+import org.wso2.carbon.device.mgt.common.DeviceManagementException;
+import org.wso2.carbon.device.mgt.common.DeviceManager;
+import org.wso2.carbon.device.mgt.common.DeviceNotFoundException;
+import org.wso2.carbon.device.mgt.common.DeviceTypeIdentifier;
+import org.wso2.carbon.device.mgt.common.EnrolmentInfo;
+import org.wso2.carbon.device.mgt.common.FeatureManager;
+import org.wso2.carbon.device.mgt.common.InitialOperationConfig;
+import org.wso2.carbon.device.mgt.common.InvalidDeviceException;
+import org.wso2.carbon.device.mgt.common.MonitoringOperation;
+import org.wso2.carbon.device.mgt.common.OperationMonitoringTaskConfig;
+import org.wso2.carbon.device.mgt.common.PaginationRequest;
+import org.wso2.carbon.device.mgt.common.PaginationResult;
+import org.wso2.carbon.device.mgt.common.TransactionManagementException;
 import org.wso2.carbon.device.mgt.common.app.mgt.Application;
 import org.wso2.carbon.device.mgt.common.configuration.mgt.PlatformConfiguration;
 import org.wso2.carbon.device.mgt.common.device.details.DeviceInfo;
@@ -37,7 +51,9 @@ import org.wso2.carbon.device.mgt.common.license.mgt.LicenseManagementException;
 import org.wso2.carbon.device.mgt.common.operation.mgt.Activity;
 import org.wso2.carbon.device.mgt.common.operation.mgt.Operation;
 import org.wso2.carbon.device.mgt.common.operation.mgt.OperationManagementException;
+import org.wso2.carbon.device.mgt.common.operation.mgt.OperationManager;
 import org.wso2.carbon.device.mgt.common.policy.mgt.PolicyMonitoringManager;
+import org.wso2.carbon.device.mgt.common.push.notification.NotificationStrategy;
 import org.wso2.carbon.device.mgt.common.spi.DeviceManagementService;
 import org.wso2.carbon.device.mgt.core.DeviceManagementConstants;
 import org.wso2.carbon.device.mgt.core.DeviceManagementPluginRepository;
@@ -307,7 +323,7 @@ public class DeviceManagementProviderServiceImpl implements DeviceManagementProv
         return enrolmentInfos;
     }
 
-        @Override
+    @Override
     public boolean disenrollDevice(DeviceIdentifier deviceId) throws DeviceManagementException {
         DeviceManager deviceManager = this.getDeviceManager(deviceId.getType());
         if (deviceManager == null) {
@@ -649,7 +665,7 @@ public class DeviceManagementProviderServiceImpl implements DeviceManagementProv
                 device.setDeviceInfo(info);
             } catch (DeviceDetailsMgtDAOException e) {
                 log.error("Error occurred while retrieving advance info of '" + device.getType() +
-                                  "' that carries the id '" + device.getDeviceIdentifier() + "'");
+                        "' that carries the id '" + device.getDeviceIdentifier() + "'");
             } catch (SQLException e) {
                 log.error("Error occurred while opening a connection to the data source", e);
             } finally {
@@ -663,7 +679,7 @@ public class DeviceManagementProviderServiceImpl implements DeviceManagementProv
                 device.setApplications(applications);
             } catch (DeviceManagementDAOException e) {
                 log.error("Error occurred while retrieving the application list of '" + device.getType() + "', " +
-                                  "which carries the id '" + device.getId() + "'", e);
+                        "which carries the id '" + device.getId() + "'", e);
             } catch (SQLException e) {
                 log.error("Error occurred while opening a connection to the data source", e);
             } finally {
@@ -674,7 +690,7 @@ public class DeviceManagementProviderServiceImpl implements DeviceManagementProv
             if (deviceManager == null) {
                 if (log.isDebugEnabled()) {
                     log.debug("Device Manager associated with the device type '" + device.getType() + "' is null. " +
-                                      "Therefore, not attempting method 'isEnrolled'");
+                            "Therefore, not attempting method 'isEnrolled'");
                 }
                 devices.add(device);
                 continue;
@@ -689,6 +705,7 @@ public class DeviceManagementProviderServiceImpl implements DeviceManagementProv
         }
         return devices;
     }
+
     @Override
     public List<Device> getAllDevices(String deviceType) throws DeviceManagementException {
         List<Device> devices = new ArrayList<>();
@@ -779,7 +796,7 @@ public class DeviceManagementProviderServiceImpl implements DeviceManagementProv
         try {
             EmailContext ctx =
                     new EmailContext.EmailContextBuilder(new ContentProviderInfo(templateName, params),
-                                                         metaInfo.getRecipients()).build();
+                            metaInfo.getRecipients()).build();
             DeviceManagementDataHolder.getInstance().getEmailSenderService().sendEmail(ctx);
         } catch (EmailSendingFailedException ex) {
             throw new DeviceManagementException("Error occurred while sending enrollment invitation", ex);
@@ -1958,6 +1975,17 @@ public class DeviceManagementProviderServiceImpl implements DeviceManagementProv
         return false;
     }
 
+    @Override
+    public NotificationStrategy getNotificationStrategyByDeviceType(String deviceType) throws DeviceManagementException {
+        int tenantId = PrivilegedCarbonContext.getThreadLocalCarbonContext().getTenantId();
+        OperationManager operationManager = pluginRepository.getOperationManager(deviceType, tenantId);
+        if (operationManager != null) {
+            return operationManager.getNotificationStrategy();
+        } else {
+            throw new DeviceManagementException("Cannot find operation manager for given device type :" + deviceType);
+        }
+    }
+
     /**
      * Change device status.
      *
@@ -2007,7 +2035,7 @@ public class DeviceManagementProviderServiceImpl implements DeviceManagementProv
             throw new DeviceManagementException("Error occurred while opening a connection to the data source", e);
         } catch (DeviceManagementDAOException e) {
             throw new DeviceManagementException("Error occurred while updating the enrollment information device for" +
-                                                        "id '" + deviceId + "' ." , e);
+                    "id '" + deviceId + "' .", e);
         } finally {
             try {
                 DeviceManagementDAOFactory.getConnection().setAutoCommit(isAutoCommit);
@@ -2085,7 +2113,7 @@ public class DeviceManagementProviderServiceImpl implements DeviceManagementProv
         List<DeviceIdentifier> deviceIdentifiers = new ArrayList<>();
         deviceIdentifiers.add(deviceIdentifier);
         if (init != null) {
-           List<String> initialOperations = init.getOperations();
+            List<String> initialOperations = init.getOperations();
 
             for (String str : initialOperations) {
                 CommandOperation operation = new CommandOperation();