Merge pull request #1259 from milanperera/master-new

Fix wso2/product-iots#1863
revert-70aa11f8
Milan Perera 6 years ago committed by GitHub
commit 39aa9c5c88
No known key found for this signature in database
GPG Key ID: 4AEE18F83AFDEB23

@ -157,8 +157,8 @@ public class OperationManagerImpl implements OperationManager {
List<DeviceIdentifier> validDeviceIds = deviceValidationResult.getValidDeviceIDList(); List<DeviceIdentifier> validDeviceIds = deviceValidationResult.getValidDeviceIDList();
if (validDeviceIds.size() > 0) { if (validDeviceIds.size() > 0) {
DeviceIDHolder deviceAuthorizationResult = this.authorizeDevices(operation, validDeviceIds); DeviceIDHolder deviceAuthorizationResult = this.authorizeDevices(operation, validDeviceIds);
List<DeviceIdentifier> authorizedDeviceList = deviceAuthorizationResult.getValidDeviceIDList(); List<DeviceIdentifier> authorizedDeviceIds = deviceAuthorizationResult.getValidDeviceIDList();
if (authorizedDeviceList.size() <= 0) { if (authorizedDeviceIds.size() <= 0) {
log.warn("User : " + getUser() + " is not authorized to perform operations on given device-list."); log.warn("User : " + getUser() + " is not authorized to perform operations on given device-list.");
Activity activity = new Activity(); Activity activity = new Activity();
//Send the operation statuses only for admin triggered operations //Send the operation statuses only for admin triggered operations
@ -186,21 +186,31 @@ public class OperationManagerImpl implements OperationManager {
org.wso2.carbon.device.mgt.core.dto.operation.mgt.Operation operationDto = org.wso2.carbon.device.mgt.core.dto.operation.mgt.Operation operationDto =
OperationDAOUtil.convertOperation(operation); OperationDAOUtil.convertOperation(operation);
int enrolmentId; int enrolmentId;
boolean hasExistingTaskOperation;
List<DeviceIdentifier> pendingDeviceList = new ArrayList<>();
String operationCode = operationDto.getCode(); String operationCode = operationDto.getCode();
for (DeviceIdentifier deviceId : authorizedDeviceList) { List<Device> authorizedDevices = new ArrayList<>();
List<Device> ignoredDevices = new ArrayList<>();
for (DeviceIdentifier deviceId : authorizedDeviceIds) {
Device device = getDevice(deviceId); Device device = getDevice(deviceId);
authorizedDevices.add(device);
}
if (operationDto.getControl() ==
org.wso2.carbon.device.mgt.core.dto.operation.mgt.Operation.Control.NO_REPEAT) {
int existingOperationID;
for (Device device : authorizedDevices) {
enrolmentId = device.getEnrolmentInfo().getId(); enrolmentId = device.getEnrolmentInfo().getId();
hasExistingTaskOperation = operationDAO.updateTaskOperation(enrolmentId, operationCode); existingOperationID = operationDAO.getExistingOperationID(enrolmentId, operationCode);
if (hasExistingTaskOperation) { if (existingOperationID > 0) {
pendingDeviceList.add(deviceId); ignoredDevices.add(device);
operation.setId(existingOperationID);
this.sendNotification(operation, device);
}
} }
} }
if (pendingDeviceList.size() > 0) { if (ignoredDevices.size() > 0) {
if (authorizedDeviceList.size() == pendingDeviceList.size()) { if (authorizedDevices.size() == ignoredDevices.size()) {
if (log.isDebugEnabled()) { if (log.isDebugEnabled()) {
log.debug("All the devices contain a pending operation for the Operation Code: " log.debug("All the devices contain a pending operation for the Operation Code: "
+ operationCode); + operationCode);
@ -212,87 +222,34 @@ public class OperationManagerImpl implements OperationManager {
deviceType)); deviceType));
return activity; return activity;
} else { } else {
authorizedDeviceList.removeAll(pendingDeviceList); authorizedDevices.removeAll(ignoredDevices);
} }
} }
int operationId = this.lookupOperationDAO(operation).addOperation(operationDto); int operationId = this.lookupOperationDAO(operation).addOperation(operationDto);
boolean isNotRepeated = false;
boolean isScheduled = false; boolean isScheduled = false;
NotificationStrategy notificationStrategy = getNotificationStrategy(); NotificationStrategy notificationStrategy = getNotificationStrategy();
// check whether device list is greater than batch size notification strategy has enable to send push // check whether device list is greater than batch size notification strategy has enable to send push
// notification using scheduler task // notification using scheduler task
if (DeviceConfigurationManager.getInstance().getDeviceManagementConfig(). if (DeviceConfigurationManager.getInstance().getDeviceManagementConfig().
getPushNotificationConfiguration().getSchedulerBatchSize() <= authorizedDeviceList.size() && getPushNotificationConfiguration().getSchedulerBatchSize() <= authorizedDeviceIds.size() &&
notificationStrategy != null) { notificationStrategy != null) {
isScheduled = notificationStrategy.getConfig().isScheduled(); isScheduled = notificationStrategy.getConfig().isScheduled();
} }
List<Device> devices = new ArrayList<>();
if (org.wso2.carbon.device.mgt.core.dto.operation.mgt.Operation.Control.NO_REPEAT == operationDto.
getControl()) {
isNotRepeated = true;
}
//TODO have to create a sql to load device details from deviceDAO using single query. //TODO have to create a sql to load device details from deviceDAO using single query.
for (DeviceIdentifier deviceId : authorizedDeviceList) { for (Device device : authorizedDevices) {
Device device = getDevice(deviceId);
devices.add(device);
enrolmentId = device.getEnrolmentInfo().getId(); enrolmentId = device.getEnrolmentInfo().getId();
//Do not repeat the task operations //Do not repeat the task operations
if (isScheduledOperation) {
hasExistingTaskOperation = operationDAO.updateTaskOperation(enrolmentId, operationCode);
if (!hasExistingTaskOperation) {
operationMappingDAO.addOperationMapping(operationId, enrolmentId, isScheduled); operationMappingDAO.addOperationMapping(operationId, enrolmentId, isScheduled);
} }
} else if (isNotRepeated) {
operationDAO.updateEnrollmentOperationsStatus(enrolmentId, operationCode,
org.wso2.carbon.device.mgt.core.dto.operation.mgt.
Operation.Status.PENDING,
org.wso2.carbon.device.mgt.core.dto.operation.mgt.
Operation.Status.REPEATED);
operationMappingDAO.addOperationMapping(operationId, enrolmentId, isScheduled);
} else {
operationMappingDAO.addOperationMapping(operationId, enrolmentId, isScheduled);
}
}
OperationManagementDAOFactory.commitTransaction(); OperationManagementDAOFactory.commitTransaction();
/* if (isScheduled) {
If notification strategy has not enable to send push notification using scheduler task we will send for (Device device : authorizedDevices) {
notification immediately. This is done in separate loop inorder to prevent overlap with DB insert this.sendNotification(operation, device);
operations with the possible db update operations trigger followed by pending operation call.
Otherwise device may call pending operation while DB is locked for write and deadlock can occur.
*/
if (notificationStrategy != null && !isScheduled) {
for (Device device : devices) {
DeviceIdentifier deviceId = new DeviceIdentifier(device.getDeviceIdentifier(), device.getType());
if (log.isDebugEnabled()) {
log.debug("Sending push notification to " + deviceId + " from add operation method.");
}
operation.setId(operationId);
operation.setActivityId(DeviceManagementConstants.OperationAttributes.ACTIVITY + operationId);
try {
notificationStrategy.execute(new NotificationContext(deviceId, operation));
} catch (PushNotificationExecutionFailedException e) {
log.error("Error occurred while sending push notifications to " + deviceId.getType() +
" device carrying id '" + deviceId + "'", e);
/*
Reschedule if push notification failed. Doing db transactions in atomic way to prevent
deadlocks.
*/
enrolmentId = device.getEnrolmentInfo().getId();
try {
operationMappingDAO.updateOperationMapping(operationId, enrolmentId, org.wso2.carbon
.device.mgt.core.dto.operation.mgt.Operation.PushNotificationStatus.SCHEDULED);
OperationManagementDAOFactory.commitTransaction();
} catch (OperationManagementDAOException ex) {
// Not throwing this exception in order to keep sending remaining notifications if any.
log.error("Error occurred while setting push notification status to SCHEDULED.", ex);
OperationManagementDAOFactory.rollbackTransaction();
}
}
} }
} }
@ -415,7 +372,7 @@ public class OperationManagerImpl implements OperationManager {
} }
private String addOperationMappings(List<DeviceIdentifier> authorizedDeviceList, org.wso2.carbon.device.mgt.core.dto.operation.mgt.Operation operationDto, int operationId, boolean isScheduledOperation, boolean isNotRepeated, boolean isScheduled, List<Device> devices) throws OperationManagementException, OperationManagementDAOException { private String addOperationMappings(List<DeviceIdentifier> authorizedDeviceList, org.wso2.carbon.device.mgt.core.dto.operation.mgt.Operation operationDto, int operationId, boolean isScheduledOperation, boolean isNotRepeated, boolean isScheduled, List<Device> devices) throws OperationManagementException, OperationManagementDAOException {
int enrolmentId; int enrolmentId;
boolean hasExistingTaskOperation;//TODO have to create a sql to load device details from deviceDAO using single query. int existingTaskOperationId;//TODO have to create a sql to load device details from deviceDAO using single query.
String operationCode = operationDto.getCode(); String operationCode = operationDto.getCode();
for (DeviceIdentifier deviceId : authorizedDeviceList) { for (DeviceIdentifier deviceId : authorizedDeviceList) {
Device device = getDevice(deviceId); Device device = getDevice(deviceId);
@ -423,8 +380,8 @@ public class OperationManagerImpl implements OperationManager {
enrolmentId = device.getEnrolmentInfo().getId(); enrolmentId = device.getEnrolmentInfo().getId();
//Do not repeat the task operations //Do not repeat the task operations
if (isScheduledOperation) { if (isScheduledOperation) {
hasExistingTaskOperation = operationDAO.updateTaskOperation(enrolmentId, operationCode); existingTaskOperationId = operationDAO.getExistingOperationID(enrolmentId, operationCode);
if (!hasExistingTaskOperation) { if (existingTaskOperationId != -1) {
operationMappingDAO.addOperationMapping(operationId, enrolmentId, isScheduled); operationMappingDAO.addOperationMapping(operationId, enrolmentId, isScheduled);
} }
} else if (isNotRepeated) { } else if (isNotRepeated) {
@ -476,6 +433,41 @@ public class OperationManagerImpl implements OperationManager {
} }
} }
private void sendNotification(Operation operation, Device device) {
NotificationStrategy notificationStrategy = getNotificationStrategy();
/*
* If notification strategy has not enable to send push notification using scheduler task we will send
* notification immediately. This is done in separate loop inorder to prevent overlap with DB insert
* operations with the possible db update operations trigger followed by pending operation call.
* Otherwise device may call pending operation while DB is locked for write and deadlock can occur.
*/
if (notificationStrategy != null) {
if (log.isDebugEnabled()) {
log.debug("Sending push notification to " + device.getDeviceIdentifier() + " from add operation method.");
}
DeviceIdentifier deviceIdentifier = new DeviceIdentifier(device.getDeviceIdentifier(), device.getType());
try {
notificationStrategy.execute(new NotificationContext(deviceIdentifier, operation));
} catch (PushNotificationExecutionFailedException e) {
log.error("Error occurred while sending push notifications to " + device.getType() +
" device carrying id '" + device.getDeviceIdentifier() + "'", e);
/*
* Reschedule if push notification failed. Doing db transactions in atomic way to prevent
* deadlocks.
*/
try {
operationMappingDAO.updateOperationMapping(operation.getId(), device.getEnrolmentInfo().getId(), org.wso2.carbon
.device.mgt.core.dto.operation.mgt.Operation.PushNotificationStatus.SCHEDULED);
OperationManagementDAOFactory.commitTransaction();
} catch (OperationManagementDAOException ex) {
// Not throwing this exception in order to keep sending remaining notifications if any.
log.error("Error occurred while setting push notification status to SCHEDULED.", ex);
OperationManagementDAOFactory.rollbackTransaction();
}
}
}
}
private List<ActivityStatus> getActivityStatus(DeviceIDHolder deviceIdValidationResult, DeviceIDHolder deviceAuthResult, private List<ActivityStatus> getActivityStatus(DeviceIDHolder deviceIdValidationResult, DeviceIDHolder deviceAuthResult,
String deviceType) { String deviceType) {
List<ActivityStatus> activityStatuses = new ArrayList<>(); List<ActivityStatus> activityStatuses = new ArrayList<>();

@ -56,7 +56,7 @@ public interface OperationDAO {
void updateEnrollmentOperationsStatus(int enrolmentId, String operationCode, Operation.Status existingStatus, void updateEnrollmentOperationsStatus(int enrolmentId, String operationCode, Operation.Status existingStatus,
Operation.Status newStatus) throws OperationManagementDAOException; Operation.Status newStatus) throws OperationManagementDAOException;
boolean updateTaskOperation(int enrolmentId, String operationCode) throws OperationManagementDAOException; int getExistingOperationID(int enrolmentId, String operationCode) throws OperationManagementDAOException;
void addOperationResponse(int enrolmentId, int operationId, Object operationResponse) void addOperationResponse(int enrolmentId, int operationId, Object operationResponse)
throws OperationManagementDAOException; throws OperationManagementDAOException;

@ -181,14 +181,14 @@ public class GenericOperationDAOImpl implements OperationDAO {
} }
@Override @Override
public boolean updateTaskOperation(int enrolmentId, String operationCode) public int getExistingOperationID(int enrolmentId, String operationCode)
throws OperationManagementDAOException { throws OperationManagementDAOException {
PreparedStatement stmt = null; PreparedStatement stmt = null;
ResultSet rs = null; ResultSet rs = null;
boolean result = false; int result = -1;
try { try {
Connection connection = OperationManagementDAOFactory.getConnection(); Connection connection = OperationManagementDAOFactory.getConnection();
String query = "SELECT EOM.ID FROM DM_ENROLMENT_OP_MAPPING EOM INNER JOIN DM_OPERATION DM " String query = "SELECT DM.ID FROM DM_ENROLMENT_OP_MAPPING 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 = connection.prepareStatement(query);
@ -198,7 +198,7 @@ public class GenericOperationDAOImpl implements OperationDAO {
// This will return only one result always. // This will return only one result always.
rs = stmt.executeQuery(); rs = stmt.executeQuery();
if (rs.next()) { if (rs.next()) {
result = true; result = rs.getInt("ID");
} }
} catch (SQLException e) { } catch (SQLException e) {
throw new OperationManagementDAOException( throw new OperationManagementDAOException(

Loading…
Cancel
Save