Add task operation with list of device ids

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

@@ -45,6 +45,8 @@ public interface OperationManager {
     Activity addOperation(Operation operation, List<DeviceIdentifier> devices) throws OperationManagementException,
             InvalidDeviceException;
 
+    void addTaskOperation(List<Device> devices, Operation operation) throws OperationManagementException;
+
     void addTaskOperation(String deviceType, Operation operation) throws OperationManagementException;
 
     /**

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

@@ -527,7 +527,7 @@ public interface DeviceDAO {
     List<GeoCluster> findGeoClusters(String deviceType, GeoCoordinate southWest, GeoCoordinate northEast,
                                      int geohashLength,int tenantId) throws DeviceManagementDAOException;
 
-    /***
+    /**
      * This method is used to identify whether given device ids are exist or not.
      *
      * @param deviceIdentifiers List of device identifiers.
@@ -539,6 +539,18 @@ public interface DeviceDAO {
     List<Device> getDevicesByIdentifiers(List<String> deviceIdentifiers, int tenantId)
             throws DeviceManagementDAOException;
 
+    /**
+     * This method is used to retrieve devices with specified device identifiers filtered with statuses.
+     *
+     * @param deviceIdentifiers List of device identifiers.
+     * @param tenantId tenant id.
+     * @return returns list of device ids that matches with device identifiers.
+     * @throws DeviceManagementDAOException throws {@link DeviceManagementDAOException} if connections establishment
+     * fails.
+     */
+    List<Device> getDevicesByIdentifiersAndStatuses(List<String> deviceIdentifiers, List<EnrolmentInfo.Status> statuses, int tenantId)
+            throws DeviceManagementDAOException;
+
     /***
      * This method is used to permanently delete devices and their related details
      * @param deviceIdentifiers List of device identifiers.

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

@@ -1829,6 +1829,65 @@ public abstract class AbstractDeviceDAOImpl implements DeviceDAO {
         }
     }
 
+    @Override
+    public List<Device> getDevicesByIdentifiersAndStatuses(List<String> deviceIdentifiers,
+                                                           List<EnrolmentInfo.Status> statuses, int tenantId)
+            throws DeviceManagementDAOException {
+        try {
+            Connection conn = this.getConnection();
+            int index = 1;
+            int counter = 0;
+            List<Device> devices = new ArrayList<>();
+
+            StringJoiner statusJoiner = new StringJoiner(",", "e.STATUS IN (", ") ");
+            while (counter < statuses.size()) {
+                statusJoiner.add("?");
+                counter++;
+            }
+
+            StringJoiner joiner = new StringJoiner(",",
+                    "SELECT "
+                            + "d1.ID AS DEVICE_ID, d1.DESCRIPTION, d1.NAME AS DEVICE_NAME, d1.DEVICE_TYPE, "
+                            + "d1.DEVICE_IDENTIFICATION, e.OWNER, e.OWNERSHIP, e.STATUS, e.IS_TRANSFERRED, "
+                            + "e.DATE_OF_LAST_UPDATE, e.DATE_OF_ENROLMENT, e.ID AS ENROLMENT_ID "
+                            + "FROM "
+                            + "DM_ENROLMENT e, "
+                            + "(SELECT d.ID, d.DESCRIPTION, d.NAME, t.NAME AS DEVICE_TYPE, d.DEVICE_IDENTIFICATION "
+                            + "FROM DM_DEVICE d, DM_DEVICE_TYPE t "
+                            + "WHERE "
+                            + "t.ID = d.DEVICE_TYPE_ID AND d.DEVICE_IDENTIFICATION IN (",
+                    ") AND d.TENANT_ID = ?) d1 "
+                            + "WHERE d1.ID = e.DEVICE_ID AND " + statusJoiner.toString()
+                            + "AND TENANT_ID = ? ORDER BY e.DATE_OF_LAST_UPDATE DESC, e.STATUS ASC");
+
+            counter = 0;
+            while (counter < deviceIdentifiers.size()) {
+                joiner.add("?");
+                counter++;
+            }
+            String query = joiner.toString();
+            try (PreparedStatement ps = conn.prepareStatement(query)) {
+                for (String identifier : deviceIdentifiers) {
+                    ps.setString(index++, identifier);
+                }
+                ps.setInt(index++, tenantId);
+                for (EnrolmentInfo.Status status : statuses) {
+                    ps.setString(index++, status.toString());
+                }
+                ps.setInt(index, tenantId);
+                try (ResultSet rs = ps.executeQuery()) {
+                    while (rs.next()) {
+                        devices.add(DeviceManagementDAOUtil.loadDevice(rs));
+                    }
+                }
+            }
+            return devices;
+        } catch (SQLException e) {
+            throw new DeviceManagementDAOException("Error occurred while obtaining the DB connection to get devices for"
+                    + " given device identifiers and statuses.", e);
+        }
+    }
+
     @Override
     public List<DeviceLocationHistorySnapshot> getDeviceLocationInfo(DeviceIdentifier deviceIdentifier, long from, long to)
             throws DeviceManagementDAOException {

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

@@ -24,15 +24,15 @@ import org.wso2.carbon.context.CarbonContext;
 import org.wso2.carbon.context.PrivilegedCarbonContext;
 import org.wso2.carbon.device.mgt.common.Device;
 import org.wso2.carbon.device.mgt.common.DeviceIdentifier;
-import org.wso2.carbon.device.mgt.common.exceptions.DeviceManagementException;
 import org.wso2.carbon.device.mgt.common.EnrolmentInfo;
-import org.wso2.carbon.device.mgt.common.exceptions.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.exceptions.TransactionManagementException;
 import org.wso2.carbon.device.mgt.common.authorization.DeviceAccessAuthorizationException;
+import org.wso2.carbon.device.mgt.common.exceptions.DeviceManagementException;
+import org.wso2.carbon.device.mgt.common.exceptions.InvalidDeviceException;
+import org.wso2.carbon.device.mgt.common.exceptions.TransactionManagementException;
 import org.wso2.carbon.device.mgt.common.group.mgt.DeviceGroupConstants;
 import org.wso2.carbon.device.mgt.common.operation.mgt.Activity;
 import org.wso2.carbon.device.mgt.common.operation.mgt.ActivityStatus;
@@ -227,67 +227,21 @@ public class OperationManagerImpl implements OperationManager {
                                     + operationCode);
                         }
                         Activity activity = new Activity();
-                        //Send the operation statuses only for admin triggered operations
-                        String deviceType = validDeviceIds.get(0).getType();
                         activity.setActivityStatus(this.getActivityStatus(deviceValidationResult,
                                 deviceAuthorizationResult));
                         return activity;
                     }
                 }
 
-                int tenantId = CarbonContext.getThreadLocalCarbonContext().getTenantId();
-                int operationId = this.lookupOperationDAO(operation).addOperation(operationDto);
-                operationDto.setId(operationId);
-
-                boolean isScheduled = false;
-                NotificationStrategy notificationStrategy = getNotificationStrategy();
-
-                // check whether device list is greater than batch size notification strategy has enable to send push
-                // notification using scheduler task
-                if (DeviceConfigurationManager.getInstance().getDeviceManagementConfig().
-                        getPushNotificationConfiguration().getSchedulerBatchSize() <= authorizedDeviceIds.size() &&
-                        notificationStrategy != null) {
-                    isScheduled = notificationStrategy.getConfig().isScheduled();
-                }
-                int failAttempts = 0;
-                while (true) {
-                    try {
-                        operationMappingDAO.addOperationMapping(operationDto,
-                                new ArrayList<>(enrolments.values()), isScheduled, tenantId);
-                        OperationManagementDAOFactory.commitTransaction();
-                        break;
-                    } catch (OperationManagementDAOException e) {
-                        OperationManagementDAOFactory.rollbackTransaction();
-                        if (++failAttempts > 3) {
-                            String msg = "Error occurred while updating operation mapping. Operation ID: " +
-                                    operationId;
-                            log.error(msg, e);
-                            throw new OperationManagementException(msg, e);
-                        }
-                        log.warn("Unable to update operation status. Operation ID: " + operationId +
-                                ", Attempt: " + failAttempts + ", Error: " + e.getMessage());
-                        try {
-                            Thread.sleep(2000);
-                        } catch (InterruptedException ignore) {
-                            break;
-                        }
-                    }
-                }
-                if (!isScheduled && notificationStrategy != null) {
-                    for (Device d : enrolments.values()) {
-                        this.sendNotification(operation, d);
-                    }
-                }
+                persistsOperation(operation, operationDto, enrolments);
 
                 Activity activity = new Activity();
-                activity.setActivityId(DeviceManagementConstants.OperationAttributes.ACTIVITY + operationId);
+                activity.setActivityId(DeviceManagementConstants.OperationAttributes.ACTIVITY + operation.getId());
                 activity.setCode(operationCode);
                 activity.setCreatedTimeStamp(new Date().toString());
                 activity.setType(Activity.Type.valueOf(operationDto.getType().toString()));
                 //For now set the operation statuses only for admin triggered operations
                 if (!isScheduledOperation) {
-                    //Get the device-type from 1st valid DeviceIdentifier. We know the 1st element is definitely there.
-                    String deviceType = validDeviceIds.get(0).getType();
                     activity.setActivityStatus(this.getActivityStatus(deviceValidationResult,
                             deviceAuthorizationResult));
                 }
@@ -305,6 +259,46 @@ public class OperationManagerImpl implements OperationManager {
         }
     }
 
+    @Override
+    public void addTaskOperation(List<Device> devices, Operation operation) throws OperationManagementException {
+        try {
+            OperationManagementDAOFactory.beginTransaction();
+            operation.setInitiatedBy(SYSTEM);
+            org.wso2.carbon.device.mgt.core.dto.operation.mgt.Operation operationDto =
+                    OperationDAOUtil.convertOperation(operation);
+            String operationCode = operationDto.getCode();
+            Map<Integer, Device> enrolments = new HashMap<>();
+            for (Device device : devices) {
+                enrolments.put(device.getEnrolmentInfo().getId(), device);
+            }
+            if (operationDto.getControl() ==
+                    org.wso2.carbon.device.mgt.core.dto.operation.mgt.Operation.Control.NO_REPEAT) {
+                Map<Integer, Integer> pendingOperationIDs = operationDAO
+                        .getExistingOperationIDs(enrolments.keySet().toArray(new Integer[0]), operationCode);
+                Device device;
+                for (Integer enrolmentId : pendingOperationIDs.keySet()) {
+                    operation.setId(pendingOperationIDs.get(enrolmentId));
+                    device = enrolments.get(enrolmentId);
+                    this.sendNotification(operation, device);
+                    //No need to keep this enrollment as it has a pending operation
+                    enrolments.remove(enrolmentId);
+                }
+                if (enrolments.size() == 0) {
+                    //No operations to be add. All are repeated.
+                    return;
+                }
+            }
+            persistsOperation(operation, operationDto, enrolments);
+        } catch (OperationManagementDAOException e) {
+            OperationManagementDAOFactory.rollbackTransaction();
+            throw new OperationManagementException("Error occurred while adding task operation", e);
+        } catch (TransactionManagementException e) {
+            throw new OperationManagementException("Error occurred while initiating the transaction", e);
+        } finally {
+            OperationManagementDAOFactory.closeConnection();
+        }
+    }
+
     @Override
     public void addTaskOperation(String deviceType, Operation operation) throws OperationManagementException {
         List<String> validStatuses = Arrays.asList(EnrolmentInfo.Status.ACTIVE.toString(),
@@ -356,49 +350,7 @@ public class OperationManagerImpl implements OperationManager {
                         break;
                     }
                 }
-
-                int operationId = this.lookupOperationDAO(operation).addOperation(operationDto);
-                operationDto.setId(operationId);
-
-                boolean isScheduled = false;
-                NotificationStrategy notificationStrategy = getNotificationStrategy();
-
-                // check whether device list is greater than batch size notification strategy has enable to send push
-                // notification using scheduler task
-                if (DeviceConfigurationManager.getInstance().getDeviceManagementConfig().
-                        getPushNotificationConfiguration().getSchedulerBatchSize() <= enrolments.size() &&
-                        notificationStrategy != null) {
-                    isScheduled = notificationStrategy.getConfig().isScheduled();
-                }
-                int failAttempts = 0;
-                while (true) {
-                    try {
-                        operationMappingDAO.addOperationMapping(operationDto,
-                                new ArrayList<>(enrolments.values()), isScheduled, tenantId);
-                        OperationManagementDAOFactory.commitTransaction();
-                        break;
-                    } catch (OperationManagementDAOException e) {
-                        OperationManagementDAOFactory.rollbackTransaction();
-                        if (++failAttempts > 3) {
-                            String msg = "Error occurred while updating operation mapping. Operation ID: " +
-                                    operationId;
-                            log.error(msg, e);
-                            throw new OperationManagementException(msg, e);
-                        }
-                        log.warn("Unable to update operation status. Operation ID: " + operationId +
-                                ", Attempt: " + failAttempts + ", Error: " + e.getMessage());
-                        try {
-                            Thread.sleep(2000);
-                        } catch (InterruptedException ignore) {
-                            break;
-                        }
-                    }
-                }
-                if (!isScheduled && notificationStrategy != null) {
-                    for (Device device : enrolments.values()) {
-                        this.sendNotification(operation, device);
-                    }
-                }
+                persistsOperation(operation, operationDto, enrolments);
                 try {
                     Thread.sleep(2000);
                 } catch (InterruptedException ignore) {
@@ -418,6 +370,57 @@ public class OperationManagerImpl implements OperationManager {
         }
     }
 
+    private void persistsOperation(Operation operation,
+                                   org.wso2.carbon.device.mgt.core.dto.operation.mgt.Operation operationDto,
+                                   Map<Integer, Device> enrolments)
+            throws OperationManagementDAOException, OperationManagementException {
+
+        int tenantId = CarbonContext.getThreadLocalCarbonContext().getTenantId();
+        int operationId = this.lookupOperationDAO(operation).addOperation(operationDto);
+        operationDto.setId(operationId);
+        operation.setId(operationId);
+
+        boolean isScheduled = false;
+        NotificationStrategy notificationStrategy = getNotificationStrategy();
+
+        // check whether device list is greater than batch size notification strategy has enable to send push
+        // notification using scheduler task
+        if (DeviceConfigurationManager.getInstance().getDeviceManagementConfig().
+                getPushNotificationConfiguration().getSchedulerBatchSize() <= enrolments.size() &&
+                notificationStrategy != null) {
+            isScheduled = notificationStrategy.getConfig().isScheduled();
+        }
+        int failAttempts = 0;
+        while (true) {
+            try {
+                operationMappingDAO.addOperationMapping(operationDto,
+                        new ArrayList<>(enrolments.values()), isScheduled, tenantId);
+                OperationManagementDAOFactory.commitTransaction();
+                break;
+            } catch (OperationManagementDAOException e) {
+                OperationManagementDAOFactory.rollbackTransaction();
+                if (++failAttempts > 3) {
+                    String msg = "Error occurred while updating operation mapping. Operation ID: " +
+                            operationId;
+                    log.error(msg, e);
+                    throw new OperationManagementException(msg, e);
+                }
+                log.warn("Unable to update operation status. Operation ID: " + operationId +
+                        ", Attempt: " + failAttempts + ", Error: " + e.getMessage());
+                try {
+                    Thread.sleep(2000);
+                } catch (InterruptedException ignore) {
+                    break;
+                }
+            }
+        }
+        if (!isScheduled && notificationStrategy != null) {
+            for (Device device : enrolments.values()) {
+                this.sendNotification(operation, device);
+            }
+        }
+    }
+
     private void sendNotification(Operation operation, Device device) {
         NotificationStrategy notificationStrategy = getNotificationStrategy();
         /*

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

@@ -25,6 +25,7 @@ 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.OperationResponse;
+import org.wso2.carbon.device.mgt.core.DeviceManagementConstants;
 import org.wso2.carbon.device.mgt.core.dto.operation.mgt.Operation;
 import org.wso2.carbon.device.mgt.core.dto.operation.mgt.OperationResponseMeta;
 import org.wso2.carbon.device.mgt.core.operation.mgt.OperationMapping;
@@ -429,6 +430,7 @@ public class GenericOperationDAOImpl implements OperationDAO {
             while (rs.next()) {
                 if (enrolmentId == 0) {
                     activity = new Activity();
+                    activity.setActivityId(DeviceManagementConstants.OperationAttributes.ACTIVITY + operationId);
                     activity.setType(Activity.Type.valueOf(rs.getString("OPERATION_TYPE")));
                     activity.setCreatedTimeStamp(new java.util.Date(rs.getLong(("CREATED_TIMESTAMP")) * 1000).toString());
                     activity.setCode(rs.getString("OPERATION_CODE"));

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

@@ -14,23 +14,23 @@
  * KIND, either express or implied. See the License for the
  * specific language governing permissions and limitations
  * under the License.
+ */
+/*
+ *  Copyright (c) 2020, Entgra (pvt) Ltd. (http://entgra.io) All Rights Reserved.
  *
+ *  Entgra (pvt) Ltd. 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
  *
- *   Copyright (c) 2019, Entgra (pvt) Ltd. (http://entgra.io) All Rights Reserved.
- *
- *   Entgra (pvt) Ltd. 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
+ *  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.
+ *  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.service;
@@ -73,6 +73,7 @@ import org.wso2.carbon.device.mgt.core.dto.DeviceType;
 import org.wso2.carbon.device.mgt.core.dto.DeviceTypeVersion;
 import org.wso2.carbon.device.mgt.core.geo.GeoCluster;
 import org.wso2.carbon.device.mgt.core.geo.geoHash.GeoCoordinate;
+import org.wso2.carbon.device.mgt.core.operation.mgt.CommandOperation;
 
 import java.sql.SQLException;
 import java.util.Date;
@@ -655,6 +656,9 @@ public interface DeviceManagementProviderService {
 
     void addTaskOperation(String deviceType, Operation operation) throws OperationManagementException;
 
+    void addTaskOperation(String type, List<Device> devices, Operation operation)
+            throws OperationManagementException;
+
     List<? extends Operation> getOperations(DeviceIdentifier deviceId) throws OperationManagementException;
 
     PaginationResult getOperations(DeviceIdentifier deviceId,
@@ -906,7 +910,6 @@ public interface DeviceManagementProviderService {
      * @param deviceIdentifiers A list of device identifiers
      * @return A list of devices
      * @throws {@link DeviceManagementException}
-     * @throws {@link InvalidDeviceException}
      */
     List<Device> getDeviceByIdList(List<String> deviceIdentifiers)
             throws DeviceManagementException;
@@ -918,4 +921,16 @@ public interface DeviceManagementProviderService {
      * @return enrollment steps of each enrollment types which are provided in the device type xml file
      */
     DeviceEnrollmentInvitationDetails getDeviceEnrollmentInvitationDetails(String deviceType);
+
+    /**
+     * This method is used to retrieve devices with specified device identifiers filtered with statuses.
+     *
+     * @param deviceIdentifiers A list of device identifiers
+     * @param statuses A list of device statuses
+     * @return A list of devices
+     * @throws {@link DeviceManagementException}
+     */
+    List<Device> getDevicesByIdentifiersAndStatuses(List<String> deviceIdentifiers, List<EnrolmentInfo.Status> statuses)
+            throws DeviceManagementException;
+
 }

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

@@ -1838,6 +1838,12 @@ public class DeviceManagementProviderServiceImpl implements DeviceManagementProv
         pluginRepository.getOperationManager(type, this.getTenantId()).addTaskOperation(type, operation);
     }
 
+    @Override
+    public void addTaskOperation(String type, List<Device> devices, Operation operation)
+            throws OperationManagementException {
+        pluginRepository.getOperationManager(type, this.getTenantId()).addTaskOperation(devices, operation);
+    }
+
     @Override
     public List<? extends Operation> getOperations(DeviceIdentifier deviceId) throws OperationManagementException {
         return pluginRepository.getOperationManager(deviceId.getType(), this.getTenantId()).getOperations(deviceId);
@@ -4212,4 +4218,26 @@ public class DeviceManagementProviderServiceImpl implements DeviceManagementProv
         DeviceManagementService dms = pluginRepository.getDeviceManagementService(deviceType, tenantId);
         return dms.getDeviceEnrollmentInvitationDetails();
     }
+
+    @Override
+    public List<Device> getDevicesByIdentifiersAndStatuses(List<String> deviceIdentifiers,
+                                                           List<EnrolmentInfo.Status> statuses)
+            throws DeviceManagementException {
+        int tenantId = this.getTenantId();
+        try {
+            DeviceManagementDAOFactory.openConnection();
+            return deviceDAO.getDevicesByIdentifiersAndStatuses(deviceIdentifiers, statuses, tenantId);
+        } catch (DeviceManagementDAOException e) {
+            String msg = "Error occurred while retrieving device list.";
+            log.error(msg, e);
+            throw new DeviceManagementException(msg, e);
+        } catch (SQLException e) {
+            String msg = "Error occurred while opening a connection to the data source";
+            log.error(msg, e);
+            throw new DeviceManagementException(msg, e);
+        } finally {
+            DeviceManagementDAOFactory.closeConnection();
+        }
+    }
+
 }